Copilot commented on code in PR #12278:
URL: https://github.com/apache/gluten/pull/12278#discussion_r3410851303
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
+ val aggOp = SortAggregateExec(
+ None,
+ isStreaming = false,
+ None,
+ Seq.empty,
+ aggregates,
+ statsAttrs,
+ 0,
+ statsResultAttrs,
+ StatisticsInputNode(dataCols))
+ val projOp = ProjectExec(statsResultAttrs, aggOp)
+ val offloads = Seq(OffloadOthers()).map(_.toStrcitRule())
Review Comment:
Likely typo in method name: `toStrcitRule()` looks misspelled and will fail
compilation unless such a method actually exists. If the intended API is
`toStrictRule()`, rename the call accordingly (and keep it consistent with
other call sites/usages of the same rule-conversion helper).
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
+ val statsAttrs =
aggregates.flatMap(_.aggregateFunction.aggBufferAttributes)
+ val statsResultAttrs =
aggregates.flatMap(_.aggregateFunction.inputAggBufferAttributes)
+ val aggOp = SortAggregateExec(
+ None,
+ isStreaming = false,
+ None,
+ Seq.empty,
+ aggregates,
+ statsAttrs,
+ 0,
+ statsResultAttrs,
+ StatisticsInputNode(dataCols))
+ val projOp = ProjectExec(statsResultAttrs, aggOp)
+ val offloads = Seq(OffloadOthers()).map(_.toStrcitRule())
Review Comment:
Likely typo in method name: `toStrcitRule()` looks misspelled and will fail
compilation unless such a method actually exists. If the intended API is
`toStrictRule()`, rename the call accordingly (and keep it consistent with
other call sites/usages of the same rule-conversion helper).
##########
backends-velox/src-delta33/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
Review Comment:
`canOffloadStats` only collects `AggregateExpression`s whose functions are
`DeclarativeAggregate`, but it does not explicitly reject the presence of
*non*-declarative aggregates. If `statsColExpr` contains any non-declarative
aggregates, this code will build a different aggregate plan than the real one
and may incorrectly return `true`, causing the native stats tracker path to be
selected and potentially reintroducing the ClassCastException/crash you’re
trying to avoid. Fix by detecting any `AggregateExpression` that is *not*
backed by a `DeclarativeAggregate` and returning `false` (fallback) in that
case.
##########
backends-velox/src-delta40/main/scala/org/apache/spark/sql/delta/stats/GlutenDeltaJobStatsTracker.scala:
##########
@@ -99,6 +110,48 @@ object GlutenDeltaJobStatsTracker extends Logging {
new GlutenDeltaJobStatsFallbackTracker(tracker)
}
+ /**
+ * Returns whether the Delta per-file statistics aggregation can be
offloaded to a Velox
+ * whole-stage transformer. This mirrors the plan that
[[GlutenDeltaTaskStatsTracker]] builds on
+ * the executors: if the aggregation/projection is not supported by Velox it
stays a vanilla
+ * [[ProjectExec]] (i.e. does not collapse into a
[[WholeStageTransformer]]), in which case the
+ * native stats tracker must not be used. Evaluated once on the driver so
the executors never
+ * allocate native resources for a plan that cannot run.
+ */
+ private def canOffloadStats(dataCols: Seq[Attribute], statsColExpr:
Expression): Boolean = {
+ try {
+ val aggregates = statsColExpr.collect {
+ case ae: AggregateExpression if
ae.aggregateFunction.isInstanceOf[DeclarativeAggregate] =>
+ ae
+ }
Review Comment:
`canOffloadStats` only collects `AggregateExpression`s whose functions are
`DeclarativeAggregate`, but it does not explicitly reject the presence of
*non*-declarative aggregates. If `statsColExpr` contains any non-declarative
aggregates, this code will build a different aggregate plan than the real one
and may incorrectly return `true`, causing the native stats tracker path to be
selected and potentially reintroducing the ClassCastException/crash you’re
trying to avoid. Fix by detecting any `AggregateExpression` that is *not*
backed by a `DeclarativeAggregate` and returning `false` (fallback) in that
case.
--
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]
