This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e5b1a676e268f84d3f93e42b90a18b8fc7e19733 Author: YangJie <[email protected]> AuthorDate: Thu Sep 17 01:20:29 2026 -0400 [core] Use full-range anchor as row representative in DE read-type pruning (#9899) --- .../operation/DataEvolutionFileStoreScan.java | 45 +++++++++--- .../apache/paimon/utils/DataEvolutionUtils.java | 2 +- .../operation/DataEvolutionFileStoreScanTest.java | 79 +++++++++++++++++++++- 3 files changed, 113 insertions(+), 13 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java index 3523b1a62e..c8ad122f5d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionFileStoreScan.java @@ -231,25 +231,46 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan { 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 @@ public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan { 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) { diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java index 329e35067b..a4e702184b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/DataEvolutionUtils.java @@ -327,7 +327,7 @@ public class DataEvolutionUtils { checkState( anchor != null, - "Data-evolution deletion vectors should have a normal anchor file in each row range group."); + "A data-evolution row range group must contain a normal (non-blob, non-vector-store) anchor file."); return anchor; } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java index 1d2c9b654c..a891ff683a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/DataEvolutionFileStoreScanTest.java @@ -38,6 +38,7 @@ import org.apache.paimon.schema.Schema; 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.concurrent.atomic.AtomicInteger; 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 class DataEvolutionFileStoreScanTest { 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 @@ public class DataEvolutionFileStoreScanTest { 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 @@ public class DataEvolutionFileStoreScanTest { 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);
