JingsongLi commented on code in PR #8334: URL: https://github.com/apache/paimon/pull/8334#discussion_r3460297953
########## paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/NestedSubfieldMergeIntoActionITCase.java: ########## @@ -0,0 +1,330 @@ +/* + * 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.paimon.flink.action; + +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.table.FileStoreTable; + +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import static org.apache.flink.table.planner.factories.TestValuesTableFactory.changelogRow; +import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_ENABLED; +import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED; +import static org.apache.paimon.CoreOptions.ROW_TRACKING_ENABLED; +import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.buildDdl; +import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init; +import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.insertInto; +import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.sEnv; +import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.testBatchRead; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * ITCase for sub-field-level data evolution via {@link DataEvolutionMergeIntoAction}: updating a + * single sub-field of a nested struct column should write an incremental file containing only that + * sub-field (a dotted write column like {@code nest.a}) aligned by row id, while the rest of the + * struct is read back from the original file. + */ +public class NestedSubfieldMergeIntoActionITCase extends ActionITCaseBase { + + @Override + public void before() throws IOException { + super.before(); + init(warehouse); + } + + private void prepareNestedTarget(boolean nestedFieldEnabled) throws Exception { + sEnv.executeSql( + buildDdl( + "T", + Arrays.asList("id INT", "nest ROW<a INT, b STRING>"), + Collections.emptyList(), + Collections.emptyList(), + new HashMap<String, String>() { + { + put(ROW_TRACKING_ENABLED.key(), "true"); + put(DATA_EVOLUTION_ENABLED.key(), "true"); + if (nestedFieldEnabled) { + put(DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true"); + } + } + })); + insertInto( + "T", + "(1, CAST(ROW(10, 'x') AS ROW<a INT, b STRING>))", + "(2, CAST(ROW(20, 'y') AS ROW<a INT, b STRING>))"); + } + + private void prepareSubFieldSource() throws Exception { + sEnv.executeSql( + buildDdl( + "S", + Arrays.asList("id INT", "newa INT"), + Collections.emptyList(), + Collections.emptyList(), + new HashMap<String, String>() { + { + put(ROW_TRACKING_ENABLED.key(), "true"); + put(DATA_EVOLUTION_ENABLED.key(), "true"); + } + })); + insertInto("S", "(1, 100)"); + } + + @Test + public void testUpdateSingleSubFieldWritesOnlyThatLeaf() throws Exception { + prepareNestedTarget(true); + prepareSubFieldSource(); + + builder(warehouse, database, "T") + .withMergeCondition("T.id=S.id") + .withMatchedUpdateSet("T.nest.a=S.newa") + .withSourceTable("S") + .withSinkParallelism(2) + .build() + .run(); + + // correctness: nest.a updated for id=1, nest.b preserved, other row untouched + testBatchRead( + "SELECT id, nest.a, nest.b FROM T", + Arrays.asList(changelogRow("+I", 1, 100, "x"), changelogRow("+I", 2, 20, "y"))); + + // feature engaged: an incremental file written by the merge only contains nest.a + assertThat(deltaWriteCols("T")).contains(Collections.singletonList("nest.a")); + } + + @Test + public void testUpdateSubFieldDisabledThrows() throws Exception { + // data-evolution.nested-field.enabled left at its default (false) + prepareNestedTarget(false); + prepareSubFieldSource(); + + assertThatThrownBy( + () -> + builder(warehouse, database, "T") + .withMergeCondition("T.id=S.id") + .withMatchedUpdateSet("T.nest.a=S.newa") + .withSourceTable("S") + .withSinkParallelism(2) + .build() + .run()) + .hasMessageContaining(DATA_EVOLUTION_NESTED_FIELD_ENABLED.key()); + } + + @Test + public void testUpdateWholeStructStillWorks() throws Exception { + prepareNestedTarget(true); + sEnv.executeSql( + buildDdl( + "S", + Arrays.asList("id INT", "nest ROW<a INT, b STRING>"), + Collections.emptyList(), + Collections.emptyList(), + new HashMap<String, String>() { + { + put(ROW_TRACKING_ENABLED.key(), "true"); + put(DATA_EVOLUTION_ENABLED.key(), "true"); + } + })); + insertInto("S", "(1, CAST(ROW(100, 'z') AS ROW<a INT, b STRING>))"); + + builder(warehouse, database, "T") + .withMergeCondition("T.id=S.id") + .withMatchedUpdateSet("T.nest=S.nest") + .withSourceTable("S") + .withSinkParallelism(2) + .build() + .run(); + + testBatchRead( + "SELECT id, nest.a, nest.b FROM T", + Arrays.asList(changelogRow("+I", 1, 100, "z"), changelogRow("+I", 2, 20, "y"))); + } + + @Test + public void testUpdateMultipleSubFieldsWritesOnlyThoseLeaves() throws Exception { + // a 3-field struct so that updating two sub-fields is a strict subset (stays + // sub-field-level + // rather than collapsing to a whole-column write) + sEnv.executeSql( + buildDdl( + "T", + Arrays.asList("id INT", "nest ROW<a INT, b STRING, c INT>"), + Collections.emptyList(), + Collections.emptyList(), + new HashMap<String, String>() { + { + put(ROW_TRACKING_ENABLED.key(), "true"); + put(DATA_EVOLUTION_ENABLED.key(), "true"); + put(DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true"); + } + })); + insertInto( + "T", + "(1, CAST(ROW(10, 'x', 30) AS ROW<a INT, b STRING, c INT>))", + "(2, CAST(ROW(20, 'y', 40) AS ROW<a INT, b STRING, c INT>))"); + + sEnv.executeSql( + buildDdl( + "S", + Arrays.asList("id INT", "newa INT", "newc INT"), + Collections.emptyList(), + Collections.emptyList(), + new HashMap<String, String>() { + { + put(ROW_TRACKING_ENABLED.key(), "true"); + put(DATA_EVOLUTION_ENABLED.key(), "true"); + } + })); + insertInto("S", "(1, 100, 300)"); + + builder(warehouse, database, "T") + .withMergeCondition("T.id=S.id") + .withMatchedUpdateSet("T.nest.a=S.newa,T.nest.c=S.newc") + .withSourceTable("S") + .withSinkParallelism(2) + .build() + .run(); + + // correctness: a and c updated for id=1, b preserved, other row untouched + testBatchRead( + "SELECT id, nest.a, nest.b, nest.c FROM T", + Arrays.asList( + changelogRow("+I", 1, 100, "x", 300), changelogRow("+I", 2, 20, "y", 40))); + + // feature engaged: the incremental file contains exactly the two touched leaves, in schema + // order (a before c) + assertThat(deltaWriteCols("T")).contains(Arrays.asList("nest.a", "nest.c")); + } + + @Test + public void testUpdateDeeplyNestedSubFieldThrows() throws Exception { + sEnv.executeSql( + buildDdl( + "T", + Arrays.asList("id INT", "nest ROW<a INT, inner ROW<x INT, y INT>>"), Review Comment: After adding the missing `@BeforeEach` locally to let this test class initialize, this DDL still fails before reaching the assertion: Flink's parser treats `inner` as a keyword (`SQL parse failed. Encountered "inner" at line 1, column 41`). Please quote the nested field name (and the matching `CAST(ROW(... ) AS ROW<...>)` below) or use a non-keyword name, otherwise `testUpdateDeeplyNestedSubFieldThrows` cannot exercise the intended deeper-than-one-level validation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
