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 4cf5e17f1d [core][python] Prune data evolution groups by merged stats
(#9114)
4cf5e17f1d is described below
commit 4cf5e17f1d39adb9d26618477d91a47fb1da9cda
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Aug 9 17:00:50 2026 +0800
[core][python] Prune data evolution groups by merged stats (#9114)
---
.../operation/DataEvolutionFileStoreScan.java | 141 ++++---
.../operation/DataEvolutionFileStoreScanTest.java | 235 ++++++++++--
.../read/scanner/data_evolution_split_generator.py | 13 +-
.../pypaimon/read/scanner/data_evolution_stats.py | 259 +++++++++++++
.../pypaimon/read/scanner/file_scanner.py | 34 +-
.../tests/data_evolution_group_stats_test.py | 420 +++++++++++++++++++++
paimon-python/pypaimon/tests/global_index_test.py | 82 ++++
7 files changed, 1104 insertions(+), 80 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 5ae7eb07c4..88053b03e3 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
@@ -47,7 +47,6 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -56,7 +55,6 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
-import java.util.function.ToLongFunction;
import java.util.stream.Collectors;
import static org.apache.paimon.format.blob.BlobFileFormat.isBlobFile;
@@ -64,6 +62,8 @@ import static
org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId;
import static org.apache.paimon.types.VectorType.isVectorStoreFile;
import static org.apache.paimon.utils.DataEvolutionUtils.fileFieldIds;
import static org.apache.paimon.utils.DataEvolutionUtils.retrieveAnchorFile;
+import static org.apache.paimon.utils.InternalRowUtils.compare;
+import static org.apache.paimon.utils.InternalRowUtils.get;
/** {@link FileStoreScan} for data-evolution enabled table. */
public class DataEvolutionFileStoreScan extends AppendOnlyFileStoreScan {
@@ -266,6 +266,21 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
Function<Long, TableSchema> scanTableSchema,
List<ManifestEntry> metas,
EvolutionStatsCache evolutionStatsCache) {
+ long groupStart =
+ metas.stream()
+ .map(ManifestEntry::file)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .mapToLong(range -> range.from)
+ .min()
+ .orElseThrow(() -> new IllegalArgumentException("Empty
evolution group."));
+ long groupEnd =
+ metas.stream()
+ .map(ManifestEntry::file)
+ .map(DataFileMeta::nonNullRowIdRange)
+ .mapToLong(range -> range.to)
+ .max()
+ .orElseThrow(() -> new IllegalArgumentException("Empty
evolution group."));
+ long groupRowCount = groupEnd - groupStart + 1;
Set<Integer> excludedFileFieldIds =
metas.stream()
.filter(
@@ -279,87 +294,91 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
.map(DataField::id))
.collect(Collectors.toSet());
// exclude blob and vector-store files, useless for predicate eval
- metas =
+ List<ManifestEntry> normalMetas =
metas.stream()
.filter(entry -> !isBlobFile(entry.file().fileName()))
.filter(entry ->
!isVectorStoreFile(entry.file().fileName()))
.collect(Collectors.toList());
- ToLongFunction<ManifestEntry> maxSeqFunc = e ->
e.file().maxSequenceNumber();
- metas.sort(Comparator.comparingLong(maxSeqFunc).reversed());
-
int[] allFields =
schema.fields().stream().mapToInt(DataField::id).toArray();
DataType[] targetTypes =
schema.fields().stream().map(DataField::type).toArray(DataType[]::new);
int fieldsCount = schema.fields().size();
int[] rowOffsets = new int[fieldsCount];
int[] fieldOffsets = new int[fieldsCount];
+ long[] latestSequences = new long[fieldsCount];
+ boolean[] tiedLatestProviders = new boolean[fieldsCount];
Arrays.fill(rowOffsets, -1);
Arrays.fill(fieldOffsets, -1);
- Set<Integer> typeMismatchedFieldIds = new HashSet<>();
+ Arrays.fill(latestSequences, Long.MIN_VALUE);
- InternalRow[] min = new InternalRow[metas.size()];
- InternalRow[] max = new InternalRow[metas.size()];
- BinaryArray[] nullCounts = new BinaryArray[metas.size()];
+ InternalRow[] min = new InternalRow[normalMetas.size()];
+ InternalRow[] max = new InternalRow[normalMetas.size()];
+ BinaryArray[] nullCounts = new BinaryArray[normalMetas.size()];
+ EvolutionStatsCache.ProjectedFileSchema[] projectedSchemas =
+ new
EvolutionStatsCache.ProjectedFileSchema[normalMetas.size()];
- for (int i = 0; i < metas.size(); i++) {
- SimpleStats stats = metas.get(i).file().valueStats();
+ for (int i = 0; i < normalMetas.size(); i++) {
+ DataFileMeta file = normalMetas.get(i).file();
+ SimpleStats stats = file.valueStats();
min[i] = stats.minValues();
max[i] = stats.maxValues();
nullCounts[i] = stats.nullCounts();
- }
-
- int unresolvedFields = fieldsCount;
- for (int i = 0; i < metas.size(); i++) {
- DataFileMeta fileMeta = metas.get(i).file();
- EvolutionStatsCache.ProjectedFileSchema projectedFileSchema =
- evolutionStatsCache.get(scanTableSchema, fileMeta);
-
+ EvolutionStatsCache.ProjectedFileSchema projected =
+ evolutionStatsCache.get(scanTableSchema, file);
+ projectedSchemas[i] = projected;
for (int j = 0; j < fieldsCount; j++) {
- if (rowOffsets[j] != -1) {
+ if (projected.fieldStats(allFields[j]) == null) {
continue;
}
- int targetFieldId = allFields[j];
- EvolutionStatsCache.FileFieldStats fileFieldStats =
- projectedFileSchema.fieldStats(targetFieldId);
- if (fileFieldStats == null) {
- continue;
- }
- if (!fileFieldStats.hasStats()) {
- rowOffsets[j] = -2;
- unresolvedFields--;
- continue;
- }
- DataType fileType = fileFieldStats.type();
- if (!fileType.equalsIgnoreFieldId(targetTypes[j])) {
- typeMismatchedFieldIds.add(targetFieldId);
- continue;
+ long sequence = file.maxSequenceNumber();
+ if (sequence > latestSequences[j]) {
+ latestSequences[j] = sequence;
+ rowOffsets[j] = i;
+ tiedLatestProviders[j] = false;
+ } else if (sequence == latestSequences[j]) {
+ tiedLatestProviders[j] = true;
}
- rowOffsets[j] = i;
- fieldOffsets[j] = fileFieldStats.index();
- unresolvedFields--;
- }
- if (unresolvedFields == 0) {
- break;
}
}
- long groupRowCount = metas.get(0).file().rowCount();
for (int j = 0; j < fieldsCount; j++) {
- if (rowOffsets[j] == -1
- && (excludedFileFieldIds.contains(allFields[j])
- || typeMismatchedFieldIds.contains(allFields[j])))
{
+ if (rowOffsets[j] == -1) {
+ if (excludedFileFieldIds.contains(allFields[j])) {
+ rowOffsets[j] = -2;
+ }
+ continue;
+ }
+ int provider = rowOffsets[j];
+ DataFileMeta file = normalMetas.get(provider).file();
+ EvolutionStatsCache.FileFieldStats fileStats =
+ projectedSchemas[provider].fieldStats(allFields[j]);
+ Range fileRange = file.nonNullRowIdRange();
+ if (tiedLatestProviders[j]
+ || !fileStats.hasStats()
+ || !fileStats.type().equalsIgnoreFieldId(targetTypes[j])
+ || fileRange.from != groupStart
+ || fileRange.to != groupEnd) {
+ rowOffsets[j] = -2;
+ continue;
+ }
+ int fieldOffset = fileStats.index();
+ if (!isValidStats(file.valueStats(), fieldOffset, targetTypes[j],
groupRowCount)) {
rowOffsets[j] = -2;
+ continue;
}
+ fieldOffsets[j] = fieldOffset;
}
- DataEvolutionRow finalMin = new DataEvolutionRow(metas.size(),
rowOffsets, fieldOffsets);
- DataEvolutionRow finalMax = new DataEvolutionRow(metas.size(),
rowOffsets, fieldOffsets);
+ DataEvolutionRow finalMin =
+ new DataEvolutionRow(normalMetas.size(), rowOffsets,
fieldOffsets);
+ DataEvolutionRow finalMax =
+ new DataEvolutionRow(normalMetas.size(), rowOffsets,
fieldOffsets);
// For null-count specifically, a field absent from every file in the
group means every
// logical row is null for that field — encode as groupRowCount so
stats predicates can
// prune non-null comparisons (e.g. `extra2 = 'x'`) instead of falling
back to
// "unknown stats -> keep" in LeafPredicate.test.
DataEvolutionArray finalNullCounts =
- new DataEvolutionArray(metas.size(), rowOffsets, fieldOffsets,
groupRowCount);
+ new DataEvolutionArray(normalMetas.size(), rowOffsets,
fieldOffsets, groupRowCount);
finalMin.setRows(min);
finalMax.setRows(max);
@@ -367,6 +386,30 @@ public class DataEvolutionFileStoreScan extends
AppendOnlyFileStoreScan {
return new EvolutionStats(groupRowCount, finalMin, finalMax,
finalNullCounts);
}
+ private static boolean isValidStats(
+ SimpleStats stats, int fieldOffset, DataType type, long rowCount) {
+ try {
+ Object min = get(stats.minValues(), fieldOffset, type);
+ Object max = get(stats.maxValues(), fieldOffset, type);
+ BinaryArray nullCounts = stats.nullCounts();
+ Long nullCount =
+ nullCounts.isNullAt(fieldOffset) ? null :
nullCounts.getLong(fieldOffset);
+ if (nullCount != null && (nullCount < 0 || nullCount > rowCount)) {
+ return false;
+ }
+ if ((min == null) != (max == null)) {
+ return false;
+ }
+ if (min == null) {
+ return true;
+ }
+ return (nullCount == null || nullCount != rowCount)
+ && compare(min, max, type.getTypeRoot()) <= 0;
+ } catch (RuntimeException e) {
+ return false;
+ }
+ }
+
/** Note: Keep this thread-safe. */
@Override
protected boolean filterByStats(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 647e6a2c8f..1d2c9b654c 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
@@ -107,8 +107,8 @@ public class DataEvolutionFileStoreScanTest {
assertThat(minRow.getString(1).toString()).isEqualTo("a");
assertThat(maxRow.getString(1).toString()).isEqualTo("z");
- assertThat(nullCounts.getInt(0)).isEqualTo(0);
- assertThat(nullCounts.getInt(1)).isEqualTo(1);
+ assertThat(nullCounts.getLong(0)).isEqualTo(0L);
+ assertThat(nullCounts.getLong(1)).isEqualTo(1L);
assertThat(minRow.getFieldCount()).isEqualTo(2);
assertThat(maxRow.getFieldCount()).isEqualTo(2);
@@ -218,7 +218,11 @@ public class DataEvolutionFileStoreScanTest {
GenericRow.of(2, 20),
GenericRow.of(4, 40),
createBinaryArray(new int[] {1, 2}),
- new int[] {0, 2}));
+ new int[] {0, 2}),
+ "newer.parquet",
+ 1L,
+ 0L,
+ 100L);
List<ManifestEntry> entries = Arrays.asList(entry2, entry1);
@@ -237,9 +241,9 @@ public class DataEvolutionFileStoreScanTest {
assertThat(maxRow.getInt(2)).isEqualTo(40);
assertThat(minRow.getString(1).toString()).isEqualTo("a");
assertThat(maxRow.getString(1).toString()).isEqualTo("c");
- assertThat(nullCounts.getInt(0)).isEqualTo(1);
- assertThat(nullCounts.getInt(1)).isEqualTo(1);
- assertThat(nullCounts.getInt(2)).isEqualTo(2);
+ assertThat(nullCounts.getLong(0)).isEqualTo(1L);
+ assertThat(nullCounts.getLong(1)).isEqualTo(1L);
+ assertThat(nullCounts.getLong(2)).isEqualTo(2L);
}
@Test
@@ -259,7 +263,11 @@ public class DataEvolutionFileStoreScanTest {
GenericRow.of(1, BinaryString.fromString("a")),
GenericRow.of(3, BinaryString.fromString("c")),
createBinaryArray(new int[] {0, 1}),
- new int[] {0, 1}));
+ new int[] {0, 1}),
+ "base-newer.parquet",
+ 1L,
+ 0L,
+ 100L);
ManifestEntry entry2 =
createManifestEntry(
@@ -290,9 +298,9 @@ public class DataEvolutionFileStoreScanTest {
assertThat(minRow.getInt(2)).isEqualTo(20);
assertThat(maxRow.getInt(2)).isEqualTo(40);
- assertThat(nullCounts.getInt(0)).isEqualTo(0);
- assertThat(nullCounts.getInt(1)).isEqualTo(1);
- assertThat(nullCounts.getInt(2)).isEqualTo(1);
+ assertThat(nullCounts.getLong(0)).isEqualTo(0L);
+ assertThat(nullCounts.getLong(1)).isEqualTo(1L);
+ assertThat(nullCounts.getLong(2)).isEqualTo(1L);
}
@Test
@@ -311,7 +319,8 @@ public class DataEvolutionFileStoreScanTest {
GenericRow.of(1, BinaryString.fromString("a")),
GenericRow.of(3, BinaryString.fromString("c")),
createBinaryArray(new int[] {0, 1}),
- new int[] {0, 1}));
+ new int[] {0, 1}),
+ 1L);
ManifestEntry entry2 =
createManifestEntryWithDifferentCols(
@@ -344,8 +353,8 @@ public class DataEvolutionFileStoreScanTest {
assertThat(minRow.isNullAt(2)).isTrue();
assertThat(maxRow.isNullAt(2)).isTrue();
- assertThat(nullCounts.getInt(0)).isEqualTo(0);
- assertThat(nullCounts.getInt(1)).isEqualTo(1);
+ assertThat(nullCounts.getLong(0)).isEqualTo(0L);
+ assertThat(nullCounts.getLong(1)).isEqualTo(1L);
assertThat(nullCounts.isNullAt(2)).isTrue();
}
@@ -384,7 +393,8 @@ public class DataEvolutionFileStoreScanTest {
newTypeMaxWriter.complete();
SimpleStats newTypeStats =
new SimpleStats(newTypeMin, newTypeMax, createBinaryArray(new
int[] {0, 0}));
- ManifestEntry newTypeEntry = createManifestEntry(1L, newTypeStats);
+ ManifestEntry newTypeEntry =
+ createManifestEntry(1L, newTypeStats, "new-type.parquet", 1L,
0L, 100L);
EvolutionStats result =
DataEvolutionFileStoreScan.evolutionStats(
@@ -494,6 +504,151 @@ public class DataEvolutionFileStoreScanTest {
.isTrue();
}
+ @Test
+ public void testNewestIncompatibleProviderIsUnknown() {
+ Schema oldSchema = createSchema("f0");
+ schemas.put(0L, TableSchema.create(0L, oldSchema));
+ TableSchema currentSchema =
+ TableSchema.create(
+ 1L, Schema.newBuilder().column("f0",
DataTypes.BIGINT()).build());
+ schemas.put(1L, currentSchema);
+
+ ManifestEntry compatible =
+ createManifestEntry(
+ 1L,
+ createSimpleStats(
+ GenericRow.of(10),
+ GenericRow.of(20),
+ createBinaryArray(new int[] {0}),
+ new int[] {0}),
+ "compatible.parquet",
+ 1L,
+ 0L,
+ 10L);
+ ManifestEntry newerIncompatible =
+ createManifestEntry(
+ 0L,
+ createSimpleStats(
+ GenericRow.of(100),
+ GenericRow.of(200),
+ createBinaryArray(new int[] {0}),
+ new int[] {0}),
+ "incompatible.parquet",
+ 2L,
+ 0L,
+ 10L);
+
+ EvolutionStats result =
+ DataEvolutionFileStoreScan.evolutionStats(
+ currentSchema,
+ scanTableSchema,
+ Arrays.asList(compatible, newerIncompatible),
+ new EvolutionStatsCache());
+
+ assertThat(result.minValues().isNullAt(0)).isTrue();
+ assertThat(result.nullCounts().isNullAt(0)).isTrue();
+ }
+
+ @Test
+ public void testPartialLatestProviderIsUnknown() {
+ Schema schema = createSchema("f0");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+ SimpleStats baseStats =
+ createSimpleStats(
+ GenericRow.of(0),
+ GenericRow.of(9),
+ createBinaryArray(new int[] {0}),
+ new int[] {0});
+ SimpleStats partialStats =
+ createSimpleStats(
+ GenericRow.of(100),
+ GenericRow.of(104),
+ createBinaryArray(new int[] {0}),
+ new int[] {0});
+
+ EvolutionStats result =
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema,
+ scanTableSchema,
+ Arrays.asList(
+ createManifestEntry(0L, baseStats,
"base.parquet", 0L, 0L, 10L),
+ createManifestEntry(
+ 0L, partialStats, "partial.parquet",
1L, 5L, 5L)),
+ new EvolutionStatsCache());
+
+ assertThat(result.minValues().isNullAt(0)).isTrue();
+ assertThat(result.nullCounts().isNullAt(0)).isTrue();
+ }
+
+ @Test
+ public void testTiedLatestProvidersAreUnknown() {
+ Schema schema = createSchema("f0");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+ SimpleStats stats =
+ createSimpleStats(
+ GenericRow.of(0),
+ GenericRow.of(9),
+ createBinaryArray(new int[] {0}),
+ new int[] {0});
+
+ EvolutionStats result =
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema,
+ scanTableSchema,
+ Arrays.asList(
+ createManifestEntry(0L, stats,
"first.parquet", 1L, 0L, 10L),
+ createManifestEntry(0L, stats,
"second.parquet", 1L, 0L, 10L)),
+ new EvolutionStatsCache());
+
+ assertThat(result.minValues().isNullAt(0)).isTrue();
+ assertThat(result.nullCounts().isNullAt(0)).isTrue();
+ }
+
+ @Test
+ public void testInvalidProviderStatsAreUnknown() {
+ Schema schema = createSchema("f0");
+ TableSchema tableSchema = TableSchema.create(0L, schema);
+ schemas.put(0L, tableSchema);
+
+ assertInvalidProviderStatsAreUnknown(tableSchema, 10, 10, -1);
+ assertInvalidProviderStatsAreUnknown(tableSchema, 10, 10, 11);
+ assertInvalidProviderStatsAreUnknown(tableSchema, null, 10, 0);
+ assertInvalidProviderStatsAreUnknown(tableSchema, 10, null, 0);
+ assertInvalidProviderStatsAreUnknown(tableSchema, 10, 1, 0);
+ assertInvalidProviderStatsAreUnknown(tableSchema, 10, 10, 10);
+ }
+
+ private void assertInvalidProviderStatsAreUnknown(
+ TableSchema tableSchema, Object min, Object max, int nullCount) {
+ SimpleStats stats =
+ createSimpleStats(
+ GenericRow.of(min),
+ GenericRow.of(max),
+ createBinaryArray(new int[] {nullCount}),
+ new int[] {0});
+ EvolutionStats result =
+ DataEvolutionFileStoreScan.evolutionStats(
+ tableSchema,
+ scanTableSchema,
+ Collections.singletonList(
+ createManifestEntry(0L, stats,
"invalid.parquet", 0L, 0L, 10L)),
+ new EvolutionStatsCache());
+
+ assertThat(result.minValues().isNullAt(0)).isTrue();
+ assertThat(result.maxValues().isNullAt(0)).isTrue();
+ assertThat(result.nullCounts().isNullAt(0)).isTrue();
+ Predicate predicate = new
PredicateBuilder(tableSchema.logicalRowType()).equal(0, 5);
+ assertThat(
+ predicate.test(
+ result.rowCount(),
+ result.minValues(),
+ result.maxValues(),
+ result.nullCounts()))
+ .isTrue();
+ }
+
@Test
public void testIntersectsRowRanges() {
List<Range> rowRanges =
@@ -523,17 +678,27 @@ public class DataEvolutionFileStoreScanTest {
}
private ManifestEntry createManifestEntry(Long schemaId, SimpleStats
stats) {
+ return createManifestEntry(schemaId, stats, "test-file.parquet", 0L,
0L, 100L);
+ }
+
+ private ManifestEntry createManifestEntry(
+ Long schemaId,
+ SimpleStats stats,
+ String fileName,
+ long sequence,
+ long firstRowId,
+ long rowCount) {
DataFileMeta fileMeta =
DataFileMeta.create(
- "test-file.parquet",
- 100L,
+ fileName,
100L,
+ rowCount,
createBinaryRow(1),
createBinaryRow(100),
stats,
stats,
- 0L,
- 0L,
+ sequence,
+ sequence,
schemaId,
0,
Collections.emptyList(),
@@ -542,7 +707,7 @@ public class DataEvolutionFileStoreScanTest {
FileSource.APPEND,
null,
null,
- null,
+ firstRowId,
null);
return ManifestEntry.create(FileKind.ADD, createBinaryRow(0), 0, 0,
fileMeta);
@@ -550,8 +715,17 @@ public class DataEvolutionFileStoreScanTest {
private ManifestEntry createManifestEntryWithDifferentCols(
Long schemaId, String[] writeCols, String[] valueStatsCols,
SimpleStats stats) {
+ return createManifestEntryWithDifferentCols(schemaId, writeCols,
valueStatsCols, stats, 0L);
+ }
+
+ private ManifestEntry createManifestEntryWithDifferentCols(
+ Long schemaId,
+ String[] writeCols,
+ String[] valueStatsCols,
+ SimpleStats stats,
+ long sequence) {
return createManifestEntryWithDifferentColsAndFileName(
- "test-file.parquet", schemaId, writeCols, valueStatsCols,
stats);
+ "test-file.parquet", schemaId, writeCols, valueStatsCols,
stats, sequence);
}
private ManifestEntry createManifestEntryWithDifferentColsAndFileName(
@@ -560,6 +734,17 @@ public class DataEvolutionFileStoreScanTest {
String[] writeCols,
String[] valueStatsCols,
SimpleStats stats) {
+ return createManifestEntryWithDifferentColsAndFileName(
+ fileName, schemaId, writeCols, valueStatsCols, stats, 0L);
+ }
+
+ private ManifestEntry createManifestEntryWithDifferentColsAndFileName(
+ String fileName,
+ Long schemaId,
+ String[] writeCols,
+ String[] valueStatsCols,
+ SimpleStats stats,
+ long sequence) {
DataFileMeta fileMeta =
DataFileMeta.create(
fileName,
@@ -569,8 +754,8 @@ public class DataEvolutionFileStoreScanTest {
createBinaryRow(100),
stats,
stats,
- 0L,
- 0L,
+ sequence,
+ sequence,
schemaId,
0,
Collections.emptyList(),
@@ -579,7 +764,7 @@ public class DataEvolutionFileStoreScanTest {
FileSource.APPEND,
Arrays.stream(valueStatsCols).collect(Collectors.toList()),
null,
- null,
+ 0L,
Arrays.stream(writeCols).collect(Collectors.toList()));
return ManifestEntry.create(FileKind.ADD, createBinaryRow(0), 0, 0,
fileMeta);
@@ -595,9 +780,9 @@ public class DataEvolutionFileStoreScanTest {
private BinaryArray createBinaryArray(int[] values) {
BinaryArray array = new BinaryArray();
- BinaryArrayWriter writer = new BinaryArrayWriter(array, values.length,
4);
+ BinaryArrayWriter writer = new BinaryArrayWriter(array, values.length,
8);
for (int i = 0; i < values.length; i++) {
- writer.writeInt(i, values[i]);
+ writer.writeLong(i, values[i]);
}
writer.complete();
return array;
diff --git
a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
index 60a8d9c700..fea6ab8687 100644
--- a/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/data_evolution_split_generator.py
@@ -39,11 +39,13 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
open_file_cost: int,
deletion_files_map=None,
row_ranges: Optional[List] = None,
- score_getter=None
+ score_getter=None,
+ group_stats_filter=None,
):
super().__init__(table, target_split_size, open_file_cost,
deletion_files_map)
self.row_ranges = row_ranges
self.score_getter = score_getter
+ self.group_stats_filter = group_stats_filter
def create_splits(self, file_entries: List[ManifestEntry]) -> List[Split]:
"""
@@ -85,6 +87,15 @@ class DataEvolutionSplitGenerator(AbstractSplitGenerator):
# Split files by firstRowId for data evolution
split_by_row_id = self._split_by_row_id(data_files)
+ if self.group_stats_filter is not None:
+ split_by_row_id = [
+ group for group in split_by_row_id
+ if self.group_stats_filter.may_match(group)
+ ]
+ split_by_row_id = [
+ [file.copy_without_stats() for file in group]
+ for group in split_by_row_id
+ ]
# Pack the split groups for optimal split sizes
packed_files = self._pack_for_ordered(
diff --git a/paimon-python/pypaimon/read/scanner/data_evolution_stats.py
b/paimon-python/pypaimon/read/scanner/data_evolution_stats.py
new file mode 100644
index 0000000000..c9de93763d
--- /dev/null
+++ b/paimon-python/pypaimon/read/scanner/data_evolution_stats.py
@@ -0,0 +1,259 @@
+# 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.
+
+from typing import Callable, Dict, List, Tuple
+
+from pypaimon.common.predicate import Predicate
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.push_down_utils import rewrite_predicate_indices
+from pypaimon.schema.data_types import DataField
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.table.row.generic_row import GenericRow
+
+
+_KNOWN = 0
+_MISSING = 1
+_UNKNOWN = 2
+
+
+class _FileLayout:
+
+ def __init__(self, data_fields, stats_offsets):
+ self.data_fields = {field.id: field for field in data_fields}
+ self.stats_offsets = stats_offsets
+
+
+class _StatsProvider:
+
+ def __init__(self, file, layout):
+ self.file = file
+ self.layout = layout
+ self.tied = False
+
+
+class DataEvolutionGroupStatsFilter:
+ """Conservatively filters logical row-id groups by merged column stats."""
+
+ def __init__(
+ self,
+ predicate: Predicate,
+ table_fields: List[DataField],
+ schema_fields: Callable[[int], List[DataField]],
+ ):
+ self.predicate = rewrite_predicate_indices(predicate, table_fields)
+ self.table_fields = table_fields
+ self.schema_fields = schema_fields
+ self._layout_cache: Dict[Tuple, _FileLayout] = {}
+
+ def may_match(self, files: List[DataFileMeta]) -> bool:
+ if not files:
+ return True
+ try:
+ stats, states, row_count = self._group_stats(files)
+ return self._predicate_may_match(
+ self.predicate, stats, states, row_count)
+ except Exception:
+ # Stats pruning is optional. Unknown schemas, corrupt stats, and
+ # incompatible types must retain the complete logical group.
+ return True
+
+ def _group_stats(self, files):
+ group_start = min(file.non_null_row_id_range().from_ for file in files)
+ group_end = max(file.non_null_row_id_range().to for file in files)
+ row_count = group_end - group_start + 1
+
+ normal_files = []
+ special_field_ids = set()
+ for file in files:
+ layout = self._layout(file)
+ if DataFileMeta.is_blob_file(file.file_name) \
+ or DataFileMeta.is_vector_file(file.file_name):
+ special_field_ids.update(layout.data_fields)
+ else:
+ normal_files.append((file, layout))
+ providers = {}
+ for file, layout in normal_files:
+ for field_id in layout.data_fields:
+ current = providers.get(field_id)
+ if (current is None
+ or file.max_sequence_number
+ > current.file.max_sequence_number):
+ providers[field_id] = _StatsProvider(file, layout)
+ elif (file.max_sequence_number
+ == current.file.max_sequence_number):
+ current.tied = True
+
+ min_values = []
+ max_values = []
+ null_counts = []
+ states = []
+ for field_index, field in enumerate(self.table_fields):
+ provider = providers.get(field.id)
+ if provider is None:
+ if field.id in special_field_ids:
+ self._append_unknown(
+ min_values, max_values, null_counts, states)
+ else:
+ min_values.append(None)
+ max_values.append(None)
+ null_counts.append(row_count)
+ states.append(_MISSING)
+ continue
+
+ if provider.tied:
+ self._append_unknown(
+ min_values, max_values, null_counts, states)
+ continue
+
+ file = provider.file
+ layout = provider.layout
+ file_range = file.non_null_row_id_range()
+ source_field = layout.data_fields[field.id]
+ # Partial-file stats do not describe the complete logical group.
+ if (file_range.from_ != group_start
+ or file_range.to != group_end
+ or source_field.type != field.type
+ or field.id not in layout.stats_offsets):
+ self._append_unknown(
+ min_values, max_values, null_counts, states)
+ continue
+
+ stats = file.value_stats
+ stats_offset = layout.stats_offsets[field.id]
+ min_value = stats.min_values.get_field(stats_offset)
+ max_value = stats.max_values.get_field(stats_offset)
+ null_count = (
+ stats.null_counts[stats_offset]
+ if (stats.null_counts is not None
+ and stats_offset < len(stats.null_counts))
+ else None
+ )
+ self._validate_stats(
+ min_value, max_value, null_count, row_count)
+ min_values.append(min_value)
+ max_values.append(max_value)
+ null_counts.append(null_count)
+ states.append(_KNOWN)
+
+ return (
+ SimpleStats(
+ GenericRow(min_values, self.table_fields),
+ GenericRow(max_values, self.table_fields),
+ null_counts,
+ ),
+ states,
+ row_count,
+ )
+
+ @staticmethod
+ def _append_unknown(min_values, max_values, null_counts, states):
+ min_values.append(None)
+ max_values.append(None)
+ null_counts.append(None)
+ states.append(_UNKNOWN)
+
+ def _layout(self, file):
+ key = (
+ file.schema_id,
+ tuple(file.write_cols) if file.write_cols is not None else None,
+ (tuple(file.value_stats_cols)
+ if file.value_stats_cols is not None else None),
+ )
+ layout = self._layout_cache.get(key)
+ if layout is not None:
+ return layout
+
+ schema_fields = self.schema_fields(file.schema_id)
+ fields_by_name = {field.name: field for field in schema_fields}
+ data_fields = self._project_fields(
+ schema_fields, fields_by_name, file.write_cols)
+ stats_fields = self._project_fields(
+ data_fields,
+ {field.name: field for field in data_fields},
+ file.value_stats_cols,
+ )
+ layout = _FileLayout(
+ data_fields,
+ {field.id: index for index, field in enumerate(stats_fields)},
+ )
+ self._layout_cache[key] = layout
+ return layout
+
+ @staticmethod
+ def _project_fields(default_fields, fields_by_name, names):
+ if names is None:
+ return default_fields
+ if len(names) != len(set(names)):
+ raise ValueError("Duplicate fields in file stats metadata.")
+ unknown = [
+ name for name in names
+ if name not in fields_by_name
+ and not SpecialFields.is_system_field(name)
+ ]
+ if unknown:
+ raise ValueError("Unknown fields in file stats metadata: %s" %
unknown)
+ return [fields_by_name[name] for name in names if name in
fields_by_name]
+
+ @staticmethod
+ def _validate_stats(min_value, max_value, null_count, row_count):
+ if (null_count is not None
+ and (isinstance(null_count, bool)
+ or not isinstance(null_count, int)
+ or null_count < 0
+ or null_count > row_count)):
+ raise ValueError("Invalid null count in file stats.")
+ if (min_value is None) != (max_value is None):
+ raise ValueError("Incomplete min/max values in file stats.")
+ if min_value is not None:
+ try:
+ ordered = min_value <= max_value
+ except TypeError as exc:
+ raise ValueError("Incomparable min/max values in file stats.")
from exc
+ if not ordered:
+ raise ValueError("Invalid min/max order in file stats.")
+ if null_count == row_count:
+ raise ValueError("All-null stats contain non-null bounds.")
+
+ def _predicate_may_match(self, predicate, stats, states, row_count):
+ if predicate.method == 'and':
+ return all(
+ self._predicate_may_match(child, stats, states, row_count)
+ for child in predicate.literals
+ )
+ if predicate.method == 'or':
+ return any(
+ self._predicate_may_match(child, stats, states, row_count)
+ for child in predicate.literals
+ )
+
+ index = predicate.index
+ if index is None or index < 0 or index >= len(states):
+ return True
+ if states[index] == _UNKNOWN:
+ return True
+ if states[index] == _MISSING:
+ tester = Predicate.testers.get(predicate.method)
+ return True if tester is None else tester.test_by_value(
+ None, predicate.literals)
+ field_type = getattr(self.table_fields[index].type, 'type', None)
+ if (field_type in ('FLOAT', 'DOUBLE')
+ and predicate.method in ('notEqual', 'notIn')):
+ # PyArrow min/max can omit NaN, which still matches negative
+ # predicates.
+ return True
+ return predicate.test_by_simple_stats(stats, row_count)
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index e92205405d..7714e7e925 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -50,6 +50,8 @@ from pypaimon.read.scanner.chunk_shuffle_split_generator
import (
)
from pypaimon.read.scanner.data_evolution_split_generator import \
DataEvolutionSplitGenerator
+from pypaimon.read.scanner.data_evolution_stats import \
+ DataEvolutionGroupStatsFilter
from pypaimon.read.scanner.primary_key_table_split_generator import \
PrimaryKeyTableSplitGenerator
from pypaimon.read.split import DataSplit
@@ -234,7 +236,7 @@ class FileScanner:
else:
self.predicate_for_stats = predicate
self.predicate_for_stats = exclude_predicate_with_fields(
- self.predicate_for_stats, {SpecialFields.ROW_ID.name})
+ self.predicate_for_stats, SpecialFields.SYSTEM_FIELD_NAMES)
# Partition columns aren't in data files, so skip them for value-stats
pruning.
self.predicate_for_stats = exclude_predicate_with_fields(
self.predicate_for_stats, set(self.table.partition_keys))
@@ -383,6 +385,12 @@ class FileScanner:
# Generate splits
splits = split_generator.create_splits(entries)
+ if self.data_evolution and self.scan_stats is not None:
+ # Data-evolution stats pruning happens on complete row-id groups
+ # inside the split generator, not in _filter_manifest_entry.
+ self.scan_stats.entries_after_stats = sum(
+ len(split.files) for split in splits)
+
if self.table.is_primary_key_table:
splits = self._apply_primary_key_sorted_indexes(splits)
@@ -469,18 +477,33 @@ class FileScanner:
{},
row_ranges,
score_getter,
+ None,
)
# Filter manifest files by row ranges if available
if row_ranges is not None:
manifest_files =
_filter_manifest_files_by_row_ranges(manifest_files, row_ranges)
- entries = self.read_manifest_entries(manifest_files,
row_ranges=row_ranges)
+ stats_predicate = getattr(self, 'predicate_for_stats', None)
+ group_stats_enabled = stats_predicate is not None and score_getter is
None
+ entries = self.read_manifest_entries(
+ manifest_files,
+ row_ranges=row_ranges,
+ keep_stats=group_stats_enabled,
+ )
# Redundant when early_record_filter ran; kept for explain mode and as
safety net.
if row_ranges is not None:
entries = _filter_manifest_entries_by_row_ranges(entries,
row_ranges)
+ group_stats_filter = None
+ if group_stats_enabled:
+ group_stats_filter = DataEvolutionGroupStatsFilter(
+ stats_predicate,
+ self.table.fields,
+ self._schema_fields,
+ )
+
return entries, DataEvolutionSplitGenerator(
self.table,
self.target_split_size,
@@ -488,6 +511,7 @@ class FileScanner:
self._deletion_files_map(entries),
row_ranges,
score_getter,
+ group_stats_filter,
)
def plan_files(self) -> List[ManifestEntry]:
@@ -536,7 +560,8 @@ class FileScanner:
return None
def read_manifest_entries(self, manifest_files: List[ManifestFileMeta],
- row_ranges=None) -> List[ManifestEntry]:
+ row_ranges=None,
+ keep_stats=False) -> List[ManifestEntry]:
max_workers =
self.table.options.scan_manifest_parallelism(os.cpu_count() or 8)
if self.scan_stats is not None:
self.scan_stats.manifest_files_total += len(manifest_files)
@@ -561,6 +586,7 @@ class FileScanner:
return self.manifest_file_manager.read_entries_parallel(
manifest_files,
self._filter_manifest_entry,
+ drop_stats=not keep_stats,
max_workers=max_workers,
early_entry_filter=self._build_early_bucket_filter(),
early_record_filter=early_row_filter,
@@ -854,8 +880,6 @@ class FileScanner:
return True
# Data evolution: file stats may be from another schema, skip
stats filter and filter in reader.
if self.data_evolution:
- if stats is not None:
- stats.entries_after_stats += 1
return True
if entry.file.value_stats_cols is None and entry.file.write_cols
is not None:
stats_fields = entry.file.write_cols
diff --git a/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
b/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
new file mode 100644
index 0000000000..4b9fbd0495
--- /dev/null
+++ b/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
@@ -0,0 +1,420 @@
+# 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.
+
+import tempfile
+import unittest
+
+import pyarrow as pa
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.read.scanner.data_evolution_split_generator import \
+ DataEvolutionSplitGenerator
+from pypaimon.read.scanner.data_evolution_stats import \
+ DataEvolutionGroupStatsFilter
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.special_fields import SpecialFields
+
+
+def _empty_stats():
+ return SimpleStats(GenericRow([], []), GenericRow([], []), [])
+
+
+def _file(
+ name,
+ first_row_id,
+ row_count,
+ fields,
+ min_values,
+ max_values,
+ null_counts=None,
+ sequence=0,
+ schema_id=0,
+ write_cols=None,
+ value_stats_cols=None,
+):
+ stats_fields = fields
+ if value_stats_cols is not None:
+ by_name = {field.name: field for field in fields}
+ stats_fields = [by_name[name] for name in value_stats_cols]
+ elif write_cols is not None:
+ by_name = {field.name: field for field in fields}
+ stats_fields = [by_name[name] for name in write_cols
+ if name in by_name]
+ return DataFileMeta.create(
+ file_name=name,
+ file_size=100,
+ row_count=row_count,
+ min_key=GenericRow([], []),
+ max_key=GenericRow([], []),
+ key_stats=_empty_stats(),
+ value_stats=SimpleStats(
+ GenericRow(min_values, stats_fields),
+ GenericRow(max_values, stats_fields),
+ ([0] * len(stats_fields)
+ if null_counts is None else null_counts),
+ ),
+ min_sequence_number=sequence,
+ max_sequence_number=sequence,
+ schema_id=schema_id,
+ level=0,
+ extra_files=[],
+ first_row_id=first_row_id,
+ write_cols=write_cols,
+ value_stats_cols=value_stats_cols,
+ )
+
+
+class DataEvolutionGroupStatsFilterTest(unittest.TestCase):
+
+ @staticmethod
+ def _filter(predicate, schemas, current_schema_id):
+ current_fields = schemas[current_schema_id]
+ return DataEvolutionGroupStatsFilter(
+ predicate,
+ current_fields,
+ lambda schema_id: schemas[schema_id],
+ )
+
+ def test_merges_stats_from_latest_file_for_each_column(self):
+ fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'left_value', AtomicType('INT')),
+ DataField(2, 'right_value', AtomicType('INT')),
+ ]
+ base = _file(
+ 'base.parquet', 0, 10, fields, [0, 10], [9, 19],
+ write_cols=['id', 'left_value'])
+ delta = _file(
+ 'delta.parquet', 0, 10, fields, [100], [109], sequence=1,
+ write_cols=['right_value'])
+ builder = PredicateBuilder(fields)
+
+ self.assertTrue(self._filter(
+ builder.and_predicates([
+ builder.equal('left_value', 15),
+ builder.equal('right_value', 105),
+ ]), {0: fields}, 0).may_match([base, delta]))
+ self.assertFalse(self._filter(
+ builder.equal('right_value', 500),
+ {0: fields}, 0).may_match([base, delta]))
+
+ def test_missing_or_corrupt_stats_fail_open(self):
+ fields = [DataField(0, 'value', AtomicType('INT'))]
+ builder = PredicateBuilder(fields)
+ without_stats = _file(
+ 'no-stats.parquet', 0, 10, fields, [], [],
+ write_cols=['value'], value_stats_cols=[])
+ corrupt = _file(
+ 'corrupt.parquet', 10, 10, fields, [10], [20])
+ corrupt.value_stats = _empty_stats()
+
+ for file in [without_stats, corrupt]:
+ with self.subTest(file=file.file_name):
+ self.assertTrue(self._filter(
+ builder.equal('value', 1000),
+ {0: fields}, 0).may_match([file]))
+
+ def test_invalid_stats_metadata_fails_open(self):
+ fields = [DataField(0, 'value', AtomicType('INT'))]
+ builder = PredicateBuilder(fields)
+ unknown_write_col = _file(
+ 'unknown-write.parquet', 0, 10, fields, [], [],
+ write_cols=['unknown'])
+ unknown_stats_col = _file(
+ 'unknown-stats.parquet', 0, 10, fields, [0], [9])
+ unknown_stats_col.value_stats_cols = ['unknown']
+ bad_null_count = _file(
+ 'bad-null-count.parquet', 0, 10, fields, [0], [9],
+ null_counts=[11])
+ reversed_min_max = _file(
+ 'reversed-min-max.parquet', 0, 10, fields, [100], [0])
+ contradictory_all_null = _file(
+ 'all-null-with-bounds.parquet', 0, 10, fields, [5], [5],
+ null_counts=[10])
+
+ cases = [
+ (unknown_write_col, builder.is_not_null('value')),
+ (unknown_stats_col, builder.equal('value', 50)),
+ (bad_null_count, builder.is_not_null('value')),
+ (reversed_min_max, builder.equal('value', 50)),
+ (contradictory_all_null, builder.not_equal('value', 5)),
+ ]
+ for file, predicate in cases:
+ with self.subTest(file=file.file_name):
+ self.assertTrue(self._filter(
+ predicate, {0: fields}, 0).may_match([file]))
+
+ def test_projected_predicate_is_rebound_by_name(self):
+ fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'b', AtomicType('INT')),
+ DataField(2, 'c', AtomicType('INT')),
+ ]
+ projected = [fields[0], fields[2]]
+ file = _file(
+ 'data.parquet', 0, 1, fields, [1, 0, 200], [1, 0, 200])
+ predicate = PredicateBuilder(projected).greater_than('c', 150)
+
+ self.assertTrue(self._filter(
+ predicate, {0: fields}, 0).may_match([file]))
+
+ def test_negative_float_predicates_fail_open_for_nan(self):
+ for type_name in ('FLOAT', 'DOUBLE'):
+ fields = [DataField(0, 'value', AtomicType(type_name))]
+ file = _file('data.parquet', 0, 2, fields, [5.0], [5.0])
+ builder = PredicateBuilder(fields)
+ for predicate in (
+ builder.not_equal('value', 5.0),
+ builder.is_not_in('value', [5.0])):
+ with self.subTest(type=type_name, method=predicate.method):
+ self.assertTrue(self._filter(
+ predicate, {0: fields}, 0).may_match([file]))
+
+ def test_projected_layout_is_cached(self):
+ fields = [DataField(0, 'value', AtomicType('INT'))]
+ schema_loads = []
+
+ def load_schema(schema_id):
+ schema_loads.append(schema_id)
+ return fields
+
+ stats_filter = DataEvolutionGroupStatsFilter(
+ PredicateBuilder(fields).equal('value', 5),
+ fields,
+ load_schema,
+ )
+ stats_filter.may_match([_file(
+ 'first.parquet', 0, 1, fields, [5], [5])])
+ stats_filter.may_match([_file(
+ 'second.parquet', 1, 1, fields, [5], [5])])
+
+ self.assertEqual([0], schema_loads)
+
+ def test_value_stats_cols_controls_covered_fields(self):
+ fields = [
+ DataField(0, 'without_stats', AtomicType('INT')),
+ DataField(1, 'with_stats', AtomicType('INT')),
+ ]
+ file = _file(
+ 'data.parquet', 0, 10, fields, [20], [29],
+ write_cols=['without_stats', 'with_stats'],
+ value_stats_cols=['with_stats'])
+ builder = PredicateBuilder(fields)
+
+ self.assertFalse(self._filter(
+ builder.equal('with_stats', 1000),
+ {0: fields}, 0).may_match([file]))
+ self.assertTrue(self._filter(
+ builder.equal('without_stats', 1000),
+ {0: fields}, 0).may_match([file]))
+
+ def test_add_column_uses_implicit_null_stats(self):
+ old_fields = [DataField(0, 'id', AtomicType('INT'))]
+ current_fields = old_fields + [
+ DataField(1, 'added', AtomicType('STRING'))]
+ old_file = _file(
+ 'old.parquet', 0, 10, old_fields, [0], [9], schema_id=0)
+ builder = PredicateBuilder(current_fields)
+ schemas = {0: old_fields, 1: current_fields}
+
+ self.assertTrue(self._filter(
+ builder.is_null('added'), schemas, 1).may_match([old_file]))
+ self.assertFalse(self._filter(
+ builder.is_not_null('added'), schemas, 1).may_match([old_file]))
+ self.assertFalse(self._filter(
+ builder.equal('added', 'x'), schemas, 1).may_match([old_file]))
+
+ def test_schema_rename_uses_field_id_and_type_change_fails_open(self):
+ old_fields = [DataField(0, 'old_name', AtomicType('INT'))]
+ renamed_fields = [DataField(0, 'new_name', AtomicType('INT'))]
+ changed_fields = [DataField(0, 'new_name', AtomicType('BIGINT'))]
+ old_file = _file(
+ 'old.parquet', 0, 10, old_fields, [0], [9], schema_id=0)
+
+ renamed_builder = PredicateBuilder(renamed_fields)
+ self.assertFalse(self._filter(
+ renamed_builder.equal('new_name', 50),
+ {0: old_fields, 1: renamed_fields}, 1).may_match([old_file]))
+
+ changed_builder = PredicateBuilder(changed_fields)
+ self.assertTrue(self._filter(
+ changed_builder.equal('new_name', 50),
+ {0: old_fields, 2: changed_fields}, 2).may_match([old_file]))
+
+ def test_blob_and_vector_files_do_not_supply_predicate_stats(self):
+ fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('BYTES')),
+ ]
+ base = _file(
+ 'base.parquet', 0, 10, fields, [0], [9],
+ write_cols=['id'])
+ builder = PredicateBuilder(fields)
+ for name in ['payload.blob', 'payload.vector.parquet']:
+ special = _file(
+ name, 0, 10, fields, [b'a'], [b'z'], sequence=1,
+ write_cols=['payload'])
+ with self.subTest(name=name):
+ stats_filter = self._filter(
+ builder.equal('id', 50), {0: fields}, 0)
+ self.assertFalse(stats_filter.may_match([base, special]))
+ stats_filter = self._filter(
+ builder.equal('payload', b'not-present'),
+ {0: fields}, 0)
+ self.assertTrue(stats_filter.may_match([base, special]))
+
+ def test_special_file_does_not_hide_normal_file_stats(self):
+ fields = [
+ DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'payload', AtomicType('BYTES')),
+ ]
+ base = _file(
+ 'base.parquet', 0, 10, fields, [0], [9],
+ write_cols=['id'])
+ vector = _file(
+ 'data.vector.parquet', 0, 10, fields,
+ [100, b'a'], [109, b'z'], sequence=1)
+ builder = PredicateBuilder(fields)
+
+ self.assertFalse(self._filter(
+ builder.equal('id', 50), {0: fields}, 0
+ ).may_match([base, vector]))
+ self.assertTrue(self._filter(
+ builder.equal('payload', b'not-present'), {0: fields}, 0
+ ).may_match([base, vector]))
+
+ def test_partial_newer_file_fails_open(self):
+ fields = [DataField(0, 'value', AtomicType('INT'))]
+ base = _file('base.parquet', 0, 10, fields, [0], [9])
+ partial_delta = _file(
+ 'delta.parquet', 3, 3, fields, [100], [102], sequence=1,
+ write_cols=['value'])
+ stats_filter = self._filter(
+ PredicateBuilder(fields).equal('value', 5), {0: fields}, 0)
+
+ self.assertTrue(stats_filter.may_match([base, partial_delta]))
+
+ def test_tied_latest_providers_fail_open(self):
+ fields = [DataField(0, 'value', AtomicType('INT'))]
+ first = _file(
+ 'first.parquet', 0, 10, fields, [0], [9], sequence=1)
+ second = _file(
+ 'second.parquet', 0, 10, fields, [100], [109], sequence=1)
+ stats_filter = self._filter(
+ PredicateBuilder(fields).equal('value', 50), {0: fields}, 0)
+
+ self.assertTrue(stats_filter.may_match([first, second]))
+
+
+class DataEvolutionGroupStatsPlanningTest(unittest.TestCase):
+
+ def test_system_field_predicate_skips_group_stats(self):
+ arrow_schema = pa.schema([
+ ('id', pa.int64()),
+ ('value', pa.int32()),
+ ])
+ with tempfile.TemporaryDirectory() as warehouse:
+ catalog = CatalogFactory.create({'warehouse': warehouse})
+ catalog.create_database('default', False)
+ catalog.create_table(
+ 'default.t',
+ Schema.from_pyarrow_schema(arrow_schema, options={
+ 'metadata.stats-mode': 'full',
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ }),
+ False,
+ )
+ table = catalog.get_table('default.t')
+
+ batch_write = table.new_batch_write_builder()
+ writer = batch_write.new_write()
+ commit = batch_write.new_commit()
+ try:
+ writer.write_arrow(pa.table({
+ 'id': [1, 2],
+ 'value': [10, 20],
+ }, schema=arrow_schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+
+ read_builder = table.new_read_builder().with_projection([
+ 'id',
+ SpecialFields.SEQUENCE_NUMBER.name,
+ ])
+ predicate = read_builder.new_predicate_builder().greater_than(
+ SpecialFields.SEQUENCE_NUMBER.name, -1)
+ read_builder.with_filter(predicate)
+
+ plan = read_builder.new_scan().plan()
+ result = read_builder.new_read().to_arrow(plan.splits())
+ self.assertEqual({
+ 'id': [1, 2],
+ SpecialFields.SEQUENCE_NUMBER.name: [1, 1],
+ }, result.to_pydict())
+
+ def test_prunes_groups_before_split_packing(self):
+ fields = [DataField(0, 'id', AtomicType('INT'))]
+ files = [
+ _file('match.parquet', 0, 10, fields, [0], [9]),
+ _file('fallback-1.parquet', 10, 10, fields, [20], [29]),
+ _file('fallback-2.parquet', 20, 10, fields, [40], [49]),
+ ]
+ entries = [ManifestEntry(
+ kind=0,
+ partition=GenericRow([], []),
+ bucket=0,
+ total_buckets=1,
+ file=file,
+ ) for file in files]
+
+ class _Options:
+ options = {}
+
+ class _Table:
+ table_path = '/tmp/table'
+ options = _Options()
+
+ predicate = PredicateBuilder(fields).equal('id', 5)
+ group_filter = DataEvolutionGroupStatsFilter(
+ predicate,
+ fields,
+ lambda schema_id: fields,
+ )
+ without_pruning = DataEvolutionSplitGenerator(
+ _Table(), 1024 * 1024, 0).create_splits(entries)
+ with_pruning = DataEvolutionSplitGenerator(
+ _Table(), 1024 * 1024, 0,
+ group_stats_filter=group_filter).create_splits(entries)
+
+ self.assertEqual(3, sum(len(split.files) for split in without_pruning))
+ self.assertEqual(1, sum(len(split.files) for split in with_pruning))
+ self.assertEqual(
+ ['match.parquet'],
+ [file.file_name for split in with_pruning for file in split.files],
+ )
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index 07adeeb25d..3e6ab7485a 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -15,12 +15,14 @@
# specific language governing permissions and limitations
# under the License.
+import tempfile
import unittest
from unittest.mock import patch
import pyarrow as pa
import pytest
+from pypaimon import CatalogFactory, Schema
from pypaimon.common.options.core_options import CoreOptions,
GlobalIndexSearchMode
from pypaimon.common.options.options import Options
from pypaimon.common.predicate import Predicate
@@ -219,6 +221,86 @@ class
DataEvolutionGlobalIndexCoverageTest(unittest.TestCase):
class GlobalIndexScalarFallbackTest(unittest.TestCase):
+ def test_full_fallback_groups_are_pruned_before_split_packing(self):
+ from pypaimon.read.scanner.file_scanner import (
+ _GlobalIndexPlanningResult,
+ )
+
+ with tempfile.TemporaryDirectory() as warehouse:
+ catalog = CatalogFactory.create({'warehouse': warehouse})
+ catalog.create_database('default', False)
+ schema = Schema.from_pyarrow_schema(
+ pa.schema([('key', pa.int64()), ('value', pa.string())]),
+ options={
+ 'data-evolution.enabled': 'true',
+ 'row-tracking.enabled': 'true',
+ 'metadata.stats-mode': 'full',
+ 'target-file-row-num': '10',
+ 'source.split.target-size': '1kb',
+ },
+ )
+ catalog.create_table('default.fallback_stats', schema, False)
+ table = catalog.get_table('default.fallback_stats')
+ for start in range(0, 100, 10):
+ write_builder = table.new_batch_write_builder()
+ writer = write_builder.new_write()
+ commit = write_builder.new_commit()
+ writer.write_arrow(pa.table({
+ 'key': list(range(start, start + 10)),
+ 'value': ['v{}'.format(i)
+ for i in range(start, start + 10)],
+ }, schema=pa.schema([
+ ('key', pa.int64()),
+ ('value', pa.string()),
+ ])))
+ commit.commit(writer.prepare_commit())
+ writer.close()
+ commit.close()
+
+ read_builder = table.new_read_builder()
+ read_builder.with_filter(
+ read_builder.new_predicate_builder().equal('key', 5))
+ index_plan = (
+ _GlobalIndexPlanningResult(
+ GlobalIndexResult.from_range(Range(5, 5)),
+ [Range(10, 99)],
+ )
+ )
+
+ baseline_scan = read_builder.new_scan()
+ baseline_scan.file_scanner._global_index_result = index_plan
+ baseline_scan.file_scanner.predicate_for_stats = None
+ baseline_plan = baseline_scan.plan()
+
+ scan = read_builder.new_scan()
+ scan.file_scanner._global_index_result = index_plan
+ plan = scan.plan()
+
+ stats_scan = read_builder.new_scan()
+ stats_scan.file_scanner._global_index_result = index_plan
+ stats_plan, stats = stats_scan.file_scanner.scan_with_stats()
+
+ self.assertEqual(
+ 10,
+ sum(len(split.files) for split in baseline_plan.splits()),
+ )
+ self.assertEqual(1, len(plan.splits()))
+ self.assertEqual(
+ 1, sum(len(split.files) for split in plan.splits()))
+ self.assertEqual(10, stats.entries_after_bucket)
+ self.assertEqual(1, stats.entries_after_stats)
+ self.assertEqual(
+ 1, sum(len(split.files) for split in stats_plan.splits()))
+ baseline_result = read_builder.new_read().to_arrow(
+ baseline_plan.splits()).to_pydict()
+ pruned_result = read_builder.new_read().to_arrow(
+ plan.splits()).to_pydict()
+ self.assertEqual(
+ {'key': [5], 'value': ['v5']},
+ baseline_result,
+ )
+ self.assertEqual(baseline_result, pruned_result)
+
def test_eval_global_index_keeps_unindexed_ranges_out_of_bitmap(self):
from pypaimon.read.scanner.file_scanner import (
FileScanner,
