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 cc1bfae3b9 [core][spark] Use DV-aware tight bounds for Spark MIN/MAX
pushdown (#8047)
cc1bfae3b9 is described below
commit cc1bfae3b91ecfba6b5f140da2adf883ea9c7053
Author: Kerwin Zhang <[email protected]>
AuthorDate: Sun May 31 21:23:15 2026 +0800
[core][spark] Use DV-aware tight bounds for Spark MIN/MAX pushdown (#8047)
Spark currently disables MIN/MAX aggregate pushdown for any table with
`deletion-vectors.enabled=true`. This is correct but too conservative:
many DV-enabled non-primary-key tables, or many splits inside them, do
not actually have deleted rows. In those cases the recorded file min/max
stats are still tight and can safely answer MIN/MAX.
This PR makes the decision based on runtime split metadata instead of
the table-level DV option. It derives whether a data file still has
tight stats from `DataFileMeta.deleteRowCount` and the paired
`DeletionFile.cardinality`, then allows Spark MIN/MAX pushdown only when
every file in the split is tight.
This keeps the existing safety behavior for files with real deletes or
unknown DV cardinality, while recovering MIN/MAX pushdown for DV-enabled
tables/splits that have no effective deletions.
---
.../apache/paimon/table/source/DvAwareStats.java | 41 ++++++++
.../apache/paimon/table/source/PushDownUtils.java | 18 ++++
.../paimon/table/source/DvAwareStatsTest.java | 76 ++++++++++++++
.../paimon/table/source/PushDownUtilsTest.java | 116 +++++++++++++++++++++
.../spark/aggregate/AggregatePushDownUtils.scala | 8 +-
.../paimon/spark/sql/PushDownAggregatesTest.scala | 19 ++++
6 files changed, 275 insertions(+), 3 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DvAwareStats.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DvAwareStats.java
new file mode 100644
index 0000000000..2d11500ab7
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DvAwareStats.java
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.io.DataFileMeta;
+
+import javax.annotation.Nullable;
+
+/** Utilities for DV-aware file statistics. */
+public final class DvAwareStats {
+
+ private DvAwareStats() {}
+
+ /** Returns whether file stats remain tight after applying delete
metadata. */
+ public static boolean isTightBounds(DataFileMeta file, @Nullable
DeletionFile dv) {
+ if (file.deleteRowCount().orElse(0L) > 0L) {
+ return false;
+ }
+ if (dv == null) {
+ return true;
+ }
+ Long card = dv.cardinality();
+ return card != null && card == 0L;
+ }
+}
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 65793014f1..cc9fc936cb 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
@@ -30,6 +30,7 @@ import org.apache.paimon.types.SmallIntType;
import org.apache.paimon.types.TinyIntType;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import static org.apache.paimon.utils.ListUtils.isNullOrEmpty;
@@ -78,4 +79,21 @@ public class PushDownUtils {
valueStatsCols == null
|| new
HashSet<>(valueStatsCols).containsAll(columns));
}
+
+ /** Returns whether every data file in the split has tight stats. */
+ public static boolean tightBoundsAvailable(Split split) {
+ if (!(split instanceof DataSplit)) {
+ return false;
+ }
+ DataSplit dataSplit = (DataSplit) split;
+ List<DataFileMeta> files = dataSplit.dataFiles();
+ List<DeletionFile> dvs = dataSplit.deletionFiles().orElse(null);
+ for (int i = 0; i < files.size(); i++) {
+ DeletionFile dv = dvs == null ? null : dvs.get(i);
+ if (!DvAwareStats.isTightBounds(files.get(i), dv)) {
+ return false;
+ }
+ }
+ return true;
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/DvAwareStatsTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/DvAwareStatsTest.java
new file mode 100644
index 0000000000..8cb07245f9
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/DvAwareStatsTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileTestUtils;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link DvAwareStats}. */
+public class DvAwareStatsTest {
+
+ @Test
+ public void testNoDvAndEmptyDeleteRowCountIsTight() {
+ DataFileMeta file = DataFileTestUtils.newFile();
+ assertThat(DvAwareStats.isTightBounds(file, null)).isTrue();
+ }
+
+ @Test
+ public void testNoDvAndPositiveDeleteRowCountIsWide() {
+ DataFileMeta file = DataFileTestUtils.newFile("f1", 0, 0, 9, 10L, 5L);
+ assertThat(DvAwareStats.isTightBounds(file, null)).isFalse();
+ }
+
+ @Test
+ public void testEmptyDvIsTight() {
+ DataFileMeta file = DataFileTestUtils.newFile();
+ DeletionFile dv = new DeletionFile("dv-1", 0L, 0L, 0L);
+ assertThat(DvAwareStats.isTightBounds(file, dv)).isTrue();
+ }
+
+ @Test
+ public void testPopulatedDvIsWide() {
+ DataFileMeta file = DataFileTestUtils.newFile();
+ DeletionFile dv = new DeletionFile("dv-1", 0L, 16L, 5L);
+ assertThat(DvAwareStats.isTightBounds(file, dv)).isFalse();
+ }
+
+ @Test
+ public void testUnknownCardinalityDvIsConservativelyWide() {
+ DataFileMeta file = DataFileTestUtils.newFile();
+ DeletionFile dv = new DeletionFile("dv-1", 0L, 16L, null);
+ assertThat(DvAwareStats.isTightBounds(file, dv)).isFalse();
+ }
+
+ @Test
+ public void testNoDvAndZeroDeleteRowCountIsTight() {
+ DataFileMeta file = DataFileTestUtils.newFile("f1", 0, 0, 9, 10L, 0L);
+ assertThat(DvAwareStats.isTightBounds(file, null)).isTrue();
+ }
+
+ @Test
+ public void testBothDeleteRowCountAndPopulatedDvIsWide() {
+ DataFileMeta file = DataFileTestUtils.newFile("f1", 0, 0, 9, 10L, 5L);
+ DeletionFile dv = new DeletionFile("dv-1", 0L, 16L, 3L);
+ assertThat(DvAwareStats.isTightBounds(file, dv)).isFalse();
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PushDownUtilsTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PushDownUtilsTest.java
new file mode 100644
index 0000000000..b1c3e5003a
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PushDownUtilsTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataFileTestUtils;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.OptionalLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link PushDownUtils}. */
+public class PushDownUtilsTest {
+
+ @Test
+ public void testNonDataSplitReturnsFalse() {
+ Split nonDataSplit =
+ new Split() {
+ @Override
+ public long rowCount() {
+ return 0;
+ }
+
+ @Override
+ public OptionalLong mergedRowCount() {
+ return OptionalLong.empty();
+ }
+ };
+ assertThat(PushDownUtils.tightBoundsAvailable(nonDataSplit)).isFalse();
+ }
+
+ @Test
+ public void testEmptySplitReturnsTrue() {
+ DataSplit split = newSplit(Collections.emptyList(), null);
+ assertThat(PushDownUtils.tightBoundsAvailable(split)).isTrue();
+ }
+
+ @Test
+ public void testAllFilesTightReturnsTrue() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ DataFileTestUtils.newFile(),
+ DataFileTestUtils.newFile(),
+ DataFileTestUtils.newFile());
+ DataSplit split = newSplit(files, null);
+ assertThat(PushDownUtils.tightBoundsAvailable(split)).isTrue();
+ }
+
+ @Test
+ public void testAnyFileWithPopulatedDvReturnsFalse() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ DataFileTestUtils.newFile(),
+ DataFileTestUtils.newFile(),
+ DataFileTestUtils.newFile());
+ List<DeletionFile> dvs =
+ Arrays.asList(
+ new DeletionFile("dv-0", 0L, 0L, 0L),
+ new DeletionFile("dv-1", 0L, 16L, 5L),
+ new DeletionFile("dv-2", 0L, 0L, 0L));
+ DataSplit split = newSplit(files, dvs);
+ assertThat(PushDownUtils.tightBoundsAvailable(split)).isFalse();
+ }
+
+ @Test
+ public void testFileWithDeleteRowCountReturnsFalse() {
+ List<DataFileMeta> files =
+ Arrays.asList(
+ DataFileTestUtils.newFile("f0", 0, 0, 9, 10L, 5L),
+ DataFileTestUtils.newFile(),
+ DataFileTestUtils.newFile());
+ List<DeletionFile> dvs =
+ Arrays.asList(
+ new DeletionFile("dv-0", 0L, 0L, 0L),
+ new DeletionFile("dv-1", 0L, 0L, 0L),
+ new DeletionFile("dv-2", 0L, 0L, 0L));
+ DataSplit split = newSplit(files, dvs);
+ assertThat(PushDownUtils.tightBoundsAvailable(split)).isFalse();
+ }
+
+ private static DataSplit newSplit(List<DataFileMeta> files,
List<DeletionFile> deletionFiles) {
+ DataSplit.Builder builder =
+ DataSplit.builder()
+ .withSnapshot(1L)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("dummy")
+ .withDataFiles(files);
+ if (deletionFiles != null) {
+ builder.withDataDeletionFiles(deletionFiles);
+ }
+ return builder.build();
+ }
+}
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 52ce726c35..8b0a57cc52 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
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.aggregate
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.table.source.{DataSplit, ReadBuilder, Split}
-import org.apache.paimon.table.source.PushDownUtils.minmaxAvailable
+import org.apache.paimon.table.source.PushDownUtils.{minmaxAvailable,
tightBoundsAvailable}
import org.apache.paimon.types._
import org.apache.spark.sql.connector.expressions.Expression
@@ -37,7 +37,6 @@ object AggregatePushDownUtils {
table: FileStoreTable,
aggregation: Aggregation,
readBuilder: ReadBuilder): Option[LocalAggregator] = {
- val options = table.coreOptions()
val rowType = table.rowType
val partitionKeys = table.partitionKeys()
@@ -53,13 +52,16 @@ object AggregatePushDownUtils {
if (columns.isEmpty) {
generateSplits(readBuilder.dropStats())
} else {
- if (options.deletionVectorsEnabled() ||
!table.primaryKeys().isEmpty) {
+ if (!table.primaryKeys().isEmpty) {
return None
}
val splits = generateSplits(readBuilder)
if (!splits.forall(minmaxAvailable(_, columns.asJava))) {
return None
}
+ if (!splits.forall(tightBoundsAvailable)) {
+ return None
+ }
splits
}
case None => return None
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 766fac386f..feb20e8da2 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
@@ -268,6 +268,25 @@ class PushDownAggregatesTest extends PaimonSparkTestBase
with AdaptiveSparkPlanH
})
}
+ test("Push down aggregate - non-primary-key DV table with tight bounds") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT)
+ |TBLPROPERTIES (
+ | 'deletion-vectors.enabled' = 'true',
+ | 'bucket-key' = 'id',
+ | 'bucket' = '1'
+ |)
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT id FROM range (0, 5000)")
+ // No deleted rows, so split stats can answer MIN/MAX.
+ runAndCheckAggregate("SELECT COUNT(*), MIN(id), MAX(id) FROM T",
Row(5000, 0, 4999) :: Nil, 0)
+ sql("DELETE FROM T WHERE id > 100 AND id <= 400")
+ // Deleted rows make file stats wide, so Spark keeps MIN/MAX aggregation.
+ runAndCheckAggregate("SELECT MIN(id), MAX(id) FROM T", Row(0, 4999) ::
Nil, 2)
+ }
+ }
+
test("Push down aggregate: group by partial partition of a multi partition
table") {
sql(s"""
|CREATE TABLE T (
