[core][spark][flink] Fix nested data evolution isolation

docs/generated/core_configuration.html

@@ -528,7 +528,7 @@
             <td><h5>data-evolution.nested-field.enabled</h5></td>
             <td style="word-wrap: break-word;">false</td>
             <td>Boolean</td>
-            <td>Whether to enable sub-field-level data evolution for nested (struct) columns. When enabled, an update that only touches some sub-fields of a nested column writes an incremental file containing just those sub-fields (aligned by row id); when disabled, the whole top-level column is rewritten. Requires data-evolution.enabled=true. Mixed-version compatibility warning: once a file's write columns record a nested sub-field path (e.g. 'nest.a'), a reader, writer, compactor, or other maintenance job on an older version cannot reconstruct it. Every such component reading or writing this table must be upgraded before enabling this option, and downgrading the binary is unsafe once such files have been committed.</td>
+            <td>Whether to enable sub-field-level data evolution for nested (struct) columns. When enabled, an update that only touches some sub-fields of a nested column writes an incremental file containing just those sub-fields (aligned by row id); when disabled, the whole top-level column is rewritten. Requires data-evolution.enabled=true. Mixed-version compatibility warning: once a file's write columns record a nested sub-field path (e.g. 'nest.a'), a reader, writer, compactor, or other maintenance job on an older version cannot reconstruct it. Every such component reading or writing this table must be upgraded before enabling this option, and downgrading the binary is unsafe once such files have been committed. This option may only be enabled through a persisted table-option change; dynamic overrides and disabling or removing the option after it has been enabled are not supported.</td>
         </tr>
         <tr>
             <td><h5>data-evolution.reassign.skip-contiguous-row-count</h5></td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -2538,7 +2538,9 @@ public String toString() {
                                     + "reconstruct it. Every such component reading or writing this "
                                     + "table must be upgraded before enabling this option, and "
                                     + "downgrading the binary is unsafe once such files have been "
-                                    + "committed.");
+                                    + "committed. This option may only be enabled through a persisted "
+                                    + "table-option change; dynamic overrides and disabling or removing "
+                                    + "the option after it has been enabled are not supported.");
 
     public static final ConfigOption<Long> DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT =
             key("data-evolution.reassign.skip-contiguous-row-count")

paimon-api/src/main/java/org/apache/paimon/schema/TableSchema.java

@@ -285,12 +285,15 @@ public TableSchema project(@Nullable List<String> writeCols) {
             return this;
         }
 
