ulysses-you commented on code in PR #57363:
URL: https://github.com/apache/spark/pull/57363#discussion_r3618935409
##########
sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDAFSuite.scala:
##########
@@ -102,28 +102,36 @@ class HiveUDAFSuite extends QueryTest
test("SPARK-24935: customized Hive UDAF with two aggregation buffers") {
withTempView("v") {
spark.range(100).createTempView("v")
- val df = sql("SELECT id % 2, mock2(id) FROM v GROUP BY id % 2")
-
- val aggs = collect(df.queryExecution.executedPlan) {
- case agg: ObjectHashAggregateExec => agg
- }
-
- // There should be two aggregate operators, one for partial aggregation,
and the other for
- // global aggregation.
- assert(aggs.length == 2)
-
- withSQLConf(SQLConf.OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD.key -> "1")
{
- checkAnswer(df, Seq(
- Row(0, Row(50, 0)),
- Row(1, Row(50, 0))
- ))
- }
-
- withSQLConf(SQLConf.OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD.key ->
"100") {
- checkAnswer(df, Seq(
- Row(0, Row(50, 0)),
- Row(1, Row(50, 0))
- ))
+ // `range(100)` is a single partition, so the partial and final
aggregate would otherwise be
+ // adjacent and merged by `CombineAdjacentAggregation`. `MockUDAF2`
deliberately uses distinct
+ // aggregation-buffer classes per aggregation mode (one for consuming
original input, another
+ // for merging partial buffers), so it does not support a single
`Complete`-mode aggregate.
+ // Keep the partial/final pair uncombined so the sort-based fallback
path below is exercised
+ // in the mode this UDAF is designed for.
+ withSQLConf(SQLConf.COMBINE_ADJACENT_AGGREGATION_ENABLED.key -> "false")
{
Review Comment:
Thanks for the detailed repro. Fixed in the latest push by making the Hive
wrapper Complete-safe rather than excluding UDAFs or pinning the test.
`HiveUDAFFunction` keeps two evaluators: `partial1HiveEvaluator` (PARTIAL1,
buffer created in `update`) and `finalHiveEvaluator` (FINAL, buffer created in
`merge`). `eval` always terminated via the FINAL evaluator, so a buffer that
only went through `update` (the Complete-mode path) was a PARTIAL1-mode buffer
handed to `terminate`, which mode-aware UDAFs reject.
`merge` already converted a PARTIAL1 buffer to a FINAL buffer on demand. I
extracted that into `toFinalBuffer` and call it from `eval` as well, so
Complete-mode (update without merge) terminates against a FINAL buffer.
SPARK-24935 now runs on the default (combined) path and asserts the single
Complete-mode `ObjectHashAggregateExec`, with both sort-fallback thresholds
still checked. I verified a mode-aware UDAF variant no longer throws
`ClassCastException`.
The change is in `sql/hive/.../hiveUDFs.scala`; `HiveUDAFSuite`,
`ObjectHashAggregateSuite`, and `UDAQuerySuite` pass.
--
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]
