Signed-off-by: Hillium maruruku@stu.csust.edu.cn

What problem does this PR solve?

Fix #419

What is changed and how it works?

We changed the interface of BatchSender, so it can run asynchronously:

type TableSink interface {
	EmitTables(tables ...CreatedTable)
	EmitError(error)
	Close()
}

// BatchSender is the abstract of how the batcher send a batch.
type BatchSender interface {
	// PutSink sets the sink of this sender, user to this interface promise
	// call this function at least once before first call to `RestoreBatch`.
	PutSink(sink TableSink)
	// RestoreBatch will send the restore request.
	RestoreBatch(ranges DrainResult)
	Close()
}

Then, we must change the call site of Batcher, and make another goroutine(known as contextCleaner) to accept tables that restored asynchronously, do the clean job, and send tables to next.

(TODO?: since the sender can be internally asynchronous, we can remove sendWorker and send requests directly at the call site. However the risk of this is that the channel size of BatchSender might block the caller, and we must pass a ctx to methods like Add since ContextManager need to be in the tree of structured concurrent too.)

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
  Internal test: real time 12m28s -> 9m5s (45000+ ranges)

Release Note

• Speed up restore by pipelining (split & scatter) and import.