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 567c8401c2 [spark] Fix bucketed scan reporting for unsupported
timestamp precision (#9216)
567c8401c2 is described below
commit 567c8401c2d17933ba7fb69080aab6dc8b04ab32
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Sat Aug 15 20:18:23 2026 +0800
[spark] Fix bucketed scan reporting for unsupported timestamp precision
(#9216)
---
.../scala/org/apache/paimon/spark/PaimonScan.scala | 10 ++++++
.../paimon/spark/sql/BucketedTableQueryTest.scala | 38 ++++++++++++++++++++++
2 files changed, 48 insertions(+)
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 7b376c2f79..b6d89cf6e8 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
@@ -21,6 +21,7 @@ package org.apache.paimon.spark
import org.apache.paimon.CoreOptions.BucketFunctionType
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, Predicate,
TopN, VectorSearch}
+import org.apache.paimon.spark.catalog.functions.BucketFunction
import org.apache.paimon.spark.commands.BucketExpression.quote
import org.apache.paimon.spark.read.VariantExtractionInfo
import org.apache.paimon.table.{BucketMode, FileStoreTable, InnerTable}
@@ -67,6 +68,15 @@ case class PaimonScan(
.bucketFunctionType() != BucketFunctionType.DEFAULT
) {
None
+ } else if (!BucketFunction.supportsTable(fileStoreTable)) {
+ // Spark tells two scans apart by the canonical name of the bound
bucket function, which
+ // is derived from Spark types. Spark's timestamp precision is fixed
to 6, so bucket keys
+ // carrying any other precision are indistinguishable there, while
Paimon lays their
+ // BinaryRow out differently and thus puts the same value into a
different bucket.
+ // Reporting a bucket transform would let Spark treat such tables as
co-partitioned and
+ // drop a shuffle that is actually required. This mirrors the same
check the write side
+ // already does in `PaimonSparkWriter`.
+ None
} else if (bucketSpec.getBucketKeys.size() > 1) {
None
} else {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BucketedTableQueryTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BucketedTableQueryTest.scala
index 16b932ed7a..745d498b6f 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BucketedTableQueryTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BucketedTableQueryTest.scala
@@ -18,7 +18,10 @@
package org.apache.paimon.spark.sql
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.schema.Schema
import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.types.DataTypes
import org.apache.spark.sql.Row
import org.apache.spark.sql.execution.SortExec
@@ -189,6 +192,41 @@ class BucketedTableQueryTest extends PaimonSparkTestBase
with AdaptiveSparkPlanH
}
}
+ test("Query on a bucketed table - join - bucket key of an unsupported
timestamp precision") {
+ assume(gteqSpark3_3)
+
+ // Spark cannot express a parameterized timestamp in DDL, so go through
the table API. Spark's
+ // timestamp precision is fixed to 6, hence a bucket key of another
precision is not
+ // distinguishable in the reported transform while Paimon buckets it
differently.
+ Seq(("ts3", DataTypes.TIMESTAMP_MILLIS()), ("ts6",
DataTypes.TIMESTAMP())).foreach {
+ case (name, tsType) =>
+ paimonCatalog.createTable(
+ Identifier.create(dbName0, name),
+ Schema.newBuilder
+ .column("ts", tsType)
+ .column("c", DataTypes.STRING())
+ .option("bucket-key", "ts")
+ .option("bucket", "2")
+ .build,
+ false
+ )
+ spark.sql(s"""
+ |INSERT INTO $name VALUES
+ |(timestamp'2024-01-01 00:00:01', 'x1'),
(timestamp'2024-01-01 00:00:02', 'x2'),
+ |(timestamp'2024-01-01 00:00:03', 'x3'),
(timestamp'2024-01-01 00:00:04', 'x4'),
+ |(timestamp'2024-01-01 00:00:05', 'x5'),
(timestamp'2024-01-01 00:00:06', 'x6'),
+ |(timestamp'2024-01-01 00:00:07', 'x7'),
(timestamp'2024-01-01 00:00:08', 'x8')
+ |""".stripMargin)
+ }
+
+ try {
+ checkAnswerAndShuffleSorts("SELECT * FROM ts3 JOIN ts6 on ts3.ts =
ts6.ts", 2, 2)
+ } finally {
+ spark.sql("DROP TABLE IF EXISTS ts3")
+ spark.sql("DROP TABLE IF EXISTS ts6")
+ }
+ }
+
test("Query on a bucketed table - other operators") {
assume(gteqSpark3_3)
