replay mo_iscp_log after clone (#23244)

replay mo_iscp_log after clone

Approved by: @XuPeng-SH

Author: jiangxinmeng1

pkg/iscp/executor.go

@@ -567,6 +567,17 @@ func (exec *ISCPTaskExecutor) applyISCPLog(ctx context.Context, from, to types.T
 		return
 	}
 	rel, err := db.Relation(ctx, MOISCPLogTableName, nil)
+
+	tid := rel.GetTableID(ctx)
+	// injection is for ut - simulate table id change
+	var injectChangeTableID bool
+	if msg, injected := objectio.ISCPExecutorInjected(); injected && msg == "tableIDChange" {
+		injectChangeTableID = true
+	}
+	if tid != exec.prevISCPTableID || injectChangeTableID {
+		err = moerr.NewErrStaleReadNoCtx("0-0", "0-0")
+		return
+	}
 	// injection is for ut
 	if msg, injected := objectio.ISCPExecutorInjected(); injected && msg == "applyISCPLog" {
 		err = moerr.NewInternalErrorNoCtx(msg)
@@ -693,7 +704,6 @@ func (exec *ISCPTaskExecutor) replay(ctx context.Context) (err error) {
 			zap.Error(err),
 		)
 	}()
-	sql := cdc.CDCSQLBuilder.ISCPLogSelectSQL()
 	txn, err := getTxn(ctx, exec.txnEngine, exec.cnTxnClient, "iscp replay")
 	if err != nil {
 		return
@@ -702,6 +712,14 @@ func (exec *ISCPTaskExecutor) replay(ctx context.Context) (err error) {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
 	defer cancel()
 	defer txn.Commit(ctx)
+
+	tid, _, err := getTableID(ctx, exec.cnUUID, txn, catalog.System_Account, catalog.MO_CATALOG, catalog.MO_ISCP_LOG)
+	if err != nil {
+		return
+	}
+	exec.prevISCPTableID = tid
+
+	sql := cdc.CDCSQLBuilder.ISCPLogSelectSQL()
 	result, err := ExecWithResult(ctx, sql, exec.cnUUID, txn)
 	if err != nil {
 		return
@@ -747,6 +765,7 @@ func (exec *ISCPTaskExecutor) replay(ctx context.Context) (err error) {
 		return true
 	})
 	exec.iscpLogWm = types.TimestampToTS(txn.SnapshotTS())
+
 	return
 }
 

pkg/iscp/types.go

@@ -137,7 +137,9 @@ type ISCPTaskExecutor struct {
 	cnUUID      string
 	txnEngine   engine.Engine
 	cnTxnClient client.TxnClient
-	iscpLogWm   types.TS
+
+	iscpLogWm       types.TS
+	prevISCPTableID uint64
 
 	rpcHandleFn func(
 		ctx context.Context,

pkg/vm/engine/test/change_handle_test.go

@@ -5409,3 +5409,131 @@ func TestPartitionChangesHandleGCKPBoundaryStaleRead(t *testing.T) {
 		}
 	}
 }
+
+func TestISCPTableIDChange(t *testing.T) {
+	catalog.SetupDefines("")
+
+	var (
+		accountId = catalog.System_Account
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ctx = context.WithValue(ctx, defines.TenantIDKey{}, accountId)
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, time.Minute*5)
+	defer cancel()
+
+	disttaeEngine, taeHandler, rpcAgent, _ := testutil.CreateEngines(ctx, testutil.TestOptions{}, t)
+	defer func() {
+		disttaeEngine.Close(ctx)
+		taeHandler.Close(true)
+		rpcAgent.Close()
+	}()
+
+	err := mock_mo_indexes(disttaeEngine, ctxWithTimeout)
+	require.NoError(t, err)
+	err = mock_mo_foreign_keys(disttaeEngine, ctxWithTimeout)
+	require.NoError(t, err)
+	err = mock_mo_intra_system_change_propagation_log(disttaeEngine, ctxWithTimeout)
+	require.NoError(t, err)
+
+	bat := CreateDBAndTableForCNConsumerAndGetAppendData(t, disttaeEngine, ctxWithTimeout, "srcdb", "src_table", 10)
+	bats := bat.Split(10)
+	defer bat.Close()
+
+	// append 1 row
+	_, rel, txn, err := disttaeEngine.GetTable(ctxWithTimeout, "srcdb", "src_table")
+	require.Nil(t, err)
+
+	tableID := rel.GetTableID(ctxWithTimeout)
+
+	err = rel.Write(ctxWithTimeout, containers.ToCNBatch(bats[0]))
+	require.Nil(t, err)
+
+	txn.Commit(ctxWithTimeout)
+
+	// init cdc executor
+	checkLeaseStub := gostub.Stub(
+		&iscp.CheckLeaseWithRetry,
+		func(
+			context.Context,
+			string,
+			engine.Engine,
+			client.TxnClient,
+		) (bool, error) {
+			return true, nil
+		},
+	)
+	defer checkLeaseStub.Reset()
+	cdcExecutor, err := iscp.NewISCPTaskExecutor(
+		ctxWithTimeout,
+		disttaeEngine.Engine,
+		disttaeEngine.GetTxnClient(),
+		"",
+		&iscp.ISCPExecutorOption{
+			GCInterval:             time.Hour,
+			GCTTL:                  time.Hour,
+			SyncTaskInterval:       time.Millisecond * 100,
+			FlushWatermarkInterval: time.Millisecond * 100,
+			RetryTimes:             1,
+		},
+		common.DebugAllocator,
+	)
+	require.NoError(t, err)
+	cdcExecutor.SetRpcHandleFn(taeHandler.GetRPCHandle().HandleGetChangedTableList)
+
+	err = cdcExecutor.Start()
+	require.NoError(t, err)
+	defer cdcExecutor.Stop()
+
+	// register index job
+	txn, err = disttaeEngine.NewTxnOperator(ctx, disttaeEngine.Engine.LatestLogtailAppliedTime())
+	require.NoError(t, err)
+	ok, err := iscp.RegisterJob(
+		ctx, "", txn,
+		&iscp.JobSpec{
+			ConsumerInfo: iscp.ConsumerInfo{
+				ConsumerType: int8(iscp.ConsumerType_CNConsumer),
+			},
+		},
+		&iscp.JobID{
+			JobName:   "test_idx",
+			DBName:    "srcdb",
+			TableName: "src_table",
+		},
+		false,
+	)
+	assert.True(t, ok)
+	assert.NoError(t, err)
+	assert.NoError(t, txn.Commit(ctxWithTimeout))
+
+	// wait for synchronization to initialize prevISCPTableID
+
+	// enable injection to trigger table id change check
+	fault.Enable()
+	defer fault.Disable()
+	rmFn, err := objectio.InjectCDCExecutor("tableIDChange")
+	assert.NoError(t, err)
+	defer rmFn()
+
+	// append more data to trigger synchronization
+	_, rel, txn, err = disttaeEngine.GetTable(ctxWithTimeout, "srcdb", "src_table")
+	require.Nil(t, err)
+
+	err = rel.Write(ctxWithTimeout, containers.ToCNBatch(bats[1]))
+	require.Nil(t, err)
+
+	txn.Commit(ctxWithTimeout)
+
+	now := taeHandler.GetDB().TxnMgr.Now()
+	testutils.WaitExpect(
+		4000,
+		func() bool {
+			ts, ok := cdcExecutor.GetWatermark(accountId, tableID, "test_idx")
+			return ok && ts.GE(&now)
+		},
+	)
+	ts, ok := cdcExecutor.GetWatermark(accountId, tableID, "test_idx")
+	assert.True(t, ok)
+	assert.True(t, ts.GE(&now))
+}