TiDB | csv文档导入，到底需要注意表名大小写问题吗？

[lightning]
level = "info"
file = "tidb-lightning.log"
index-concurrency = 2
table-concurrency = 5
io-concurrency = 5


[tikv-importer]
backend = "local"
sorted-kv-dir = "/tmp/tidb/lightning_dir"


[mydumper]
data-source-dir = "/tmp/tidb/data"
no-schema = true
filter = ['*.*']


[mydumper.csv]
# 字段分隔符，支持一个或多个字符，默认值为 ','。
separator = '|'
# 引用定界符，设置为空表示字符串未加引号。
delimiter = ''
# 行尾定界字符，支持一个或多个字符。设置为空（默认值）表示 "\n"（换行）和 "\r\n" （回车+换行），均表示行尾。
terminator = ""
# CSV 文件是否包含表头。
# 如果 header = true，将跳过首行。
header = false
# CSV 文件是否包含 NULL。
# 如果 not-null = true，CSV 所有列都不能解析为 NULL。
not-null = false
# 如果 not-null = false（即 CSV 可以包含 NULL），
# 为以下值的字段将会被解析为 NULL。
null = '\N'
# 是否对字段内“\“进行转义
backslash-escape = true
# 如果有行以分隔符结尾，删除尾部分隔符。
trim-last-separator = false


[tidb]
host = "x.x.x.x"
port = 4000
user = "root"
password = ""
status-port = 10080
pd-addr = "x.x.x.x:2379"


[checkpoint]
enable = false


[post-restore]
checksum = false
analyze = false

func (rc *Controller) Run(ctx context.Context) error {
  opts := []func(context.Context) error{
    rc.setGlobalVariables,
    rc.restoreSchema,
    rc.preCheckRequirements,
    rc.restoreTables,
    rc.fullCompact,
    rc.switchToNormalMode,
    rc.cleanCheckpoints,
  }
    ....
   for i, process := range opts {
    err = process(ctx)
    ....
  }
    ....
}

func (l *Lightning) run(taskCtx context.Context, taskCfg *config.Config, g glue.Glue) (err error) {
  ...
  dbMetas := mdl.GetDatabases()
  web.BroadcastInitProgress(dbMetas)


  var procedure *restore.Controller
  procedure, err = restore.NewRestoreController(ctx, dbMetas, taskCfg, s, g)
  if err != nil {
    log.L().Error("restore failed", log.ShortError(err))
    return errors.Trace(err)
  }
  defer procedure.Close()


  err = procedure.Run(ctx)
  return errors.Trace(err)
}

/// br > pkg > lightning > lightning.go
func (l *Lightning) RunOnce(taskCtx context.Context, taskCfg *config.Config, glue glue.Glue) error {
  if err := taskCfg.Adjust(taskCtx); err != nil {
    return err
  }


  taskCfg.TaskID = time.Now().UnixNano()
  ...
  return l.run(taskCtx, taskCfg, glue)
}


/// br > cmd > tidb-lightning > main.go
func main() {
    globalCfg := config.Must(config.LoadGlobalConfig(os.Args[1:], nil))
  ....
  err = func() error {
    if globalCfg.App.ServerMode {
      return app.RunServer()
    }
    cfg := config.NewConfig()
    if err := cfg.LoadFromGlobal(globalCfg); err != nil {
      return err
    }
    return app.RunOnce(context.Background(), cfg, nil)
  }()
  ....
}

getTableFunc := rc.backend.FetchRemoteTableModels
  ....
  err := worker.makeJobs(rc.dbMetas, getTableFunc)
  ....
  dbInfos, err := LoadSchemaInfo(ctx, rc.dbMetas, getTableFunc)
  if err != nil {
    return errors.Trace(err)
  }
  rc.dbInfos = dbInfos
  ....

  // 2. restore tables, execute statements concurrency
  for _, dbMeta := range dbMetas {
    // we can ignore error here, and let check failed later if schema not match
    tables, _ := getTables(worker.ctx, dbMeta.Name)
    tableMap := make(map[string]struct{})
    for _, t := range tables {
      tableMap[t.Name.L] = struct{}{}
    }
    for _, tblMeta := range dbMeta.Tables {
      if _, ok := tableMap[strings.ToLower(tblMeta.Name)]; ok {
        // we already has this table in TiDB.
        // we should skip ddl job and let SchemaValid check.
        continue
      } else if tblMeta.SchemaFile.FileMeta.Path == "" {
        return errors.Errorf("table `%s`.`%s` schema not found", dbMeta.Name, tblMeta.Name)
      }
            ...
        }
        ...

func LoadSchemaInfo(
  ctx context.Context,
  schemas []*mydump.MDDatabaseMeta,
  getTables func(context.Context, string) ([]*model.TableInfo, error),
) (map[string]*checkpoints.TidbDBInfo, error) {
  result := make(map[string]*checkpoints.TidbDBInfo, len(schemas))
  for _, schema := range schemas {
    tables, err := getTables(ctx, schema.Name)
    if err != nil {
      return nil, err
    }


    tableMap := make(map[string]*model.TableInfo, len(tables))
    for _, tbl := range tables {
      tableMap[tbl.Name.L] = tbl
    }


    dbInfo := &checkpoints.TidbDBInfo{
      Name:   schema.Name,
      Tables: make(map[string]*checkpoints.TidbTableInfo),
    }


    for _, tbl := range schema.Tables {
      tblInfo, ok := tableMap[strings.ToLower(tbl.Name)]
      if !ok {
        return nil, errors.Errorf("table '%s' schema not found", tbl.Name)
      }
      tableName := tblInfo.Name.String()
      if tblInfo.State != model.StatePublic {
        err := errors.Errorf("table [%s.%s] state is not public", schema.Name, tableName)
        metric.RecordTableCount(metric.TableStatePending, err)
        return nil, err
      }
      metric.RecordTableCount(metric.TableStatePending, err)
      if err != nil {
        return nil, errors.Trace(err)
      }
      tableInfo := &checkpoints.TidbTableInfo{
        ID:   tblInfo.ID,
        DB:   schema.Name,
        Name: tableName,
        Core: tblInfo,
      }
      dbInfo.Tables[tableName] = tableInfo
    }


    result[schema.Name] = dbInfo
  }
  return result, nil
}

// CIStr is case insensitive string.
type CIStr struct {
  O string `json:"O"` // Original string.
  L string `json:"L"` // Lower case string.
}


// String implements fmt.Stringer interface.
func (cis CIStr) String() string {
  return cis.O
}

本期内容就到这里啦

有更好的办法或疑问请

⬇欢迎加入社群一起讨论哦⬇