add BulkInsert3 (default 128)

Author: ffffwh

driver/common/binlog.go

@@ -162,6 +162,7 @@ type ApplierTableItem struct {
 	PsInsert0 []*sql.Stmt
 	PsInsert1 []*sql.Stmt
 	PsInsert2 []*sql.Stmt
+	PsInsert3 []*sql.Stmt
 	PsDelete  []*sql.Stmt
 	PsUpdate    []*sql.Stmt
 	ColumnMapTo []string
@@ -173,6 +174,7 @@ func NewApplierTableItem(parallelWorkers int) *ApplierTableItem {
 		PsInsert0: make([]*sql.Stmt, parallelWorkers),
 		PsInsert1: make([]*sql.Stmt, parallelWorkers),
 		PsInsert2: make([]*sql.Stmt, parallelWorkers),
+		PsInsert3: make([]*sql.Stmt, parallelWorkers),
 		PsDelete:  make([]*sql.Stmt, parallelWorkers),
 		PsUpdate:  make([]*sql.Stmt, parallelWorkers),
 	}
@@ -190,6 +192,7 @@ func (ait *ApplierTableItem) Reset() {
 	closeStmts(ait.PsInsert0)
 	closeStmts(ait.PsInsert1)
 	closeStmts(ait.PsInsert2)
+	closeStmts(ait.PsInsert3)
 	closeStmts(ait.PsDelete)
 	closeStmts(ait.PsUpdate)
 

driver/common/taskconfig.go

@@ -56,6 +56,7 @@ type DtleTaskConfig struct {
 	WaitOnJob            string        `codec:"WaitOnJob"`
 	BulkInsert1          int           `codec:"BulkInsert1"`
 	BulkInsert2          int           `codec:"BulkInsert2"`
+	BulkInsert3          int           `codec:"BulkInsert3"`
 	SlaveNetWriteTimeout int           `codec:"SlaveNetWriteTimeout"`
 	BigTxSrcQueue        int32         `codec:"BigTxSrcQueue"`
 

driver/driver.go

@@ -144,6 +144,8 @@ var (
 			hclspec.NewLiteral(`4`)),
 		"BulkInsert2": hclspec.NewDefault(hclspec.NewAttr("BulkInsert2", "number", false),
 			hclspec.NewLiteral(`8`)),
+		"BulkInsert3": hclspec.NewDefault(hclspec.NewAttr("BulkInsert3", "number", false),
+			hclspec.NewLiteral(`128`)),
 		"BigTxSrcQueue": hclspec.NewDefault(hclspec.NewAttr("BigTxSrcQueue", "number", false),
 			hclspec.NewLiteral(`3`)),
 		"KafkaConfig": hclspec.NewBlock("KafkaConfig", false, hclspec.NewObject(map[string]*hclspec.Spec{

driver/mysql/applier_incr.go

@@ -597,7 +597,11 @@ func (a *ApplierIncr) ApplyBinlogEvent(workerIdx int, binlogEntryCtx *common.Ent
 				for i := 0; i < nRows; {
 					var pstmt **gosql.Stmt
 					var rows [][]interface{}
-					if nRows-i >= a.mysqlContext.BulkInsert2 {
+					if nRows-i >= a.mysqlContext.BulkInsert3 {
+						pstmt = &tableItem.PsInsert3[workerIdx]
+						rows = event.Rows[i : i+a.mysqlContext.BulkInsert3]
+						i += a.mysqlContext.BulkInsert3
+					} else if nRows-i >= a.mysqlContext.BulkInsert2 {
 						pstmt = &tableItem.PsInsert2[workerIdx]
 						rows = event.Rows[i : i+a.mysqlContext.BulkInsert2]
 						i += a.mysqlContext.BulkInsert2