[
https://issues.apache.org/jira/browse/FLINK-40739?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18117939#comment-18117939
]
Yanquan Lv commented on FLINK-40739:
------------------------------------
I would like to take it.
> [Bug][pipeline-connector][maxcompute] Partition values extracted from wrong
> column index when partition columns are not contiguous at the end of schema
> -------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40739
> URL: https://issues.apache.org/jira/browse/FLINK-40739
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.2.0, cdc-3.3.0, cdc-3.2.1, cdc-3.4.0, cdc-3.5.0
> Reporter: Yanquan Lv
> Priority: Minor
>
> h2. Description
> h3. Problem
> In {{{}SessionManageOperator#extractPartition{}}}, the MaxCompute pipeline
> sink computes the index of a partition column by assuming that all partition
> columns are contiguous and located at the end of the schema:
> [https://github.com/apache/flink-cdc/blob/master/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-maxcompute/src/main/java/org/apache/flink/cdc/connectors/maxcompute/coordinator/SessionManageOperator.java#L268]
> {code:java}
> for (int i = 0; i < partitionKeyCount; i++) {
> RecordData.FieldGetter fieldGetter =
> fieldGetters.get(columnCount - partitionKeyCount - 1 + i);
> ...
> } {code}
> The {{-1}} offset is incorrect. When partition columns are *not* contiguously
> placed at the end of the column list, the sink reads values from the wrong
> field getters and writes data into the wrong MaxCompute partitions.
> h3. Example
> Consider a table whose columns are ordered as:
> ||Index||Column||Is Partition Key||
> |0|id| |
> |1|pt|yes|
> |2|name| |
> |3|dt|yes|
> Here {{columnCount = 4}} and {{{}partitionKeyCount = 2{}}}.
> With the current code:
> * For {{{}i = 0{}}}: index = {{4 - 2 - 1 + 0 = 1}} → reads {{pt}} (correct
> by coincidence)
> * For {{{}i = 1{}}}: index = {{4 - 2 - 1 + 1 = 2}} → reads {{name}} (wrong!
> should read {{dt}} at index 3)
> As a result, the partition spec will contain {{{}pt=<correct>, dt=<value of
> name>{}}}, which silently corrupts data.
> h3. Impact
> * Data is written into incorrect partitions when the source schema has
> partition columns interleaved with regular columns.
> * This can happen after schema evolution (e.g., {{{}ADD COLUMN{}}}) or when
> the user explicitly defines columns in a non-trailing order.
> h3. Expected Behavior
> The sink should resolve each partition column by its name instead of by a
> fixed trailing offset, so that it works regardless of column ordering.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
