[FLINK-36981][cdc] Fix the bug of transform merging with route in YAML task

[aiwenmo]

Situation

In FlinkPipelineComposerITCase.testTransformMergingWithRoute:

// Create test dataset:
// Create table 1 [id, name, age]
// Table 1: +I[1, Alice, 18]
// Table 1: +I[2, Bob, 20]
// Table 1: -U[2, Bob, 20] +U[2, Bob, 30]
// Create table 2 [id, name, age, description]
// Table 2: +I[3, Charlie, 15, student]
// Table 2: +I[4, Donald, 25, student]
// Table 2: -D[4, Donald, 25, student]
// Rename column for table 1: name -> last_name
// Add column for table 2: gender
// Table 1: +I[5, Eliza, 24]
// Table 2: +I[6, Frank, 30, student, male] 

This is executed correctly.

// Create test dataset:
// Create table 1 [id, name, age]
// Create table 2 [id, name, age, description]
// Table 1: +I[1, Alice, 18]
// Table 1: +I[2, Bob, 20]
// Table 1: -U[2, Bob, 20] +U[2, Bob, 30]
// Table 2: +I[3, Charlie, 15, student]
// Table 2: +I[4, Donald, 25, student]
// Table 2: -D[4, Donald, 25, student]
// Rename column for table 1: name -> last_name
// Add column for table 2: gender
// Table 1: +I[5, Eliza, 24]
// Table 2: +I[6, Frank, 30, student, male] 

This will cause an exception.

Caused by: org.codehaus.commons.compiler.CompileException: Line 1, Column 76: Assignment conversion not possible from type "java.lang.Integer" to type "java.lang.Long" 

Cause

When a conversion rule of transform matches multiple tables, only tables with the same structure are supported. However, in the scenario of sharding databases and tables, it is possible to match multiple tables with different structures.
In the Map<Tuple2<TableId, TransformProjection>, TransformProjectionProcessor> transformProjectionProcessorMap ,TransformProjection is wrongly shared, resulting in Schema always having the schema of the latest matched table.

Fix

1.Change Map<Tuple2<TableId, TransformProjection>, TransformProjectionProcessor> transformProjectionProcessorMap into Map<Tuple2<TableId, String>, TransformProjectionProcessor> transformProjectionProcessorMap.
2.When transformProjectionProcessor has none TableInfo, set TransformProjection into new transformProjectionProcessor from existed transformProjectionProcessor.

[aiwenmo]

PTAL @yuxiqian @lvyanquan

[yuxiqian]

1.Change Map<Tuple2<TableId, TransformProjection>, TransformProjectionProcessor> transformProjectionProcessorMap into Map<Tuple2<TableId, String>, TransformProjectionProcessor> transformProjectionProcessorMap.

Seems you're changing the key of this map from [TableId, TransformProjection] to [Transform, TransformProjection#projection]. However the problem is Projection string is a field of TransformProjection. If TransformProjection might collide among different rules, how can the ProjectionString be unique?

---

Off topic, but I doubt if transform module has to be that complicated.

[yuxiqian]

As discussed with @aiwenmo, we'll keep this PR minimal, focusing on bug-fix and leave transform module cleaning-up work later.

Could @leonardBang please trigger the CI?

[leonardBang]

Thanks @aiwenmo for the contribution and @yuxiqian for the review, LGTM