table: fix some txn assertion error when executing DDL and DML (#55314)

close #55313

pkg/table/tables/BUILD.bazel

@@ -77,7 +77,7 @@ go_test(
     ],
     embed = [":tables"],
     flaky = True,
-    shard_count = 33,
+    shard_count = 34,
     deps = [
         "//pkg/ddl",
         "//pkg/domain",

pkg/table/tables/index.go

@@ -246,7 +246,7 @@ func (c *index) create(sctx table.MutateContext, txn kv.Transaction, indexedValu
 			return nil, err
 		}
 
-		opt.IgnoreAssertion = opt.IgnoreAssertion || c.idxInfo.State != model.StatePublic
+		ignoreAssertion := opt.IgnoreAssertion || c.idxInfo.State != model.StatePublic
 
 		if !distinct || skipCheck || untouched {
 			val := idxVal
@@ -271,7 +271,7 @@ func (c *index) create(sctx table.MutateContext, txn kv.Transaction, indexedValu
 					return nil, err
 				}
 			}
-			if !opt.IgnoreAssertion && !untouched {
+			if !ignoreAssertion && !untouched {
 				if opt.DupKeyCheck == table.DupKeyCheckLazy && !txn.IsPessimistic() {
 					err = txn.SetAssertion(key, kv.SetAssertUnknown)
 				} else {
@@ -346,7 +346,7 @@ func (c *index) create(sctx table.MutateContext, txn kv.Transaction, indexedValu
 					return nil, err
 				}
 			}
-			if opt.IgnoreAssertion {
+			if ignoreAssertion {
 				continue
 			}
 			if lazyCheck && !txn.IsPessimistic() {

pkg/table/tables/index_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/kv"
@@ -281,3 +282,72 @@ func TestGenIndexValueWithLargePaddingSize(t *testing.T) {
 	require.False(t, handle.IsInt())
 	require.Equal(t, commonHandle.Encoded(), handle.Encoded())
 }
+
+// See issue: https://github.com/pingcap/tidb/issues/55313
+func TestTableOperationsInDDLDropIndexWriteOnly(t *testing.T) {
+	store, do := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a int, b int, key a(a), key(b))")
+	tk.MustExec("insert into t values(1, 1), (2, 2), (3, 3)")
+	// use MDL to block drop index DDL in `StateWriteOnly`
+	tk.MustExec("set @@global.tidb_enable_metadata_lock='ON'")
+	tk.MustExec("begin pessimistic")
+	tk.MustQuery("select * from t order by a asc").Check(testkit.Rows("1 1", "2 2", "3 3"))
+	tk2 := testkit.NewTestKit(t, store)
+	tk2.MustExec("use test")
+
+	ch := make(chan struct{})
+	go func() {
+		defer close(ch)
+		// run `DROP INDEX` in background, because a transaction is started,
+		// the DDL will hang in state `StateWriteOnly` until all transactions are committed or rollback.
+		tk3 := testkit.NewTestKit(t, store)
+		tk3.MustExec("use test")
+		tk3.MustExec("alter table t drop index a")
+	}()
+
+	defer func() {
+		// after test case, clear transactions and wait background goroutine exit.
+		tk.MustExec("rollback")
+		tk2.MustExec("rollback")
+		select {
+		case <-ch:
+		case <-time.After(time.Minute):
+			require.FailNow(t, "timeout")
+		}
+	}()
+
+	start := time.Now()
+	for {
+		time.Sleep(20 * time.Millisecond)
+		// wait the DDL state change to `StateWriteOnly`
+		tblInfo, err := do.InfoSchema().TableInfoByName(model.NewCIStr("test"), model.NewCIStr("t"))
+		require.NoError(t, err)
+		if state := tblInfo.Indices[0].State; state != model.StatePublic {
+			require.Equal(t, model.StateWriteOnly, state)
+			break
+		}
+		if time.Since(start) > time.Minute {
+			require.FailNow(t, "timeout")
+		}
+	}
+
+	// tk2 is used to do some operations when DDL is in state `WriteOnly`.
+	// In this state, the dropping index is still written.
+	// We set `@@tidb_txn_assertion_level='STRICT'` to detect any inconsistency.
+	tk2.MustExec("set @@tidb_txn_assertion_level='STRICT'")
+	tk2.MustExec("begin pessimistic")
+	// insert new values.
+	tk2.MustExec("insert into t values(4, 4), (5, 5), (6, 6)")
+	// delete some rows: 1 in storage, 1 in memory buffer.
+	tk2.MustExec("delete from t where a in (1, 4)")
+	// update some rows: 1 in storage, 1 in memory buffer.
+	tk2.MustExec("update t set a = a + 10 where a in (2, 6)")
+	// should be tested in `StateWriteOnly` state.
+	tblInfo, err := tk2.Session().GetInfoSchema().TableInfoByName(model.NewCIStr("test"), model.NewCIStr("t"))
+	require.NoError(t, err)
+	require.Equal(t, model.StateWriteOnly, tblInfo.Indices[0].State)
+	// commit should success without any assertion fail.
+	tk2.MustExec("commit")
+}