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 322d270cdc [Spark] Quote primary keys when reporting scan output
ordering (#10659)
322d270cdc is described below
commit 322d270cdcce4e489fa3f340fd79223bed18e0e2
Author: Jiwen liu <[email protected]>
AuthorDate: Tue Oct 20 15:11:17 2026 +0800
[Spark] Quote primary keys when reporting scan output ordering (#10659)
---
.../scala/org/apache/paimon/spark/PaimonScan.scala | 1 +
.../paimon/spark/sql/BucketedTableQueryTest.scala | 69 +++++++++++++++++++++-
2 files changed, 69 insertions(+), 1 deletion(-)
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 83f2476a58..96d65bbcb7 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
@@ -163,6 +163,7 @@ case class PaimonScan(
// Multi-primary keys are fine:
// `Array(a, b)` satisfies the required ordering `Array(a)`
primaryKeys
+ .map(quote)
.map(Expressions.identity)
.map {
sortExpr =>
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 7eaf27f957..3e0e5af86d 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
@@ -19,13 +19,18 @@
package org.apache.paimon.spark.sql
import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.data.{BinaryString, GenericRow}
import org.apache.paimon.schema.Schema
-import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.spark.{PaimonBucketedInputPartition, PaimonScan,
PaimonSparkTestBase}
+import org.apache.paimon.spark.commands.BucketExpression.quote
+import org.apache.paimon.table.source.DataSplit
import org.apache.paimon.types.DataTypes
import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.expressions.SortDirection
import org.apache.spark.sql.execution.SortExec
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
class BucketedTableQueryTest extends PaimonSparkTestBase with
AdaptiveSparkPlanHelper {
@@ -255,6 +260,68 @@ class BucketedTableQueryTest extends PaimonSparkTestBase
with AdaptiveSparkPlanH
}
}
+ for (key <- Seq("id", "#id", "a.b", "a`b")) {
+ test(s"Report scan output ordering - primary key $key") {
+ assume(gteqSpark3_4)
+
+ withTable("t") {
+ val quotedKey = quote(key)
+ sql(s"""CREATE TABLE t ($quotedKey STRING, seq INT)
+ |TBLPROPERTIES ('primary-key'='seq,$key', 'bucket-key'='seq',
'bucket'='1')
+ |""".stripMargin)
+ // Prepare data through the table API to keep this test focused on
scan output ordering.
+ val writeBuilder = loadTable("t").newBatchWriteBuilder()
+ val write = writeBuilder.newWrite()
+ val commit = writeBuilder.newCommit()
+ try {
+ write.write(GenericRow.of(BinaryString.fromString("b"), 1))
+ write.write(GenericRow.of(BinaryString.fromString("a"), 1))
+ write.write(GenericRow.of(BinaryString.fromString("a"), 2))
+ commit.commit(write.prepareCommit())
+ } finally {
+ write.close()
+ commit.close()
+ }
+
+ for (preserve <- Seq(false, true); v2 <- Seq(false, true)) {
+ withSparkSQLConf(
+ "spark.paimon.scan.preserve-data-grouping" -> preserve.toString,
+ "spark.sql.sources.v2.bucketing.enabled" -> v2.toString) {
+ val df = sql(s"SELECT $quotedKey, seq FROM t")
+ checkAnswer(df, Seq(Row("a", 1), Row("b", 1), Row("a", 2)))
+ val scans = collect(df.queryExecution.executedPlan) {
+ case batch: BatchScanExec if batch.scan.isInstanceOf[PaimonScan]
=>
+ batch.scan.asInstanceOf[PaimonScan]
+ }
+ assert(scans.size == 1)
+ val scan = scans.head
+ val ordering = scan.outputOrdering()
+ if (preserve && v2) {
+ val partitions = scan.inputPartitions
+ assert(partitions.size == 1)
+
assert(partitions.head.isInstanceOf[PaimonBucketedInputPartition])
+ val splits =
partitions.head.asInstanceOf[PaimonBucketedInputPartition].splits
+ assert(splits.size == 1)
+ assert(splits.head.asInstanceOf[DataSplit].dataFiles().size() ==
1)
+ assert(
+ ordering
+
.map(_.expression().references().map(_.fieldNames().toSeq).toSeq)
+ .toSeq == Seq(Seq(Seq("seq")), Seq(Seq(key))))
+ assert(ordering.forall(_.direction() == SortDirection.ASCENDING))
+ } else {
+ assert(ordering.isEmpty)
+ }
+ }
+ }
+
+ checkAnswerAndShuffleSorts(
+ s"SELECT l.$quotedKey, l.seq FROM t l JOIN t r ON l.seq = r.seq AND
l.$quotedKey = r.$quotedKey",
+ 0,
+ 0)
+ }
+ }
+ }
+
test("Report scan output ordering - rawConvertible") {
assume(gteqSpark3_3)
