sink(ticdc): fix incorrect table info in delete event (#11975)

ref #11879
pingcap · Jan 7, 2025 · 99e00c0 · 99e00c0
1 parent 80f49c6
commit 99e00c0
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 9 deletions.
diff --git a/cmd/pulsar-consumer/main.go b/cmd/pulsar-consumer/main.go
@@ -524,14 +524,15 @@ func (c *Consumer) HandleMsg(msg pulsar.Message) error {
 				// todo: mark the offset after the DDL is fully synced to the downstream mysql.
 				continue
 			}
-			var partitionID int64
-			if row.TableInfo.IsPartitionTable() {
-				partitionID = row.GetTableID()
-			}
+			tableID := row.GetTableID()
 			// use schema, table and tableID to identify a table
-			tableID := c.fakeTableIDGenerator.
-				generateFakeTableID(row.TableInfo.GetSchemaName(), row.TableInfo.GetTableName(), partitionID)
-			row.TableInfo.TableName.TableID = tableID
+			switch c.option.protocol {
+			case config.ProtocolCanalJSON:
+			default:
+				tableID := c.fakeTableIDGenerator.
+					generateFakeTableID(row.TableInfo.GetSchemaName(), row.TableInfo.GetTableName(), tableID)
+				row.PhysicalTableID = tableID
+			}
 
 			group, ok := c.eventGroups[tableID]
 			if !ok {

diff --git a/pkg/sink/codec/canal/canal_json_message.go b/pkg/sink/codec/canal/canal_json_message.go
@@ -220,6 +220,8 @@ func (b *batchDecoder) canalJSONMessage2RowChange() (*model.RowChangedEvent, err
 	result.CommitTs = msg.getCommitTs()
 	result.PhysicalTableID = msg.getPhysicalTableID()
 	mysqlType := msg.getMySQLType()
+	result.TableInfo.TableName.IsPartition = msg.isPartition()
+	result.TableInfo.TableName.TableID = msg.getTableID()
 
 	var err error
 	if msg.eventType() == canal.EventType_DELETE {
@@ -236,8 +238,6 @@ func (b *batchDecoder) canalJSONMessage2RowChange() (*model.RowChangedEvent, err
 	if err != nil {
 		return nil, err
 	}
-	result.TableInfo.TableName.IsPartition = msg.isPartition()
-	result.TableInfo.TableName.TableID = msg.getTableID()
 
 	// for `UPDATE`, `old` contain old data, set it as the `PreColumns`
 	if msg.eventType() == canal.EventType_UPDATE {