+        RowType rowType = new RowType(fields);
+        List<DataField> projectedFields =
+                new CoreOptions(options).dataEvolutionNestedFieldEnabled()
+                        ? rowType.projectByPaths(writeCols).getFields()
+                        : rowType.project(writeCols).getFields();
         return new TableSchema(
                 version,
                 id,
-                // writeCols may contain nested dotted paths (e.g. "nest.a") for sub-field-level
-                // data evolution; projectByPaths handles both plain top-level names and paths
-                new RowType(fields).projectByPaths(writeCols).getFields(),
+                projectedFields,
                 highestFieldId,
                 partitionKeys,
                 primaryKeys,

paimon-api/src/main/java/org/apache/paimon/types/RowType.java

@@ -416,7 +416,7 @@ private static RowType projectTypeByPaths(RowType type, List<String> paths) {
      * covers some sub-fields is expanded into dotted leaf paths. This is the inverse of {@link
      * #projectByPaths(List)} and is used to derive {@code writeCols}.
      */
-    public List<String> leafPaths(RowType fullType) {
+    public List<String> collectLeafPaths(RowType fullType) {
         List<String> result = new ArrayList<>();
         collectLeafPaths(getFields(), fullType, fullType, "", result);
         return result;

paimon-api/src/test/java/org/apache/paimon/schema/TableSchemaTest.java

@@ -0,0 +1,92 @@
+/*
+ * 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.schema;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TableSchema}. */
+class TableSchemaTest {
+
+    @Test
+    void testNestedProjectionRequiresEnabledOption() {
+        TableSchema disabled = nestedSchema(Collections.emptyMap());
+        assertThatThrownBy(() -> disabled.project(Collections.singletonList("nest.a")))
+                .isInstanceOf(IndexOutOfBoundsException.class);
+
+        Map<String, String> enabledOptions = new HashMap<>();
+        enabledOptions.put(CoreOptions.DATA_EVOLUTION_NESTED_FIELD_ENABLED.key(), "true");
+        TableSchema enabled = nestedSchema(enabledOptions);
+        RowType projected = enabled.project(Collections.singletonList("nest.a")).logicalRowType();
+
+        assertThat(projected.getFieldNames()).containsExactly("nest");
+        assertThat(((RowType) projected.getTypeAt(0)).getFieldNames()).containsExactly("a");
+    }
+
+    @Test
+    void testDisabledProjectionTreatsDotAsPartOfTopLevelName() {
+        TableSchema schema =
+                new TableSchema(
+                        1L,
+                        Arrays.asList(
+                                new DataField(1, "nest.a", DataTypes.INT()),
+                                new DataField(
+                                        2,
+                                        "nest",
+                                        DataTypes.ROW(new DataField(3, "a", DataTypes.STRING())))),
+                        3,
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        Collections.emptyMap(),
+                        "");
+
+        assertThat(schema.project(Collections.singletonList("nest.a")).fields())
+                .extracting(DataField::id)
+                .containsExactly(1);
+    }
+
+    private static TableSchema nestedSchema(Map<String, String> options) {
+        return new TableSchema(
+                1L,
+                Collections.singletonList(
+                        new DataField(
+                                1,
+                                "nest",
+                                DataTypes.ROW(
+                                        new DataField(2, "a", DataTypes.INT()),
+                                        new DataField(3, "b", DataTypes.STRING())))),
+                3,
+                Collections.emptyList(),
+                Collections.emptyList(),
+                options,
+                "");
+    }
+}

paimon-api/src/test/java/org/apache/paimon/types/RowTypeTest.java

@@ -26,11 +26,11 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
-/** Tests for {@link RowType#leafPaths} and {@link RowType#projectByPaths}. */
+/** Tests for {@link RowType#collectLeafPaths} and {@link RowType#projectByPaths}. */
 class RowTypeTest {
 
     @Test
-    void leafPathsRejectsDottedPathCollidingWithTopLevelFieldName() {
+    void collectLeafPathsRejectsDottedPathCollidingWithTopLevelFieldName() {
         // fullType has a top-level field literally named "a.b" (id 5), plus a struct "a" (id 6)
         // with children x (id 7) and b (id 8).
         RowType fullType =
@@ -58,13 +58,13 @@ void leafPathsRejectsDottedPathCollidingWithTopLevelFieldName() {
                                                 Arrays.asList(
                                                         new DataField(8, "b", new IntType()))))));
 
-        assertThatThrownBy(() -> writeType.leafPaths(fullType))
+        assertThatThrownBy(() -> writeType.collectLeafPaths(fullType))
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessageContaining("a.b");
     }
 
     @Test
-    void leafPathsPreservesReorderedFullStructWriteOrder() {
+    void collectLeafPathsPreservesReorderedFullStructWriteOrder() {
         // fullType: nest<a INT, b STRING> declared in order a, b.
         RowType nestFull =
                 new RowType(
@@ -81,7 +81,7 @@ void leafPathsPreservesReorderedFullStructWriteOrder() {
         // Even though every sub-field of "nest" is present, the reordered layout must not
         // collapse to the bare top-level name "nest" (that would silently discard the physical
         // write order and let a reader reconstruct schema-declaration order instead).
-        List<String> leafPaths = writeType.leafPaths(fullType);
+        List<String> leafPaths = writeType.collectLeafPaths(fullType);
         assertThat(leafPaths).containsExactly("nest.b", "nest.a");
 
         RowType reconstructed = fullType.projectByPaths(leafPaths);
@@ -90,7 +90,7 @@ void leafPathsPreservesReorderedFullStructWriteOrder() {
     }
 
     @Test
-    void leafPathsCollapsesToWholeFieldWhenOrderMatches() {
+    void collectLeafPathsCollapsesToWholeFieldWhenOrderMatches() {
         // Same schema, but written in declaration order: coversFully should still collapse to
         // the bare top-level name, since nothing is ambiguous or reordered here.
         RowType nestFull =
@@ -101,6 +101,6 @@ void leafPathsCollapsesToWholeFieldWhenOrderMatches() {
         RowType fullType = new RowType(Arrays.asList(new DataField(1, "nest", nestFull)));
 
         RowType writeType = fullType.projectByPaths(Arrays.asList("nest.a", "nest.b"));
-        assertThat(writeType.leafPaths(fullType)).containsExactly("nest");
+        assertThat(writeType.collectLeafPaths(fullType)).containsExactly("nest");
     }
 }

paimon-common/src/main/java/org/apache/paimon/reader/DataEvolutionFileReader.java

@@ -86,7 +86,9 @@ public DataEvolutionFileReader(
     @Nullable
     public RecordIterator<InternalRow> readBatch() throws IOException {
         DataEvolutionRow row = new DataEvolutionRow(readers.length, rowOffsets, fieldOffsets);
-        row.setNested(nested);
+        if (nested != null) {
+            row.setNested(nested);
+        }
         RecordIterator<InternalRow>[] iterators = new RecordIterator[readers.length];
         for (int i = 0; i < readers.length; i++) {
             RecordReader<InternalRow> reader = readers[i];

paimon-common/src/main/java/org/apache/paimon/reader/DataEvolutionRow.java

@@ -43,6 +43,10 @@ public class DataEvolutionRow implements InternalRow {
      */
     private NestedField[] nested;
 
+    // Only nested-field composition installs nullable source rows. Ordinary union rows retain the
+    // legacy invariant that every referenced source row is present.
+    private boolean rowsMayBeNull;
+
     private RowKind rowKind;
 
     public DataEvolutionRow(int rowNumber, int[] rowOffsets, int[] fieldOffsets) {
@@ -72,6 +76,7 @@ public void setRow(int pos, InternalRow row) {
     }
 
     private void setRowsAllowNull(InternalRow[] newRows) {
+        rowsMayBeNull = true;
         for (int i = 0; i < newRows.length; i++) {
             this.rows[i] = newRows[i];
             if (rowKind == null && newRows[i] != null) {
@@ -139,7 +144,7 @@ public boolean isNullAt(int pos) {
             return true;
         }
         InternalRow row = chooseRow(pos);
-        return row == null || row.isNullAt(offsetInRow(pos));
+        return (rowsMayBeNull && row == null) || row.isNullAt(offsetInRow(pos));
     }
 
     @Override

paimon-common/src/test/java/org/apache/paimon/types/DataTypesTest.java

@@ -297,21 +297,21 @@ void testProjectByPathsMatchesLegacyProjectIncludingOrder() {
         RowType reordered = type.projectByPaths(Arrays.asList("nest.b", "nest.a"));
         Assertions.assertThat(((RowType) reordered.getField("nest").type()).getFieldNames())
                 .containsExactly("b", "a");
-        // ... and leafPaths does not collapse it to the whole column name: coversFully requires
-        // the same physical order as the full type, not just the same field ids, so a
+        // ... and collectLeafPaths does not collapse it to the whole column name: coversFully
+        // requires the same physical order as the full type, not just the same field ids, so a
         // complete-but-reordered struct is still treated as a partial (order-preserving) write.
         // Collapsing it to "nest" would let a reader reconstruct schema-declaration order (a, b)
         // instead of the physical write order (b, a).
-        Assertions.assertThat(reordered.leafPaths(type)).containsExactly("nest.b", "nest.a");
+        Assertions.assertThat(reordered.collectLeafPaths(type)).containsExactly("nest.b", "nest.a");
     }
 
     /**
-     * leafPaths replaced getFieldNames() on the write path, and its result is persisted into the
-     * manifest as writeCols. Metadata written there is permanent, so for any write type that
+     * collectLeafPaths replaced getFieldNames() on the write path, and its result is persisted into
+     * the manifest as writeCols. Metadata written there is permanent, so for any write type that
      * contains no partially-written struct the two must produce exactly the same list.
      */
     @Test
-    void testLeafPathsEqualsFieldNamesWithoutPartialStruct() {
+    void testCollectLeafPathsEqualsFieldNamesWithoutPartialStruct() {
         RowType full =
                 new RowType(
                         Arrays.asList(
@@ -331,8 +331,10 @@ void testLeafPathsEqualsFieldNamesWithoutPartialStruct() {
                         Collections.singletonList("nest"),
                         Collections.singletonList("id"))) {
             RowType writeType = full.projectByPaths(names);
-            Assertions.assertThat(writeType.leafPaths(full))
-                    .as("leafPaths must equal getFieldNames for whole-column write type %s", names)
+            Assertions.assertThat(writeType.collectLeafPaths(full))
+                    .as(
+                            "collectLeafPaths must equal getFieldNames for whole-column write type %s",
+                            names)
                     .isEqualTo(writeType.getFieldNames());
         }
 
@@ -342,17 +344,17 @@ void testLeafPathsEqualsFieldNamesWithoutPartialStruct() {
                         Arrays.asList(
                                 new DataField(0, "id", DataTypes.INT()),
                                 new DataField(-1, "_ROW_ID", DataTypes.BIGINT())));
-        Assertions.assertThat(withSystemField.leafPaths(full))
+        Assertions.assertThat(withSystemField.collectLeafPaths(full))
                 .isEqualTo(withSystemField.getFieldNames());
 
         // a top-level column whose name contains a dot is still emitted whole, not rejected
         RowType dotted =
                 new RowType(Collections.singletonList(new DataField(0, "a.b", DataTypes.INT())));
-        Assertions.assertThat(dotted.leafPaths(dotted)).containsExactly("a.b");
+        Assertions.assertThat(dotted.collectLeafPaths(dotted)).containsExactly("a.b");
     }
 
     @Test
-    void testLeafPaths() {
+    void testCollectLeafPaths() {
         RowType full =
                 new RowType(
                         Arrays.asList(
@@ -374,21 +376,23 @@ void testLeafPaths() {
                                                                         DataTypes.INT())))))));
 
         // a full write collapses to top-level names (no dotted paths)
-        Assertions.assertThat(full.leafPaths(full)).containsExactly("id", "nest");
+        Assertions.assertThat(full.collectLeafPaths(full)).containsExactly("id", "nest");
 
         // one level of partial nesting: a direct sub-field of a top-level struct
         Assertions.assertThat(
-                        full.projectByPaths(Collections.singletonList("nest.a")).leafPaths(full))
+                        full.projectByPaths(Collections.singletonList("nest.a"))
+                                .collectLeafPaths(full))
                 .containsExactly("nest.a");
 
         // a whole sub-struct under a partial top-level struct is still one level
         Assertions.assertThat(
-                        full.projectByPaths(Collections.singletonList("nest.sub")).leafPaths(full))
+                        full.projectByPaths(Collections.singletonList("nest.sub"))
+                                .collectLeafPaths(full))
                 .containsExactly("nest.sub");
 
         // deeper than one level (a partial sub-struct) is rejected so it can never be committed
         RowType deepPartial = full.projectByPaths(Collections.singletonList("nest.sub.x"));
-        assertThatThrownBy(() -> deepPartial.leafPaths(full))
+        assertThatThrownBy(() -> deepPartial.collectLeafPaths(full))
                 .isInstanceOf(UnsupportedOperationException.class)
                 .hasMessageContaining("one level");
     }

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -303,6 +303,7 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                                 bucketMode(),
                                 options.deletionVectorsEnabled(),
                                 options.dataEvolutionEnabled(),
+                                options.dataEvolutionNestedFieldEnabled(),
                                 options.pkClusteringOverride(),
                                 newIndexFileHandler(),
                                 snapshotManager,

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionNormalCompactTask.java

@@ -55,7 +55,6 @@
 import static org.apache.paimon.types.VectorType.isVectorStoreFile;
 import static org.apache.paimon.utils.DataEvolutionUtils.checkContiguousRowRange;
 import static org.apache.paimon.utils.DataEvolutionUtils.fieldMaxSequenceNumber;
-import static org.apache.paimon.utils.DataEvolutionUtils.fileFields;
 import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Compacts normal structured files of a data evolution table. */
@@ -197,4 +196,13 @@ private long[] compactedColumnMaxSequenceNumbers(
         }
         return result;
     }
+
+    private static List<DataField> fileFields(
+            Function<Long, TableSchema> schemaLoader, DataFileMeta file) {
+        TableSchema fileSchema = schemaLoader.apply(file.schemaId());
+        boolean nestedFieldEnabled =
+                new CoreOptions(fileSchema.options()).dataEvolutionNestedFieldEnabled();
+        return org.apache.paimon.utils.DataEvolutionUtils.fileFields(
+                fileSchema.fields(), file, nestedFieldEnabled);
+    }
 }