[jira] [Updated] (FLINK-40690) Fluss sink applies top-level column mapping to nested ROW fields

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ 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)