fix: apply sink routing to new table ID after TRUNCATE TABLE

For TRUNCATE TABLE, job.TableID is the OLD table ID which no longer
exists in the snapshot after the DDL is applied. The snapshot only
contains the NEW table with a different ID from job.BinlogInfo.TableInfo.ID.

This fix ensures sink routing is applied to the correct (new) table ID
so that DML events after TRUNCATE are routed to the correct destination.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: Alexandra Cong

cdc/entry/schema_storage.go

@@ -244,7 +244,13 @@ func (s *schemaStorage) HandleDDLJob(job *timodel.Job) error {
 	// Apply sink routing to the TableInfo if sinkRouter is configured
 	// This ensures DML events get the routed schema/table for SQL generation
 	if s.sinkRouter != nil && job.TableID > 0 {
-		applySinkRoutingToTable(snap, job.TableID, s.sinkRouter)
+		tableIDForRouting := job.TableID
+		// For TRUNCATE TABLE, job.TableID is the OLD table ID which no longer exists
+		// in the snapshot. We need to use the NEW table ID from BinlogInfo.TableInfo.
+		if job.Type == timodel.ActionTruncateTable && job.BinlogInfo.TableInfo != nil {
+			tableIDForRouting = job.BinlogInfo.TableInfo.ID
+		}
+		applySinkRoutingToTable(snap, tableIDForRouting, s.sinkRouter)
 	}
 
 	s.snaps = append(s.snaps, snap)

cdc/entry/schema_storage_test.go

@@ -1162,4 +1162,41 @@ func TestSchemaStorageWithSinkRouting(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, "unmapped_db", t5.TableName.Schema)
 	require.Equal(t, "", t5.TableName.TargetSchema)
+
+	// Test TRUNCATE TABLE - this creates a new table with a new ID
+	// The new table should also have routing applied
+	tk.MustExec("truncate table source_db1.t4")
+
+	jobs, err = getAllHistoryDDLJob(store, f)
+	require.Nil(t, err)
+
+	var truncateT4Job *timodel.Job
+	for i := len(jobs) - 1; i >= 0; i-- {
+		if jobs[i].Type == timodel.ActionTruncateTable && jobs[i].SchemaName == "source_db1" && jobs[i].TableName == "t4" {
+			truncateT4Job = jobs[i]
+			break
+		}
+	}
+	require.NotNil(t, truncateT4Job)
+
+	// Note: For TRUNCATE, job.TableID is the OLD table ID, and job.BinlogInfo.TableInfo.ID is the NEW table ID
+	oldTableID := truncateT4Job.TableID
+	newTableID := truncateT4Job.BinlogInfo.TableInfo.ID
+	require.NotEqual(t, oldTableID, newTableID, "TRUNCATE should create a new table ID")
+
+	err = schemaStorage.HandleDDLJob(truncateT4Job)
+	require.NoError(t, err)
+
+	newSnap3, err := schemaStorage.GetSnapshot(ctx, truncateT4Job.BinlogInfo.FinishedTS)
+	require.NoError(t, err)
+
+	// The old table ID should not exist anymore
+	_, ok = newSnap3.PhysicalTableByID(oldTableID)
+	require.False(t, ok, "old table ID should not exist after truncate")
+
+	// The new table should exist and have routing applied
+	t4New, ok := newSnap3.PhysicalTableByID(newTableID)
+	require.True(t, ok, "new table ID should exist after truncate")
+	require.Equal(t, "source_db1", t4New.TableName.Schema)
+	require.Equal(t, "target_db1", t4New.TableName.TargetSchema, "new table after truncate should have routing applied")
 }

tests/integration_tests/sink_routing/data/test.sql

@@ -79,8 +79,9 @@ INSERT INTO truncate_test VALUES (2, 'also truncated');
 
 TRUNCATE TABLE truncate_test;
 
--- Insert new data after truncate
-INSERT INTO truncate_test VALUES (10, 'after truncate');
+-- NOTE: INSERT after truncate is done separately in run.sh to ensure
+-- the TRUNCATE DDL is fully processed before the INSERT arrives.
+-- This is necessary to properly test DML routing after TRUNCATE.
 
 -- ============================================
 -- DDL: DROP TABLE

tests/integration_tests/sink_routing/run.sh

@@ -89,7 +89,54 @@ function run() {
 	# ============================================
 	echo "Verifying TRUNCATE TABLE routing..."
 	check_table_exists target_db.truncate_test_routed ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	# After truncate, only 1 row should exist (inserted after truncate)
+	# Wait for TRUNCATE to complete by checking the table is empty
+	# (the pre-truncate rows should be gone)
+	echo "Waiting for TRUNCATE to complete..."
+	i=0
+	while [ $i -lt 60 ]; do
+		run_sql "SELECT COUNT(*) as cnt FROM target_db.truncate_test_routed" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+		if [ "$(cat $OUT_DIR/sql_res.$TEST_NAME.txt | grep -c 'cnt: 0')" -eq 1 ]; then
+			echo "TRUNCATE completed, table is empty"
+			break
+		fi
+		echo "Table not empty yet, current state:"
+		cat $OUT_DIR/sql_res.$TEST_NAME.txt
+		sleep 1
+		i=$((i + 1))
+	done
+	if [ $i -ge 60 ]; then
+		echo "Timeout waiting for TRUNCATE to complete"
+		echo "Final table state:"
+		run_sql "SELECT * FROM target_db.truncate_test_routed" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+		exit 1
+	fi
+
+	# Now insert data AFTER truncate is confirmed complete
+	# This ensures the INSERT uses the NEW table ID and tests DML routing after TRUNCATE
+	echo "Inserting data after TRUNCATE..."
+	run_sql "INSERT INTO source_db.truncate_test VALUES (10, 'after truncate')" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	# Wait for the INSERT to be replicated and verify it arrived at the routed destination
+	echo "Waiting for INSERT after TRUNCATE to be replicated..."
+	i=0
+	while [ $i -lt 60 ]; do
+		run_sql "SELECT COUNT(*) as cnt FROM target_db.truncate_test_routed WHERE id = 10" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+		if [ "$(cat $OUT_DIR/sql_res.$TEST_NAME.txt | grep -c 'cnt: 1')" -eq 1 ]; then
+			echo "INSERT after TRUNCATE successfully routed"
+			break
+		fi
+		sleep 1
+		i=$((i + 1))
+	done
+	if [ $i -ge 60 ]; then
+		echo "ERROR: INSERT after TRUNCATE was not routed to target_db.truncate_test_routed"
+		echo "This indicates DML routing is broken after TRUNCATE TABLE"
+		run_sql "SELECT COUNT(*) as cnt FROM target_db.truncate_test_routed" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+		cat $OUT_DIR/sql_res.$TEST_NAME.txt
+		exit 1
+	fi
+
+	# Final verification
 	run_sql "SELECT COUNT(*) as cnt FROM target_db.truncate_test_routed" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_contains "cnt: 1"
 	run_sql "SELECT id FROM target_db.truncate_test_routed" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}