[GLUTEN-12474][VL] Route WriteFiles required-ordering through the shim to fix Spark 3.3 build.

littlelittlewhite09 · littlelittlewhite09 · commit d4a00e83b988 · 2026-07-15T22:20:07.000+08:00
EnsureRowBasedWriteFilesOrdering imported V1WritesUtils directly, but that class
only exists since Spark 3.4 (planned write, SPARK-37287). Since the rule lives in
gluten-substrait, which is compiled against every supported Spark profile,
building with -Pspark-3.3 failed to resolve the import.

Move the version-specific `getSortOrder` behind SparkShims:
  - Add SparkShims.getWriteFilesRequiredOrdering, defaulting to Nil (covers
    Spark 3.2/3.3, where the planned-write WriteFilesExec path is not used at
    runtime, so the rule is a no-op there).
  - Override it in the Spark 3.4/3.5/4.0/4.1 shims to delegate to
    V1WritesUtils.getSortOrder, keeping behavior identical (including the vivo
    MergeSmallFilesContext and concurrent-writer cases).
  - Inline isOrderingMatched in the rule; it is pure catalyst and identical
    across versions, so it needs no shim.
diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureRowBasedWriteFilesOrdering.scala b/gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureRowBasedWriteFilesOrdering.scala
@@ -17,11 +17,12 @@
 package org.apache.gluten.extension.columnar
 
 import org.apache.gluten.execution.{SortExecTransformer, TransformSupport}
+import org.apache.gluten.sql.shims.SparkShimLoader
 
-import org.apache.spark.sql.catalyst.expressions.SortOrder
+import org.apache.spark.sql.catalyst.expressions.{Expression, SortOrder}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.{ColumnarWriteFilesExec, SortExec, SparkPlan}
-import org.apache.spark.sql.execution.datasources.{V1WritesUtils, WriteFilesExec}
+import org.apache.spark.sql.execution.datasources.WriteFilesExec
 
 /**
  * Re-adds the local sort on dynamic partition (and bucket) columns that a row-based
@@ -46,9 +47,13 @@ import org.apache.spark.sql.execution.datasources.{V1WritesUtils, WriteFilesExec
  * `WriteFilesExec` that [[ColumnarWriteFilesExec]] injects for the native path is skipped by
  * checking for its [[ColumnarWriteFilesExec.NoopLeaf]] child.
  *
- * The required ordering and the "already satisfied" check reuse [[V1WritesUtils]] so that the
- * behavior stays identical to vanilla `FileFormatWriter` (including the concurrent-writers and
- * small-file-merge cases where `getSortOrder` returns empty).
+ * The required ordering is obtained through [[SparkShimLoader]] from `V1WritesUtils.getSortOrder`
+ * so that the behavior stays identical to vanilla `FileFormatWriter` (including the
+ * concurrent-writers and small-file-merge cases where `getSortOrder` returns empty). It is fetched
+ * via the shim because `V1WritesUtils` only exists since Spark 3.4; on Spark 3.2/3.3 the shim
+ * returns `Nil` and this rule is a no-op (the planned-write `WriteFilesExec` path is not used
+ * there). The "already satisfied" check mirrors `V1WritesUtils.isOrderingMatched`, which is pure
+ * catalyst and version-agnostic, so it is inlined here.
  *
  * This rule must run before [[org.apache.gluten.extension.columnar.transition.InsertTransitions]]:
  * when the write's child is an offloaded transformer we insert a native [[SortExecTransformer]]
@@ -58,22 +63,38 @@ import org.apache.spark.sql.execution.datasources.{V1WritesUtils, WriteFilesExec
 object EnsureRowBasedWriteFilesOrdering extends Rule[SparkPlan] {
   override def apply(plan: SparkPlan): SparkPlan = plan.transformUp {
     case w: WriteFilesExec if !w.child.isInstanceOf[ColumnarWriteFilesExec.NoopLeaf] =>
-      val requiredOrdering: Seq[SortOrder] = V1WritesUtils.getSortOrder(
-        w.child.output,
-        w.partitionColumns,
-        w.bucketSpec,
-        w.options,
-        w.staticPartitions.size)
+      val requiredOrdering: Seq[SortOrder] =
+        SparkShimLoader.getSparkShims.getWriteFilesRequiredOrdering(
+          w.child.output,
+          w.partitionColumns,
+          w.bucketSpec,
+          w.options,
+          w.staticPartitions.size)
       if (
         requiredOrdering.isEmpty ||
-        V1WritesUtils.isOrderingMatched(requiredOrdering.map(_.child), w.child.outputOrdering)
+        isOrderingMatched(requiredOrdering.map(_.child), w.child.outputOrdering)
       ) {
         w
       } else {
         w.withNewChildren(sortPlan(w.child, requiredOrdering) :: Nil)
       }
   }
 
+  // Mirrors `V1WritesUtils.isOrderingMatched`. Pure catalyst and identical across Spark versions,
+  // so it is inlined to avoid depending on `V1WritesUtils`, which is absent before Spark 3.4.
+  private def isOrderingMatched(
+      requiredOrdering: Seq[Expression],
+      outputOrdering: Seq[SortOrder]): Boolean = {
+    if (requiredOrdering.length > outputOrdering.length) {
+      false
+    } else {
+      requiredOrdering.zip(outputOrdering).forall {
+        case (requiredOrder, outputOrder) =>
+          outputOrder.satisfies(outputOrder.copy(child = requiredOrder))
+      }
+    }
+  }
+
   private def sortPlan(child: SparkPlan, requiredOrdering: Seq[SortOrder]): SparkPlan =
     child match {
       case c: TransformSupport =>
diff --git a/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala b/shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala
@@ -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
@@ -129,6 +130,19 @@ trait SparkShims {
 
   def enableNativeWriteFilesByDefault(): Boolean = false
 
+  /**
+   * The local sort a row-based `WriteFilesExec` requires on its child for dynamic partition (and
+   * bucket) writes. Delegates to `V1WritesUtils.getSortOrder`, which only exists since Spark 3.4;
+   * on Spark 3.2/3.3 the planned-write `WriteFilesExec` path is not used at runtime, so the default
+   * returns `Nil`.
+   */
+  def getWriteFilesRequiredOrdering(
+      outputColumns: Seq[Attribute],
+      partitionColumns: Seq[Attribute],
+      bucketSpec: Option[BucketSpec],
+      options: Map[String, String],
+      numStaticPartitionCols: Int): Seq[SortOrder] = Nil
+
   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.
