[core] Use full-range anchor as row representative in DE read-type pruning

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

@@ -231,25 +231,46 @@ private List<ManifestEntry> pruneByReadType(List<ManifestEntry> group) {
         if (readType == null || group.size() <= 1) {
             return group;
         }
-        ManifestEntry anchor =
-                deletionVectorsEnabled ? retrieveAnchorFile(group, ManifestEntry::file) : null;
-        Set<Integer> readFieldIds = new HashSet<>();
-        for (DataField f : readType.getFields()) {
-            readFieldIds.add(f.id());
-        }
+        Set<Integer> filterFieldIds = Collections.emptySet();
         if (inputFilter != null) {
             // executeFilter may need columns absent from the output projection. Keep their latest
             // files too, otherwise widening the reader could see an older value or a null.
+            filterFieldIds = new HashSet<>();
             Set<String> filterFields = PredicateVisitor.collectFieldNames(inputFilter);
             for (DataField field : schema.fields()) {
                 if (filterFields.contains(field.name())) {
-                    readFieldIds.add(field.id());
+                    filterFieldIds.add(field.id());
                 }
             }
         }
+        return pruneByReadType(
+                group,
+                readType,
+                filterFieldIds,
+                deletionVectorsEnabled,
+                this::fileFieldIdsForEntry);
+    }
+
+    @VisibleForTesting
+    static List<ManifestEntry> pruneByReadType(
+            List<ManifestEntry> group,
+            RowType readType,
+            Set<Integer> filterFieldIds,
+            boolean deletionVectorsEnabled,
+            Function<ManifestEntry, Set<Integer>> fileFieldIds) {
+        ManifestEntry anchor =
+                deletionVectorsEnabled ? retrieveAnchorFile(group, ManifestEntry::file) : null;
+        Set<Integer> readFieldIds = new HashSet<>();
+        for (DataField f : readType.getFields()) {
+            readFieldIds.add(f.id());
+        }
+        // The caller (which has the schema) folds in any fields referenced only by the filter, so
+        // their latest files are kept too; otherwise widening the reader could see an older value
+        // or a null.
+        readFieldIds.addAll(filterFieldIds);
         List<ManifestEntry> kept = new ArrayList<>(group.size());
         for (ManifestEntry entry : group) {
-            Set<Integer> fileIds = fileFieldIdsForEntry(entry);
+            Set<Integer> fileIds = fileFieldIds.apply(entry);
             for (int id : readFieldIds) {
                 if (fileIds.contains(id)) {
                     kept.add(entry);
@@ -261,8 +282,12 @@ private List<ManifestEntry> pruneByReadType(List<ManifestEntry> group) {
             kept.add(anchor);
         }
         // Group must contribute at least one file so the reader sees rowCount and can NULL-fill
-        // missing columns for the projection's rows.
-        return kept.isEmpty() ? Collections.singletonList(group.get(0)) : kept;
+        // missing columns for the projection's rows. The representative must be a full-range
+        // normal file: a blob or vector-store file covers only a sub-range of the group's row
+        // ids, so the split would silently emit fewer rows than the group contains.
+        return kept.isEmpty()
+                ? Collections.singletonList(retrieveAnchorFile(group, ManifestEntry::file))
+                : kept;
     }
 
     private Set<Integer> fileFieldIdsForEntry(ManifestEntry entry) {

paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java

@@ -38,6 +38,7 @@
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.stats.SimpleStats;
 import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RowRangeIndex;
 
@@ -53,6 +54,7 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link DataEvolutionFileStoreScan}. */
@@ -67,6 +69,66 @@ public void setUp() {
         scanTableSchema = schemas::get;
     }
 
+    @Test
+    public void testReadTypePruningKeepsAnchorAsRowRepresentative() {
+        // query reads a freshly added column no file in the group writes: the group must
+        // still contribute one row-count representative, and that representative is the
+        // oldest full-range normal file — not group.get(0), which can be a blob file
+        // covering only part of the group's row ids
+        Schema schema = createSchema("v", "b");
+        TableSchema tableSchema = TableSchema.create(0L, schema);
+        schemas.put(0L, tableSchema);
+
+        // blob file first in the group, covering only row ids [0, 1]
+        ManifestEntry blob =
+                createManifestEntryWithDifferentColsAndFileName(
+                        "data-blob-0.blob",
+                        0L,
+                        new String[] {"b"},
+                        new String[] {"b"},
+                        null,
+                        0L,
+                        0L,
+                        2L);
+        // normal files covering the whole group range [0, 9]; write only "v", not "c"
+        ManifestEntry newer =
+                createManifestEntryWithDifferentColsAndFileName(
+                        "data-newer.parquet",
+                        0L,
+                        new String[] {"v"},
+                        new String[] {"v"},
+                        null,
+                        5L,
+                        0L,
+                        10L);
+        ManifestEntry older =
+                createManifestEntryWithDifferentColsAndFileName(
+                        "data-older.parquet",
+                        0L,
+                        new String[] {"v"},
+                        new String[] {"v"},
+                        null,
+                        1L,
+                        0L,
+                        10L);
+
+        RowType readType =
+                DataTypes.ROW(
+                        DataTypes.FIELD(tableSchema.highestFieldId() + 1, "c", DataTypes.INT()));
+
+        List<ManifestEntry> pruned =
+                DataEvolutionFileStoreScan.pruneByReadType(
+                        Arrays.asList(blob, newer, older),
+                        readType,
+                        Collections.emptySet(),
+                        false,
+                        entry -> fileFieldIds(schemas.get(entry.file().schemaId()), entry.file()));
+
+        assertThat(pruned)
+                .extracting(e -> e.file().fileName())
+                .containsExactly("data-older.parquet");
+    }
+
     @Test
     public void testEvolutionStatsSingleFile() {
         Schema schema = createSchema("f0", "f1");
@@ -745,11 +807,24 @@ private ManifestEntry createManifestEntryWithDifferentColsAndFileName(
             String[] valueStatsCols,
             SimpleStats stats,
             long sequence) {
+        return createManifestEntryWithDifferentColsAndFileName(
+                fileName, schemaId, writeCols, valueStatsCols, stats, sequence, 0L, 100L);
+    }
+
+    private ManifestEntry createManifestEntryWithDifferentColsAndFileName(
+            String fileName,
+            Long schemaId,
+            String[] writeCols,
+            String[] valueStatsCols,
+            SimpleStats stats,
+            long sequence,
+            long firstRowId,
+            long rowCount) {
         DataFileMeta fileMeta =
                 DataFileMeta.create(
                         fileName,
                         100L,
-                        100L,
+                        rowCount,
                         createBinaryRow(1),
                         createBinaryRow(100),
                         stats,
@@ -764,7 +839,7 @@ private ManifestEntry createManifestEntryWithDifferentColsAndFileName(
                         FileSource.APPEND,
                         Arrays.stream(valueStatsCols).collect(Collectors.toList()),
                         null,
-                        0L,
+                        firstRowId,
                         Arrays.stream(writeCols).collect(Collectors.toList()));
 
         return ManifestEntry.create(FileKind.ADD, createBinaryRow(0), 0, 0, fileMeta);