[
https://issues.apache.org/jira/browse/FLINK-40690?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40690:
-----------------------------------
Labels: pull-request-available (was: )
> Fluss sink applies top-level column mapping to nested ROW fields
> ----------------------------------------------------------------
>
> Key: FLINK-40690
> URL: https://issues.apache.org/jira/browse/FLINK-40690
> Project: Flink
> Issue Type: Bug
> Reporter: Xiaobing Fang
> Priority: Major
> Labels: pull-request-available
>
> h2. Description
> The Fluss pipeline sink reuses the top-level column mapping when reading
> fields inside a nested ROW. This can silently reorder nested values even when
> the source and sink nested types are identical.
> The problem also occurs without column reordering: if a nested ROW has more
> fields than the top-level row, child positions absent from the parent mapping
> are incorrectly reported as null.
> h2. Steps to reproduce
> Use source columns {{[nested ROW<a INT, b INT>, id INT]}} and sink columns
> {{[id INT, nested ROW<a INT, b INT>]}}. The input {{((7, 8), 42)}} should
> become {{(42, (7, 8))}}, but the nested fields are read as {{(8, 7)}}.
> The adapter-level reproduction is:
> {code:java}
> GenericRecordData input =
> GenericRecordData.of(GenericRecordData.of(7, 8), 42);
> InternalRow output = CdcAsFlussRow.replace(input, 2, Map.of(0, 1, 1, 0));
> InternalRow nested = output.getRow(1, 2);
> // nested.getInt(0) returns 8 instead of 7.
> // nested.getInt(1) returns 7 instead of 8.
> {code}
> Reproduced with a unit test on master commit {{93a40138}}. A pipeline
> integration regression also reproduces the issue: pre-create a primary-key
> Fluss table with {{[id, nested]}}, run a CDC pipeline using source order
> {{[nested, id]}}, then read the complete result through Flink SQL. The old
> implementation returns {{(42, (8, 7))}} instead of {{(42, (7, 8))}}.
> h2. Expected behavior and proposed fix
> The top-level mapping should select the nested column only. Nested fields
> should retain their own positions, as their source and sink types are already
> required to match.
> In {{CdcAsFlussRow.getRow()}}, wrap the selected nested record using the
> existing identity-mapping {{replace(RecordData)}} overload, consistent with
> {{CdcAsFlussArray.getRow()}}. This does not add support for nested schema
> evolution.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
