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 d70d0a0218 [core] Do not push down MIN/MAX onto a column that has no
min/max statistics (#10593)
d70d0a0218 is described below
commit d70d0a0218b031091c4682538f5a9a371b4ac78a
Author: cxzl25 <[email protected]>
AuthorDate: Fri Oct 16 13:00:22 2026 +0800
[core] Do not push down MIN/MAX onto a column that has no min/max
statistics (#10593)
---
.../apache/paimon/table/source/PushDownUtils.java | 48 ++++++++++++++++++++++
.../source/aggregate/AggregatePushDownUtils.java | 7 +++-
.../apache/paimon/flink/BatchFileStoreITCase.java | 25 +++++++++++
.../spark/aggregate/AggregatePushDownUtils.scala | 6 ++-
.../paimon/spark/sql/PushDownAggregatesTest.scala | 26 ++++++++++++
5 files changed, 110 insertions(+), 2 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PushDownUtils.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PushDownUtils.java
index cc9fc936cb..fd3946705a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PushDownUtils.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PushDownUtils.java
@@ -19,15 +19,21 @@
package org.apache.paimon.table.source;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.stats.SimpleStatsEvolution;
+import org.apache.paimon.stats.SimpleStatsEvolutions;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.DateType;
import org.apache.paimon.types.DoubleType;
import org.apache.paimon.types.FloatType;
import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.types.SmallIntType;
import org.apache.paimon.types.TinyIntType;
+import org.apache.paimon.utils.InternalRowUtils;
import java.util.HashSet;
import java.util.List;
@@ -81,6 +87,48 @@ public class PushDownUtils {
}
/** Returns whether every data file in the split has tight stats. */
+ /**
+ * Like {@link #minmaxAvailable(Split, Set)}, and additionally requires
every data file of the
+ * split to carry a min and a max for each column, or to hold only nulls
in that column. A file
+ * written with {@code stats-mode} {@code counts} for the column is listed
in its {@code
+ * valueStatsCols} but has no min and max, and {@link DataSplit#minValue}
could not tell it
+ * apart from a column that is null in every row.
+ */
+ public static boolean minmaxAvailable(
+ Split split, Set<String> columns, RowType rowType,
SimpleStatsEvolutions evolutions) {
+ if (!minmaxAvailable(split, columns)) {
+ return false;
+ }
+ DataSplit dataSplit = (DataSplit) split;
+ for (String column : columns) {
+ int fieldIndex = rowType.getFieldIndex(column);
+ DataField field = rowType.getField(column);
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ if (file.rowCount() == 0) {
+ continue;
+ }
+ SimpleStatsEvolution.Result stats =
+ evolutions
+ .getOrCreate(file.schemaId())
+ .evolution(
+ file.valueStats(), file.rowCount(),
file.valueStatsCols());
+ Object min = InternalRowUtils.get(stats.minValues(),
fieldIndex, field.type());
+ Object max = InternalRowUtils.get(stats.maxValues(),
fieldIndex, field.type());
+ if (min != null && max != null) {
+ continue;
+ }
+ Long nullCount =
+ (Long)
+ InternalRowUtils.get(
+ stats.nullCounts(), fieldIndex,
DataTypes.BIGINT());
+ if (nullCount == null || nullCount != file.rowCount()) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
public static boolean tightBoundsAvailable(Split split) {
if (!(split instanceof DataSplit)) {
return false;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/aggregate/AggregatePushDownUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/aggregate/AggregatePushDownUtils.java
index e2862729d1..17eeb55274 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/aggregate/AggregatePushDownUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/aggregate/AggregatePushDownUtils.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.flink.LogicalTypeConversion;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.stats.SimpleStatsEvolutions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
@@ -159,6 +160,9 @@ public class AggregatePushDownUtils {
List<Split> splits = readBuilder.newScan().plan().splits();
List<DataSplit> dataSplits = new ArrayList<>(splits.size());
+ SimpleStatsEvolutions evolutions =
+ new SimpleStatsEvolutions(
+ id -> table.schemaManager().schema(id).fields(),
table.schema().id());
for (Split split : splits) {
if (!(split instanceof DataSplit)) {
return null;
@@ -169,7 +173,8 @@ public class AggregatePushDownUtils {
return null;
}
- if (!minMaxColumns.isEmpty() && !minmaxAvailable(split,
minMaxColumns)) {
+ if (!minMaxColumns.isEmpty()
+ && !minmaxAvailable(split, minMaxColumns, table.rowType(),
evolutions)) {
return null;
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 94f223f3de..288f72518e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -928,6 +928,31 @@ public class BatchFileStoreITCase extends
CatalogITCaseBase {
validateAggregateNotPushDown(sql, "MinAggFunction");
}
+ @Test
+ public void testMinMaxCountsStatsModeNotPushDown() {
+ // a column with stats-mode `counts` has a null count but no min/max
+ sql(
+ "CREATE TABLE min_max_counts (f0 INT, f1 INT) "
+ + "WITH ('fields.f0.stats-mode' = 'counts')");
+ sql("INSERT INTO min_max_counts VALUES (1, 10), (2, 20), (3, CAST(NULL
AS INT))");
+
+ String sql = "SELECT MIN(f0), MAX(f0) FROM min_max_counts";
+ assertThat(sql(sql)).containsOnly(Row.of(1, 3));
+ validateAggregateNotPushDown(sql, "MinAggFunction", "MaxAggFunction");
+
+ sql = "SELECT MIN(f1), MAX(f1) FROM min_max_counts";
+ assertThat(sql(sql)).containsOnly(Row.of(10, 20));
+ validateAggregatePushDown(sql, "MinAggFunction", "MaxAggFunction");
+
+ sql(
+ "CREATE TABLE min_max_counts_all (f0 INT, f1 INT) "
+ + "WITH ('metadata.stats-mode' = 'counts')");
+ sql("INSERT INTO min_max_counts_all VALUES (1, 10), (2, 20)");
+ sql = "SELECT MIN(f0), MAX(f1) FROM min_max_counts_all";
+ assertThat(sql(sql)).containsOnly(Row.of(1, 20));
+ validateAggregateNotPushDown(sql, "MinAggFunction", "MaxAggFunction");
+ }
+
@Test
public void testMinMaxPKNotPushDown() {
sql("CREATE TABLE min_max_pk (f0 INT PRIMARY KEY NOT ENFORCED, f1
INT)");
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/aggregate/AggregatePushDownUtils.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/aggregate/AggregatePushDownUtils.scala
index c2c82989e6..1e9e721e65 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/aggregate/AggregatePushDownUtils.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/aggregate/AggregatePushDownUtils.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.aggregate
+import org.apache.paimon.stats.SimpleStatsEvolutions
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.source.{DataSplit, ReadBuilder, Split}
import org.apache.paimon.table.source.PushDownUtils.{minmaxAvailable,
tightBoundsAvailable}
@@ -56,7 +57,10 @@ object AggregatePushDownUtils {
return None
}
val splits = generateSplits(readBuilder)
- if (!splits.forall(minmaxAvailable(_, columns.asJava))) {
+ val evolutions = new SimpleStatsEvolutions(
+ sid => table.schemaManager().schema(sid).fields(),
+ table.schema().id())
+ if (!splits.forall(minmaxAvailable(_, columns.asJava, rowType,
evolutions))) {
return None
}
if (!splits.forall(tightBoundsAvailable)) {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PushDownAggregatesTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PushDownAggregatesTest.scala
index 875fd28577..33f4afad10 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PushDownAggregatesTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PushDownAggregatesTest.scala
@@ -227,6 +227,32 @@ class PushDownAggregatesTest extends PaimonSparkTestBase
with AdaptiveSparkPlanH
}
}
+ test("Push down aggregate - column without min/max statistics") {
+ // A column whose stats-mode is `counts` carries a null count but no
min/max, which must not
+ // be mistaken for a column that is all null.
+ Seq("'fields.c1.stats-mode' = 'counts'", "'metadata.stats-mode' =
'counts'").foreach {
+ statsMode =>
+ withTable("T") {
+ spark.sql(s"CREATE TABLE T (c1 INT, c2 INT)
TBLPROPERTIES($statsMode)")
+ spark.sql("INSERT INTO T VALUES (1, 10), (2, 20), (3, null)")
+ runAndCheckAggregate("SELECT COUNT(*) FROM T", Row(3) :: Nil, 0)
+ runAndCheckAggregate("SELECT MIN(c1), MAX(c1) FROM T", Row(1, 3) ::
Nil, 2)
+ runAndCheckAggregate("SELECT MIN(c1), MAX(c2) FROM T", Row(1, 20) ::
Nil, 2)
+ }
+ }
+
+ withTable("T") {
+ spark.sql("CREATE TABLE T (c1 INT, c2 INT)
TBLPROPERTIES('fields.c1.stats-mode' = 'counts')")
+ spark.sql("INSERT INTO T VALUES (1, 10), (2, 20), (3, null)")
+ // c2 still has full statistics
+ runAndCheckAggregate("SELECT MIN(c2), MAX(c2) FROM T", Row(10, 20) ::
Nil, 0)
+ // A column added after the data was written is null in every existing
file
+ spark.sql("ALTER TABLE T ADD COLUMN c3 INT")
+ spark.sql("INSERT INTO T VALUES (4, 40, 400)")
+ runAndCheckAggregate("SELECT MIN(c3), MAX(c3) FROM T", Row(400, 400) ::
Nil, 0)
+ }
+ }
+
test("Push down aggregate - primary key table with deletion vector") {
Seq(true, false).foreach(
deletionVectorsEnabled => {
