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 41f6a77f82 [spark] Push down physical bucket predicates (#10638)
41f6a77f82 is described below
commit 41f6a77f82949ba5b31fa0a932cb2748c6ab4e91
Author: Ran Tao <[email protected]>
AuthorDate: Sun Oct 18 16:48:05 2026 +0800
[spark] Push down physical bucket predicates (#10638)
---
docs/docs/spark/sql-query.md | 19 +
.../paimon/spark/PaimonBaseScanBuilder.scala | 35 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 4 +-
.../spark/sql/PhysicalBucketPushDownTest.scala | 51 ++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../spark/sql/PhysicalBucketPushDownTest.scala | 21 +
.../spark/sql/PhysicalBucketPushDownTest.scala | 21 +
.../spark/sql/PhysicalBucketPushDownTest.scala | 21 +
.../spark/sql/PhysicalBucketPushDownTest.scala | 21 +
.../spark/sql/PhysicalBucketPushDownTest.scala | 21 +
.../paimon/spark/PaimonBaseScanBuilder.scala | 40 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 3 +-
.../apache/paimon/spark/PhysicalBucketFilter.scala | 70 +++
.../org/apache/paimon/spark/read/BaseScan.scala | 13 +-
.../paimon/spark/PhysicalBucketFilterTest.scala | 174 +++++++
.../spark/sql/PhysicalBucketPushDownTestBase.scala | 543 +++++++++++++++++++++
18 files changed, 1045 insertions(+), 21 deletions(-)
diff --git a/docs/docs/spark/sql-query.md b/docs/docs/spark/sql-query.md
index bd3a12f368..7e70553126 100644
--- a/docs/docs/spark/sql-query.md
+++ b/docs/docs/spark/sql-query.md
@@ -61,6 +61,25 @@ For example:
SELECT *, __paimon_file_path, __paimon_partition, __paimon_bucket,
__paimon_row_index FROM t;
```
+Supported equality, membership and range predicates on `__paimon_bucket` are
pushed down to
+select physical buckets before reading data files. For example, to read only
buckets 0 and 1:
+
+```sql
+SELECT * FROM t WHERE __paimon_bucket IN (0, 1);
+```
+
+Bucket IDs refer to the physical buckets of the stored files, including
historical buckets
+from partitions written with a different bucket count. Other predicates can be
combined with
+the bucket condition as usual. A supported bucket predicate combined with
other conditions
+using `AND` can still be pushed down. An `OR` expression that mixes bucket and
business
+columns, such as `__paimon_bucket = 1 OR id = 42`, is evaluated by Spark and
does not itself
+enable bucket pruning. Unsupported bucket expressions, such as
`__paimon_bucket % 2 = 0`,
+are also evaluated by Spark.
+
+Bucket predicate pushdown reduces the data files selected for reading.
Membership and range
+predicates may still require reading manifests to identify matching files;
they do not
+guarantee that manifest files are skipped.
+
### Scan Layout and Storage Partition Joins
By default, Paimon plans batch read tasks using its regular split packing. It
does not report
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
index f6e3b97c81..d4af72faee 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
@@ -23,6 +23,7 @@ import org.apache.paimon.partition.PartitionPredicate
import
org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates
import org.apache.paimon.predicate.{PartitionPredicateVisitor, Predicate,
TopN, VectorSearch}
import org.apache.paimon.predicate.FullTextSearch
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.BUCKET_COLUMN
import org.apache.paimon.table.SpecialFields.rowTypeWithRowTracking
import org.apache.paimon.table.Table
import org.apache.paimon.types.RowType
@@ -51,6 +52,7 @@ abstract class PaimonBaseScanBuilder
protected var pushedPartitionFilters: Array[PartitionPredicate] = Array.empty
protected var pushedDataFilters: Array[Predicate] = Array.empty
+ protected var pushedBucketFilters: Array[Predicate] = Array.empty
protected var pushedLimit: Option[Int] = None
protected var pushedTopN: Option[TopN] = None
protected var pushedVectorSearch: Option[VectorSearch] = None
@@ -72,6 +74,8 @@ abstract class PaimonBaseScanBuilder
val pushablePartitionDataFilters = mutable.ArrayBuffer.empty[Predicate]
val pushableDataFilters = mutable.ArrayBuffer.empty[Predicate]
val postScan = mutable.ArrayBuffer.empty[Filter]
+ val bucketFilters = mutable.ArrayBuffer.empty[Predicate]
+ val bucketConverter = new
SparkFilterConverter(PhysicalBucketFilter.rowType)
var newRowType = rowType
if (coreOptions.rowTrackingEnabled() &&
coreOptions.dataEvolutionEnabled()) {
@@ -79,22 +83,39 @@ abstract class PaimonBaseScanBuilder
}
val converter = new SparkFilterConverter(newRowType)
val partitionPredicateVisitor = new
PartitionPredicateVisitor(partitionKeys)
+ val canPushBucketFilters = PhysicalBucketFilter.supports(table)
filters.foreach {
filter =>
- val predicate = converter.convertIgnoreFailure(filter)
- if (predicate == null) {
+ val bucketPredicate =
+ if (
+ canPushBucketFilters &&
+ filter.references.nonEmpty && filter.references.forall(_ ==
BUCKET_COLUMN)
+ ) {
+ bucketConverter.convertIgnoreFailure(filter)
+ } else {
+ null
+ }
+ if (bucketPredicate != null) {
+ bucketFilters.append(bucketPredicate)
+ pushable.append(filter)
postScan.append(filter)
} else {
- pushable.append(filter)
- if (predicate.visit(partitionPredicateVisitor)) {
- pushablePartitionDataFilters.append(predicate)
- } else {
- pushableDataFilters.append(predicate)
+ val predicate = converter.convertIgnoreFailure(filter)
+ if (predicate == null) {
postScan.append(filter)
+ } else {
+ pushable.append(filter)
+ if (predicate.visit(partitionPredicateVisitor)) {
+ pushablePartitionDataFilters.append(predicate)
+ } else {
+ pushableDataFilters.append(predicate)
+ postScan.append(filter)
+ }
}
}
}
+ this.pushedBucketFilters = bucketFilters.toArray
if (pushable.nonEmpty) {
this.pushedSparkFilters = pushable.toArray
}
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index d655c483e3..991cc8b8d4 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -37,5 +37,6 @@ case class PaimonScan(
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
- preserveDataGrouping: Boolean = false)
+ preserveDataGrouping: Boolean = false,
+ override val pushedBucketFilters: Seq[Predicate] = Seq.empty)
extends PaimonBaseScan(table) {}
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index e451a433cd..7e465a06c3 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -59,6 +59,8 @@ class PaimonScanBuilder(val table: InnerTable) extends
PaimonBaseScanBuilder {
pushedTopN,
vectorSearch,
hybridSearch,
- fullTextSearch)
+ fullTextSearch,
+ pushedBucketFilters = pushedBucketFilters.toList
+ )
}
}
diff --git
a/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..a423b29c03
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.2/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.spark.sql
+
+import org.apache.paimon.spark.{PhysicalBucketFilter, SparkFilterConverter}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.BUCKET_COLUMN
+
+import org.apache.spark.sql.sources.{And, EqualNullSafe, EqualTo, In, Not, Or}
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {
+
+ test("preserve NULL semantics in V1 compound bucket filters") {
+ // SQL may fold these predicates before pushdown, so exercise the V1
converter directly.
+ val converter = new SparkFilterConverter(PhysicalBucketFilter.rowType)
+ val in = In(BUCKET_COLUMN, Array[Any](1, null))
+ val equalNull = EqualTo(BUCKET_COLUMN, null)
+ val nullSafe = EqualNullSafe(BUCKET_COLUMN, null)
+ val equalThree = EqualTo(BUCKET_COLUMN, 3)
+ Seq(
+ in -> Set(1),
+ Not(in) -> Set.empty[Int],
+ equalNull -> Set.empty[Int],
+ Not(equalNull) -> Set.empty[Int],
+ nullSafe -> Set.empty[Int],
+ Not(nullSafe) -> Set(-1, 0, 1, 2, 3),
+ Or(in, equalThree) -> Set(1, 3),
+ Not(Or(in, equalThree)) -> Set.empty[Int],
+ Not(And(in, equalThree)) -> Set(-1, 0, 1, 2)
+ ).foreach {
+ case (source, expected) =>
+ val filter = PhysicalBucketFilter(converter.convert(source))
+ assert((-1 to 3).filter(bucket => filter.test(bucket)).toSet ==
expected, source.toString)
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index f0476f6c49..bb674f3dd4 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -41,7 +41,8 @@ case class PaimonScan(
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
- preserveDataGrouping: Boolean = false)
+ preserveDataGrouping: Boolean = false,
+ override val pushedBucketFilters: Seq[Predicate] = Seq.empty)
extends PaimonBaseScan(table)
with SupportsReportPartitioning {
diff --git
a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..74a0c6697e
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..74a0c6697e
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {}
diff --git
a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..74a0c6697e
--- /dev/null
+++
b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {}
diff --git
a/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..74a0c6697e
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.0/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {}
diff --git
a/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
new file mode 100644
index 0000000000..74a0c6697e
--- /dev/null
+++
b/paimon-spark/paimon-spark-4.1/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTest.scala
@@ -0,0 +1,21 @@
+/*
+ * 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.spark.sql
+
+class PhysicalBucketPushDownTest extends PhysicalBucketPushDownTestBase {}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
index c249169ae7..82df5242f9 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala
@@ -23,6 +23,7 @@ import org.apache.paimon.partition.PartitionPredicate
import
org.apache.paimon.partition.PartitionPredicate.splitPartitionPredicatesAndDataPredicates
import org.apache.paimon.predicate.{PartitionPredicateVisitor, Predicate,
TopN, VectorSearch}
import org.apache.paimon.predicate.FullTextSearch
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.BUCKET_COLUMN
import org.apache.paimon.table.{SpecialFields, Table}
import org.apache.paimon.types.RowType
@@ -51,6 +52,7 @@ abstract class PaimonBaseScanBuilder
protected var pushedPartitionFilters: Array[PartitionPredicate] = Array.empty
protected var pushedDataFilters: Array[Predicate] = Array.empty
+ protected var pushedBucketFilters: Array[Predicate] = Array.empty
protected var pushedLimit: Option[Int] = None
protected var pushedTopN: Option[TopN] = None
protected var pushedVectorSearch: Option[VectorSearch] = None
@@ -67,6 +69,8 @@ abstract class PaimonBaseScanBuilder
val pushablePartitionDataFilters = mutable.ArrayBuffer.empty[Predicate]
val pushableDataFilters = mutable.ArrayBuffer.empty[Predicate]
val postScan = mutable.ArrayBuffer.empty[SparkPredicate]
+ val bucketFilters = mutable.ArrayBuffer.empty[Predicate]
+ val bucketConverter = SparkV2FilterConverter(PhysicalBucketFilter.rowType)
var newRowType = rowType
if (
@@ -77,22 +81,42 @@ abstract class PaimonBaseScanBuilder
}
val converter = SparkV2FilterConverter(newRowType)
val partitionPredicateVisitor = new
PartitionPredicateVisitor(partitionKeys)
+ val canPushBucketFilters = PhysicalBucketFilter.supports(table)
predicates.foreach {
predicate =>
- converter.convert(predicate) match {
+ val bucketPredicate =
+ if (
+ canPushBucketFilters &&
+ predicate.references().nonEmpty &&
+
predicate.references().forall(_.fieldNames().sameElements(Array(BUCKET_COLUMN)))
+ ) {
+ bucketConverter.convert(predicate)
+ } else {
+ None
+ }
+ bucketPredicate match {
case Some(paimonPredicate) =>
+ bucketFilters.append(paimonPredicate)
pushable.append(predicate)
- if (paimonPredicate.visit(partitionPredicateVisitor)) {
- pushablePartitionDataFilters.append(paimonPredicate)
- } else {
- pushableDataFilters.append(paimonPredicate)
- postScan.append(predicate)
- }
- case None =>
+ // Keep the residual so aggregate and TopN shortcuts cannot bypass
bucket filtering.
postScan.append(predicate)
+ case None =>
+ converter.convert(predicate) match {
+ case Some(paimonPredicate) =>
+ pushable.append(predicate)
+ if (paimonPredicate.visit(partitionPredicateVisitor)) {
+ pushablePartitionDataFilters.append(paimonPredicate)
+ } else {
+ pushableDataFilters.append(paimonPredicate)
+ postScan.append(predicate)
+ }
+ case None =>
+ postScan.append(predicate)
+ }
}
}
+ this.pushedBucketFilters = bucketFilters.toArray
if (pushable.nonEmpty) {
this.pushedSparkPredicates = pushable.toArray
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index a2e1ede366..83f2476a58 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -48,7 +48,8 @@ case class PaimonScan(
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
override val pushedMapSelectedKeys: Map[String, Seq[String]] = Map.empty,
- preserveDataGrouping: Boolean = false)
+ preserveDataGrouping: Boolean = false,
+ override val pushedBucketFilters: Seq[Predicate] = Seq.empty)
extends PaimonBaseScan(table)
with SupportsReportPartitioning
with SupportsReportOrdering {
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 03e4ad8c69..8f6d56d80f 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
@@ -182,7 +182,8 @@ class PaimonScanBuilder(val table: InnerTable)
hybridSearch,
fullTextSearch,
acceptedVariantExtractions,
- preserveDataGrouping = preserveDataGrouping
+ preserveDataGrouping = preserveDataGrouping,
+ pushedBucketFilters = pushedBucketFilters.toList
)
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PhysicalBucketFilter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PhysicalBucketFilter.scala
new file mode 100644
index 0000000000..c7fe06df55
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PhysicalBucketFilter.scala
@@ -0,0 +1,70 @@
+/*
+ * 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.spark
+
+import org.apache.paimon.data.GenericRow
+import org.apache.paimon.predicate.{Equal, LeafPredicate, Predicate,
PredicateBuilder}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.BUCKET_COLUMN
+import org.apache.paimon.table.{FileStoreTable, Table}
+import org.apache.paimon.table.source.ReadBuilder
+import org.apache.paimon.types.{DataType, DataTypes, RowType}
+import org.apache.paimon.utils.Filter
+
+import scala.collection.JavaConverters._
+
+/** Evaluates metadata predicates on physical bucket IDs, independently of the
current bucket count. */
+private[spark] case class PhysicalBucketFilter(predicate: Predicate)
+ extends Filter[Integer]
+ with Serializable {
+
+ override def test(bucket: Integer): Boolean =
predicate.test(GenericRow.of(bucket))
+}
+
+private[spark] object PhysicalBucketFilter {
+
+ val rowType: RowType =
+ RowType.of(Array[DataType](DataTypes.INT()), Array(BUCKET_COLUMN))
+
+ def supports(table: Table): Boolean = {
+ table.isInstanceOf[FileStoreTable] &&
+ !table.rowType().containsField(BUCKET_COLUMN) &&
+ !PostponeMergeOnRead.usesCustomSource(table)
+ }
+
+ def push(readBuilder: ReadBuilder, predicates: Seq[Predicate], scanBucket:
Option[Int]): Unit = {
+ if (predicates.nonEmpty) {
+ val predicate = PredicateBuilder.and(predicates.asJava)
+ // A plain equality can also use the existing single-bucket manifest
pruning.
+ PredicateBuilder
+ .splitAnd(predicate)
+ .asScala
+ .collectFirst {
+ case leaf: LeafPredicate
+ if leaf.function() == Equal.INSTANCE &&
leaf.fieldRefOptional().isPresent &&
+ leaf.literals().get(0).isInstanceOf[Integer] =>
+ leaf.literals().get(0).asInstanceOf[Integer].intValue()
+ }
+ // Preserve scan.bucket when it conflicts with the equality; the full
filter below
+ // intersects with the configured bucket and produces an empty scan.
+ .filter(bucket => scanBucket.forall(_ == bucket))
+ .foreach(readBuilder.withBucket)
+ readBuilder.withBucketFilter(PhysicalBucketFilter(predicate))
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
index b7854ea4e2..e2dd7e6ace 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
@@ -21,7 +21,7 @@ package org.apache.paimon.spark.read
import org.apache.paimon.CoreOptions
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, Predicate,
TopN, VectorSearch}
-import org.apache.paimon.spark.{PaimonBatch, PaimonInputPartition,
PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric,
PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric,
SparkTypeUtils}
+import org.apache.paimon.spark.{PaimonBatch, PaimonInputPartition,
PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric,
PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric,
PhysicalBucketFilter, SparkTypeUtils}
import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.spark.schema.PaimonMetadataColumn._
import org.apache.paimon.spark.util.{OptionUtils, SplitUtils}
@@ -49,6 +49,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
// Push down
def pushedPartitionFilters: Seq[PartitionPredicate]
def pushedDataFilters: Seq[Predicate]
+ def pushedBucketFilters: Seq[Predicate] = Seq.empty
def pushedLimit: Option[Int] = None
def pushedTopN: Option[TopN] = None
def pushedVectorSearch: Option[VectorSearch] = None
@@ -127,6 +128,10 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
if (pushedDataFilters.nonEmpty) {
_readBuilder.withFilter(pushedDataFilters.asJava)
}
+ PhysicalBucketFilter.push(
+ _readBuilder,
+ pushedBucketFilters,
+ Option(coreOptions.scanBucket()).map(_.intValue()))
pushedLimit.foreach(_readBuilder.withLimit)
pushedTopN.foreach(_readBuilder.withTopN)
_readBuilder.dropStats()
@@ -210,6 +215,11 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
} else {
""
}
+ val pushedBucketFiltersStr = if (pushedBucketFilters.nonEmpty) {
+ s", BucketFilters: [${pushedBucketFilters.mkString(",")}]"
+ } else {
+ ""
+ }
val pushedVariantsStr =
if (pushedVariantExtractions.isEmpty) ""
else
@@ -228,6 +238,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
pushedPartitionFiltersStr +
pushedRuntimePartitionFiltersStr +
pushedDataFiltersStr +
+ pushedBucketFiltersStr +
pushedTopN.map(topN => s", TopN: [$topN]").getOrElse("") +
pushedLimit.map(limit => s", Limit: [$limit]").getOrElse("") +
pushedVectorSearch.map(vs => s", VectorSearch: [$vs]").getOrElse("") +
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PhysicalBucketFilterTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PhysicalBucketFilterTest.scala
new file mode 100644
index 0000000000..65d3431029
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/PhysicalBucketFilterTest.scala
@@ -0,0 +1,174 @@
+/*
+ * 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.spark
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.predicate.PredicateBuilder
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.BUCKET_COLUMN
+import org.apache.paimon.table.{BucketMode, FileStoreTable, InnerTable}
+import org.apache.paimon.table.source.ReadBuilder
+import org.apache.paimon.types.{DataField, DataTypes, RowType}
+import org.apache.paimon.utils.InstantiationUtil
+
+import org.apache.spark.sql.connector.expressions.{Expression, Expressions,
Literal}
+import org.apache.spark.sql.connector.expressions.filter.{And, Not, Or,
Predicate => SparkPredicate}
+import org.apache.spark.sql.types.{DataType, IntegerType}
+import org.mockito.ArgumentMatchers.anyInt
+import org.mockito.Mockito.{mock, never, verify, when}
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+
+class PhysicalBucketFilterTest extends AnyFunSuite {
+
+ private val bucket = Expressions.column(BUCKET_COLUMN)
+ private val converter = SparkV2FilterConverter(PhysicalBucketFilter.rowType)
+
+ private def predicate(name: String, values: Integer*): SparkPredicate = {
+ val literals = values.map {
+ v =>
+ new Literal[Integer] {
+ override def value(): Integer = v
+ override def dataType(): DataType = IntegerType
+ override def children(): Array[Expression] = Array.empty
+ override def describe(): String = String.valueOf(v)
+ }
+ }
+ new SparkPredicate(name, Array[Expression](bucket) ++ literals)
+ }
+
+ test("preserve NULL semantics and serialize compound bucket filters") {
+ val in = predicate("IN", 1, null)
+ val equalNull = predicate("=", null)
+ val nullSafe = predicate("<=>", null)
+ Seq(
+ in -> Set(1),
+ new Not(in) -> Set.empty[Int],
+ equalNull -> Set.empty[Int],
+ new Not(equalNull) -> Set.empty[Int],
+ nullSafe -> Set.empty[Int],
+ new Not(nullSafe) -> Set(-1, 0, 1, 2, 3),
+ new Or(in, predicate("=", 3)) -> Set(1, 3),
+ new Not(new Or(in, predicate("=", 3))) -> Set.empty[Int],
+ new Not(new And(in, predicate("=", 3))) -> Set(-1, 0, 1, 2)
+ ).foreach {
+ case (source, expected) =>
+ val filter = PhysicalBucketFilter(converter.convert(source).get)
+ assert((-1 to 3).filter(b => filter.test(b)).toSet == expected,
source.toString)
+ }
+
+ val filter = PhysicalBucketFilter(converter.convert(new Or(in,
predicate("=", 3))).get)
+ val restored = InstantiationUtil.deserializeObject[PhysicalBucketFilter](
+ InstantiationUtil.serializeObject(filter),
+ getClass.getClassLoader)
+ assert(restored == filter)
+ assert((-1 to 3).filter(b => restored.test(b)).toSet == Set(1, 3))
+ }
+
+ test("support negative and boundary bucket IDs in IN filters") {
+ val filter =
+ PhysicalBucketFilter(
+ converter.convert(predicate("IN", Int.MinValue, -2, 3,
Int.MaxValue)).get)
+ Seq(Int.MinValue, -2, 3, Int.MaxValue).foreach(b => assert(filter.test(b)))
+ Seq(-1, 0, 2, 4).foreach(b => assert(!filter.test(b)))
+ }
+
+ test("use withBucket only for compatible equality predicates and retain the
full filter") {
+ // (predicates, configured scan.bucket, expected withBucket argument)
+ Seq(
+ (Seq(predicate("=", 1)), None, Some(1)),
+ (Seq(predicate("=", 1)), Some(1), Some(1)),
+ (Seq(predicate("<=>", 1)), None, Some(1)),
+ (Seq(new And(predicate(">=", 0), predicate("=", 1))), None, Some(1)),
+ (Seq(predicate(">=", 0), predicate("=", 1)), None, Some(1)),
+ (Seq(predicate("IN", 1, 3)), None, None),
+ (Seq(new And(predicate(">=", 1), predicate("<", 3))), None, None),
+ (Seq(new Or(predicate("=", 1), predicate("=", 3))), None, None),
+ (Seq(predicate("=", null)), None, None),
+ (Seq(predicate("<=>", null)), None, None),
+ (Seq(predicate("=", 1)), Some(0), None),
+ (Seq(predicate("<=>", 1)), Some(0), None)
+ ).foreach {
+ case (sources, scanBucket, expectedBucket) =>
+ withClue(s"$sources, scan.bucket=$scanBucket: ") {
+ val readBuilder = mock(classOf[ReadBuilder])
+ val predicates = sources.map(source => converter.convert(source).get)
+ PhysicalBucketFilter.push(readBuilder, predicates, scanBucket)
+
+ expectedBucket match {
+ case Some(bucket) => verify(readBuilder).withBucket(bucket)
+ case None => verify(readBuilder, never()).withBucket(anyInt())
+ }
+ verify(readBuilder)
+
.withBucketFilter(PhysicalBucketFilter(PredicateBuilder.and(predicates.asJava)))
+ }
+ }
+ }
+
+ test("leave bucket filters unpushed for non-file-store tables") {
+ val builder = scanBuilder(mock(classOf[InnerTable]), Map.empty)
+ val predicates = Array(predicate("=", 1), predicate("IN", 1, 3),
predicate(">=", 1))
+ assert(builder.pushPredicates(predicates).sameElements(predicates))
+ assert(builder.pushedPredicates.isEmpty)
+ val scan = builder.build().asInstanceOf[PaimonScan]
+ assert(scan.pushedBucketFilters.isEmpty)
+ assert(scan.pushedDataFilters.isEmpty)
+ }
+
+ test("push bucket filters only when postpone merge uses the standard scan") {
+ Seq(
+ (false, "latest-full", true),
+ (true, "latest-full", false),
+ (true, "compacted-full", true)
+ ).foreach {
+ case (mergeOnRead, scanMode, canPush) =>
+ val options = Map(
+ "bucket" -> "-2",
+ "postpone.merge-on-read" -> mergeOnRead.toString,
+ "scan.mode" -> scanMode)
+ val table = mock(classOf[FileStoreTable])
+ val builder = scanBuilder(table, options)
+
when(table.coreOptions()).thenReturn(CoreOptions.fromMap(options.asJava))
+ when(table.bucketMode()).thenReturn(BucketMode.POSTPONE_MODE)
+ when(table.primaryKeys()).thenReturn(Collections.singletonList("id"))
+
+ val predicates = Array(predicate("=", 1), predicate("IN", 1, 3),
predicate(">=", 1))
+ assert(builder.pushPredicates(predicates).sameElements(predicates))
+ val scan = builder.build().asInstanceOf[PaimonScan]
+ assert(scan.pushedDataFilters.isEmpty)
+ if (canPush) {
+ assert(builder.pushedPredicates.sameElements(predicates))
+ assert(scan.pushedBucketFilters.size == predicates.length)
+ } else {
+ assert(builder.pushedPredicates.isEmpty)
+ assert(scan.pushedBucketFilters.isEmpty)
+ }
+ }
+ }
+
+ private def scanBuilder(table: InnerTable, options: Map[String, String]):
PaimonScanBuilder = {
+ when(table.options()).thenReturn(options.asJava)
+ when(table.rowType()).thenReturn(RowType.of(new DataField(0, "id",
DataTypes.INT())))
+ when(table.partitionKeys()).thenReturn(Collections.emptyList[String]())
+ when(table.primaryKeys()).thenReturn(Collections.emptyList[String]())
+ new PaimonScanBuilder(table)
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTestBase.scala
new file mode 100644
index 0000000000..294fce08e2
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PhysicalBucketPushDownTestBase.scala
@@ -0,0 +1,543 @@
+/*
+ * 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.spark.sql
+
+import org.apache.paimon.spark.{PaimonScan, PaimonSparkTestBase}
+import org.apache.paimon.table.FallbackReadFileStoreTable.FallbackDataSplit
+import org.apache.paimon.table.source.DataSplit
+
+import org.apache.spark.sql.{DataFrame, Row}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
+import org.apache.spark.sql.functions.expr
+
+import scala.collection.JavaConverters._
+
+abstract class PhysicalBucketPushDownTestBase extends PaimonSparkTestBase {
+
+ Seq(false, true).foreach {
+ hasPrimaryKey =>
+ test(s"physical bucket predicates prune data files:
hasPrimaryKey=$hasPrimaryKey") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey)
+ if (hasPrimaryKey) {
+ sql("INSERT INTO T SELECT CAST(id AS INT), 'updated' FROM range(0,
100)")
+ }
+ val all = sql("SELECT id, payload, __paimon_bucket FROM T")
+ val reference = materialize(all)
+ val allSplits =
scanOf(all).inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(allSplits.map(_.bucket()).toSet == Set(0, 1, 2, 3))
+
+ // Exercise SQL conversion once; the unit tests cover the
NULL/boolean matrix.
+ val extraConditions = if (hasPrimaryKey) {
+ Seq.empty
+ } else {
+ // Type NULL explicitly to preserve metadata columns in older
Spark analyzers.
+ Seq(
+ "__paimon_bucket IN (1, CAST(NULL AS INT), 3)" -> Set(1, 3),
+ "__paimon_bucket <=> 1" -> Set(1),
+ "__paimon_bucket NOT IN (1, 3)" -> Set(0, 2),
+ "__paimon_bucket = 1 OR __paimon_bucket = 3" -> Set(1, 3),
+ "__paimon_bucket = 1 OR id = 42" -> Set(0, 1, 2, 3)
+ )
+ }
+ val conditions = Seq(
+ "__paimon_bucket = 1" -> Set(1),
+ "__paimon_bucket IN (1, 3)" -> Set(1, 3),
+ "__paimon_bucket >= 1 AND __paimon_bucket < 3" -> Set(1, 2),
+ "__paimon_bucket = 99" -> Set.empty[Int]
+ ) ++ extraConditions
+ conditions.foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"SELECT id, payload, __paimon_bucket FROM T WHERE
$condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ assertBucketFiles(scan, expectedBuckets, allSplits)
+ }
+
+ if (!hasPrimaryKey) {
+ // Isolate the unsupported expression from inferred IS NOT NULL
predicates, which can
+ // independently be pushed down as bucket filters.
+ withSparkSQLConf("spark.sql.constraintPropagation.enabled" ->
"false") {
+ val condition = "__paimon_bucket % 2 = 0"
+ val query = s"SELECT id, payload, __paimon_bucket FROM T WHERE
$condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ assert(scan.pushedBucketFilters.isEmpty)
+ assertBucketFiles(scan, Set(0, 1, 2, 3), allSplits)
+ }
+ }
+
+ if (hasPrimaryKey) {
+ val condition = "__paimon_bucket = 1 AND payload = 'updated'"
+ val query = s"SELECT id, payload, __paimon_bucket FROM T WHERE
$condition"
+ val expected = reference.filter(expr(condition)).collect().toSeq
+ assert(expected.nonEmpty)
+ val actual = sql(query)
+ checkAnswer(actual, expected)
+ val scan = scanOf(actual)
+ assert(scan.pushedBucketFilters.nonEmpty)
+ assert(scan.pushedDataFilters.nonEmpty)
+
assert(scan.inputSplits.map(_.asInstanceOf[DataSplit].bucket()).toSet == Set(1))
+ }
+
+ // Filtering must also work when the metadata column is not
selected, and must not be
+ // bypassed by metadata-only aggregation or TopN/limit pushdown.
+ val selected = reference.filter("__paimon_bucket = 1")
+ Seq(
+ "SELECT id FROM T WHERE __paimon_bucket = 1" ->
selected.select("id"),
+ "SELECT count(*) FROM T WHERE __paimon_bucket = 1" ->
selected.selectExpr("count(*)"),
+ "SELECT id FROM T WHERE __paimon_bucket = 1 ORDER BY id LIMIT 5" ->
+ selected.select("id").orderBy("id").limit(5)
+ ).foreach {
+ case (query, expected) =>
+ val scan = checkAnswerAndGetScan(query, expected)
+ assert(scan.pushedBucketFilters.nonEmpty)
+ assert(scan.pushedTopN.isEmpty)
+ assertBucketFiles(scan, Set(1), allSplits)
+ }
+
+ if (!hasPrimaryKey) {
+ val bucketOneScan = getPaimonScan("SELECT id FROM T WHERE
__paimon_bucket = 1")
+ val bucketThreeScan = getPaimonScan("SELECT id FROM T WHERE
__paimon_bucket = 3")
+ val copiedScan = bucketOneScan.copy()
+ assert(bucketOneScan != bucketThreeScan)
+ assert(copiedScan == bucketOneScan)
+ assertBucketFiles(copiedScan, Set(1), allSplits)
+ assert(bucketOneScan.description().contains("BucketFilters:"))
+ }
+ }
+ }
+
+ test(
+ s"physical bucket predicates intersect bucket-key equality:
hasPrimaryKey=$hasPrimaryKey") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey)
+ val reference = materialize(sql("SELECT id, __paimon_bucket FROM T"))
+ val expectedKey = reference.filter("id = 42")
+ val keyBucket =
expectedKey.select("__paimon_bucket").head().getInt(0)
+ val otherBuckets = (0 until 4).filter(_ != keyBucket)
+ val keyQuery = "SELECT id, __paimon_bucket FROM T WHERE id = 42"
+ val keySplits =
getPaimonScan(keyQuery).inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(keySplits.map(_.bucket()).toSet == Set(keyBucket))
+
+ Seq(
+ s"__paimon_bucket = $keyBucket" -> Set(keyBucket),
+ s"__paimon_bucket = ${otherBuckets.head}" -> Set.empty[Int],
+ s"__paimon_bucket IN ($keyBucket, ${otherBuckets.head})" ->
Set(keyBucket),
+ s"__paimon_bucket IN (${otherBuckets(0)}, ${otherBuckets(1)})" ->
Set.empty[Int]
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"$keyQuery AND $condition"
+ val scan = checkAnswerAndGetScan(query,
expectedKey.filter(expr(condition)))
+ assert(scan.pushedBucketFilters.nonEmpty, condition)
+ assert(scan.pushedDataFilters.nonEmpty, condition)
+ assertBucketFiles(scan, expectedBuckets, keySplits)
+ }
+ }
+ }
+
+ test(s"physical bucket predicates with unordered limit:
hasPrimaryKey=$hasPrimaryKey") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey)
+ val reference = materialize(sql("SELECT id, __paimon_bucket FROM T"))
+ assert(reference.filter("__paimon_bucket = 1").count() > 5)
+
+ Seq(1, 99).foreach {
+ bucket =>
+ val query = s"SELECT id FROM T WHERE __paimon_bucket = $bucket
LIMIT 5"
+ val expected = reference
+ .filter(s"__paimon_bucket = $bucket")
+ .select("id")
+ .collect()
+ .toSet
+ val result = sql(query)
+ val actual = result.collect().toSeq
+ // LIMIT without ORDER BY may return any subset of the matching
rows.
+ assert(actual.size == math.min(5, expected.size), query)
+ assert(actual.distinct.size == actual.size, query)
+ assert(actual.forall(expected.contains), query)
+
+ val scan = scanOf(result)
+ assert(scan.pushedBucketFilters.nonEmpty, query)
+ val splits = scan.inputSplits.map(_.asInstanceOf[DataSplit])
+ val expectedBuckets = if (expected.isEmpty) Set.empty[Int] else
Set(bucket)
+ assert(splits.map(_.bucket()).toSet == expectedBuckets, query)
+ }
+ }
+ }
+ }
+
+ test("physical bucket predicates prune files with unsupported residual
expressions") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey = false)
+ val all = sql("SELECT id, payload, __paimon_bucket FROM T")
+ val reference = materialize(all)
+ val allSplits = scanOf(all).inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(allSplits.map(_.bucket()).toSet == Set(0, 1, 2, 3))
+
+ // Disable inferred IS NOT NULL filters so only the bucket predicate can
be pushed down.
+ withSparkSQLConf("spark.sql.constraintPropagation.enabled" -> "false") {
+ val bucketCondition = "__paimon_bucket IN (1, 3)"
+ val bucketRows = reference.filter(bucketCondition).collect().toSeq
+ val expected = bucketRows.filter(_.getInt(0) % 2 == 0)
+ assert(expected.nonEmpty && expected.size < bucketRows.size)
+
+ val actual =
+ sql(s"SELECT id, payload, __paimon_bucket FROM T WHERE
$bucketCondition AND id % 2 = 0")
+ checkAnswer(actual, expected)
+ val scan = scanOf(actual)
+ assert(scan.pushedBucketFilters.nonEmpty)
+ assert(scan.pushedDataFilters.isEmpty)
+ assertBucketFiles(scan, Set(1, 3), allSplits)
+ }
+ }
+ }
+
+ if (gteqSpark3_3) {
+ test("physical bucket predicates preserve grouped aggregates and joins") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey = true)
+ sql("INSERT INTO T SELECT CAST(id AS INT), 'updated' FROM range(100)")
+ val all = sql("SELECT id, payload, __paimon_bucket FROM T")
+ val reference = materialize(all)
+ val allSplits = scanOf(all).inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(allSplits.map(_.bucket()).toSet == Set(0, 1, 2, 3))
+
+ withSparkSQLConf(
+ "spark.paimon.scan.preserve-data-grouping" -> "true",
+ "spark.sql.sources.v2.bucketing.enabled" -> "true",
+ "spark.sql.adaptive.enabled" -> "false",
+ "spark.sql.autoBroadcastJoinThreshold" -> "-1"
+ ) {
+ Seq(
+ "__paimon_bucket IN (1, 3)" -> Set(1, 3),
+ "__paimon_bucket = 99" -> Set.empty[Int]
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"SELECT id, count(*) FROM T WHERE $condition GROUP
BY id"
+ val scan =
+ checkAnswerAndGetScan(
+ query,
+ reference.filter(expr(condition)).groupBy("id").count())
+ assert(scan.pushedBucketFilters.nonEmpty, condition)
+ assertBucketFiles(scan, expectedBuckets, allSplits)
+ assert(scan.inputPartitions.nonEmpty ==
expectedBuckets.nonEmpty, condition)
+ assert(scan.inputPartitions.forall(_.bucketed), condition)
+ }
+
+ // The two scans retain different bucket sets, so the join must
align matching keys.
+ // Spark 3.4+ can fill missing bucket partitions without shuffling
either side.
+ val pushPartValuesOptions = if (gteqSpark3_4) Seq(false, true) else
Seq(false)
+ pushPartValuesOptions.foreach {
+ pushPartValues =>
+ withSparkSQLConf(
+ "spark.sql.sources.v2.bucketing.pushPartValues.enabled" ->
pushPartValues.toString) {
+ val joined =
+ sql("""SELECT l.id, l.payload, r.payload FROM T l JOIN T r
ON l.id = r.id
+ |WHERE l.__paimon_bucket IN (1, 3) AND
r.__paimon_bucket IN (0, 1)
+ |""".stripMargin)
+ checkAnswer(
+ joined,
+ reference.filter("__paimon_bucket = 1").selectExpr("id",
"payload", "payload"))
+ val plan = joined.queryExecution.executedPlan
+ val shuffles = plan.collect { case exchange:
ShuffleExchangeLike => exchange }
+ assert(
+ shuffles.isEmpty == pushPartValues,
+ s"pushPartValues=$pushPartValues:\n$plan")
+
+ val scans = joined.queryExecution.optimizedPlan.collect {
+ case relation: DataSourceV2ScanRelation =>
relation.scan.asInstanceOf[PaimonScan]
+ }
+ assert(scans.size == 2)
+ val scanBuckets = scans.map {
+ scan => (scan,
scan.inputSplits.map(_.asInstanceOf[DataSplit].bucket()).toSet)
+ }
+ assert(scanBuckets.map(_._2).toSet == Set(Set(1, 3), Set(0,
1)))
+ scanBuckets.foreach {
+ case (scan, buckets) =>
+ assert(scan.pushedBucketFilters.nonEmpty)
+ assertBucketFiles(scan, buckets, allSplits)
+ assert(scan.inputPartitions.nonEmpty)
+ assert(scan.inputPartitions.forall(_.bucketed))
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("physical bucket filters preserve partition filters and historical
bucket IDs") {
+ withTable("T") {
+ sql("""CREATE TABLE T (id INT, p INT) PARTITIONED BY (p)
+ |TBLPROPERTIES ('bucket' = '4', 'bucket-key' =
'id')""".stripMargin)
+ sql("INSERT INTO T SELECT CAST(id AS INT), 0 FROM range(400)")
+ sql("ALTER TABLE T SET TBLPROPERTIES ('bucket' = '2')")
+ sql("INSERT INTO T SELECT CAST(id AS INT), 1 FROM range(400)")
+ val all = sql("SELECT id, p, __paimon_bucket FROM T")
+ val reference = materialize(all)
+ val allSplits = scanOf(all).inputSplits.map(_.asInstanceOf[DataSplit])
+ Seq(
+ "__paimon_bucket = 3" -> Set((0, 3)),
+ "__paimon_bucket IN (1, 3)" -> Set((0, 1), (0, 3), (1, 1)),
+ "__paimon_bucket >= 2" -> Set((0, 2), (0, 3))
+ ).foreach {
+ case (condition, expectedPartitionBuckets) =>
+ val query = s"SELECT id, p, __paimon_bucket FROM T WHERE $condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ val splits = scan.inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(
+ splits
+ .map(s => (s.partition().getInt(0), s.bucket()))
+ .toSet == expectedPartitionBuckets,
+ condition)
+ val expectedSplits = allSplits.filter(
+ s => expectedPartitionBuckets.contains((s.partition().getInt(0),
s.bucket())))
+ assert(dataFileNames(splits) == dataFileNames(expectedSplits),
condition)
+ }
+
+ val query =
+ "SELECT id, p, __paimon_bucket FROM T WHERE p = 1 AND __paimon_bucket
= 1 AND id >= 100"
+ val scan =
+ checkAnswerAndGetScan(
+ query,
+ reference.filter(expr("p = 1 AND __paimon_bucket = 1 AND id >=
100")))
+ val splits = scan.inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(splits.nonEmpty)
+ assert(splits.forall(s => s.bucket() == 1 && s.partition().getInt(0) ==
1))
+ }
+ }
+
+ test("physical bucket predicates preserve fallback partition ownership") {
+ withTable("T") {
+ sql("""CREATE TABLE T (id INT, p INT, payload STRING) PARTITIONED BY (p)
+ |TBLPROPERTIES ('bucket' = '4', 'bucket-key' =
'id')""".stripMargin)
+ sql("CALL sys.create_branch('test.T', 'fallback')")
+ sql("""INSERT INTO `T$branch_fallback`
+ |SELECT CAST(ids.id AS INT), CAST(parts.id AS INT), 'fallback'
+ |FROM range(100) ids CROSS JOIN range(3) parts""".stripMargin)
+ val fallback =
+ materialize(sql("SELECT id, p, payload, __paimon_bucket FROM
`T$branch_fallback`"))
+
+ sql("INSERT INTO T SELECT CAST(id AS INT), 0, 'main' FROM range(100)")
+ // The main branch owns partition 2, but has no rows in bucket 1. Bucket
pruning must not
+ // make it fall back to the older partition, which does contain rows in
that bucket.
+ val bucketZeroId =
+ fallback.filter("p = 2 AND __paimon_bucket =
0").select("id").head().getInt(0)
+ sql(s"INSERT INTO T VALUES ($bucketZeroId, 2, 'main')")
+ val main = materialize(sql("SELECT id, p, payload, __paimon_bucket FROM
T"))
+ val reference = main.union(fallback.filter("p = 1"))
+ assert(fallback.filter("p = 2 AND __paimon_bucket = 1").count() > 0)
+
+ sql("ALTER TABLE T SET TBLPROPERTIES ('scan.fallback-branch' =
'fallback')")
+ val query = "SELECT id, p, payload, __paimon_bucket FROM T"
+ val allSplits =
+ checkAnswerAndGetScan(query,
reference).inputSplits.map(_.asInstanceOf[FallbackDataSplit])
+ assert(
+ allSplits.filter(!_.isFallback).map(s => (s.partition().getInt(0),
s.bucket())).toSet ==
+ (0 until 4).map(b => (0, b)).toSet + ((2, 0)))
+ assert(
+ allSplits.filter(_.isFallback).map(s => (s.partition().getInt(0),
s.bucket())).toSet ==
+ (0 until 4).map(b => (1, b)).toSet)
+
+ Seq(
+ "__paimon_bucket = 1" -> Set(1),
+ "__paimon_bucket IN (1, 3)" -> Set(1, 3)
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val scan =
+ checkAnswerAndGetScan(s"$query WHERE $condition",
reference.filter(expr(condition)))
+ assert(scan.pushedBucketFilters.nonEmpty, condition)
+ val splits = scan.inputSplits.map(_.asInstanceOf[FallbackDataSplit])
+ val expectedSplits = allSplits.filter(s =>
expectedBuckets.contains(s.bucket()))
+ assert(
+ splits.map(s => (s.partition().getInt(0), s.bucket(),
s.isFallback)).toSet ==
+ expectedSplits.map(s => (s.partition().getInt(0), s.bucket(),
s.isFallback)).toSet,
+ condition
+ )
+ assert(dataFileNames(splits) == dataFileNames(expectedSplits),
condition)
+ }
+
+ val missingBucket = sql(s"$query WHERE p = 2 AND __paimon_bucket = 1")
+ checkAnswer(missingBucket, Seq.empty[Row])
+ assert(scanOf(missingBucket).inputSplits.isEmpty)
+ }
+ }
+
+ test("physical bucket predicates preserve the configured scan bucket") {
+ withTable("T") {
+ createFixedBucketTable(hasPrimaryKey = true)
+ sql("ALTER TABLE T SET TBLPROPERTIES ('scan.bucket' = '0')")
+ val reference = materialize(sql("SELECT id, __paimon_bucket FROM T"))
+ assert(reference.collect().map(_.getInt(1)).toSet == Set(0))
+
+ Seq(
+ "__paimon_bucket = 0" -> Set(0),
+ "__paimon_bucket = 1" -> Set.empty[Int],
+ "__paimon_bucket IN (0, 1)" -> Set(0),
+ "__paimon_bucket IN (1, 3)" -> Set.empty[Int]
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"SELECT id, __paimon_bucket FROM T WHERE $condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ val splits = scan.inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(splits.map(_.bucket()).toSet == expectedBuckets, condition)
+ }
+ }
+ }
+
+ Seq(false, true).foreach {
+ hasPrimaryKey =>
+ test(s"physical bucket predicates with bucket=-1:
hasPrimaryKey=$hasPrimaryKey") {
+ withTable("T") {
+ val primaryKeyOptions = if (hasPrimaryKey) {
+ """, 'primary-key' = 'id', 'dynamic-bucket.initial-buckets' = '3',
+ |'dynamic-bucket.assigner-parallelism' = '3'""".stripMargin
+ } else {
+ ""
+ }
+ sql(s"""CREATE TABLE T (id INT, payload INT)
+ |TBLPROPERTIES ('bucket' = '-1'
$primaryKeyOptions)""".stripMargin)
+ sql("INSERT INTO T SELECT CAST(id AS INT), CAST(id AS INT) FROM
range(100)")
+ if (hasPrimaryKey) {
+ sql("INSERT INTO T SELECT CAST(id AS INT), -1 FROM range(10)")
+ }
+
+ val all = sql("SELECT id, payload, __paimon_bucket FROM T")
+ val reference = materialize(all)
+ val allSplits =
scanOf(all).inputSplits.map(_.asInstanceOf[DataSplit])
+ val physicalBuckets = allSplits.map(_.bucket()).toSet
+ // bucket=-1 selects the bucket mode; stored files still have
non-negative physical IDs.
+ if (hasPrimaryKey) {
+ assert(physicalBuckets.size > 1 && physicalBuckets.forall(_ >= 0))
+ } else {
+ assert(physicalBuckets == Set(0))
+ }
+ assert(reference.collect().map(_.getInt(2)).toSet == physicalBuckets)
+ val selectedBucket = physicalBuckets.min
+ Seq(
+ s"__paimon_bucket = $selectedBucket" -> Set(selectedBucket),
+ s"__paimon_bucket > $selectedBucket" -> (physicalBuckets -
selectedBucket),
+ "__paimon_bucket = -1" -> Set.empty[Int]
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"SELECT id, payload, __paimon_bucket FROM T WHERE
$condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ assert(scan.pushedBucketFilters.nonEmpty, condition)
+ assertBucketFiles(scan, expectedBuckets, allSplits)
+ }
+ }
+ }
+ }
+
+ test("physical bucket predicates prune postpone tables in compacted-full
mode") {
+ withTable("T") {
+ sql("""CREATE TABLE T (id INT, payload STRING) TBLPROPERTIES (
+ |'primary-key' = 'id', 'bucket-key' = 'id', 'bucket' = '-2',
+ |'postpone.default-bucket-num' = '4',
+ |'postpone.batch-write-fixed-bucket' = 'false')""".stripMargin)
+ sql("INSERT INTO T SELECT CAST(id AS INT), 'compacted' FROM range(400)")
+ checkAnswer(sql("SELECT DISTINCT bucket FROM `T$buckets`"), Seq(Row(-2)))
+ sql("CALL sys.compact(table => 'T')")
+
+ val reference = materialize(sql("SELECT id, payload, __paimon_bucket
FROM T"))
+ checkAnswer(reference.select("id", "payload"), (0 until 400).map(id =>
Row(id, "compacted")))
+
+ // Leave both updates and new keys in the postpone bucket after the
compacted snapshot.
+ sql("INSERT INTO T SELECT CAST(id AS INT), 'pending' FROM range(410)")
+ checkAnswer(
+ sql("SELECT DISTINCT bucket FROM `T$buckets`"),
+ Seq(Row(-2), Row(0), Row(1), Row(2), Row(3)))
+ checkAnswer(sql("SELECT record_count FROM `T$buckets` WHERE bucket =
-2"), Seq(Row(410L)))
+
+ withSparkSQLConf(
+ "spark.paimon.postpone.merge-on-read" -> "true",
+ "spark.paimon.scan.mode" -> "compacted-full") {
+ val allScan = checkAnswerAndGetScan("SELECT id, payload,
__paimon_bucket FROM T", reference)
+ assert(allScan.planPostponeMerge().isEmpty)
+ val allSplits = allScan.inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(allSplits.map(_.bucket()).toSet == Set(0, 1, 2, 3))
+
+ Seq(
+ "__paimon_bucket = 1" -> Set(1),
+ "__paimon_bucket IN (1, 3)" -> Set(1, 3),
+ "__paimon_bucket = -2" -> Set.empty[Int]
+ ).foreach {
+ case (condition, expectedBuckets) =>
+ val query = s"SELECT id, payload, __paimon_bucket FROM T WHERE
$condition"
+ val scan = checkAnswerAndGetScan(query,
reference.filter(expr(condition)))
+ assert(scan.planPostponeMerge().isEmpty, condition)
+ assert(scan.pushedBucketFilters.nonEmpty, condition)
+ assertBucketFiles(scan, expectedBuckets, allSplits)
+ }
+ }
+ }
+ }
+
+ test("a business column named __paimon_bucket is not a physical bucket
predicate") {
+ withTable("T") {
+ sql("""CREATE TABLE T (id INT, __paimon_bucket INT)
+ |TBLPROPERTIES ('bucket' = '4', 'bucket-key' =
'id')""".stripMargin)
+ sql("INSERT INTO T SELECT CAST(id AS INT), 99 FROM range(100)")
+ val actual = sql("SELECT * FROM T WHERE __paimon_bucket = 99")
+ assert(actual.collect().length == 100)
+ val scan = scanOf(actual)
+ assert(scan.pushedBucketFilters.isEmpty)
+ assert(scan.inputSplits.map(_.asInstanceOf[DataSplit].bucket()).toSet ==
Set(0, 1, 2, 3))
+ }
+ }
+
+ private def checkAnswerAndGetScan(query: String, expected: DataFrame):
PaimonScan = {
+ val actual = sql(query)
+ checkAnswer(actual, expected)
+ scanOf(actual)
+ }
+
+ private def scanOf(df: DataFrame): PaimonScan = {
+ df.queryExecution.optimizedPlan.collectFirst {
+ case relation: DataSourceV2ScanRelation =>
relation.scan.asInstanceOf[PaimonScan]
+ }.get
+ }
+
+ private def createFixedBucketTable(hasPrimaryKey: Boolean): Unit = {
+ val primaryKeyOption = if (hasPrimaryKey) ", 'primary-key' = 'id'" else ""
+ sql(s"""CREATE TABLE T (id INT, payload STRING)
+ |TBLPROPERTIES ('bucket' = '4', 'bucket-key' = 'id',
+ |'file.format' = 'parquet' $primaryKeyOption)
+ |""".stripMargin)
+ sql("INSERT INTO T SELECT CAST(id AS INT), concat('v', CAST(id AS STRING))
FROM range(400)")
+ }
+
+ private def assertBucketFiles(
+ scan: PaimonScan,
+ expectedBuckets: Set[Int],
+ allSplits: Array[DataSplit]): Unit = {
+ val splits = scan.inputSplits.map(_.asInstanceOf[DataSplit])
+ assert(splits.map(_.bucket()).toSet == expectedBuckets, scan.description())
+ val expectedSplits = allSplits.filter(s =>
expectedBuckets.contains(s.bucket()))
+ assert(dataFileNames(splits) == dataFileNames(expectedSplits),
scan.description())
+ }
+
+ private def dataFileNames(splits: Array[_ <: DataSplit]): Seq[String] = {
+ splits.flatMap(_.dataFiles().asScala.map(_.fileName())).sorted.toSeq
+ }
+
+ private def materialize(df: DataFrame): DataFrame = {
+ spark.createDataFrame(spark.sparkContext.parallelize(df.collect().toSeq),
df.schema)
+ }
+}
