sunchao commented on code in PR #57363:
URL: https://github.com/apache/spark/pull/57363#discussion_r3616717627
##########
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:
[P1] Please do not disable this regression test to accommodate the new
default. A Complete object aggregate calls `HiveUDAFFunction.update`, which
creates a PARTIAL1-mode buffer, and then `eval`, which passes that buffer to
the separately initialized FINAL evaluator. Mode-aware Hive UDAFs may legally
use different buffer classes in those modes, so the default rewrite can turn a
previously supported query into a `ClassCastException`. I reproduced this and
also verified that a variant which correctly handles Hive's native COMPLETE
lifecycle still fails through Spark's mixed-evaluator path. Please make the
Hive wrapper Complete-safe or exclude these UDAFs before enabling the rule by
default, and keep this test exercising the default path.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
- Since Spark 4.3, the configuration key
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` has
been renamed to
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` to
reflect that it now applies to storage-partitioned joins, aggregates, and
windows. The old key continues to work as an alias.
- Since Spark 4.3, the Spark Thrift Server rejects setting JVM system
properties through the `set:system:` session configuration overlay (for
example, in a JDBC connection string). To restore the previous behavior, set
`spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties` to `true`.
- Since Spark 4.3, the adaptive execution rule
`org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` has been renamed
to `DemoteBroadcastHashJoin`, which now only demotes broadcast hash joins
(emitting `NO_BROADCAST_HASH`). Its selection of shuffled hash join over sort
merge join has moved to a new physical rule gated by
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` (default
`true`). If you previously disabled the shuffled-hash-join preference by
listing `org.apache.spark.sql.execution.adaptive.DynamicJoinSelection` in
`spark.sql.adaptive.optimizer.excludedRules`, that name no longer matches any
rule (unknown names are silently ignored); set
`spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled` to `false`
instead.
+- Since Spark 4.3, `spark.sql.execution.replaceHashWithSortAgg` defaults to
`true`. Spark now replaces a hash-based aggregate with a sort aggregate when
the aggregate's child is already sorted on the grouping keys. To restore the
previous behavior, set `spark.sql.execution.replaceHashWithSortAgg` to `false`.
Review Comment:
[P2] Please document `spark.sql.execution.combineAdjacentAggregation` here
as well. Although the config itself is introduced during 4.3 development, its
default-on execution behavior is new for users upgrading from 4.2. The two
settings are now independent, so following this rollback instruction and
disabling only `replaceHashWithSortAgg` still leaves adjacent Partial/Final
aggregation combined. Please restore the removed entry and state that both
settings may need to be false to recover the previous staging.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -3006,18 +3006,17 @@ object SQLConf {
.doc("Whether to replace hash aggregate node with sort aggregate based on
children's ordering")
.version("3.3.0")
.booleanConf
- .createWithDefault(false)
+ .createWithDefault(true)
val COMBINE_ADJACENT_AGGREGATION_ENABLED =
buildConf("spark.sql.execution.combineAdjacentAggregation")
.internal()
.doc("When true, combine adjacent aggregation with `Partial` and `Final`
to `Complete` " +
- "mode. This defaults to the value of
`spark.sql.execution.replaceHashWithSortAgg` since " +
- "combining adjacent aggregation subsumes the partial-and-final merge
that " +
- "`replaceHashWithSortAgg` used to perform on its own.")
+ "mode.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .fallbackConf(REPLACE_HASH_WITH_SORT_AGG_ENABLED)
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
[P2] Enabling this rewrite changes built-in aggregate results, not only the
physical plan. With AQE off and a single-partition group containing two finite
`1e155` values, disabling combining returns `NaN` for `var_pop`, `covar_pop`,
and `regr_sxy`, while Complete returns `0.0`. The old NaN comes from
overflowing the first merge into the empty Final buffer, so 0.0 is
mathematically preferable, but a default physical optimization should not
silently introduce config-dependent results. Please fix and test the
empty-buffer merge as a deliberate correctness change, or keep these aggregates
out of combining until the modes are equivalent.
--
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]
