DM: Get DB flavor from source instead of guessing it

Author: dveeden

dm/loader/util.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tiflow/dm/config"
+	"github.com/pingcap/tiflow/dm/pkg/conn"
 	"github.com/pingcap/tiflow/dm/pkg/dumpling"
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
@@ -45,14 +46,26 @@ func percent(a int64, b int64, finish bool) string {
 }
 
 func getMydumpMetadata(ctx context.Context, cli *clientv3.Client, cfg *config.SubTaskConfig, workerName string) (string, string, error) {
+	flavor := ""
+	baseDB, err := conn.GetUpstreamDB(&cfg.From)
+	if err != nil {
+		log.L().Warn("set up db connect failed", zap.Any("db", cfg.From), zap.Error(err))
+	} else {
+		flavor, err = conn.GetFlavor(ctx, baseDB)
+		if err != nil {
+			log.L().Warn("failed to get database flavor", zap.Any("db", cfg.From), zap.Error(err))
+		}
+		baseDB.Close()
+	}
+
 	metafile := "metadata"
 	failpoint.Inject("TestRemoveMetaFile", func() {
 		err := storage.RemoveAll(ctx, cfg.LoaderConfig.Dir, nil)
 		if err != nil {
 			log.L().Warn("TestRemoveMetaFile Error", log.ShortError(err))
 		}
 	})
-	loc, _, err := dumpling.ParseMetaData(ctx, cfg.LoaderConfig.Dir, metafile, cfg.ExtStorage)
+	loc, _, err := dumpling.ParseMetaData(ctx, cfg.LoaderConfig.Dir, metafile, cfg.ExtStorage, flavor)
 	if err == nil {
 		return loc.Position.String(), loc.GTIDSetStr(), nil
 	}

dm/pkg/dumpling/utils.go

@@ -48,18 +48,19 @@ func ParseMetaData(
 	dir string,
 	filename string,
 	extStorage brstorage.ExternalStorage,
+	flavor string,
 ) (*binlog.Location, *binlog.Location, error) {
 	fd, err := storage.OpenFile(ctx, dir, filename, extStorage)
 	if err != nil {
 		return nil, nil, err
 	}
 	defer fd.Close()
 
-	return ParseMetaDataByReader(filename, fd)
+	return ParseMetaDataByReader(filename, fd, flavor)
 }
 
 // ParseMetaDataByReader parses mydumper's output meta file by created reader and returns binlog location.
-func ParseMetaDataByReader(filename string, rd io.Reader) (*binlog.Location, *binlog.Location, error) {
+func ParseMetaDataByReader(filename string, rd io.Reader, flavor string) (*binlog.Location, *binlog.Location, error) {
 	invalidErr := fmt.Errorf("file %s invalid format", filename)
 
 	var (
@@ -156,7 +157,7 @@ func ParseMetaDataByReader(filename string, rd io.Reader) (*binlog.Location, *bi
 		return nil, nil, terror.ErrMetadataNoBinlogLoc.Generate(filename)
 	}
 
-	gset, err := gtid.ParserGTID("", gtidStr)
+	gset, err := gtid.ParserGTID(flavor, gtidStr)
 	if err != nil {
 		return nil, nil, invalidErr
 	}

dm/syncer/checkpoint.go

@@ -1302,7 +1302,20 @@ func (cp *RemoteCheckPoint) genUpdateSQL(cpSchema, cpTable string, location binl
 func (cp *RemoteCheckPoint) parseMetaData(ctx context.Context) (*binlog.Location, *binlog.Location, error) {
 	// `metadata` is mydumper's output meta file name
 	filename := "metadata"
-	loc, loc2, err := dumpling.ParseMetaData(ctx, cp.cfg.LoaderConfig.Dir, filename, cp.cfg.ExtStorage)
+
+	flavor := ""
+	baseDB, err := conn.GetUpstreamDB(&cp.cfg.From)
+	if err != nil {
+		log.L().Warn("set up db connect failed", zap.Any("db", cp.cfg.From), zap.Error(err))
+	} else {
+		flavor, err = conn.GetFlavor(ctx, baseDB)
+		if err != nil {
+			log.L().Warn("failed to get database flavor", zap.Any("db", cp.cfg.From), zap.Error(err))
+		}
+		baseDB.Close()
+	}
+
+	loc, loc2, err := dumpling.ParseMetaData(ctx, cp.cfg.LoaderConfig.Dir, filename, cp.cfg.ExtStorage, flavor)
 	if err != nil {
 		toPrint, err2 := storage.ReadFile(ctx, cp.cfg.LoaderConfig.Dir, filename, nil)
 		if err2 != nil {