ddl: fill in original default for extra writable columns in batch ins…

…ert (#40198) (#51145)

close #40192

ddl/column_test.go

@@ -985,3 +985,39 @@ func TestIssue39080(t *testing.T) {
 		"  UNIQUE KEY `authorIdx` (`authorId`)\n"+
 		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
 }
+
+func TestWriteDataWriteOnlyMode(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, dbTestLease)
+
+	tk := testkit.NewTestKit(t, store)
+	tk2 := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk2.MustExec("use test")
+	tk.MustExec("CREATE TABLE t (`col1` bigint(20) DEFAULT 1,`col2` float,UNIQUE KEY `key1` (`col1`))")
+
+	originalCallback := dom.DDL().GetHook()
+	defer dom.DDL().SetHook(originalCallback)
+
+	hook := &ddl.TestDDLCallback{Do: dom}
+	hook.OnJobRunBeforeExported = func(job *model.Job) {
+		if job.SchemaState != model.StateWriteOnly {
+			return
+		}
+		tk2.MustExec("insert ignore into t values (1, 2)")
+		tk2.MustExec("insert ignore into t values (2, 2)")
+	}
+	dom.DDL().SetHook(hook)
+	tk.MustExec("alter table t change column `col1` `col1` varchar(20)")
+
+	hook = &ddl.TestDDLCallback{Do: dom}
+	hook.OnJobRunBeforeExported = func(job *model.Job) {
+		if job.SchemaState != model.StateWriteOnly {
+			return
+		}
+		tk2.MustExec("insert ignore into t values (1)")
+		tk2.MustExec("insert ignore into t values (2)")
+	}
+	dom.DDL().SetHook(hook)
+	tk.MustExec("alter table t drop column `col1`")
+	dom.DDL().SetHook(originalCallback)
+}

executor/batch_checker.go

@@ -146,8 +146,33 @@ func getKeysNeedCheckOneRow(ctx sessionctx.Context, t table.Table, row []types.D
 		}
 	}
 
-	// addChangingColTimes is used to fetch values while processing "modify/change column" operation.
-	addChangingColTimes := 0
+	// extraColumns is used to fetch values while processing "add/drop/modify/change column" operation.
+	extraColumns := 0
+	for _, col := range t.WritableCols() {
+		// if there is a changing column, append the dependency column for index fetch values
+		if col.ChangeStateInfo != nil && col.State != model.StatePublic {
+			value, err := table.CastValue(ctx, row[col.DependencyColumnOffset], col.ColumnInfo, false, false)
+			if err != nil {
+				return nil, err
+			}
+			row = append(row, value)
+			extraColumns++
+			continue
+		}
+
+		if col.State != model.StatePublic {
+			// only append origin default value for index fetch values
+			if col.Offset >= len(row) {
+				value, err := table.GetColOriginDefaultValue(ctx, col.ToInfo())
+				if err != nil {
+					return nil, err
+				}
+
+				row = append(row, value)
+				extraColumns++
+			}
+		}
+	}
 	// append unique keys and errors
 	for _, v := range t.Indices() {
 		if !tables.IsIndexWritable(v) {
@@ -159,12 +184,6 @@ func getKeysNeedCheckOneRow(ctx sessionctx.Context, t table.Table, row []types.D
 		if t.Meta().IsCommonHandle && v.Meta().Primary {
 			continue
 		}
-		if len(row) < len(t.WritableCols()) && addChangingColTimes == 0 {
-			if col := tables.FindChangingCol(t.WritableCols(), v.Meta()); col != nil {
-				row = append(row, row[col.DependencyColumnOffset])
-				addChangingColTimes++
-			}
-		}
 		colVals, err1 := v.FetchValues(row, nil)
 		if err1 != nil {
 			return nil, err1
@@ -194,9 +213,7 @@ func getKeysNeedCheckOneRow(ctx sessionctx.Context, t table.Table, row []types.D
 			commonHandle: t.Meta().IsCommonHandle,
 		})
 	}
-	if addChangingColTimes == 1 {
-		row = row[:len(row)-1]
-	}
+	row = row[:len(row)-extraColumns]
 	result = append(result, toBeCheckedRow{
 		row:        row,
 		handleKey:  handleKey,