reproduce data and columnID count not match

Author: lcwangchao

pkg/ddl/column.go

@@ -55,6 +55,8 @@ import (
 	"go.uber.org/zap"
 )
 
+var WaitChan chan struct{}
+
 // InitAndAddColumnToTable initializes the ColumnInfo in-place and adds it to the table.
 func InitAndAddColumnToTable(tblInfo *model.TableInfo, colInfo *model.ColumnInfo) *model.ColumnInfo {
 	cols := tblInfo.Columns
@@ -171,6 +173,10 @@ func onAddColumn(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error)
 	case model.StateWriteReorganization:
 		// reorganization -> public
 		// Adjust table column offset.
+		if columnInfo.Name.L == "wait_notify" {
+			WaitChan <- struct{}{}
+			<-WaitChan
+		}
 		offset, err := LocateOffsetToMove(columnInfo.Offset, pos, tblInfo)
 		if err != nil {
 			return ver, errors.Trace(err)

pkg/executor/executor_txn_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/executor"
 	"github.com/pingcap/tidb/pkg/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/pkg/testkit"
@@ -694,3 +695,32 @@ func TestSavepointWithBinlog(t *testing.T) {
 	tk.MustExec("commit")
 	tk.MustQuery("select * from t").Check(testkit.Rows("1 1"))
 }
+
+func TestColumnNotMatchError(t *testing.T) {
+	ddl.WaitChan = make(chan struct{})
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.Session().GetSessionVars().BinlogClient = binloginfo.MockPumpsClient(&testkit.MockPumpClient{})
+	tk.MustExec("set @@global.tidb_enable_metadata_lock=0")
+	tk.MustExec("use test")
+	tk2 := testkit.NewTestKit(t, store)
+	tk2.MustExec("use test")
+	tk.MustExec("create table t(id int primary key, a int)")
+	tk.MustExec("insert into t values(1, 2)")
+	go func() {
+		tk2.MustExec("alter table t add column wait_notify int")
+		close(ddl.WaitChan)
+	}()
+	// wait the ddl stage to the reorg
+	<-ddl.WaitChan
+	// begin to make sure the table t in txn contains a no public column
+	tk.MustExec("begin")
+	// continue ddl and wait it finished to avoid some necessary error log
+	ddl.WaitChan <- struct{}{}
+	<-ddl.WaitChan
+	// this statement will fail because:
+	// - schema in the plan includes the non-public column -> data contains non-public column
+	// - table.Cols() does not contain the non-public column
+	tk.MustExec("delete from t where id=1")
+	tk.MustExec("commit")
+}