This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 52b2099f7f [VL][Iceberg] Add global iceberg sorted write (#13199)
52b2099f7f is described below
commit 52b2099f7f300504f4ca168ed803592a8ff40a13
Author: inf / Hazmi <[email protected]>
AuthorDate: Fri Oct 23 11:12:06 2026 +0300
[VL][Iceberg] Add global iceberg sorted write (#13199)
---
.../gluten/backendsapi/bolt/BoltRuleApi.scala | 2 +-
.../execution/AbstractIcebergWriteExec.scala | 2 +
.../enhanced/VeloxIcebergSortOrderSuite.scala | 259 +++++++++++++++++++++
.../execution/enhanced/VeloxIcebergSuite.scala | 26 ---
.../gluten/backendsapi/velox/VeloxRuleApi.scala | 2 +-
docs/backends/velox/get-started/VeloxIceberg.md | 27 ++-
.../apache/gluten/execution/IcebergWriteExec.scala | 9 +-
.../iceberg/spark/source/IcebergWriteUtil.scala | 31 ++-
.../columnar/EnsureLocalSortRequirements.scala | 4 +-
9 files changed, 321 insertions(+), 41 deletions(-)
diff --git
a/backends-bolt/src/main/scala/org/apache/gluten/backendsapi/bolt/BoltRuleApi.scala
b/backends-bolt/src/main/scala/org/apache/gluten/backendsapi/bolt/BoltRuleApi.scala
index f1fcf222b2..d89201899d 100644
---
a/backends-bolt/src/main/scala/org/apache/gluten/backendsapi/bolt/BoltRuleApi.scala
+++
b/backends-bolt/src/main/scala/org/apache/gluten/backendsapi/bolt/BoltRuleApi.scala
@@ -110,6 +110,7 @@ object BoltRuleApi {
injector.injectPostTransform(_ => RemoveNativeWriteFilesSortAndProject())
injector.injectPostTransform(_ => PushDownFilterToScan)
injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
+ injector.injectPostTransform(_ => V2WritePostRule())
injector.injectPostTransform(_ => EnsureLocalSortRequirements)
injector.injectPostTransform(_ => EliminateLocalSort)
injector.injectPostTransform(_ => PullOutDuplicateProject)
@@ -118,7 +119,6 @@ object BoltRuleApi {
injector.injectPostTransform(c =>
HashAggregateIgnoreNullKeysRule.apply(c.session))
injector.injectPostTransform(_ => CollectLimitTransformerRule())
injector.injectPostTransform(_ => CollectTailTransformerRule())
- injector.injectPostTransform(_ => V2WritePostRule())
injector.injectPostTransform(c =>
InsertTransitions.create(c.outputsColumnar, BoltBatchType))
// Gluten columnar: Fallback policies.
diff --git
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala
index 2fe2ce06ff..76143c8938 100644
---
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala
+++
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/execution/AbstractIcebergWriteExec.scala
@@ -33,6 +33,8 @@ import scala.collection.JavaConverters._
abstract class AbstractIcebergWriteExec extends IcebergWriteExec {
+ override protected def supportsSortedWrite: Boolean = true
+
private val parquetPageRowLimitSession =
"spark.gluten.sql.columnar.backend.velox.parquet_writer_page_row_limit"
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSortOrderSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSortOrderSuite.scala
new file mode 100644
index 0000000000..e6172af37e
--- /dev/null
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSortOrderSuite.scala
@@ -0,0 +1,259 @@
+/*
+ * 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.gluten.execution.enhanced
+
+import org.apache.gluten.execution._
+import org.apache.gluten.tags.EnhancedFeaturesTest
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.execution.CommandResultExec
+
+import org.apache.hadoop.fs.Path
+import org.apache.iceberg.{DataFile, SortOrderComparators, Table}
+import org.apache.iceberg.data.{GenericRecord, Record}
+import org.apache.iceberg.shaded.org.apache.parquet.example.data.Group
+import org.apache.iceberg.shaded.org.apache.parquet.hadoop.ParquetReader
+import
org.apache.iceberg.shaded.org.apache.parquet.hadoop.example.GroupReadSupport
+import org.apache.iceberg.spark.Spark3Util
+
+import scala.collection.JavaConverters._
+
+@EnhancedFeaturesTest
+class VeloxIcebergSortOrderSuite extends WholeStageTransformerSuite {
+
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+
+ override protected def sparkConf: SparkConf = super.sparkConf
+ .set("spark.gluten.sql.enable.enhancedFeatures", "true")
+ .set("spark.gluten.sql.columnar.maxBatchSize", "64")
+ .set("spark.sql.ansi.enabled", "false")
+ .set("spark.sql.shuffle.partitions", "4")
+ .set("spark.sql.adaptive.enabled", "true")
+ .set("spark.sql.adaptive.coalescePartitions.enabled", "false")
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set(
+ "spark.sql.extensions",
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
+ .set("spark.sql.catalog.spark_catalog",
"org.apache.iceberg.spark.SparkCatalog")
+ .set("spark.sql.catalog.spark_catalog.type", "hadoop")
+ .set(
+ "spark.sql.catalog.spark_catalog.warehouse",
+
s"file://${getClass.getResource("/").getPath}/iceberg-sort-order-warehouse")
+
+ private def input: DataFrame = spark
+ .range(0, 1024, 1, 4)
+ .selectExpr("CAST(pmod(id * 37, 1024) AS INT) AS id")
+ .selectExpr(
+ "id",
+ "CASE WHEN id % 7 = 0 THEN NULL ELSE id % 5 END AS category",
+ "named_struct('rank', id % 11) AS nested",
+ "concat(sha2(CAST(id AS STRING), 256), sha2(CAST(id + 1024 AS STRING),
256)) AS payload"
+ )
+
+ private def createTable(partitioning: String = ""): Unit = {
+ spark.sql(s"""
+ |CREATE TABLE iceberg_sort_order (
+ | id INT, category INT, nested STRUCT<rank: INT>, payload
STRING
+ |) USING iceberg $partitioning
+ |TBLPROPERTIES ('write.parquet.compression-codec' =
'uncompressed')
+ |""".stripMargin)
+ }
+
+ private def nativeWrite(sql: String): IcebergWriteExec = {
+ val result = spark.sql(sql)
+ val plan =
+
result.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan
+ assert(plan.isInstanceOf[IcebergWriteExec], s"Expected native Iceberg
write: $plan")
+ plan.asInstanceOf[IcebergWriteExec]
+ }
+
+ private def readFiles(table: Table): Vector[(DataFile, Vector[Record])] = {
+ table.refresh()
+ val tasks = table.newScan().planFiles()
+ val files =
+ try {
+ tasks.asScala.map(_.file().copy()).toVector
+ } finally {
+ tasks.close()
+ }
+ files.map {
+ file =>
+ val reader = ParquetReader
+ .builder[Group](new GroupReadSupport(), new
Path(file.path().toString))
+ .withConf(spark.sessionState.newHadoopConf())
+ .build()
+ val rows =
+ try {
+ Iterator.continually(reader.read()).takeWhile(_ != null).map {
+ group =>
+ val row = GenericRecord.create(table.schema())
+ row.setField("id", Int.box(group.getInteger("id", 0)))
+ if (group.getFieldRepetitionCount("category") > 0) {
+ row.setField("category",
Int.box(group.getInteger("category", 0)))
+ }
+ val nested =
GenericRecord.create(table.schema().findType("nested").asStructType())
+ nested.setField("rank", Int.box(group.getGroup("nested",
0).getInteger("rank", 0)))
+ row.setField("nested", nested)
+ row.setField("payload", group.getString("payload", 0))
+ row: Record
+ }.toVector
+ } finally {
+ reader.close()
+ }
+ assert(rows.nonEmpty)
+ (file, rows)
+ }
+ }
+
+ private def assertSortedFiles(global: Boolean): Vector[(DataFile,
Vector[Record])] = {
+ val table = Spark3Util.loadIcebergTable(spark, "iceberg_sort_order")
+ val files = readFiles(table)
+ assert(files.size > 1, "Sorting must be tested across multiple output
files")
+ val comparator = SortOrderComparators.forSchema(table.schema(),
table.sortOrder())
+ files.foreach {
+ case (file, rows) =>
+ assert(file.sortOrderId() == table.sortOrder().orderId())
+ rows.sliding(2).filter(_.size == 2).foreach {
+ pair =>
+ assert(comparator.compare(pair.head, pair.last) <= 0, s"Unsorted
file: ${file.path()}")
+ }
+ }
+ val ids =
files.flatMap(_._2.map(_.getField("id").asInstanceOf[Int])).sorted
+ assert(ids == (0 until 1024).toVector)
+
+ if (global) {
+ files.groupBy(_._1.partition().toString).values.foreach {
+ partitionFiles =>
+ val ranges = partitionFiles.map(_._2).sortWith {
+ (left, right) => comparator.compare(left.head, right.head) < 0
+ }
+ ranges.sliding(2).filter(_.size == 2).foreach {
+ pair =>
+ assert(
+ comparator.compare(pair.head.last, pair.last.head) <= 0,
+ "Output files have overlapping sort-key ranges")
+ }
+ }
+ }
+ files
+ }
+
+ for {
+ (name, partitioning, ordering, global) <- Seq(
+ ("global", "", "ORDERED BY category ASC NULLS LAST, id DESC NULLS
FIRST", true),
+ ("local", "", "LOCALLY ORDERED BY category DESC NULLS FIRST, id ASC
NULLS LAST", false),
+ (
+ "partitioned",
+ "PARTITIONED BY (bucket(2, id))",
+ "ORDERED BY category DESC NULLS LAST, id ASC",
+ true),
+ (
+ "transformed",
+ "",
+ "ORDERED BY truncate(3, category) DESC NULLS FIRST, id ASC",
+ true),
+ ("nested", "", "ORDERED BY nested.rank ASC, id DESC", true)
+ )
+ } {
+ test(s"iceberg native $name sort order") {
+ withTable("iceberg_sort_order") {
+ withTempView("sort_order_input") {
+ createTable(partitioning)
+ spark.sql(s"ALTER TABLE iceberg_sort_order WRITE $ordering")
+ if (!global) {
+ spark.sql("""
+ |ALTER TABLE iceberg_sort_order SET TBLPROPERTIES (
+ | 'write.distribution-mode' = 'none')
+ |""".stripMargin)
+ }
+ input.createOrReplaceTempView("sort_order_input")
+ val write = nativeWrite("INSERT INTO iceberg_sort_order SELECT *
FROM sort_order_input")
+ if (name == "global" || name == "local") {
+ assert(collect(write.query) { case s: SortExecTransformer => s
}.nonEmpty)
+ }
+ assertSortedFiles(global)
+ }
+ }
+ }
+ }
+
+ test("iceberg sort order survives file rotation and overwrite") {
+ withSQLConf("spark.sql.adaptive.enabled" -> "false") {
+ withTable("iceberg_sort_order") {
+ withTempView("sort_order_input") {
+ createTable()
+ spark.sql("ALTER TABLE iceberg_sort_order WRITE ORDERED BY id DESC")
+ spark.sql("""
+ |ALTER TABLE iceberg_sort_order SET TBLPROPERTIES (
+ | 'write.target-file-size-bytes' = '8192',
+ | 'write.parquet.page-size-bytes' = '1024')
+ |""".stripMargin)
+ input.createOrReplaceTempView("sort_order_input")
+ nativeWrite("INSERT INTO iceberg_sort_order SELECT * FROM
sort_order_input")
+ val files = assertSortedFiles(global = true)
+ assert(files.size > 4, "Expected file rotation within write tasks")
+
+ spark.sql("ALTER TABLE iceberg_sort_order WRITE ORDERED BY category,
id")
+ nativeWrite("INSERT OVERWRITE iceberg_sort_order SELECT * FROM
sort_order_input")
+ assertSortedFiles(global = true)
+ }
+ }
+ }
+ }
+
+ test("iceberg copy-on-write UPDATE preserves sort order") {
+ withTable("iceberg_sort_order") {
+ withTempView("sort_order_input") {
+ createTable()
+ spark.sql(
+ "ALTER TABLE iceberg_sort_order WRITE ORDERED BY category DESC NULLS
LAST, id ASC")
+ input.createOrReplaceTempView("sort_order_input")
+ nativeWrite("INSERT INTO iceberg_sort_order SELECT * FROM
sort_order_input")
+ nativeWrite("UPDATE iceberg_sort_order SET category = 20 - category
WHERE id % 2 = 0")
+ assertSortedFiles(global = false)
+ }
+ }
+ }
+
+ for (aqe <- Seq(false, true)) {
+ test(s"iceberg local sort order survives native hash aggregation:
aqe=$aqe") {
+ withSQLConf("spark.sql.adaptive.enabled" -> aqe.toString) {
+ withTable("iceberg_sort_order") {
+ withTempView("sort_order_input") {
+ createTable()
+ spark.sql("ALTER TABLE iceberg_sort_order WRITE LOCALLY ORDERED BY
id")
+ spark.sql("""
+ |ALTER TABLE iceberg_sort_order SET TBLPROPERTIES (
+ | 'write.distribution-mode' = 'none')
+ |""".stripMargin)
+ input.createOrReplaceTempView("sort_order_input")
+ val write = nativeWrite("""
+ |INSERT INTO iceberg_sort_order
+ |SELECT id, first(category),
first(nested), max(payload)
+ |FROM sort_order_input GROUP BY id
+ |""".stripMargin)
+ assert(collect(write.query) { case a: HashAggregateExecTransformer
=> a }.nonEmpty)
+ assertSortedFiles(global = false)
+ assert(collect(write.query) { case s: SortExecTransformer => s
}.nonEmpty)
+ }
+ }
+ }
+ }
+ }
+}
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
index 145e9f6a49..94c8844612 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/enhanced/VeloxIcebergSuite.scala
@@ -24,7 +24,6 @@ import org.apache.gluten.tags.EnhancedFeaturesTest
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.execution.{CommandExecutionMode, CommandResultExec}
-import org.apache.spark.sql.execution.GlutenImplicits._
import org.apache.spark.sql.execution.datasources.v2.AppendDataExec
import org.apache.spark.sql.execution.streaming.MemoryStream
import org.apache.spark.sql.gluten.TestUtils
@@ -463,31 +462,6 @@ class VeloxIcebergSuite extends IcebergSuite {
}
}
- test("iceberg native write fallback when validation fails - sort order") {
- withTable("iceberg_sorted_tbl") {
- spark.sql("CREATE TABLE iceberg_sorted_tbl (a INT, b STRING) USING
iceberg")
- spark.sql("ALTER TABLE iceberg_sorted_tbl WRITE ORDERED BY a")
-
- val df = spark.sql("INSERT INTO iceberg_sorted_tbl VALUES (1, 'hello'),
(2, 'world')")
-
- // Should fallback to vanilla Spark's AppendDataExec.
- val commandPlan =
-
df.queryExecution.executedPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan
- assert(commandPlan.isInstanceOf[AppendDataExec])
- assert(!commandPlan.isInstanceOf[VeloxIcebergAppendDataExec])
-
- checkAnswer(
- spark.sql("SELECT * FROM iceberg_sorted_tbl ORDER BY a"),
- Seq(Row(1, "hello"), Row(2, "world")))
-
- // Verify fallbackSummary reports the sort order fallback reason.
- val summary = df.fallbackSummary()
- assert(
- summary.fallbackNodeToReason.exists(
- _.values.exists(_.contains("Not support write table with sort
order"))))
- }
- }
-
test("iceberg read cow table - update after schema evolution") {
withTable("iceberg_cow_update_evolved_tb") {
spark.sql("""
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
index 63c8880e8a..07de0ab0f0 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -128,13 +128,13 @@ object VeloxRuleApi {
injector.injectPostTransform(_ => PushDownFilterToScan)
injector.injectPostTransform(_ => ScanMapKeyPruning)
injector.injectPostTransform(_ => PushDownInputFileExpression.PostOffload)
+ injector.injectPostTransform(_ => V2WritePostRule())
injector.injectPostTransform(_ => EnsureLocalSortRequirements)
injector.injectPostTransform(_ => EliminateLocalSort)
injector.injectPostTransform(_ => CollapseProjectExecTransformer)
injector.injectPostTransform(c =>
FlushableHashAggregateRule.apply(c.session))
injector.injectPostTransform(_ => CollectLimitTransformerRule())
injector.injectPostTransform(_ => CollectTailTransformerRule())
- injector.injectPostTransform(_ => V2WritePostRule())
injector.injectPostTransform(c =>
InsertTransitions.create(c.outputsColumnar, VeloxBatchType))
// Gluten columnar: Fallback policies.
diff --git a/docs/backends/velox/get-started/VeloxIceberg.md
b/docs/backends/velox/get-started/VeloxIceberg.md
index b9f18256a4..929f3fda8b 100644
--- a/docs/backends/velox/get-started/VeloxIceberg.md
+++ b/docs/backends/velox/get-started/VeloxIceberg.md
@@ -30,16 +30,27 @@ Fallback
Fallback
## Writing
-Fallback
-````
+Supported Parquet writes can offload when
+`spark.gluten.sql.enable.enhancedFeatures` and
+`spark.gluten.sql.columnar.iceberg.enableNativeWrite` are enabled.
+```sql
INSERT INTO local.db.table VALUES (1, 'a'), (2, 'b'), (3, 'c');
-````
-PartialOffload
-
-The write is fallback while read is offload.
-````
INSERT INTO local.db.table SELECT id, data FROM source WHERE length(data) = 1;
-````
+```
+
+### Sorted native writes
+
+With `spark.gluten.sql.enable.enhancedFeatures` and
+`spark.gluten.sql.columnar.iceberg.enableNativeWrite` enabled, the Velox writer
+supports Iceberg table sort orders. Spark plans the required distribution and
+ordering before the native write:
+
+```sql
+ALTER TABLE local.db.table WRITE ORDERED BY category ASC NULLS LAST, id DESC;
+-- Sort within each task without requiring global range distribution:
+ALTER TABLE local.db.table WRITE LOCALLY ORDERED BY category, id;
+ALTER TABLE local.db.table SET TBLPROPERTIES ('write.distribution-mode' =
'none');
+```
## Reading
### Read data
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala
index cf5c047514..f67490f28e 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergWriteExec.scala
@@ -18,6 +18,8 @@ package org.apache.gluten.execution
import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.spark.sql.catalyst.expressions.SortOrder
+
import org.apache.iceberg.{FileFormat, PartitionField, PartitionSpec, Schema,
TableProperties}
import org.apache.iceberg.TableProperties._
import org.apache.iceberg.avro.AvroSchemaUtil
@@ -30,6 +32,11 @@ import scala.collection.JavaConverters._
trait IcebergWriteExec extends ColumnarV2TableWriteExec {
+ protected def supportsSortedWrite: Boolean = false
+
+ override lazy val requiredChildOrdering: Seq[Seq[SortOrder]] =
+ Seq(IcebergWriteUtil.getRequiredOrdering(write, query.output))
+
protected def getFileFormat(format: FileFormat): Int = {
format match {
case FileFormat.PARQUET => 1
@@ -130,7 +137,7 @@ trait IcebergWriteExec extends ColumnarV2TableWriteExec {
"Not support write unsupported partition type, or is nested
partition column")
}
}
- if (IcebergWriteUtil.getTable(write).sortOrder().isSorted) {
+ if (!supportsSortedWrite && IcebergWriteUtil.getSortOrder(write).isSorted)
{
return ValidationResult.failed("Not support write table with sort order")
}
val format = IcebergWriteUtil.getFileFormat(write)
diff --git
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
index 83ecb07037..e8792556d6 100644
---
a/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/iceberg/spark/source/IcebergWriteUtil.scala
@@ -16,10 +16,15 @@
*/
package org.apache.iceberg.spark.source
-import org.apache.spark.sql.connector.write.{Write, WriterCommitMessage}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SortOrder =>
CatalystSortOrder}
+import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, Sort}
+import org.apache.spark.sql.connector.distributions.{Distribution,
Distributions}
+import org.apache.spark.sql.connector.expressions.{SortOrder =>
ConnectorSortOrder}
+import org.apache.spark.sql.connector.write.{RequiresDistributionAndOrdering,
Write, WriterCommitMessage}
+import
org.apache.spark.sql.execution.datasources.v2.DistributionAndOrderingUtils
import org.apache.iceberg._
-import org.apache.iceberg.spark.SparkWriteConf
+import org.apache.iceberg.spark.{SparkFunctionCatalog, SparkWriteConf}
import org.apache.iceberg.spark.source.SparkWrite.TaskCommit
import org.apache.iceberg.types.Type
import org.apache.iceberg.types.Type.TypeID
@@ -114,7 +119,27 @@ object IcebergWriteUtil {
}
def getSortOrder(write: Write): SortOrder = {
- getTable(write).sortOrder()
+ if (write.asInstanceOf[SparkWrite].requiredOrdering().nonEmpty) {
+ getTable(write).sortOrder()
+ } else {
+ SortOrder.unsorted()
+ }
+ }
+
+ def getRequiredOrdering(write: Write, output: Seq[Attribute]):
Seq[CatalystSortOrder] = {
+ // Distribution may reference metadata columns, such as _file, absent from
the writer's input.
+ val orderingOnly = new RequiresDistributionAndOrdering {
+ override def requiredDistribution(): Distribution =
Distributions.unspecified()
+ override def requiredOrdering(): Array[ConnectorSortOrder] = write match
{
+ case ordered: RequiresDistributionAndOrdering =>
ordered.requiredOrdering()
+ case _ => Array.empty
+ }
+ }
+ DistributionAndOrderingUtils
+ .prepareQuery(orderingOnly, LocalRelation(output),
Some(SparkFunctionCatalog.get())) match {
+ case sort: Sort => sort.order
+ case _ => Nil
+ }
}
def getPartitionSpec(write: Write): PartitionSpec = {
diff --git
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala
index 008735f760..bc2d762209 100644
---
a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala
+++
b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.extension.columnar
-import org.apache.gluten.execution.GlutenPlan
+import org.apache.gluten.execution.{ColumnarV2TableWriteExec, GlutenPlan}
import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
import org.apache.spark.sql.catalyst.expressions.SortOrder
@@ -75,6 +75,8 @@ object EnsureLocalSortRequirements extends Rule[SparkPlan] {
transform.apply(newChild)
case (parent: GlutenPlan, _) if parent.supportsColumnar =>
transform.apply(newChild)
+ case (_: ColumnarV2TableWriteExec, _) =>
+ transform.apply(newChild)
case _ =>
newChild
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
