Skip to content

[GLUTEN-12474][CORE] Preserve V1 write ordering for dynamic partition writes #12514

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wForget merged 6 commits into from
Jul 23, 2026
Merged

[GLUTEN-12474][CORE] Preserve V1 write ordering for dynamic partition writes #12514

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3716fcf
test GLUTEN-12474
wForget Jul 15, 2026
19d6bb6
Preserve V1 write ordering when inserting local sorts
wForget Jul 15, 2026
0bbfcd5
add unit tests for other spark version
wForget Jul 15, 2026
4d776e1
fix
wForget Jul 15, 2026
9cfff27
fix spotless check
wForget Jul 15, 2026
bc63bb3
Enhance local sort handling for GlutenPlan support
wForget Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 44 additions & 4 deletions ...ait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,15 @@
*/
package org.apache.gluten.extension.columnar

import org.apache.gluten.execution.GlutenPlan
import org.apache.gluten.extension.columnar.heuristic.HeuristicTransform
import org.apache.gluten.sql.shims.SparkShimLoader

import org.apache.spark.sql.catalyst.expressions.SortOrder
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.{SortExec, SparkPlan}
import org.apache.spark.sql.execution.{ColumnarWriteFilesExec, SortExec, SparkPlan}
import org.apache.spark.sql.execution.datasources.WriteFilesExec
import org.apache.spark.sql.internal.SQLConf

