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 3523b1a62e63..c8ad122f5d7b 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 @@ private List pruneByReadType(List group) { if (readType == null || group.size() <= 1) { return group; } - ManifestEntry anchor = - deletionVectorsEnabled ? retrieveAnchorFile(group, ManifestEntry::file) : null; - Set readFieldIds = new HashSet<>(); - for (DataField f : readType.getFields()) { - readFieldIds.add(f.id()); - } + Set 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 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 pruneByReadType( + List group, + RowType readType, + Set filterFieldIds, + boolean deletionVectorsEnabled, + Function> fileFieldIds) { + ManifestEntry anchor = + deletionVectorsEnabled ? retrieveAnchorFile(group, ManifestEntry::file) : null; + Set 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 kept = new ArrayList<>(group.size()); for (ManifestEntry entry : group) { - Set fileIds = fileFieldIdsForEntry(entry); + Set fileIds = fileFieldIds.apply(entry); for (int id : readFieldIds) { if (fileIds.contains(id)) { kept.add(entry); @@ -261,8 +282,12 @@ private List pruneByReadType(List 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 fileFieldIdsForEntry(ManifestEntry entry) { 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 1d2c9b654c48..a891ff683ac0 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.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 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);