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 2024/10/11 03:02:47 UTC
(paimon) branch master updated: [core][spark] Support push down limit for primary key table (#4299)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 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 9bbf62b6b [core][spark] Support push down limit for primary key table (#4299)
9bbf62b6b is described below
commit 9bbf62b6b6bf3db968cbc3f297971a84c4d497bf
Author: Xiduo You <ul...@gmail.com>
AuthorDate: Fri Oct 11 11:02:42 2024 +0800
[core][spark] Support push down limit for primary key table (#4299)
---
.../paimon/table/source/DataTableBatchScan.java | 22 +++++++++++++++++++++
.../apache/paimon/spark/PaimonScanBuilder.scala | 5 ++---
.../paimon/spark/sql/PaimonPushDownTest.scala | 23 ++++++++++++++++++++--
3 files changed, 45 insertions(+), 5 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
index 29baac9a5..d1d455040 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
@@ -96,6 +96,28 @@ public class DataTableBatchScan extends AbstractDataTableScan {
long scannedRowCount = 0;
SnapshotReader.Plan plan = ((ScannedResult) result).plan();
List<DataSplit> splits = plan.dataSplits();
+ if (splits.isEmpty()) {
+ return result;
+ }
+
+ // We first add the rawConvertible split to avoid merging, and if the row count
+ // is still less than limit number, then add split which is not rawConvertible.
+ splits.sort(
+ (x, y) -> {
+ if (x.rawConvertible() && y.rawConvertible()) {
+ return 0;
+ } else if (x.rawConvertible()) {
+ return -1;
+ } else {
+ return 1;
+ }
+ });
+
+ // fast return if there is no rawConvertible split
+ if (!splits.get(0).rawConvertible()) {
+ return result;
+ }
+
List<Split> limitedSplits = new ArrayList<>();
for (DataSplit dataSplit : splits) {
long splitRowCount = getRowCountForSplit(dataSplit);
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index 8f4bc27bf..d8b66e1cd 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -34,9 +34,8 @@ class PaimonScanBuilder(table: Table)
private var localScan: Option[Scan] = None
override def pushLimit(limit: Int): Boolean = {
- if (table.primaryKeys().isEmpty) {
- pushDownLimit = Some(limit)
- }
+ // It is safe, since we will do nothing if it is the primary table and the split is not `rawConvertible`
+ pushDownLimit = Some(limit)
// just make a best effort to push down limit
false
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
index 8f62a5b33..59968b555 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/PaimonPushDownTest.scala
@@ -129,7 +129,7 @@ class PaimonPushDownTest extends PaimonSparkTestBase {
Assertions.assertEquals(1, spark.sql("SELECT * FROM T LIMIT 1").count())
}
- test("Paimon pushDown: limit for change-log tables") {
+ test("Paimon pushDown: limit for primary key table") {
spark.sql(s"""
|CREATE TABLE T (a INT, b STRING, c STRING)
|TBLPROPERTIES ('primary-key'='a')
@@ -141,8 +141,27 @@ class PaimonPushDownTest extends PaimonSparkTestBase {
val scanBuilder = getScanBuilder()
Assertions.assertTrue(scanBuilder.isInstanceOf[SupportsPushDownLimit])
- // Tables with primary keys can't support the push-down limit.
+ val dataSplitsWithoutLimit = scanBuilder.build().asInstanceOf[PaimonScan].getOriginSplits
+ Assertions.assertEquals(4, dataSplitsWithoutLimit.length)
+
+ // It still return false even it can push down limit.
Assertions.assertFalse(scanBuilder.asInstanceOf[SupportsPushDownLimit].pushLimit(1))
+
+ val dataSplitsWithLimit = scanBuilder.build().asInstanceOf[PaimonScan].getOriginSplits
+ Assertions.assertEquals(1, dataSplitsWithLimit.length)
+
+ Assertions.assertEquals(1, spark.sql("SELECT * FROM T LIMIT 1").count())
+
+ spark.sql("UPDATE T SET b = 'x' WHERE a = 1")
+
+ val scanBuilder2 = getScanBuilder()
+ val dataSplitsWithoutLimit2 = scanBuilder2.build().asInstanceOf[PaimonScan].getOriginSplits
+ Assertions.assertEquals(4, dataSplitsWithoutLimit2.length)
+
+ val dataSplitsWithLimit2 = scanBuilder2.build().asInstanceOf[PaimonScan].getOriginSplits
+ Assertions.assertEquals(4, dataSplitsWithLimit2.length)
+
+ Assertions.assertEquals(1, spark.sql("SELECT * FROM T LIMIT 1").count())
}
test("Paimon pushDown: runtime filter") {