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 260fcc757b [spark] Support partition sink strategy for Format Table
(#8786)
260fcc757b is described below
commit 260fcc757bda851fc6e9b7c00c16f75687fc5c24
Author: Zouxxyy <[email protected]>
AuthorDate: Tue Jul 21 23:47:27 2026 +0800
[spark] Support partition sink strategy for Format Table (#8786)
---
.../paimon/spark/format/PaimonFormatTable.scala | 12 ++++-
.../spark/write/PaimonWriteRequirement.scala | 28 +++++++---
.../sql/FormatTableWriteDistributeModeTest.scala | 61 ++++++++++++++++++++++
3 files changed, 93 insertions(+), 8 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
index 776fc98ca1..eecaed851b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.CoreOptions
import org.apache.paimon.format.csv.CsvOptions
import org.apache.paimon.fs.Path
import org.apache.paimon.spark.{BaseTable, FormatTableScanBuilder}
-import org.apache.paimon.spark.write.BaseV2WriteBuilder
+import org.apache.paimon.spark.write.{BaseV2WriteBuilder,
PaimonWriteRequirement}
import org.apache.paimon.table.FormatTable
import org.apache.paimon.table.format.FormatTablePartitionManager
import org.apache.paimon.types.RowType
@@ -31,6 +31,8 @@ import org.apache.paimon.utils.PartitionPathUtils
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
TableCapability, TableCatalog}
import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ,
BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
+import org.apache.spark.sql.connector.distributions.Distribution
+import org.apache.spark.sql.connector.expressions.SortOrder
import org.apache.spark.sql.connector.read.ScanBuilder
import org.apache.spark.sql.connector.write._
import org.apache.spark.sql.connector.write.streaming.StreamingWrite
@@ -268,7 +270,13 @@ case class PaimonFormatTableWriterBuilder(table:
FormatTable, writeSchema: Struc
override def partitionRowType(): RowType = table.partitionType
- override def build: Write = new Write() {
+ override def build: Write = new Write with RequiresDistributionAndOrdering {
+ private val writeRequirement = PaimonWriteRequirement(table)
+
+ override def requiredDistribution(): Distribution =
writeRequirement.distribution
+
+ override def requiredOrdering(): Array[SortOrder] =
writeRequirement.ordering
+
override def toBatch: BatchWrite = {
SparkShimLoader.shim
.createFormatTableBatchWrite(table, overwriteDynamic,
overwritePartitions, writeSchema)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteRequirement.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteRequirement.scala
index 1f95c19146..be661807e4 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteRequirement.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteRequirement.scala
@@ -18,10 +18,11 @@
package org.apache.paimon.spark.write
+import org.apache.paimon.CoreOptions
import org.apache.paimon.CoreOptions.PartitionSinkStrategy
import org.apache.paimon.spark.commands.BucketExpression.quote
+import org.apache.paimon.table.{FileStoreTable, FormatTable}
import org.apache.paimon.table.BucketMode._
-import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.connector.distributions.{ClusteredDistribution,
Distribution, Distributions}
import org.apache.spark.sql.connector.expressions.{Expression, Expressions,
SortOrder}
@@ -39,7 +40,7 @@ object PaimonWriteRequirement {
def apply(table: FileStoreTable): PaimonWriteRequirement = {
val bucketSpec = table.bucketSpec()
- val bucketTransforms = bucketSpec.getBucketMode match {
+ val bucketTransforms: Seq[Expression] = bucketSpec.getBucketMode match {
case HASH_FIXED =>
Seq(
Expressions.bucket(
@@ -52,15 +53,30 @@ object PaimonWriteRequirement {
s"Unsupported bucket mode ${bucketSpec.getBucketMode}")
}
+ create(
+ table.schema().partitionKeys().asScala.toSeq,
+ bucketTransforms,
+ table.coreOptions().partitionSinkStrategy())
+ }
+
+ def apply(table: FormatTable): PaimonWriteRequirement = {
+ create(
+ table.partitionKeys().asScala.toSeq,
+ Seq.empty,
+ CoreOptions.fromMap(table.options()).partitionSinkStrategy())
+ }
+
+ private def create(
+ partitionKeys: Seq[String],
+ bucketTransforms: Seq[Expression],
+ partitionSinkStrategy: PartitionSinkStrategy): PaimonWriteRequirement = {
val partitionTransforms =
- table.schema().partitionKeys().asScala.map(key =>
Expressions.identity(quote(key)))
+ partitionKeys.map(key => Expressions.identity(quote(key)))
val clusteringExpressions =
(partitionTransforms ++
bucketTransforms).map(identity[Expression]).toArray
if (
- clusteringExpressions.isEmpty || (bucketTransforms.isEmpty && table
- .coreOptions()
- .partitionSinkStrategy()
+ clusteringExpressions.isEmpty || (bucketTransforms.isEmpty &&
partitionSinkStrategy
.equals(PartitionSinkStrategy.NONE))
) {
EMPTY
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableWriteDistributeModeTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableWriteDistributeModeTest.scala
new file mode 100644
index 0000000000..946252fa42
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableWriteDistributeModeTest.scala
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.spark.PaimonHiveTestBase
+
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.CommandResultExec
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeLike
+
+class FormatTableWriteDistributeModeTest extends PaimonHiveTestBase with
AdaptiveSparkPlanHelper {
+
+ override protected def beforeEach(): Unit = {
+ sql(s"USE $paimonHiveCatalogName")
+ sql(s"USE $hiveDbName")
+ }
+
+ test("Write distribute mode: write partitioned format table") {
+ for (distributeMode <- Seq("none", "hash")) {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, pt STRING) USING parquet PARTITIONED BY
(pt) " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon')")
+ val query = "INSERT INTO t VALUES (1, 'p1'), (2, 'p2')"
+
+ withSparkSQLConf("spark.paimon.partition.sink-strategy" ->
distributeMode) {
+ val df = spark.sql(query)
+ val shuffleNodes = collect(
+
df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan)
{
+ case shuffle: ShuffleExchangeLike => shuffle
+ }
+
+ if (distributeMode == "none") {
+ assert(shuffleNodes.isEmpty)
+ } else {
+ assert(shuffleNodes.size == 1)
+ }
+
+ checkAnswer(spark.sql("SELECT * FROM t ORDER BY id"), Seq(Row(1,
"p1"), Row(2, "p2")))
+ }
+ }
+ }
+ }
+}
