ulysses-you commented on PR #59823: URL: https://github.com/apache/spark/pull/59823#issuecomment-6436941615
**0 blocking, 2 non-blocking.** Both non-blocking items are documentation. The behavior section is there so we review the same semantics, not as a request for change. | Item | Severity | |---|---| | `minRows = 0` behavior is undocumented | non-blocking (doc) | | `minRows` entry contradicts the new config | non-blocking (doc) | I traced both execution paths at `4d6fdc7` against base `f784da87e6d`: they are line-for-line equivalent, all four window counters reset with `processedRows` at a spill, and the pass-through ordering invariant is untouched. ### Definitions - **Epoch**: one in-memory aggregation map's lifetime. It starts at task start or right after a spill, where `processedRows`, `checkRows`/`checkKeys` and `windowRows`/`windowKeys` all reset to 0 (`HashAggregateExec.scala:1078-1084`, `TungstenAggregationIterator.scala:271-276`). - **M** = `minRows`, default 100000. - **R**: rows successfully aggregated into the current epoch's map when the spill check runs. The row that fails insertion counts on neither side. - **K(R)**: the epoch's key count at that point; with the two-level map, `fastNumKeys - fastKeysAtSpill + regularNumKeys` (`HashAggregateExec.scala:1016-1021`). ### The two check points | | Periodic | Spill | |---|---|---| | Trigger | `processedRows == nextCheckRow`, every M rows | buffer allocation fails | | Ratio | whole epoch | recent window | | Threshold | `minCompaction` (1.05) | `max(minCompaction, minCompactionToSpill)` (2.0) | | Floor | - | `processedRows >= max(M, 1)` | ### Window evolution, per epoch The window start lags the last check by one interval. At the first check there is no previous check to lag to, so it stays at the epoch start. | R at the fill | periodic checks run | `windowRows` | spill-check window | |---|---|---|---| | `R < M` | 0 | 0 | not evaluated; spills | | `M <= R < 2M` | 1 | 0 | `[0, R]`, the whole epoch | | `2M <= R < 3M` | 2 | M | `[M, R]` | | `kM <= R < (k+1)M` | k | (k-1)M | `[(k-1)M, R]` | So the recent-rows measurement needs an epoch of at least `2 * minRows` rows. Every spill restarts the epoch, so a task whose map fills in `[M, 2M)` is judged whole-epoch on every epoch, not only the first. I understand and do not object to the whole-epoch fallback in `[M, 2M)`. Two consequences are worth stating where users read them: the whole-epoch ratio is systematically lower than the recent one, because the epoch's early rows are the ones that add new keys, so the band biases toward pass-through; and the decision then depends on the map capacity rather than the data. Same data (200K distinct keys per task, 10 rows per key) with `minRows = 100000`: a fill at 220000 rows sees 120000 / 54700 = 2.19 and keeps aggregating; a fill at 150000 rows sees 150000 / 105500 = 1.42 and passes through. ### Non-blocking, documentation **1. Document `minRows = 0`.** Today the periodic check never fires, so the window never advances and every spill check judges the whole epoch against `max(minCompaction, minCompactionToSpill)`, with a floor of one row. **2. Align the `minRows` entry.** It says setting it to 0 leaves "only the evaluation when the map is about to spill", without saying that evaluation then measures the whole epoch. Adding ", which then measures all rows and keys of the current map epoch" keeps the two configs from contradicting each other. -- 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]
