Merge branch 'master' into 4.22.07.x

Author: ffffwh

api/docs/docs.go

@@ -3017,6 +3017,9 @@ var doc = `{
                 "binlog_relay": {
                     "type": "boolean"
                 },
+                "dump_entry_limit": {
+                    "type": "integer"
+                },
                 "expand_syntax_support": {
                     "type": "boolean"
                 },

api/docs/swagger.json

@@ -3001,6 +3001,9 @@
                 "binlog_relay": {
                     "type": "boolean"
                 },
+                "dump_entry_limit": {
+                    "type": "integer"
+                },
                 "expand_syntax_support": {
                     "type": "boolean"
                 },

api/docs/swagger.yaml

@@ -591,6 +591,8 @@ definitions:
         type: boolean
       binlog_relay:
         type: boolean
+      dump_entry_limit:
+        type: integer
       expand_syntax_support:
         type: boolean
       gtid:

api/handler/v2/job.go

@@ -592,6 +592,7 @@ func buildDatabaseSrcTaskConfigMap(config *models.SrcTaskConfig) map[string]inte
 		addNotRequiredParamToMap(taskConfigInNomadFormat, config.MysqlSrcTaskConfig.BinlogRelay, "BinlogRelay")
 		addNotRequiredParamToMap(taskConfigInNomadFormat, config.MysqlSrcTaskConfig.Gtid, "Gtid")
 		addNotRequiredParamToMap(taskConfigInNomadFormat, config.MysqlSrcTaskConfig.ExpandSyntaxSupport, "ExpandSyntaxSupport")
+		addNotRequiredParamToMap(taskConfigInNomadFormat, config.MysqlSrcTaskConfig.DumpEntryLimit, "DumpEntryLimit")
 		taskConfigInNomadFormat["ConnectionConfig"] = buildMysqlConnectionConfigMap(config.ConnectionConfig)
 	}
 	// for Oracle
@@ -634,8 +635,8 @@ func buildMysqlTableConfigMap(configs []*models.TableConfig) []map[string]interf
 		configMap := make(map[string]interface{})
 		if len(c.ColumnMapFrom) != 0 {
 			configMap["ColumnMapFrom"] = c.ColumnMapFrom
-			if len(c.ColumnMapTO) != 0 {
-				configMap["ColumnMapTo"] = c.ColumnMapTO
+			if len(c.ColumnMapTo) != 0 {
+				configMap["ColumnMapTo"] = c.ColumnMapTo
 			}
 		}
 		addNotRequiredParamToMap(configMap, c.TableName, "TableName")
@@ -799,6 +800,7 @@ func buildBasicTaskProfile(logger g.LoggerType, jobId string, srcTaskDetail *mod
 				BinlogRelay:         srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.BinlogRelay,
 				WaitOnJob:           srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.WaitOnJob,
 				AutoGtid:            srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.AutoGtid,
+				DumpEntryLimit:      srcTaskDetail.TaskConfig.MysqlSrcTaskConfig.DumpEntryLimit,
 			}
 		} else if srcTaskDetail.TaskConfig.OracleSrcTaskConfig != nil {
 			srcConfig.OracleSrcTaskConfig = &models.OracleSrcTaskConfig{
@@ -951,7 +953,7 @@ func buildSrcTaskDetail(taskName string, internalTaskConfig common.DtleTaskConfi
 					TableRegex:    tb.TableRegex,
 					TableRename:   tb.TableRename,
 					ColumnMapFrom: tb.ColumnMapFrom,
-					ColumnMapTO:   tb.ColumnMapTo,
+					ColumnMapTo:   tb.ColumnMapTo,
 					Where:         tb.Where,
 				})
 			}
@@ -1000,6 +1002,7 @@ func buildSrcTaskDetail(taskName string, internalTaskConfig common.DtleTaskConfi
 			Gtid:                internalTaskConfig.Gtid,
 			BinlogRelay:         internalTaskConfig.BinlogRelay,
 			WaitOnJob:           internalTaskConfig.WaitOnJob,
+			DumpEntryLimit:      internalTaskConfig.DumpEntryLimit,
 		}
 	}
 	srcTaskDetail.TaskConfig.ConnectionConfig = connectionConfig
@@ -1932,6 +1935,7 @@ func ReverseJobV2(c echo.Context, filterJobType DtleJobType) error {
 				BinlogRelay:         originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.BinlogRelay,
 				WaitOnJob:           consulJobItem.JobId,
 				AutoGtid:            true,
+				DumpEntryLimit:      originalJob.BasicTaskProfile.Configuration.SrcConfig.MysqlSrcTaskConfig.DumpEntryLimit,
 			},
 		}
 		reverseJobParam.DestTask = &models.DestTaskConfig{

api/models/job_v2.go

@@ -152,6 +152,7 @@ type MysqlSrcTaskConfig struct {
 	BinlogRelay         bool   `json:"binlog_relay"`
 	WaitOnJob           string `json:"wait_on_job"`
 	AutoGtid            bool   `json:"auto_gtid"`
+	DumpEntryLimit      int    `json:"dump_entry_limit"`
 }
 
 type OracleSrcTaskConfig struct {
@@ -186,7 +187,7 @@ type TableConfig struct {
 	TableRegex    string   `json:"table_regex"`
 	TableRename   string   `json:"table_rename"`
 	ColumnMapFrom []string `json:"column_map_from"`
-	ColumnMapTO   []string `json:"column_map_to"`
+	ColumnMapTo   []string `json:"column_map_to"`
 	Where         string   `json:"where"`
 }
 type DatabaseConnectionConfig struct {

driver/common/dumper.go

@@ -27,7 +27,10 @@ type Dumper struct {
 	Table              *Table
 	Iteration          int64
 	Columns            string
+	// ResultsChannel should be closed after writing all entries.
 	ResultsChannel     chan *DumpEntry
+	// Set Err (if there is) before closing ResultsChannel.
+	Err                error
 	shutdown           bool
 	ShutdownCh         chan struct{}
 	shutdownLock       sync.Mutex
@@ -71,6 +74,7 @@ func (d *Dumper) Dump() error {
 
 			nRows, err := d.GetChunkData()
 			if err != nil {
+				d.Err = err
 				d.Logger.Error("error at dump", "err", err)
 				break
 			}

driver/common/store.go

@@ -142,6 +142,17 @@ func (sm *StoreManager) PutSourceType(jobName, sourceType string) error {
 	return nil
 }
 
+func (sm *StoreManager) GetNatsIfExist(jobName string) (string, bool, error) {
+	natsKey := fmt.Sprintf("dtle/%v/NatsAddr", jobName)
+	kv, err := sm.consulStore.Get(natsKey)
+	if err == store.ErrKeyNotFound {
+		return "", false, nil
+	} else if err != nil {
+		return "", false, err
+	}
+	return string(kv.Value), true, nil
+}
+
 func (sm *StoreManager) DstPutNats(jobName string, natsAddr string, stopCh chan struct{}, onWatchError func(error)) error {
 	sm.logger.Debug("DstPutNats")
 

driver/common/taskconfig.go

@@ -64,6 +64,7 @@ type DtleTaskConfig struct {
 	DependencyHistorySize int  `codec:"DependencyHistorySize"`
 	UseMySQLDependency    bool `codec:"UseMySQLDependency"`
 	ForeignKeyChecks      bool `codec:"ForeignKeyChecks"`
+	DumpEntryLimit        int  `codec:"DumpEntryLimit"`
 
 	SkipCreateDbTable    bool                          `codec:"SkipCreateDbTable"`
 	SkipPrivilegeCheck   bool                          `codec:"SkipPrivilegeCheck"`

driver/common/type.schema

@@ -7,7 +7,6 @@ struct DumpEntry {
 	TbSQL           []string
 	ValuesX    [][]*[]byte
 	TotalCount int64
-	Err        string
 	Table      []byte
 	ColumnMapTo []string
 }

driver/common/type.schema.gen.go

@@ -21,7 +21,6 @@ type DumpEntry struct {
 	TbSQL           []string
 	ValuesX         [][]*[]byte
 	TotalCount      int64
-	Err             string
 	Table           []byte
 	ColumnMapTo     []string
 }
@@ -228,21 +227,6 @@ func (d *DumpEntry) Size() (s uint64) {
 		}
 
 	}
-	{
-		l := uint64(len(d.Err))
-
-		{
-
-			t := l
-			for t >= 0x80 {
-				t >>= 7
-				s++
-			}
-			s++
-
-		}
-		s += l
-	}
 	{
 		l := uint64(len(d.Table))
 
@@ -560,25 +544,6 @@ func (d *DumpEntry) Marshal(buf []byte) ([]byte, error) {
 		buf[i+7+0] = byte(d.TotalCount >> 56)
 
 	}
-	{
-		l := uint64(len(d.Err))
-
-		{
-
-			t := uint64(l)
-
-			for t >= 0x80 {
-				buf[i+8] = byte(t) | 0x80
-				t >>= 7
-				i++
-			}
-			buf[i+8] = byte(t)
-			i++
-
-		}
-		copy(buf[i+8:], d.Err)
-		i += l
-	}
 	{
 		l := uint64(len(d.Table))
 
@@ -924,26 +889,6 @@ func (d *DumpEntry) Unmarshal(buf []byte) (uint64, error) {
 	{
 		l := uint64(0)
 
-		{
-
-			bs := uint8(7)
-			t := uint64(buf[i+8] & 0x7F)
-			for buf[i+8]&0x80 == 0x80 {
-				i++
-				t |= uint64(buf[i+8]&0x7F) << bs
-				bs += 7
-			}
-			i++
-
-			l = t
-
-		}
-		d.Err = string(buf[i+8 : i+8+l])
-		i += l
-	}
-	{
-		l := uint64(0)
-
 		{
 
 			bs := uint8(7)