executor: do not append extra cols to the old row when updateDupRow (…

…#33656) (#35165)

close #33608

executor/executor_test.go

@@ -114,7 +114,7 @@ var _ = SerialSuites(&testSuiteJoinSerial{&baseTestSuite{}})
 var _ = Suite(&testSuiteAgg{baseTestSuite: &baseTestSuite{}})
 var _ = Suite(&testSuite6{&baseTestSuite{}})
 var _ = Suite(&testSuite7{&baseTestSuite{}})
-var _ = Suite(&testSuite8{&baseTestSuite{}})
+var _ = SerialSuites(&testSuite8{&baseTestSuite{}})
 var _ = SerialSuites(&testShowStatsSuite{&baseTestSuite{}})
 var _ = Suite(&testBypassSuite{})
 var _ = Suite(&testUpdateSuite{})

executor/insert.go

@@ -193,13 +193,13 @@ func (e *InsertExec) updateDupRow(ctx context.Context, idxInBatch int, txn kv.Tr
 	if err != nil {
 		return err
 	}
-	// get the extra columns from the SELECT clause and get the final `oldRow`.
+	// get the extra columns from the SELECT clause.
+	var extraCols []types.Datum
 	if len(e.ctx.GetSessionVars().CurrInsertBatchExtraCols) > 0 {
-		extraCols := e.ctx.GetSessionVars().CurrInsertBatchExtraCols[idxInBatch]
-		oldRow = append(oldRow, extraCols...)
+		extraCols = e.ctx.GetSessionVars().CurrInsertBatchExtraCols[idxInBatch]
 	}
 
-	err = e.doDupRowUpdate(ctx, handle, oldRow, row.row, e.OnDuplicate)
+	err = e.doDupRowUpdate(ctx, handle, oldRow, row.row, extraCols, e.OnDuplicate)
 	if e.ctx.GetSessionVars().StmtCtx.DupKeyAsWarning && kv.ErrKeyExists.Equal(err) {
 		e.ctx.GetSessionVars().StmtCtx.AppendWarning(err)
 		return nil
@@ -374,16 +374,18 @@ func (e *InsertExec) initEvalBuffer4Dup() {
 
 // doDupRowUpdate updates the duplicate row.
 func (e *InsertExec) doDupRowUpdate(ctx context.Context, handle kv.Handle, oldRow []types.Datum, newRow []types.Datum,
-	cols []*expression.Assignment) error {
+	extraCols []types.Datum, cols []*expression.Assignment) error {
 	assignFlag := make([]bool, len(e.Table.WritableCols()))
 	// See http://dev.mysql.com/doc/refman/5.7/en/miscellaneous-functions.html#function_values
 	e.curInsertVals.SetDatums(newRow...)
 	e.ctx.GetSessionVars().CurrInsertValues = e.curInsertVals.ToRow()
 	// NOTE: In order to execute the expression inside the column assignment,
-	// we have to put the value of "oldRow" before "newRow" in "row4Update" to
-	// be consistent with "Schema4OnDuplicate" in the "Insert" PhysicalPlan.
+	// we have to put the value of "oldRow" and "extraCols" before "newRow" in
+	// "row4Update" to be consistent with "Schema4OnDuplicate" in the "Insert"
+	// PhysicalPlan.
 	e.row4Update = e.row4Update[:0]
 	e.row4Update = append(e.row4Update, oldRow...)
+	e.row4Update = append(e.row4Update, extraCols...)
 	e.row4Update = append(e.row4Update, newRow...)
 
 	// Update old row when the key is duplicated.

executor/insert_test.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/meta/autoid"
@@ -39,6 +40,17 @@ import (
 
 func (s *testSuite8) TestInsertOnDuplicateKey(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
+	testInsertOnDuplicateKey(c, tk)
+}
+
+func (s *testSuite8) TestInsertOnDuplicateKeyWithBinlog(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	failpoint.Enable("github.com/pingcap/tidb/table/tables/forceWriteBinlog", "return")
+	defer failpoint.Disable("github.com/pingcap/tidb/table/tables/forceWriteBinlog")
+	testInsertOnDuplicateKey(c, tk)
+}
+
+func testInsertOnDuplicateKey(c *C, tk *testkit.TestKit) {
 	tk.MustExec("use test")
 
 	tk.MustExec(`drop table if exists t1, t2;`)

table/tables/tables.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/meta/autoid"
@@ -1471,6 +1472,11 @@ func (t *TableCommon) Type() table.Type {
 }
 
 func shouldWriteBinlog(ctx sessionctx.Context, tblInfo *model.TableInfo) bool {
+	failpoint.Inject("forceWriteBinlog", func() {
+		// Just to cover binlog related code in this package, since the `BinlogClient` is
+		// still nil, mutations won't be written to pump on commit.
+		failpoint.Return(true)
+	})
 	if ctx.GetSessionVars().BinlogClient == nil {
 		return false
 	}