This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 7ecab6a8f [FLINK-40690][cdc-connect][fluss] Preserve nested row field
positions (#4549)
7ecab6a8f is described below
commit 7ecab6a8f0b8887bb18ac77988a5c9c1f4c01513
Author: Xiaobing Fang <[email protected]>
AuthorDate: Mon Sep 21 10:41:52 2026 +0800
[FLINK-40690][cdc-connect][fluss] Preserve nested row field positions
(#4549)
---
.../connectors/fluss/sink/row/CdcAsFlussRow.java | 3 +-
.../fluss/sink/FlussSinkPipelineITCase.java | 45 ++++++++
.../fluss/sink/row/CdcAsFlussRowTest.java | 117 +++++++++++++++++++++
3 files changed, 163 insertions(+), 2 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRow.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRow.java
index 4a97ebc22..b107e897b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRow.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/main/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRow.java
@@ -181,8 +181,7 @@ public class CdcAsFlussRow implements InternalRow {
@Override
public InternalRow getRow(int i, int numFields) {
- return new CdcAsFlussRow(
- cdcRecord.getRow(indexMapping.get(i), numFields), numFields,
indexMapping);
+ return CdcAsFlussRow.replace(cdcRecord.getRow(indexMapping.get(i),
numFields));
}
@VisibleForTesting
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
index f371cdd0e..455f1e8cf 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/FlussSinkPipelineITCase.java
@@ -18,6 +18,7 @@
package org.apache.flink.cdc.connectors.fluss.sink;
import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.data.GenericRecordData;
import org.apache.flink.cdc.common.data.binary.BinaryStringData;
import org.apache.flink.cdc.common.event.AddColumnEvent;
import org.apache.flink.cdc.common.event.CreateTableEvent;
@@ -75,6 +76,7 @@ import static
org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOA
import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
import static
org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
import static
org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** ITCase for Fluss Pipeline. */
@@ -508,6 +510,49 @@ public class FlussSinkPipelineITCase {
checkResult(TABLE_2, Arrays.asList("+I[1, 1]", "+I[2, 2]", "+I[3,
3]"));
}
+ @Test
+ void testNestedRowWithReorderedColumns() throws Exception {
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "CREATE TABLE %s.%s (id INT, nested ROW<a INT,
b INT>,"
+ + " PRIMARY KEY (id) NOT ENFORCED)",
+ TABLE_1.getSchemaName(),
TABLE_1.getTableName()))
+ .await();
+
+ RowType nestedType =
+ DataTypes.ROW(
+ DataTypes.FIELD("a", DataTypes.INT()),
+ DataTypes.FIELD("b", DataTypes.INT()));
+ Schema schema =
+ Schema.newBuilder()
+ .physicalColumn("nested", nestedType)
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .primaryKey("id")
+ .build();
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(RowType.of(nestedType,
DataTypes.INT().notNull()));
+ List<Event> events =
+ Arrays.asList(
+ new CreateTableEvent(TABLE_1, schema),
+ DataChangeEvent.insertEvent(
+ TABLE_1,
+ generator.generate(new Object[]
{GenericRecordData.of(7, 8), 42})));
+
+ composeAndExecuteInEvolveMode(Collections.singletonList(events));
+ List<Row> results = new ArrayList<>();
+ try (CloseableIterator<Row> rows =
+ tBatchEnv
+ .executeSql(
+ String.format(
+ "SELECT id, nested FROM %s.%s",
+ TABLE_1.getSchemaName(),
TABLE_1.getTableName()))
+ .collect()) {
+ rows.forEachRemaining(results::add);
+ }
+ assertThat(results).containsExactly(Row.of(42, Row.of(7, 8)));
+ }
+
@Test
void testInsertExistTableWithMoreColumns() throws Exception {
// create a fluss table with 2 columns but then cdc read a source
table with 3 columns
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRowTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRowTest.java
new file mode 100644
index 000000000..fcd44278f
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-fluss/src/test/java/org/apache/flink/cdc/connectors/fluss/sink/row/CdcAsFlussRowTest.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file distributed
with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.fluss.sink.row;
+
+import org.apache.flink.cdc.common.data.GenericRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.types.DataTypes;
+import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
+
+import org.apache.fluss.row.InternalRow;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link CdcAsFlussRow}. */
+class CdcAsFlussRowTest {
+
+ @Test
+ void testNestedRowWithIndexMapping() {
+ GenericRecordData input = GenericRecordData.of(GenericRecordData.of(7,
8), 42);
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input, 2, Map.of(0, 1, 1,
0));
+
+ assertThat(row.getInt(0)).isEqualTo(42);
+
+ InternalRow nested = row.getRow(1, 2);
+ assertThat(nested.getInt(0)).isEqualTo(7);
+ assertThat(nested.getInt(1)).isEqualTo(8);
+ }
+
+ @Test
+ void testNestedRowWithMoreChildrenThanOuterRow() {
+ GenericRecordData input = GenericRecordData.of(GenericRecordData.of(7,
8, 9), 42);
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input);
+
+ InternalRow nested = row.getRow(0, 3);
+ assertThat(nested.isNullAt(2)).isFalse();
+ assertThat(nested.getInt(0)).isEqualTo(7);
+ assertThat(nested.getInt(1)).isEqualTo(8);
+ assertThat(nested.getInt(2)).isEqualTo(9);
+ }
+
+ @Test
+ void testSparseOuterMappingKeepsNestedRowIntact() {
+ GenericRecordData input = GenericRecordData.of(GenericRecordData.of(7,
8), 42);
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input, 3, Map.of(0, 0, 2,
1));
+
+ assertThat(row.isNullAt(1)).isTrue();
+
+ InternalRow nested = row.getRow(0, 2);
+ assertThat(nested.isNullAt(1)).isFalse();
+ assertThat(nested.getInt(0)).isEqualTo(7);
+ assertThat(nested.getInt(1)).isEqualTo(8);
+ }
+
+ @Test
+ void testNestedRowWithNullChild() {
+ GenericRecordData input = GenericRecordData.of(GenericRecordData.of(7,
null, 8), 42);
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input, 2, Map.of(0, 1, 1,
0));
+
+ InternalRow nested = row.getRow(1, 3);
+ assertThat(nested.isNullAt(0)).isFalse();
+ assertThat(nested.getInt(0)).isEqualTo(7);
+ assertThat(nested.isNullAt(1)).isTrue();
+ assertThat(nested.isNullAt(2)).isFalse();
+ assertThat(nested.getInt(2)).isEqualTo(8);
+ }
+
+ @Test
+ void testNestedRowsWithDifferentArities() {
+ GenericRecordData innerA = GenericRecordData.of(7, 8, 9);
+ GenericRecordData innerB = GenericRecordData.of(10, 11);
+ GenericRecordData input =
GenericRecordData.of(GenericRecordData.of(innerA, innerB), 42);
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input, 2, Map.of(0, 1, 1,
0));
+
+ InternalRow middle = row.getRow(1, 2);
+ InternalRow first = middle.getRow(0, 3);
+ InternalRow second = middle.getRow(1, 2);
+ assertThat(first.getInt(0)).isEqualTo(7);
+ assertThat(first.getInt(1)).isEqualTo(8);
+ assertThat(first.getInt(2)).isEqualTo(9);
+ assertThat(second.getInt(0)).isEqualTo(10);
+ assertThat(second.getInt(1)).isEqualTo(11);
+ }
+
+ @Test
+ void testBinaryRecordDataNestedRow() {
+ BinaryRecordDataGenerator generator =
+ new BinaryRecordDataGenerator(
+ DataTypes.ROW(
+ DataTypes.ROW(DataTypes.INT(),
DataTypes.INT()), DataTypes.INT()));
+ BinaryRecordData input = generator.generate(new Object[]
{GenericRecordData.of(7, 8), 42});
+ CdcAsFlussRow row = CdcAsFlussRow.replace(input, 2, Map.of(0, 1, 1,
0));
+
+ assertThat(row.getInt(0)).isEqualTo(42);
+
+ InternalRow nested = row.getRow(1, 2);
+ assertThat(nested.getInt(0)).isEqualTo(7);
+ assertThat(nested.getInt(1)).isEqualTo(8);
+ }
+}
