From 88daea49fae7babe9378fe7f5e48a9bb33c4e1ef Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Fri, 14 Aug 2026 15:16:32 +0800 Subject: [PATCH] [spark] Fix bucketed scan reporting for unsupported timestamp precision Spark's timestamp precision is fixed to 6, so a bucket key of TIMESTAMP(3) and one of TIMESTAMP(6) are indistinguishable in the transform reported through SupportsReportPartitioning: the bound bucket function derives its canonical name from Spark types, and Spark compares exactly that canonical name to decide whether two scans are co-partitioned. Paimon, however, lays those two out differently in BinaryRow - a timestamp is stored compactly when the precision is <= 3 - so the same value hashes differently and lands in a different bucket. Joining such tables with spark.sql.sources.v2.bucketing.enabled therefore let Spark drop a shuffle that was actually required, silently returning fewer rows. Skip reporting the bucket transform for these tables, which is the same BucketFunction.supportsTable check the write side already performs in PaimonSparkWriter. Tables whose bucket key carries a precision other than 6 now fall back to a shuffle instead of producing wrong results. --- .../org/apache/paimon/spark/PaimonScan.scala | 10 +++++ .../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 7b376c2f7918..b6d89cf6e888 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 16b932ed7ae6..745d498b6f7d 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)