add one ut to cover hash partition table scenario

pkg/sink/codec/canal/canal_json_message.go

@@ -244,6 +244,22 @@ func (b *batchDecoder) setPhysicalTableID(event *model.RowChangedEvent, physical
 		return fmt.Errorf("cannot found partition for column value %s", columnValue)
 	// todo: support following rule if meet the corresponding workload
 	case pmodel.PartitionTypeHash:
+		targetColumnID := event.TableInfo.ForceGetColumnIDByName(strings.ReplaceAll(event.TableInfo.Partition.Expr, "`", ""))
+		columns := event.Columns
+		if columns == nil {
+			columns = event.PreColumns
+		}
+		var columnValue int64
+		for _, col := range columns {
+			if col.ColumnID == targetColumnID {
+				columnValue = col.Value.(int64)
+				break
+			}
+		}
+		result := columnValue % int64(len(event.TableInfo.Partition.Definitions))
+		partitionID := event.TableInfo.GetPartitionInfo().Definitions[result].ID
+		event.PhysicalTableID = partitionID
+		return nil
 	case pmodel.PartitionTypeKey:
 	case pmodel.PartitionTypeList:
 	case pmodel.PartitionTypeNone:

pkg/sink/codec/canal/canal_json_row_event_encoder_test.go

@@ -729,6 +729,59 @@ func TestCanalJSONContentCompatibleE2E(t *testing.T) {
 	}
 }
 
+func TestE2EPartitionTableByHash(t *testing.T) {
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	helper.Tk().MustExec("use test")
+
+	createTableDDLEvent := helper.DDL2Event(`CREATE TABLE t (a INT,PRIMARY KEY(a)) PARTITION BY HASH (a) PARTITIONS 5`)
+	require.NotNil(t, createTableDDLEvent)
+	insertEvent := helper.DML2Event(`insert into t values (5)`, "test", "t", "p0")
+	require.NotNil(t, insertEvent)
+
+	ctx := context.Background()
+	codecConfig := common.NewConfig(config.ProtocolCanalJSON)
+
+	builder, err := NewJSONRowEventEncoderBuilder(ctx, codecConfig)
+	require.NoError(t, err)
+	encoder := builder.Build()
+
+	decoder, err := NewBatchDecoder(ctx, codecConfig, nil)
+	require.NoError(t, err)
+
+	message, err := encoder.EncodeDDLEvent(createTableDDLEvent)
+	require.NoError(t, err)
+
+	err = decoder.AddKeyValue(message.Key, message.Value)
+	require.NoError(t, err)
+
+	tp, hasNext, err := decoder.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, model.MessageTypeDDL, tp)
+
+	decodedDDL, err := decoder.NextDDLEvent()
+	require.NoError(t, err)
+	require.NotNil(t, decodedDDL)
+
+	err = encoder.AppendRowChangedEvent(ctx, "", insertEvent, nil)
+	require.NoError(t, err)
+	message = encoder.Build()[0]
+
+	err = decoder.AddKeyValue(message.Key, message.Value)
+	require.NoError(t, err)
+	tp, hasNext, err = decoder.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, model.MessageTypeRow, tp)
+
+	decodedEvent, err := decoder.NextRowChangedEvent()
+	require.NoError(t, err)
+	require.NotZero(t, decodedEvent.GetTableID())
+	require.Equal(t, decodedEvent.GetTableID(), decodedEvent.TableInfo.GetPartitionInfo().Definitions[0].ID)
+}
+
 func TestE2EPartitionTableByRange(t *testing.T) {
 	helper := entry.NewSchemaTestHelper(t)
 	defer helper.Close()