add support rollback running txn to 1.2 (#18775)

add support rollback running txn

Approved by: @reusee, @sukki37

Author: zhangxu19830126

pkg/incrservice/column_cache.go

@@ -495,9 +495,7 @@ func (col *columnCache) waitPrevAllocatingLocked(ctx context.Context) error {
 }
 
 func (col *columnCache) close() error {
-	col.Lock()
-	defer col.Unlock()
-	return col.waitPrevAllocatingLocked(context.Background())
+	return nil
 }
 
 func insertAutoValues[T constraints.Integer](

pkg/txn/client/operator.go

@@ -241,6 +241,7 @@ type txnOperator struct {
 	fprints             footPrints
 
 	waitActiveCost time.Duration
+	runningSQL     atomic.Bool
 }
 
 func newTxnOperator(
@@ -517,6 +518,10 @@ func (tc *txnOperator) WriteAndCommit(ctx context.Context, requests []txn.TxnReq
 }
 
 func (tc *txnOperator) Commit(ctx context.Context) (err error) {
+	if tc.runningSQL.Load() {
+		util.GetLogger().Fatal("commit on running txn")
+	}
+
 	tc.commitCounter.addEnter()
 	defer tc.commitCounter.addExit()
 	txn := tc.getTxnMeta(false)
@@ -552,6 +557,10 @@ func (tc *txnOperator) Commit(ctx context.Context) (err error) {
 }
 
 func (tc *txnOperator) Rollback(ctx context.Context) (err error) {
+	if tc.runningSQL.Load() {
+		util.GetLogger().Fatal("commit on running txn")
+	}
+
 	tc.rollbackCounter.addEnter()
 	defer tc.rollbackCounter.addExit()
 	v2.TxnRollbackCounter.Inc()
@@ -1262,10 +1271,12 @@ func (tc *txnOperator) doCostAction(
 }
 
 func (tc *txnOperator) EnterRunSql() {
+	tc.runningSQL.Store(true)
 	tc.runSqlCounter.addEnter()
 }
 
 func (tc *txnOperator) ExitRunSql() {
+	tc.runningSQL.Store(false)
 	tc.runSqlCounter.addExit()
 }
 

pkg/txn/client/operator_test.go

@@ -545,6 +545,109 @@ func TestWaitCommittedLogAppliedInRCMode(t *testing.T) {
 		nil)
 }
 
+func TestCannotCommitRunningSQLTxn(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			defer func() {
+				if err := recover(); err != nil {
+					require.NotNil(t, err)
+				}
+			}()
+
+			tc.EnterRunSql()
+			_ = tc.Commit(ctx)
+		},
+	)
+}
+
+func TestCannotRollbackRunningSQLTxn(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			defer func() {
+				if err := recover(); err != nil {
+					require.NotNil(t, err)
+				}
+			}()
+
+			tc.EnterRunSql()
+			_ = tc.Rollback(ctx)
+		},
+	)
+}
+
+func TestEmptyLockSkipped(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			require.False(t, tc.LockSkipped(1, lock.LockMode_Exclusive))
+		},
+	)
+}
+
+func TestLockSkipped(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			require.True(t, tc.LockSkipped(1, lock.LockMode_Exclusive))
+			require.False(t, tc.LockSkipped(1, lock.LockMode_Shared))
+			require.False(t, tc.LockSkipped(2, lock.LockMode_Exclusive))
+		},
+		WithTxnSkipLock(
+			[]uint64{1},
+			[]lock.LockMode{lock.LockMode_Exclusive},
+		),
+	)
+}
+
+func TestHasLockTable(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			require.NoError(t, tc.AddLockTable(lock.LockTable{Table: 1}))
+			require.True(t, tc.HasLockTable(1))
+			require.False(t, tc.HasLockTable(2))
+		},
+	)
+}
+
+func TestBase(t *testing.T) {
+	runOperatorTests(
+		t,
+		func(
+			ctx context.Context,
+			tc *txnOperator,
+			_ *testTxnSender,
+		) {
+			require.NotNil(t, tc.TxnRef())
+			require.Equal(t, tc.Txn().SnapshotTS, tc.SnapshotTS())
+			require.NotEqual(t, timestamp.Timestamp{}, tc.CreateTS())
+			require.Equal(t, txn.TxnStatus_Active, tc.Status())
+		},
+	)
+}
+
 func runOperatorTests(
 	t *testing.T,
 	tc func(context.Context, *txnOperator, *testTxnSender),