peter-toth commented on code in PR #57742:
URL: https://github.com/apache/spark/pull/57742#discussion_r3758036592
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -191,29 +209,67 @@ class TungstenAggregationIterator(
}
} else {
var i = 0
- while (inputIter.hasNext) {
+ var processedRows = 0L
+ val minRows = adaptiveMinRows
+ // The processed-row count at which the compaction ratio is evaluated
next. It advances by
+ // `minRows` after every check, and restarts after a spill so the new
in-memory map epoch is
+ // judged on its own rows. `minRows = 0` disables the periodic check:
the count is only ever
+ // compared after being incremented past 0, so it never matches and only
the spill check
+ // below remains.
+ var nextCheckRow = minRows
+ // The partial aggregation is ineffective when it does not collapse
`minCompaction` rows into
+ // one key. There is no fast map on this path, so the map's keys are all
the operator holds.
+ def ineffective(): Boolean =
+ processedRows < hashMap.getNumKeys().toDouble * adaptiveMinCompaction
+ while (inputIter.hasNext && !passThrough) {
Review Comment:
**Finding 15.** Fair enough — the cost list is accurate and the trade is
yours. Closing this from my side; finding 1 then has to be settled inside the
freeze design.
--
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]
