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 a3121b7a1c [spark] Test count with DV merge-on-read level-0 files
(#8894)
a3121b7a1c is described below
commit a3121b7a1cb7b1e9c86da3cb21bd857e282841a9
Author: Zouxxyy <[email protected]>
AuthorDate: Wed Jul 29 08:11:03 2026 +0800
[spark] Test count with DV merge-on-read level-0 files (#8894)
---
.../paimon/spark/sql/PushDownAggregatesTest.scala | 33 ++++++++++++++++++++++
1 file changed, 33 insertions(+)
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 feb20e8da2..3c195d150f 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
@@ -239,6 +239,39 @@ class PushDownAggregatesTest extends PaimonSparkTestBase
with AdaptiveSparkPlanH
})
}
+ test("Count with deletion-vector merge-on-read and level-0 files") {
+ withTable("T") {
+ sql("""
+ |CREATE TABLE T (id INT, value STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'deletion-vectors.merge-on-read' = 'true',
+ | 'write-only' = 'true'
+ |)
+ |""".stripMargin)
+
+ sql("INSERT INTO T VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ checkAnswer(sql("SELECT COUNT(*), MIN(level), MAX(level) FROM
`T$files`"), Row(1, 0, 0))
+ // A single non-overlapping L0 file has an exact row count.
+ runAndCheckAggregate("SELECT COUNT(*) FROM T", Row(3) :: Nil, 0)
+
+ sql("DELETE FROM T WHERE id = 2")
+ checkAnswer(sql("SELECT COUNT(*), MIN(level), MAX(level) FROM
`T$files`"), Row(2, 0, 0))
+ // The L0 delete record overlaps with the original file and requires MOR.
+ runAndCheckAggregate("SELECT COUNT(*) FROM T", Row(2) :: Nil, 2)
+
+ sql("INSERT INTO T VALUES (1, 'updated'), (4, 'd')")
+ checkAnswer(sql("SELECT COUNT(*), MIN(level), MAX(level) FROM
`T$files`"), Row(3, 0, 0))
+ checkAnswer(
+ sql("SELECT * FROM T ORDER BY id"),
+ Row(1, "updated") :: Row(3, "c") :: Row(4, "d") :: Nil)
+ // Overlapping L0 files require MOR again, so Spark executes the
aggregate.
+ runAndCheckAggregate("SELECT COUNT(*) FROM T", Row(3) :: Nil, 2)
+ }
+ }
+
test("Push down aggregate - table with deletion vector") {
Seq(true, false).foreach(
deletionVectorsEnabled => {
