peter-toth commented on code in PR #57815:
URL: https://github.com/apache/spark/pull/57815#discussion_r3763462611
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/windowExpressions.scala:
##########
@@ -337,6 +337,23 @@ object WindowExpression {
e.find(_.isInstanceOf[WindowExpression]).isDefined
}
+ private[sql] def distinctAggregateChildren(function: AggregateFunction):
Seq[Expression] = {
+ function.children.filterNot(_.foldable).map {
+ case sortOrder: SortOrder => sortOrder.child
+ case expression => expression
+ }.distinctBy(_.canonicalized)
+ }
+
+ private[sql] def isSupportedDistinctAggregate(
Review Comment:
**Finding 11.** This predicate admits DISTINCT aggregates for which DISTINCT
cannot change the result, and they then pay for the full dedup path — a
`BytesToBytesMap` and, on a growing frame, two `UnsafeKVExternalSorter`s per
window partition — to produce what `UnboundedPrecedingWindowFunctionFrame`
(`WindowFunctionFrame.scala:565`) produces by streaming rows into the buffer.
The set of functions is already enumerated in the optimizer:
`EliminateDistinct.isDuplicateAgnostic`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:612-621`)
lists `Max`, `Min`, `BitAndAgg`, `BitOrAgg`, `CollectSet`, `First`, `Last`.
But `apply` matches `case agg: Aggregate` only (`:597-610`), so a window
expression never reaches it. Before this PR that didn't matter — `max(DISTINCT
x) OVER (...)` failed analysis; now it silently takes the slow path.
For `Max`/`Min`/`BitAndAgg`/`BitOrAgg`/`CollectSet` dropping `isDistinct` is
exactly equivalent on any frame, so widening the rule is safe. The pruning
predicate needs widening too, since `Window` carries `WINDOW`, not `AGGREGATE`
(`basicLogicalOperators.scala:1328`):
```scala
override def apply(plan: LogicalPlan): LogicalPlan =
plan.transformWithPruning(
_.containsAnyPattern(AGGREGATE, WINDOW)) {
case agg: Aggregate => ... // unchanged
case w: Window =>
w.transformExpressionsWithPruning(_.containsPattern(AGGREGATE_EXPRESSION)) {
case ae: AggregateExpression if ae.isDistinct &&
isDuplicateAgnostic(ae.aggregateFunction) =>
ae.copy(isDistinct = false)
}
}
```
`First`/`Last` want a second look before they join that branch: dedup feeds
a growing frame in first-occurrence order, so over `a, b, a` the third row of
`last(DISTINCT v) OVER (ORDER BY id ROWS BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW)` is `b` today, while `last(v)` gives `a`. Both are defensible for
an unordered distinct set, but it is a visible difference — so I'd leave those
two out or handle them deliberately.
Fine as a separate JIRA rather than growing this PR — I'm flagging it
because this PR is what makes the slow path reachable.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -4678,6 +4678,15 @@ object SQLConf {
.version("4.1.0")
.fallbackConf(SHUFFLE_SPILL_MAX_SIZE_FORCE_SPILL_THRESHOLD)
+ val WINDOW_EXEC_DISTINCT_HASH_FALLBACK_THRESHOLD =
+ buildConf("spark.sql.windowExec.distinct.hash.fallbackThreshold")
+ .internal()
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .doc("Maximum number of distinct keys kept in the in-memory hash map. A
new key after " +
+ "this threshold makes a window DISTINCT aggregate fall back to
external sorting")
+ .version("4.4.0")
+ .fallbackConf(WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD)
Review Comment:
**Finding 10.** This inherits
`spark.sql.windowExec.buffer.in.memory.threshold`, so the map is capped at 4096
distinct keys by default — and the cap is a cliff, not a soft limit. Once a
partition crosses it, `usingMap` is false for the rest of that partition, so
every remaining candidate row goes into the distinct-key sorter, duplicates
included (`DistinctWindowFunctionFrame.scala:161-164`, after `:174-177` drains
the map into it). A partition with 1M rows and 20k distinct values then does a
1M-record external sort, which will normally spill, where the hash path would
have held 20k short keys in a couple of MB.
The two configs also measure unrelated things: the fallback conf bounds
*rows the window operator keeps in memory before spilling its input buffer*,
this one bounds *distinct keys in a hash map*. As written, raising the row
buffer silently raises the distinct hash cap.
And the key count isn't what keeps the map safe — memory is:
`getTotalMemoryConsumption >= spillSizeThreshold`, `Location.append` returning
false (reached only after `allocatePage` has already asked other consumers to
spill), and the constructor `SparkOutOfMemoryError` catch you added at
`:149-153`. Hash aggregation relies on exactly that with no key-count cap at
all — `TungstenAggregationIterator.processInputs:197-214` switches to
sort-based aggregation only when `getAggregationBufferFromUnsafeRow` returns
null.
So I'd give this conf its own default, in the spirit of
`spark.shuffle.spill.numElementsForceSpillThreshold` ("By default it's
Integer.MAX_VALUE, which means we never force the sorter to spill, until we
reach some limitations"):
```suggestion
.intConf
.createWithDefault(Int.MaxValue)
```
If you'd rather keep a finite valve, pick a number tied to the map rather
than to the row buffer, and say so in the `doc`.
--
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]
