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)