Copilot commented on code in PR #12653:
URL: https://github.com/apache/gluten/pull/12653#discussion_r3673891683
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/MergeTwoPhasesHashBaseAggregate.scala:
##########
@@ -59,6 +64,26 @@ case class MergeTwoPhasesHashBaseAggregate(session:
SparkSession)
}
}
+ /**
+ * Builds Complete mode aggregate expressions from the final aggregate. The
physical final
+ * aggregate no longer carries the FILTER predicate (see `isPartialAgg`), so
the FILTER is
+ * restored from the partial aggregate, whose expressions align one-to-one
with the final ones.
+ */
+ private def toCompleteAggregateExpressions(
+ partialAgg: BaseAggregateExec,
+ finalAggExpressions: Seq[AggregateExpression]): Seq[AggregateExpression]
= {
+ require(
+ finalAggExpressions.length == partialAgg.aggregateExpressions.length,
+ s"Expected partial and final aggregate expression lists to align 1:1,
but got " +
+ s"${partialAgg.aggregateExpressions.length} partial and " +
+ s"${finalAggExpressions.length} final expressions"
+ )
+ finalAggExpressions.zip(partialAgg.aggregateExpressions).map {
+ case (finalExpr, partialExpr) =>
+ finalExpr.copy(mode = Complete, filter = partialExpr.filter)
+ }
Review Comment:
This restores `FILTER` by positional alignment only. If Spark ever reorders
(or otherwise rewrites) `partialAgg.aggregateExpressions` vs
`finalAggExpressions` while keeping the same length, the wrong `FILTER`
predicate can be attached to the wrong aggregate expression, producing
incorrect results (especially for mixed filtered + non-filtered aggregates). A
more robust approach is to validate pairwise identity before copying the
`filter` (e.g., match on `resultId` / `aggregateFunction` semantic equality)
and skip merging if a mismatch is detected.
##########
gluten-substrait/src/main/scala/org/apache/gluten/extension/columnar/MergeTwoPhasesHashBaseAggregate.scala:
##########
@@ -59,6 +64,26 @@ case class MergeTwoPhasesHashBaseAggregate(session:
SparkSession)
}
}
+ /**
+ * Builds Complete mode aggregate expressions from the final aggregate. The
physical final
+ * aggregate no longer carries the FILTER predicate (see `isPartialAgg`), so
the FILTER is
+ * restored from the partial aggregate, whose expressions align one-to-one
with the final ones.
+ */
+ private def toCompleteAggregateExpressions(
+ partialAgg: BaseAggregateExec,
+ finalAggExpressions: Seq[AggregateExpression]): Seq[AggregateExpression]
= {
+ require(
+ finalAggExpressions.length == partialAgg.aggregateExpressions.length,
+ s"Expected partial and final aggregate expression lists to align 1:1,
but got " +
+ s"${partialAgg.aggregateExpressions.length} partial and " +
+ s"${finalAggExpressions.length} final expressions"
+ )
Review Comment:
`require(...)` in a Spark planner/optimizer rule can fail the entire query
planning/execution path with an exception. Since this rule is an optimization,
it’s generally safer to “decline to merge” and fall back to the original plan
when invariants aren’t met. Consider replacing the `require` with a graceful
fallback (e.g., return `finalAggExpressions.map(_.copy(mode = Complete))` or
simply skip merging earlier) to avoid turning a non-critical optimization into
a hard failure.
--
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]
