lightning: restore W.I.P

lightning/restore/tidb.go

@@ -135,32 +135,42 @@ func InitSchema(ctx context.Context, g glue.Glue, database string, tablesSchema
  if err != nil {
  return errors.Trace(err)
  }
-
  task := logger.Begin(zap.InfoLevel, "create tables")
  var sqlCreateStmts []string
+ stmtQueue := make(chan string, len(g.tableWorkers.GetLimit())))
+ createTableContext, finishCreateTable := context.WithCancel(ctx)
+ createTable := func (ctx ctx context.Context, database, table, stmt string, logger log.Logger) error {
+ log := zap.String("table", common.UniqueTable(database, tbl))
+ err = sqlExecutor.ExecuteWithLog(ctx, stmt, "create table", logger.With(log))
+ if err != nil {
+ return error
+ }
+ return nil
+ )
+ go func (ctx context.Context, database, table, string logger log.Logger) {
+ for {
+ switch {
+ case stmt := <-stmtQueue:
+ go createTable(ctx, stmt, database, table, logger)
+ case <-ctx.Done():
+ return
+ }
+ }
+ }(createTableContext, database, tbl, logger)
  for tbl, sqlCreateTable := range tablesSchema {
  task.Debug("create table", zap.String("schema", sqlCreateTable))
-
  sqlCreateStmts, err = createTableIfNotExistsStmt(g.GetParser(), sqlCreateTable, database, tbl)
  if err != nil {
  break
  }
-
  //TODO: maybe we should put these createStems into a transaction
- for _, s := range sqlCreateStmts {
- err = sqlExecutor.ExecuteWithLog(
- ctx,
- s,
- "create table",
- logger.With(zap.String("table", common.UniqueTable(database, tbl))),
- )
- if err != nil {
- break
- }
+ for _, stmt := range sqlCreateStmts {
+ stmtQueue <- stmt
  }
  }
+ finishCreateTable()
+ close(stmtQueue)
  task.End(zap.ErrorLevel, err)
-
  return errors.Trace(err)
 }
 

lightning/worker/worker.go

@@ -63,3 +63,7 @@ func (pool *Pool) Recycle(worker *Worker) {
 func (pool *Pool) HasWorker() bool {
  return len(pool.workers) > 0
 }
+
+func (pool *Pool) GetLimit() int {
+ return pool.limit
+}