This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 41e7738f54 [core] Use full-range anchor as row representative in DE
read-type pruning (#9899)
41e7738f54 is described below
commit 41e7738f543d41aa3f09588815538ea1bfea28c3
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);
