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 712060744d [core] Do not prune TopN splits with unknown sort-column
statistics (#10039)
712060744d is described below
commit 712060744d9d4a7c7b90bbc0bef4e3e4aeb56df5
Author: YangJie <[email protected]>
AuthorDate: Mon Sep 21 03:05:08 2026 -0400
[core] Do not prune TopN splits with unknown sort-column statistics (#10039)
---
.../table/source/TopNDataSplitEvaluator.java | 99 +++++++++++++++++++++-
.../apache/paimon/table/source/TableScanTest.java | 78 +++++++++++++++--
2 files changed, 166 insertions(+), 11 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/TopNDataSplitEvaluator.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/TopNDataSplitEvaluator.java
index c67c05f094..c010256f76 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/TopNDataSplitEvaluator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/TopNDataSplitEvaluator.java
@@ -18,13 +18,17 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.predicate.CompareUtils;
import org.apache.paimon.predicate.SortValue;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.stats.SimpleStatsEvolution;
import org.apache.paimon.stats.SimpleStatsEvolutions;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.InternalRowUtils;
import java.util.ArrayList;
import java.util.Collections;
@@ -77,10 +81,16 @@ public class TopNDataSplitEvaluator {
}
DataSplit dataSplit = (DataSplit) split;
- Object min = dataSplit.minValue(index, field, evolutions);
- Object max = dataSplit.maxValue(index, field, evolutions);
- Long nullCount = dataSplit.nullCount(index, evolutions);
- richSplits.add(new RichSplit(dataSplit, min, max, nullCount));
+ SplitStats stats = extractStats(dataSplit, index, field,
evolutions);
+ if (!stats.usableForPruning()) {
+ // Some file of the split lacks min/max/null-count statistics
for the sort
+ // column (stats.mode=counts/none, or files written before the
column was
+ // added). The aggregate bound is then not a true bound, so
ordering the
+ // split against the others could drop a split holding a top
row — read it.
+ results.add(dataSplit);
+ continue;
+ }
+ richSplits.add(new RichSplit(dataSplit, stats.min, stats.max,
stats.nullCount));
}
// pick the TopN splits
@@ -90,6 +100,87 @@ public class TopNDataSplitEvaluator {
return results;
}
+ /**
+ * Aggregated sort-column statistics of a split, computed per file so that
missing statistics of
+ * ANY file are visible. {@link DataSplit#minValue} and friends aggregate
by skipping files
+ * without statistics, which produces a value that looks known but is not
a true bound once a
+ * single file is unknown.
+ */
+ private static class SplitStats {
+
+ private final Object min;
+ private final Object max;
+ private final Long nullCount;
+ private final boolean complete;
+
+ private SplitStats(Object min, Object max, Long nullCount, boolean
complete) {
+ this.min = min;
+ this.max = max;
+ this.nullCount = nullCount;
+ this.complete = complete;
+ }
+
+ /** Whether the split can safely participate in ordering-based
pruning. */
+ private boolean usableForPruning() {
+ return complete;
+ }
+ }
+
+ private SplitStats extractStats(
+ DataSplit split, int fieldIndex, DataField field,
SimpleStatsEvolutions evolutions) {
+ Object min = null;
+ Object max = null;
+ Long nullCount = null;
+ boolean complete = true;
+ for (DataFileMeta file : split.dataFiles()) {
+ SimpleStatsEvolution evolution =
evolutions.getOrCreate(file.schemaId());
+ Long fileNullCount =
+ (Long)
+ InternalRowUtils.get(
+ evolution.evolution(
+ file.valueStats().nullCounts(),
+ file.rowCount(),
+ file.valueStatsCols()),
+ fieldIndex,
+ DataTypes.BIGINT());
+
+ if (fileNullCount != null && fileNullCount.longValue() ==
file.rowCount()) {
+ // provably no non-null value in this file: it legitimately
contributes
+ // nothing to min/max (all-null column, or file written before
the column
+ // was added)
+ nullCount = nullCount == null ? fileNullCount : nullCount +
fileNullCount;
+ continue;
+ }
+
+ Object fileMin =
+ InternalRowUtils.get(
+ evolution.evolution(
+ file.valueStats().minValues(),
file.valueStatsCols()),
+ fieldIndex,
+ field.type());
+ Object fileMax =
+ InternalRowUtils.get(
+ evolution.evolution(
+ file.valueStats().maxValues(),
file.valueStatsCols()),
+ fieldIndex,
+ field.type());
+ if (fileMin == null || fileMax == null || fileNullCount == null) {
+ // statistics not collected for this file
(stats.mode=counts/none): the
+ // aggregate bound would not be a true bound
+ complete = false;
+ continue;
+ }
+ nullCount = nullCount == null ? fileNullCount : nullCount +
fileNullCount;
+ if (min == null || CompareUtils.compareLiteral(field.type(),
fileMin, min) < 0) {
+ min = fileMin;
+ }
+ if (max == null || CompareUtils.compareLiteral(field.type(),
fileMax, max) > 0) {
+ max = fileMax;
+ }
+ }
+ return new SplitStats(min, max, nullCount, complete);
+ }
+
/**
* Orders splits by their best row under the query's sort order and keeps
the first {@code
* limit} ones. In the NULLS LAST branches a split whose sort column is
provably all null
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
index 6d706d5b9a..9839bd5cdd 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/TableScanTest.java
@@ -931,10 +931,12 @@ public class TableScanTest extends ScannerTestBase {
FieldRef ref = new FieldRef(1, field.name(), field.type());
// stats-mode=counts-like split: min/max unknown (null) but only 2 of
5 rows are null,
- // so it is NOT provably all-null and must stay ahead of splits with
known bounds
+ // so it is NOT provably all-null. Its bound is unknown, so it can
never be pruned;
+ // the best known-bound split still wins the remaining limit slot and
the worst
+ // known split is dropped
DataSplit unknownSplit = newTestSplitWithField1Stats("unknown", null,
null, 2L, 5);
- DataSplit realSplit = newTestSplit("real", 10, 19, null);
- DataSplit realMaxSplit = newTestSplit("real-max", 100, 109, null);
+ DataSplit lowSplit = newTestSplit("low", 10, 19, null);
+ DataSplit highSplit = newTestSplit("high", 100, 109, null);
TopN ascTopN = new TopN(ref, ASCENDING, NULLS_LAST, 1);
List<Split> ascResult =
@@ -942,8 +944,8 @@ public class TableScanTest extends ScannerTestBase {
.evaluate(
ascTopN.orders().get(0),
ascTopN.limit(),
- Arrays.asList(unknownSplit, realSplit));
- assertThat(ascResult).containsExactly(unknownSplit);
+ Arrays.asList(unknownSplit, lowSplit,
highSplit));
+ assertThat(ascResult).containsExactlyInAnyOrder(unknownSplit,
lowSplit);
TopN descTopN = new TopN(ref, DESCENDING, NULLS_LAST, 1);
List<Split> descResult =
@@ -951,8 +953,70 @@ public class TableScanTest extends ScannerTestBase {
.evaluate(
descTopN.orders().get(0),
descTopN.limit(),
- Arrays.asList(unknownSplit, realMaxSplit));
- assertThat(descResult).containsExactly(unknownSplit);
+ Arrays.asList(unknownSplit, lowSplit,
highSplit));
+ assertThat(descResult).containsExactlyInAnyOrder(unknownSplit,
highSplit);
+ }
+
+ @Test
+ public void testPushDownTopNMultiFileSplitWithMixedStatsIsAlwaysRead()
throws Exception {
+ createAppendOnlyTable();
+
+ DataField field = table.schema().fields().get(1);
+ FieldRef ref = new FieldRef(1, field.name(), field.type());
+
+ // one split holding two files: a full-statistics file whose min is
large (50) and a
+ // counts-mode file with no min/max. DataSplit.minValue skipped the
counts file and
+ // reported 50 as the split's bound, so ASC ordering ranked the split
last and pruned it
+ // at LIMIT 1 — yet the counts file may hold a value below every other
split's min, i.e.
+ // the true top row. Treating the split as unknown reads it instead of
dropping it.
+ DataFileMeta fullStatsFile =
+ newTestSplitWithField1Stats("full", 50, 59, 2L,
10).dataFiles().get(0);
+ DataFileMeta countsFile =
+ newTestSplitWithField1Stats("counts", null, null, 1L,
3).dataFiles().get(0);
+ DataSplit mixed =
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("dummy")
+ .rawConvertible(true)
+ .withDataFiles(Arrays.asList(fullStatsFile,
countsFile))
+ .build();
+ DataSplit lowSplit = newTestSplit("low", 10, 19, null);
+ DataSplit midSplit = newTestSplit("mid", 20, 29, null);
+
+ TopN topN = new TopN(ref, ASCENDING, NULLS_LAST, 1);
+ List<Split> result =
+ new TopNDataSplitEvaluator(table.schema(),
table.schemaManager())
+ .evaluate(
+ topN.orders().get(0),
+ topN.limit(),
+ Arrays.asList(mixed, lowSplit, midSplit));
+ // mixed is always read (its stats are incomplete); the best
true-bound split (low) still
+ // wins the remaining limit slot. Before the fix mixed's fabricated
min of 50 ranked it
+ // last and LIMIT 1 kept only low, dropping the split that may hold
the top row.
+ assertThat(result).containsExactlyInAnyOrder(mixed, lowSplit);
+ }
+
+ @Test
+ public void testPushDownTopNCountsModeStatsDisablesPruning() throws
Exception {
+ createAppendOnlyTable();
+
+ DataField field = table.schema().fields().get(1);
+ FieldRef ref = new FieldRef(1, field.name(), field.type());
+
+ // stats.mode=counts-like table: every split knows only its null
count, so no split
+ // bound is known — pruning would keep an arbitrary subset and can
drop the split
+ // holding the true top row, so all splits must be read
+ DataSplit a = newTestSplitWithField1Stats("a", null, null, 1L, 3);
+ DataSplit b = newTestSplitWithField1Stats("b", null, null, 1L, 3);
+ DataSplit c = newTestSplitWithField1Stats("c", null, null, 1L, 3);
+
+ TopN topN = new TopN(ref, DESCENDING, NULLS_LAST, 1);
+ List<Split> result =
+ new TopNDataSplitEvaluator(table.schema(),
table.schemaManager())
+ .evaluate(topN.orders().get(0), topN.limit(),
Arrays.asList(a, b, c));
+ assertThat(result).containsExactlyInAnyOrder(a, b, c);
}
@Test
