jayzhan211 commented on issue #20773: URL: https://github.com/apache/datafusion/issues/20773#issuecomment-5762104712
I prototyped the second half of this idea (draft PR #25567) — a Final aggregation that, once its table outgrows the cache, moves its state and all further input into 64 hash buckets and aggregates them one at a time — and measured where the time goes. Two findings seem relevant to the design here. **1. Re-partitioning a table that was already built is what loses.** Emitting the first table, scattering it and interning its groups again costs about as much as building it did. With a threshold of `T` groups, every query that ends at 1-4x `T` groups per partition pays that without earning it back, and raising `T` only moves the loss to other queries: | threshold | slower than main | |---|---| | 256k | Q13 1.18x, Q14 1.11x | | 512k | Q13 1.60x, Q14 1.26x | | 1M | Q33 1.23x, Q31 1.15x, Q35 1.12x | | 2M | Q16 1.09x | while the large aggregations are best with an early switch (Q18 0.39x at 256k, 0.51x at 2M). DuckDB's ICDE 2024 paper makes the same point against the HyPer / BLU design: "Rather than partitioning tuples when the hash table is reset ... our implementation directly materializes tuples into partitions ... we avoid copying tuples more than once", and a full table only has its pointer array reset "while the tuples stay in place". I would aim for that shape — rows go to their partition once, from the first batch — rather than "one table, radix-repartition it when it grows". **2. The second scatter is 35-55% of the bucketed Final's compute.** Per Final input row: | query | one table | bucketed = routing + aggregation | |---|---|---| | Q31 (2 int keys) | 85 ns | 44 = 24.5 + 19.5 | | Q18 (int + string keys) | 229 | 71 = 27 + 44 | | Q14 (string key) | 79-85 | 91-99 = 42-46 + 49-54 | | Q12 (string key) | 91 | 97 = 39 + 58 | `RepartitionExec` has already hashed, gathered and copied each of these rows once; the Final does it again. If the exchange produced `partitions x K` sub-buckets from the same hash, the routing column would disappear: the mid-size string-key queries would turn from 1.03-1.07x into roughly 0.92x, and the queries that already win would gain about as much again. A standalone harness (no Arrow; one copy into 64 buckets, hash kept with the row) confirms that partitioning pays for string keys from 250k groups per thread: 24-byte keys 0.63x / 0.56x / 0.48x / 0.34x of a single table at 250k / 500k / 1M / 4M groups, 12 threads. Re-hashing in the bucket instead of carrying the hash costs little for short keys and a lot for long ones (60 bytes at 1M groups: 0.54x -> 0.78x), which ties in with #11680. Numbers are from an Apple M4 Pro, 12 partitions, ClickBench `hits_partitioned`, interleaved runs. I am prototyping the "repartition produces the buckets" variant next and will report what it measures here. -- 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]
