Skip to content

[GLUTEN-11752] Fix AdaptiveSparkPlanExec accessibility in columnar write optimization #11753

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

Open
wangyum wants to merge 6 commits into
base: main
Choose a base branch
from
+39 −34
Open

[GLUTEN-11752] Fix AdaptiveSparkPlanExec accessibility in columnar write optimization #11753

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
e498f03
[CORE] Fix AdaptiveSparkPlanExec accessibility in columnar write opti…
wangyum Mar 13, 2026
d81d958
fix format
wangyum Mar 13, 2026
6b2afb4
fix
wangyum Mar 14, 2026
9f63ddf
fix
wangyum Mar 14, 2026
df71eb9
fix:
wangyum Mar 14, 2026
a66aff7
Remove test
wangyum Mar 15, 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
25 changes: 15 additions & 10 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/GlutenWriterColumnarRules.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,22 +79,27 @@ object GlutenWriterColumnarRules {
// So FakeRowAdaptor will always consumes columnar data,
// thus avoiding the case of c2r->aqe->r2c->writer
case aqe: AdaptiveSparkPlanExec =>
command.withNewChildren(
Array(
BackendsApiManager.getSparkPlanExecApiInstance.genColumnarToCarrierRow(
AdaptiveSparkPlanExec(
aqe.inputPlan,
aqe.context,
aqe.preprocessingRules,
aqe.isSubquery,
supportsColumnar = true
))))
val newChild = BackendsApiManager.getSparkPlanExecApiInstance
.genColumnarToCarrierRow(aqe.inputPlan)
command.withNewChildren(Array(wrapColumnarToRowWithAqe(newChild, aqe)))
case other =>
command.withNewChildren(
Array(BackendsApiManager.getSparkPlanExecApiInstance.genColumnarToCarrierRow(other)))
}
}

private def wrapColumnarToRowWithAqe(
newChild: SparkPlan,
aqe: AdaptiveSparkPlanExec): AdaptiveSparkPlanExec = {
aqe.inputPlan.logicalLink.foreach(newChild.setLogicalLink)
AdaptiveSparkPlanExec(
newChild,
aqe.context,
aqe.preprocessingRules,
aqe.isSubquery,
supportsColumnar = false)
}

case class NativeWritePostRule(session: SparkSession) extends Rule[SparkPlan] {

override def apply(p: SparkPlan): SparkPlan = p match {
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1255,12 +1255,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1259,12 +1259,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1208,12 +1208,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
Copy link
Contributor

@JkSelf JkSelf Mar 18, 2026

Choose a reason for hiding this comment

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

GlutenWriterColumnarRules should be only used in spark 32 and spark 33. I have filed PR#11787 to fix. After merging, we can remove thoese changes in spark 34 to spark 41.

assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1208,12 +1208,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1214,12 +1214,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
8 changes: 4 additions & 4 deletions ...est/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1218,12 +1218,12 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT
sparkContext.listenerBus.waitUntilEmpty()
assert(plan.isInstanceOf[V2TableWriteExec])
val childPlan = plan.asInstanceOf[V2TableWriteExec].child
assert(childPlan.isInstanceOf[ColumnarToCarrierRowExecBase])
assert(childPlan.isInstanceOf[AdaptiveSparkPlanExec])
assert(
childPlan
.asInstanceOf[ColumnarToCarrierRowExecBase]
.child
.isInstanceOf[AdaptiveSparkPlanExec])
.asInstanceOf[AdaptiveSparkPlanExec]
.inputPlan
.isInstanceOf[ColumnarToCarrierRowExecBase])

spark.listenerManager.unregister(listener)
}
Expand Down
Loading