ulysses-you commented on code in PR #12514:
URL: https://github.com/apache/gluten/pull/12514#discussion_r3627753644
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala:
##########
@@ -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.
+ 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 {
Review Comment:
I do not get this change, what's wrong with previous code?
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/EnsureLocalSortRequirements.scala:
##########
@@ -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.
Review Comment:
It seems we can make `WriteFilesExec` supports require child ordering native
in Spark upstream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
