[core][spark][flink] Support sub-field-level data evolution for nested columns

docs/generated/core_configuration.html

@@ -512,6 +512,12 @@
             <td>Boolean</td>
             <td>Whether to persist source when process merge into action on data evolution table.</td>
         </tr>
+        <tr>
+            <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>
+        </tr>
         <tr>
             <td><h5>data-evolution.reassign.skip-contiguous-row-count</h5></td>
             <td style="word-wrap: break-word;">1000000000</td>

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

@@ -2493,6 +2493,25 @@ public String toString() {
                     .defaultValue(false)
                     .withDescription("Whether enable data evolution for row tracking table.");
 
+    public static final ConfigOption<Boolean> DATA_EVOLUTION_NESTED_FIELD_ENABLED =
+            key("data-evolution.nested-field.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "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.");
+
     public static final ConfigOption<Long> DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT =
             key("data-evolution.reassign.skip-contiguous-row-count")
                     .longType()
@@ -4238,6 +4257,10 @@ public boolean dataEvolutionEnabled() {
         return options.get(DATA_EVOLUTION_ENABLED);
     }
 
+    public boolean dataEvolutionNestedFieldEnabled() {
+        return options.get(DATA_EVOLUTION_NESTED_FIELD_ENABLED);
+    }
+
     public long dataEvolutionReassignSkipContiguousRowCount() {
         long threshold = options.get(DATA_EVOLUTION_REASSIGN_SKIP_CONTIGUOUS_ROW_COUNT);
         checkArgument(

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

@@ -288,7 +288,9 @@ public TableSchema project(@Nullable List<String> writeCols) {
         return new TableSchema(
                 version,
                 id,
-                new RowType(fields).project(writeCols).getFields(),
+                // 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(),
                 highestFieldId,
                 partitionKeys,
                 primaryKeys,

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

@@ -37,6 +37,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -333,6 +334,194 @@ public RowType project(String... names) {
         return project(Arrays.asList(names));
     }
 
+    /**
+     * Project this row type by a list of (possibly nested) dotted paths, e.g. {@code ["f0",
+     * "nest.a"]}. A path without a dot selects the whole top-level field (same as {@link
+     * #project(List)}); a dotted path selects only the addressed sub-field of a nested {@link
+     * RowType}, preserving field ids and nullability of every level. Fields are emitted in the
+     * order the paths are given (exactly like {@link #project(List)}), not in schema declaration
+     * order. This is used by data evolution to reconstruct the partial nested schema of a
+     * column-group file from its {@code writeCols}.
+     */
+    public RowType projectByPaths(List<String> paths) {
+        return projectTypeByPaths(this, paths);
+    }
+
+    private static RowType projectTypeByPaths(RowType type, List<String> paths) {
+        // group paths by their immediate child name, keeping the order in which the paths are
+        // given; a child appearing without a tail (or also with a tail) is selected as a whole
+        // field
+        Map<String, List<String>> childToSubPaths = new LinkedHashMap<>();
+        Set<String> wholeChildren = new HashSet<>();
+        Map<String, DataField> fieldByName = new LinkedHashMap<>();
+        for (DataField field : type.getFields()) {
+            fieldByName.put(field.name(), field);
+        }
+        for (String path : paths) {
+            int dot = path.indexOf('.');
+            // Prefer an exact field-name match so a column whose name itself contains a dot (and
+            // any
+            // plain top-level name) is selected whole; only split into head.tail for genuine nested
+            // sub-field paths that do not name a field directly. This keeps backward compatibility
+            // with the legacy exact-name project(List).
+            if (dot < 0 || fieldByName.containsKey(path)) {
+                childToSubPaths.computeIfAbsent(path, k -> new ArrayList<>());
+                wholeChildren.add(path);
+            } else {
+                String head = path.substring(0, dot);
+                String tail = path.substring(dot + 1);
+                childToSubPaths.computeIfAbsent(head, k -> new ArrayList<>()).add(tail);
+            }
+        }
+
+        // Emit fields in the order the paths were given, exactly like project(List). Callers such
+        // as TableSchema.project(writeCols) rebuild the physical layout of a data file from its
+        // writeCols, and that order is not necessarily the schema declaration order; reordering
+        // here would silently describe the file with the columns permuted.
+        List<DataField> result = new ArrayList<>();
+        for (Map.Entry<String, List<String>> entry : childToSubPaths.entrySet()) {
+            String name = entry.getKey();
+            DataField field = fieldByName.get(name);
+            if (field == null) {
+                throw new IllegalArgumentException(
+                        "Cannot project by paths, unknown field '" + name + "' in " + type);
+            }
+            List<String> subPaths = entry.getValue();
+            if (wholeChildren.contains(name) || subPaths.isEmpty()) {
+                result.add(field);
+            } else if (field.type() instanceof RowType) {
+                RowType prunedChild =
+                        projectTypeByPaths((RowType) field.type(), subPaths)
+                                .copy(field.type().isNullable());
+                result.add(field.newType(prunedChild));
+            } else {
+                // a dotted path addresses a sub-field, but this field is not a ROW; reject rather
+                // than silently selecting the whole field, so invalid dotted paths surface early
+                throw new IllegalArgumentException(
+                        "Cannot project sub-field(s) "
+                                + subPaths
+                                + " of non-ROW field '"
+                                + name
+                                + "' in "
+                                + type);
+            }
+        }
+        return new RowType(type.isNullable(), result);
+    }
+
+    /**
+     * Compute the dotted paths describing this (possibly partially nested) write type relative to a
+     * full row type. A top-level field, or a nested field whose structure fully covers the
+     * corresponding field in {@code fullType}, is emitted by its name; a nested field that only
+     * 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) {
+        List<String> result = new ArrayList<>();
+        collectLeafPaths(getFields(), fullType, fullType, "", result);
+        return result;
+    }
+
+    private static void collectLeafPaths(
+            List<DataField> writeFields,
+            RowType fullType,
+            RowType topLevelFullType,
+            String prefix,
+            List<String> out) {
+        for (DataField writeField : writeFields) {
+            String path = prefix.isEmpty() ? writeField.name() : prefix + "." + writeField.name();
+            // A field absent from the reference type (e.g. the _ROW_ID / _SEQUENCE_NUMBER special
+            // fields added by row tracking, which are not part of the table's logical row type) has
+            // no sub-field split: emit it whole by name, matching the legacy getFieldNames()
+            // output.
+            if (!fullType.containsField(writeField.id())) {
+                out.add(path);
+                continue;
+            }
+            DataField fullField = fullType.getField(writeField.id());
+            boolean willExpand =
+                    writeField.type() instanceof RowType
+                            && fullField.type() instanceof RowType
+                            && !coversFully(
+                                    (RowType) writeField.type(), (RowType) fullField.type());
+            // A dotted path is only unambiguous if no name segment contains a literal '.'. A name
+            // with a dot is fine when emitted whole at top level (projectByPaths matches it
+            // exactly),
+            // but not when it participates in a multi-segment nested path.
+            if (writeField.name().indexOf('.') >= 0 && (!prefix.isEmpty() || willExpand)) {
+                throw new UnsupportedOperationException(
+                        "Sub-field-level data evolution does not support a nested field whose name "
+                                + "contains '.': "
+                                + path);
+            }
+            if (willExpand) {
+                // A partial struct nested inside another partial struct (a path deeper than one
+                // level, e.g. nest.sub.x) cannot be composed back on read — the data-evolution read
+                // path only assembles one nested level. Reject it here so such a file is never
+                // written/committed and later breaks full-table reads.
+                if (!prefix.isEmpty()) {
+                    throw new UnsupportedOperationException(
+                            "Sub-field-level data evolution supports only one level of partial "
+                                    + "nesting; the nested sub-field '"
+                                    + path
+                                    + "' cannot be partially written. Write the whole '"
+                                    + path
+                                    + "' sub-field instead.");
+                }
+                collectLeafPaths(
+                        ((RowType) writeField.type()).getFields(),
+                        (RowType) fullField.type(),
+                        topLevelFullType,
+                        path,
+                        out);
+            } else {
+                // A dotted leaf path is only unambiguous if it cannot also be read as a literal
+                // top-level field name: projectByPaths prefers an exact top-level name match, so
+                // a schema that has both a nested leaf "a.b" and a literal top-level field named
+                // "a.b" cannot be told apart once flattened into this same string. Reject such a
+                // write up front rather than silently reconstructing the wrong field on read.
+                if (!prefix.isEmpty() && topLevelFullType.containsField(path)) {
+                    throw new UnsupportedOperationException(
+                            "Sub-field-level data evolution cannot write the nested sub-field '"
+                                    + path
+                                    + "' because a top-level field named '"
+                                    + path
+                                    + "' already exists in the table, making the encoded write "
+                                    + "path ambiguous. Rename one of the two fields.");
+                }
+                out.add(path);
+            }
+        }
+    }
+
+    /**
+     * Whether {@code part} contains every (recursively nested) field of {@code full} in the same
+     * physical order. Order matters here: {@code part} describes the physical layout a file was
+     * actually written with, and a caller that finds this returns {@code true} collapses the leaf
+     * path to the bare top-level field name, relying on {@code full}'s declared order to describe
+     * that physical layout on read.
+     */
+    private static boolean coversFully(RowType part, RowType full) {
+        if (part.getFieldCount() != full.getFieldCount()) {
+            return false;
+        }
+        List<DataField> partFields = part.getFields();
+        List<DataField> fullFields = full.getFields();
+        for (int i = 0; i < fullFields.size(); i++) {
+            DataField partField = partFields.get(i);
+            DataField fullField = fullFields.get(i);
+            if (partField.id() != fullField.id()) {
+                return false;
+            }
+            if (partField.type() instanceof RowType && fullField.type() instanceof RowType) {
+                if (!coversFully((RowType) partField.type(), (RowType) fullField.type())) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
+
     private Map<String, DataField> nameToField() {
         Map<String, DataField> nameToField = this.laziedNameToField;
         if (nameToField == null) {

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

@@ -0,0 +1,106 @@
+/*
+ * 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.types;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+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}. */
+class RowTypeTest {
+
+    @Test
+    void leafPathsRejectsDottedPathCollidingWithTopLevelFieldName() {
+        // 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 =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(5, "a.b", new IntType()),
+                                new DataField(
+                                        6,
+                                        "a",
+                                        new RowType(
+                                                Arrays.asList(
+                                                        new DataField(7, "x", new IntType()),
+                                                        new DataField(8, "b", new IntType()))))));
+
+        // Partial write: only the "b" child of struct "a" is written (id 8), not "x". Its
+        // flattened dotted path "a.b" would collide with the literal top-level field of the same
+        // name.
+        RowType writeType =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(
+                                        6,
+                                        "a",
+                                        new RowType(
+                                                Arrays.asList(
+                                                        new DataField(8, "b", new IntType()))))));
+
+        assertThatThrownBy(() -> writeType.leafPaths(fullType))
+                .isInstanceOf(UnsupportedOperationException.class)
+                .hasMessageContaining("a.b");
+    }
+
+    @Test
+    void leafPathsPreservesReorderedFullStructWriteOrder() {
+        // fullType: nest<a INT, b STRING> declared in order a, b.
+        RowType nestFull =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(2, "a", new IntType()),
+                                new DataField(3, "b", VarCharType.STRING_TYPE)));
+        RowType fullType = new RowType(Arrays.asList(new DataField(1, "nest", nestFull)));
+
+        // Physically written in reversed order: nest<b, a>.
+        RowType writeType = fullType.projectByPaths(Arrays.asList("nest.b", "nest.a"));
+        RowType writtenNest = (RowType) writeType.getFields().get(0).type();
+        assertThat(writtenNest.getFieldNames()).containsExactly("b", "a");
+
+        // 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);
+        assertThat(leafPaths).containsExactly("nest.b", "nest.a");
+
+        RowType reconstructed = fullType.projectByPaths(leafPaths);
+        RowType reconstructedNest = (RowType) reconstructed.getFields().get(0).type();
+        assertThat(reconstructedNest.getFieldNames()).isEqualTo(writtenNest.getFieldNames());
+    }
+
+    @Test
+    void leafPathsCollapsesToWholeFieldWhenOrderMatches() {
+        // 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 =
+                new RowType(
+                        Arrays.asList(
+                                new DataField(2, "a", new IntType()),
+                                new DataField(3, "b", VarCharType.STRING_TYPE)));
+        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");
+    }
+}