/**
* This rule is similar with `EnsureRequirements` but only handle local `SortExec`.
Expand All @@ -33,25 +37,61 @@ import org.apache.spark.sql.execution.{SortExec, SparkPlan}
object EnsureLocalSortRequirements extends Rule[SparkPlan] {
private lazy val transform: HeuristicTransform = HeuristicTransform.static()

private def numStaticPartitionCols(writeFiles: WriteFilesExec): Int = {
// HadoopFs writes include static partition columns in partitionColumns, while Hive writes may
// only include the partition columns that are present in the write query.
val resolver = SQLConf.get.resolver
val staticPartitionNames = writeFiles.staticPartitions.keys
writeFiles.partitionColumns.takeWhile {
partitionColumn => staticPartitionNames.exists(resolver(_, partitionColumn.name))
}.size
}

private def requiredChildOrdering(plan: SparkPlan): Seq[Seq[SortOrder]] = {
plan match {
// V1Writes assumes that the logical ordering it prepared is preserved in the physical plan,
// so WriteFilesExec does not expose requiredChildOrdering itself. Gluten may invalidate that
// ordering when it replaces a SortAggregateExec with a hash aggregate.

@ulysses-you ulysses-you Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems we can make WriteFilesExec supports require child ordering native in Spark upstream.

case writeFiles: WriteFilesExec
if ColumnarWriteFilesExec.OnNoopLeafPath.unapply(writeFiles).isEmpty =>
Seq(
SparkShimLoader.getSparkShims.getV1WriteRequiredOrdering(
writeFiles.child.output,
writeFiles.partitionColumns,
writeFiles.bucketSpec,
writeFiles.options,
numStaticPartitionCols(writeFiles)))
case _ => plan.requiredChildOrdering
}
}

private def addLocalSort(
plan: SparkPlan,
originalChild: SparkPlan,
requiredOrdering: Seq[SortOrder]): SparkPlan = {
// FIXME: HeuristicTransform is costly. Re-applying it may cause performance issues.
val newChild = SortExec(requiredOrdering, global = false, child = originalChild)
transform.apply(newChild)
(plan, originalChild) match {

@ulysses-you ulysses-you Jul 22, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I do not get this change, what's wrong with previous code?

@wForget wForget Jul 22, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Avoid generating expensive plan like:

  ProjectExec
      |
     R2C
      |
SortExecTransformer
      |
     C2R
      |
 WriteFileExec

Also fixed the failure at #12514 (comment)

case (_, child: GlutenPlan) if child.supportsColumnar =>
transform.apply(newChild)
case (parent: GlutenPlan, _) if parent.supportsColumnar =>
transform.apply(newChild)
case _ =>
newChild
}
}

override def apply(plan: SparkPlan): SparkPlan = {
plan.transformUp {
case p =>
val newChildren = p.children.zip(p.requiredChildOrdering).map {
val newChildren = p.children.zip(requiredChildOrdering(p)).map {
case (child, requiredOrdering) =>
// If child.outputOrdering already satisfies the requiredOrdering,
// we do not need to sort.
if (SortOrder.orderingSatisfies(child.outputOrdering, requiredOrdering)) {
child
} else {
addLocalSort(child, requiredOrdering)
addLocalSort(p, child, requiredOrdering)
}
}
p.withNewChildren(newChildren)
Expand Down
35 changes: 35 additions & 0 deletions ...src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,41 @@ class GlutenV1WriteCommandSuite
with GlutenV1WriteCommandSuiteBase
with GlutenSQLTestsBaseTrait {

testGluten("GLUTEN-12474: preserve ordering for dynamic partition writes") {
withSQLConf(
"spark.sql.maxConcurrentOutputFileWriters" -> "0",
"spark.sql.sources.partitionOverwriteMode" -> "DYNAMIC") {
withTable("gluten_12474_src", "gluten_12474_tgt") {
sql(
"""
|CREATE TABLE gluten_12474_src USING ORC AS
|SELECT concat('k', id) AS k, format_string('v%02d', id) AS v,
| if(id % 2 = 1, '2026-06-01', '2026-06-02') AS day
|FROM range(0, 10)
|""".stripMargin)

sql(
"""
|CREATE TABLE gluten_12474_tgt (k STRING, m STRING)
|USING ORC
|PARTITIONED BY (day STRING)
|""".stripMargin)

sql(
"""
|INSERT OVERWRITE TABLE gluten_12474_tgt PARTITION (day)
|SELECT k, max(v) AS m, day
|FROM gluten_12474_src
|GROUP BY day, k
|""".stripMargin)

checkAnswer(
sql("SELECT k, m, day FROM gluten_12474_tgt"),
sql("SELECT k, v, day FROM gluten_12474_src"))
}
}
}

testGluten(
"SPARK-41914: v1 write with AQE and in-partition sorted - non-string partition column") {
withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
Expand Down
35 changes: 35 additions & 0 deletions ...src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,41 @@ class GlutenV1WriteCommandSuite
with GlutenSQLTestsBaseTrait
with GlutenColumnarWriteTestSupport {

testGluten("GLUTEN-12474: preserve ordering for dynamic partition writes") {
withSQLConf(
"spark.sql.maxConcurrentOutputFileWriters" -> "0",
"spark.sql.sources.partitionOverwriteMode" -> "DYNAMIC") {
withTable("gluten_12474_src", "gluten_12474_tgt") {
sql(
"""
|CREATE TABLE gluten_12474_src USING ORC AS
|SELECT concat('k', id) AS k, format_string('v%02d', id) AS v,
| if(id % 2 = 1, '2026-06-01', '2026-06-02') AS day
|FROM range(0, 10)
|""".stripMargin)

sql(
"""
|CREATE TABLE gluten_12474_tgt (k STRING, m STRING)
|USING ORC
|PARTITIONED BY (day STRING)
|""".stripMargin)

sql(
"""
|INSERT OVERWRITE TABLE gluten_12474_tgt PARTITION (day)
|SELECT k, max(v) AS m, day
|FROM gluten_12474_src
|GROUP BY day, k
|""".stripMargin)

checkAnswer(
sql("SELECT k, m, day FROM gluten_12474_tgt"),
sql("SELECT k, v, day FROM gluten_12474_src"))
}
}
}

testGluten(
"SPARK-41914: v1 write with AQE and in-partition sorted - non-string partition column") {
withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
Expand Down
35 changes: 35 additions & 0 deletions ...src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,41 @@ class GlutenV1WriteCommandSuite
with GlutenSQLTestsBaseTrait
with GlutenColumnarWriteTestSupport {

testGluten("GLUTEN-12474: preserve ordering for dynamic partition writes") {
withSQLConf(
"spark.sql.maxConcurrentOutputFileWriters" -> "0",
"spark.sql.sources.partitionOverwriteMode" -> "DYNAMIC") {
withTable("gluten_12474_src", "gluten_12474_tgt") {
sql(
"""
|CREATE TABLE gluten_12474_src USING ORC AS
|SELECT concat('k', id) AS k, format_string('v%02d', id) AS v,
| if(id % 2 = 1, '2026-06-01', '2026-06-02') AS day
|FROM range(0, 10)
|""".stripMargin)

sql(
"""
|CREATE TABLE gluten_12474_tgt (k STRING, m STRING)
|USING ORC
|PARTITIONED BY (day STRING)
|""".stripMargin)

sql(
"""
|INSERT OVERWRITE TABLE gluten_12474_tgt PARTITION (day)
|SELECT k, max(v) AS m, day
|FROM gluten_12474_src
|GROUP BY day, k
|""".stripMargin)

checkAnswer(
sql("SELECT k, m, day FROM gluten_12474_tgt"),
sql("SELECT k, v, day FROM gluten_12474_src"))
}
}
}

// TODO: fix in Spark-4.0
ignoreGluten(
"SPARK-41914: v1 write with AQE and in-partition sorted - non-string partition column") {
Expand Down
35 changes: 35 additions & 0 deletions ...src/test/scala/org/apache/spark/sql/execution/datasources/GlutenV1WriteCommandSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,41 @@ class GlutenV1WriteCommandSuite
with GlutenSQLTestsBaseTrait
with GlutenColumnarWriteTestSupport {

testGluten("GLUTEN-12474: preserve ordering for dynamic partition writes") {
withSQLConf(
"spark.sql.maxConcurrentOutputFileWriters" -> "0",
"spark.sql.sources.partitionOverwriteMode" -> "DYNAMIC") {
withTable("gluten_12474_src", "gluten_12474_tgt") {
sql(
"""
|CREATE TABLE gluten_12474_src USING ORC AS
|SELECT concat('k', id) AS k, format_string('v%02d', id) AS v,
| if(id % 2 = 1, '2026-06-01', '2026-06-02') AS day
|FROM range(0, 10)
|""".stripMargin)

sql(
"""
|CREATE TABLE gluten_12474_tgt (k STRING, m STRING)
|USING ORC
|PARTITIONED BY (day STRING)
|""".stripMargin)

sql(
"""
|INSERT OVERWRITE TABLE gluten_12474_tgt PARTITION (day)
|SELECT k, max(v) AS m, day
|FROM gluten_12474_src
|GROUP BY day, k
|""".stripMargin)

checkAnswer(
sql("SELECT k, m, day FROM gluten_12474_tgt"),
sql("SELECT k, v, day FROM gluten_12474_src"))
}
}
}

// TODO: fix in Spark-4.0
ignoreGluten(
"SPARK-41914: v1 write with AQE and in-partition sorted - non-string partition column") {
Expand Down
13 changes: 12 additions & 1 deletion shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,8 @@ import org.apache.spark.broadcast.Broadcast
import org.apache.spark.internal.io.FileCommitProtocol
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryArithmetic, Expression, InputFileBlockLength, InputFileBlockStart, InputFileName, RaiseError, UnBase64}
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions.{Attribute, BinaryArithmetic, Expression, InputFileBlockLength, InputFileBlockStart, InputFileName, RaiseError, SortOrder, UnBase64}
import org.apache.spark.sql.catalyst.plans.JoinType
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
Expand Down Expand Up @@ -129,6 +130,16 @@ trait SparkShims {

def enableNativeWriteFilesByDefault(): Boolean = false

// Planned V1 writes were introduced in Spark 3.4. Older versions do not expose a required
// ordering utility and keep the default empty ordering.
// TODO: Remove this shim after dropping Spark 3.3 support.
def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
numStaticPartitionCols: Int): Seq[SortOrder] = Seq.empty

def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T] = {
// Since Spark 3.4, the `sc.broadcast` has been optimized to use `sc.broadcastInternal`.
// More details see SPARK-39983.
Expand Down
15 changes: 15 additions & 0 deletions shims/spark34/src/main/scala/org/apache/gluten/sql/shims/spark34/Spark34Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.DecimalPrecision
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans.QueryPlan
Expand Down Expand Up @@ -208,6 +209,20 @@ class Spark34Shims extends SparkShims {

override def enableNativeWriteFilesByDefault(): Boolean = true

override def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
numStaticPartitionCols: Int): Seq[SortOrder] = {
V1WritesUtils.getSortOrder(
outputColumns,
partitionColumns,
bucketSpec,
options,
numStaticPartitionCols)
}

override def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T] = {
SparkContextUtils.broadcastInternal(sc, value)
}
Expand Down
15 changes: 15 additions & 0 deletions shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.{ExtendedAnalysisException, InternalRow}
import org.apache.spark.sql.catalyst.analysis.DecimalPrecision
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans.QueryPlan
Expand Down Expand Up @@ -249,6 +250,20 @@ class Spark35Shims extends SparkShims {

override def enableNativeWriteFilesByDefault(): Boolean = true

override def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
numStaticPartitionCols: Int): Seq[SortOrder] = {
V1WritesUtils.getSortOrder(
outputColumns,
partitionColumns,
bucketSpec,
options,
numStaticPartitionCols)
}

override def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T] = {
SparkContextUtils.broadcastInternal(sc, value)
}
Expand Down
15 changes: 15 additions & 0 deletions shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.{ExtendedAnalysisException, InternalRow}
import org.apache.spark.sql.catalyst.analysis.DecimalPrecisionTypeCoercion
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans.{JoinType, LeftSingle}
Expand Down Expand Up @@ -254,6 +255,20 @@ class Spark40Shims extends SparkShims {

override def enableNativeWriteFilesByDefault(): Boolean = true

override def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
numStaticPartitionCols: Int): Seq[SortOrder] = {
V1WritesUtils.getSortOrder(
outputColumns,
partitionColumns,
bucketSpec,
options,
numStaticPartitionCols)
}

override def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T] = {
SparkContextUtils.broadcastInternal(sc, value)
}
Expand Down
15 changes: 15 additions & 0 deletions shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.{AnalysisException, SparkSession}
import org.apache.spark.sql.catalyst.{ExtendedAnalysisException, InternalRow}
import org.apache.spark.sql.catalyst.analysis.DecimalPrecisionTypeCoercion
import org.apache.spark.sql.catalyst.catalog.BucketSpec
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.catalyst.plans.{JoinType, LeftSingle}
Expand Down Expand Up @@ -253,6 +254,20 @@ class Spark41Shims extends SparkShims {

override def enableNativeWriteFilesByDefault(): Boolean = true

override def getV1WriteRequiredOrdering(
outputColumns: Seq[Attribute],
partitionColumns: Seq[Attribute],
bucketSpec: Option[BucketSpec],
options: Map[String, String],
numStaticPartitionCols: Int): Seq[SortOrder] = {
V1WritesUtils.getSortOrder(
outputColumns,
partitionColumns,
bucketSpec,
options,
numStaticPartitionCols)
}

override def broadcastInternal[T: ClassTag](sc: SparkContext, value: T): Broadcast[T] = {
SparkContextUtils.broadcastInternal(sc, value)
}
Expand Down
Loading