[spark] Fix bucketed scan reporting for unsupported timestamp precision

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 {

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)