You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/29 00:11:09 UTC
(paimon) branch master updated: [spark] Test count with DV merge-on-read level-0 files (#8894)
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 <zo...@alibaba-inc.com>
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 => {