diff --git a/shims/spark34/src/main/scala/org/apache/gluten/sql/shims/spark34/Spark34Shims.scala b/shims/spark34/src/main/scala/org/apache/gluten/sql/shims/spark34/Spark34Shims.scala
@@ -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
@@ -208,6 +209,19 @@ class Spark34Shims extends SparkShims {
 
   override def enableNativeWriteFilesByDefault(): Boolean = true
 
+  override def getWriteFilesRequiredOrdering(
+      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)
   }
diff --git a/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala b/shims/spark35/src/main/scala/org/apache/gluten/sql/shims/spark35/Spark35Shims.scala
@@ -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
@@ -249,6 +250,19 @@ class Spark35Shims extends SparkShims {
 
   override def enableNativeWriteFilesByDefault(): Boolean = true
 
+  override def getWriteFilesRequiredOrdering(
+      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)
   }
diff --git a/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala b/shims/spark40/src/main/scala/org/apache/gluten/sql/shims/spark40/Spark40Shims.scala
@@ -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}
@@ -254,6 +255,19 @@ class Spark40Shims extends SparkShims {
 
   override def enableNativeWriteFilesByDefault(): Boolean = true
 
+  override def getWriteFilesRequiredOrdering(
+      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)
   }
diff --git a/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala b/shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala
@@ -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}
@@ -253,6 +254,19 @@ class Spark41Shims extends SparkShims {
 
   override def enableNativeWriteFilesByDefault(): Boolean = true
 
+  override def getWriteFilesRequiredOrdering(
+      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)
   }
