[core] Introduce a vector read planner for data evolution

paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java

@@ -278,14 +278,62 @@ private RecordReader<InternalRow> createReader(IndexedSplit indexedSplit) throws
                 createReader(dataSplit, rowRanges, info.actualReadType), info);
     }
 
-    private DataEvolutionFileReader createUnionReader(
+    private RecordReader<InternalRow> createUnionReader(
             List<DataFileMeta> needMergeFiles,
             BinaryRow partition,
             DataFilePathFactory dataFilePathFactory,
             List<Range> rowRanges,
             RowType readRowType,
             @Nullable DeletionVectorWithRange deletionVector)
             throws IOException {
+        List<DataEvolutionVectorReadPlanner.ReadRange> vectorRanges =
+                DataEvolutionVectorReadPlanner.plan(
+                        needMergeFiles,
+                        readRowType,
+                        file ->
+                                schemaFetcher
+                                        .apply(file.schemaId())
+                                        .dataFileSchema(file.writeCols())
+                                        .logicalRowType());
+        if (vectorRanges != null) {
+            List<FieldBunch> nonVectorBunches =
+                    splitFieldBunches(
+                            needMergeFiles.stream()
+                                    .filter(file -> !isVectorStoreFile(file.fileName()))
+                                    .collect(Collectors.toList()),
+                            file -> schemaFetcher.apply(file.schemaId()).logicalRowType(),
+                            rowRanges != null);
+            List<Range> selectedRanges = Range.sortAndMergeOverlap(rowRanges, true);
+            List<ReaderSupplier<InternalRow>> suppliers = new ArrayList<>();
+            for (DataEvolutionVectorReadPlanner.ReadRange vectorRange : vectorRanges) {
+                List<Range> ranges = Collections.singletonList(vectorRange.range);
+                if (rowRanges != null) {
+                    ranges = Range.and(ranges, selectedRanges);
+                }
+                if (ranges.isEmpty()) {
+                    continue;
+                }
+                // Union readers have fixed field offsets. Apply the same selection to every
+                // column reader so their rows remain aligned when a vector provider changes.
+                List<Range> readRanges = ranges;
+                List<FieldBunch> bunches = new ArrayList<>(nonVectorBunches);
+                // Providers are newest-first within this range. Keep fields from the same
+                // physical file together so the column planner opens that file only once.
+                vectorRange.files.forEach(file -> bunches.add(new DataBunch(file)));
+                suppliers.add(
+                        () ->
+                                createUnionReader(
+                                        bunches,
+                                        needMergeFiles,
+                                        partition,
+                                        dataFilePathFactory,
+                                        readRanges,
+                                        readRowType,
+                                        deletionVector));
+            }
+            return ConcatRecordReader.create(suppliers);
+        }
+
         List<FieldBunch> fieldsFiles =
                 splitFieldBunches(
                         needMergeFiles,
@@ -298,6 +346,26 @@ private DataEvolutionFileReader createUnionReader(
                         },
                         rowRanges != null);
 
+        return createUnionReader(
+                fieldsFiles,
+                needMergeFiles,
+                partition,
+                dataFilePathFactory,
+                rowRanges,
+                readRowType,
+                deletionVector);
+    }
+
+    private RecordReader<InternalRow> createUnionReader(
+            List<FieldBunch> fieldsFiles,
+            List<DataFileMeta> needMergeFiles,
+            BinaryRow partition,
+            DataFilePathFactory dataFilePathFactory,
+            List<Range> rowRanges,
+            RowType readRowType,
+            @Nullable DeletionVectorWithRange deletionVector)
+            throws IOException {
+
         long rowCount = fieldsFiles.get(0).rowCount();
         long firstRowId = bunchFirstRowId(fieldsFiles.get(0));
 
@@ -331,6 +399,21 @@ private DataEvolutionFileReader createUnionReader(
         DataEvolutionReadPlanner.DataEvolutionReadPlan plan =
                 new DataEvolutionReadPlanner(readRowType, bunchAvailTypes, nestedFieldEnabled)
                         .plan();
+        if (plan.bunchReadFields.stream().allMatch(List::isEmpty)) {
+            // For example, a newly added vector column may cover only part of the normal file's
+            // row range. Projecting only that column leaves no fields to merge in uncovered ranges,
+            // but the selected rows must still be emitted.
+            // Read one bunch and let schema evolution fill the missing fields with NULL.
+            return createMissingFieldsReader(
+                    partition,
+                    fieldsFiles.get(0),
+                    bunchDataSchemas[0],
+                    dataFilePathFactory,
+                    formatBuilder,
+                    rowRanges,
+                    readRowType,
+                    deletionVector);
+        }
 
         // Build the per-bunch readers from the planned partial read row types.
         for (int i = 0; i < numBunches; i++) {
@@ -379,6 +462,35 @@ private DataEvolutionFileReader createUnionReader(
                         plan.rowOffsets, plan.fieldOffsets, fileRecordReaders);
     }
 
+    private RecordReader<InternalRow> createMissingFieldsReader(
+            BinaryRow partition,
+            FieldBunch bunch,
+            TableSchema dataSchema,
+            DataFilePathFactory dataFilePathFactory,
+            Builder formatBuilder,
+            List<Range> rowRanges,
+            RowType readRowType,
+            @Nullable DeletionVectorWithRange deletionVector)
+            throws IOException {
+        DataFileMeta firstFile = bunch.files().get(0);
+        // Use the physical schema: the full table schema may declare columns this file never wrote.
+        FormatReaderMapping mapping =
+                formatBuilder.build(
+                        readTarget(firstFile, dataFilePathFactory, rowRanges).formatIdentifier,
+                        schema,
+                        dataSchema,
+                        readRowType.getFields(),
+                        false);
+        return createFieldBunchReader(
+                partition,
+                bunch,
+                dataFilePathFactory,
+                mapping,
+                rowRanges,
+                readRowType,
+                deletionVector);
+    }
+
     private boolean nestedFieldEnabledFor(List<DataFileMeta> files) {
         if (coreOptions.dataEvolutionNestedFieldEnabled()) {
             return true;

paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionVectorReadPlanner.java

@@ -0,0 +1,183 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.operation.DataEvolutionSplitRead.VectorStoreBunchKey;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
+import static org.apache.paimon.types.VectorType.isVectorStoreFile;
+
+/** Plans ranges with fixed vector field providers before positional column merging. */
+class DataEvolutionVectorReadPlanner {
+
+    /** Returns null when the existing sequential column-group readers suffice. */
+    @Nullable
+    static List<ReadRange> plan(
+            List<DataFileMeta> files,
+            RowType readType,
+            Function<DataFileMeta, RowType> fileToRowType) {
+        Set<Integer> readIds =
+                readType.getFields().stream().map(DataField::id).collect(Collectors.toSet());
+        boolean vectorOnly = files.stream().allMatch(file -> isVectorStoreFile(file.fileName()));
+        Map<Integer, VectorStoreBunchKey> fieldGroups = new HashMap<>();
+        Set<Integer> vectorReadIds = new HashSet<>();
+        List<Candidate> candidates = new ArrayList<>();
+        boolean overlappingGroups = false;
+        Range logicalRange = null;
+        long firstRowId = Long.MAX_VALUE;
+        long lastRowId = Long.MIN_VALUE;
+        for (DataFileMeta file : files) {
+            Range range = file.nonNullRowIdRange();
+            firstRowId = Math.min(firstRowId, range.from);
+            lastRowId = Math.max(lastRowId, range.to);
+            if (!isVectorStoreFile(file.fileName())) {
+                if (!isBlobFile(file.fileName())) {
+                    logicalRange = range;
+                }
+                continue;
+            }
+            RowType rowType = fileToRowType.apply(file);
+            VectorStoreBunchKey key =
+                    new VectorStoreBunchKey(
+                            file.schemaId(),
+                            DataFilePathFactory.formatIdentifier(file.fileName()),
+                            file.writeCols(),
+                            rowType);
+            Set<Integer> fieldIds = new HashSet<>();
+            for (DataField field : rowType.getFields()) {
+                // Match historical fields by id: a rename must not create a different provider.
+                VectorStoreBunchKey previous = fieldGroups.putIfAbsent(field.id(), key);
+                overlappingGroups |= previous != null && !previous.equals(key);
+                if (readIds.contains(field.id())) {
+                    fieldIds.add(field.id());
+                    vectorReadIds.add(field.id());
+                }
+            }
+            if (!fieldIds.isEmpty() || vectorOnly) {
+                candidates.add(new Candidate(file, fieldIds));
+            }
+        }
+
+        // Keep the sequential path for disjoint column groups, including rolled vector files.
+        if (!overlappingGroups) {
+            return null;
+        }
+        if (logicalRange == null) {
+            logicalRange = new Range(firstRowId, lastRowId);
+        }
+
+        // Resolve the original files before VectorFileBunch can discard older overlapping files.
+        // A bunch may contain different sequences in adjacent ranges, so sorting whole bunches
+        // by their maximum sequence cannot establish the latest provider for every row.
+        TreeMap<Long, List<Candidate>> boundaries = new TreeMap<>();
+        boundaries.put(logicalRange.from, new ArrayList<>());
+        boundaries.put(logicalRange.to + 1, new ArrayList<>());
+        // Candidate ranges lie within the normal anchor, or the enclosing range computed above.
+        for (Candidate candidate : candidates) {
+            Range range = candidate.file.nonNullRowIdRange();
+            boundaries.computeIfAbsent(range.from, ignored -> new ArrayList<>()).add(candidate);
+            boundaries.computeIfAbsent(range.to + 1, ignored -> new ArrayList<>()).add(candidate);
+        }
+
+        TreeSet<Candidate> active =
+                new TreeSet<>(
+                        Comparator.<Candidate>comparingLong(c -> c.file.maxSequenceNumber())
+                                .reversed()
+                                .thenComparing(c -> c.file.fileName()));
+        List<ReadRange> result = new ArrayList<>();
+        long start = logicalRange.from;
+        for (Map.Entry<Long, List<Candidate>> boundary : boundaries.entrySet()) {
+            long end = boundary.getKey();
+            if (start < end) {
+                Set<Integer> assigned = new HashSet<>();
+                List<DataFileMeta> providers = new ArrayList<>();
+                for (Candidate candidate : active) {
+                    // Presence in writeCols is what matters. An explicit NULL in the newest
+                    // file must overwrite the old value, just like any other partial update.
+                    if (assigned.addAll(candidate.fieldIds)) {
+                        providers.add(candidate.file);
+                    }
+                    if (assigned.size() == vectorReadIds.size()) {
+                        break;
+                    }
+                }
+                if (providers.isEmpty() && vectorOnly && !active.isEmpty()) {
+                    // Column pruning can remove the normal anchor. Retain a row-count provider
+                    // for ranges where the projected vector has not been populated yet.
+                    providers.add(active.first().file);
+                }
+                ReadRange previous = result.isEmpty() ? null : result.get(result.size() - 1);
+                if (previous != null && previous.files.equals(providers)) {
+                    previous.range = new Range(previous.range.from, end - 1);
+                } else {
+                    result.add(new ReadRange(new Range(start, end - 1), providers));
+                }
+            }
+            // Each file enters at its first row id and leaves just after its last row id.
+            for (Candidate candidate : boundary.getValue()) {
+                if (!active.remove(candidate)) {
+                    active.add(candidate);
+                }
+            }
+            start = end;
+        }
+        return result;
+    }
+
+    static class ReadRange {
+
+        Range range;
+        final List<DataFileMeta> files;
+
+        private ReadRange(Range range, List<DataFileMeta> files) {
+            this.range = range;
+            this.files = files;
+        }
+    }
+
+    private static class Candidate {
+
+        final DataFileMeta file;
+        final Set<Integer> fieldIds;
+
+        private Candidate(DataFileMeta file, Set<Integer> fieldIds) {
+            this.file = file;
+            this.fieldIds = fieldIds;
+        }
+    }
+}