Xiaobing Fang created FLINK-40690:
-------------------------------------
Summary: 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
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)
