comphead opened a new pull request, #7345: URL: https://github.com/apache/datafusion-comet/pull/7345
## Which issue does this PR close? Part of #7341. This is the first item of the solution proposed there. The `GroupsAccumulator` and the `datasketches` changes it also lists are not in this PR, so #7341 stays open. ## Rationale for this change `HllSketchAccumulator::merge_batch` merged each partial sketch through `SparkHllSketch::merge_sketch`, which built a new `SparkHllUnion` for the row, merged the accumulator and the partial into it, and turned the union back into a sketch. Once a group's sketch holds the dense register array, `datasketches` 0.3.0 copies that array into the empty union and rescans all 2^lgConfigK registers to rebuild its estimator. So every merged row costs as much as the whole array, 262,144 registers at lgConfigK 18. The issue has the profile and the numbers from a Spark 4.0 job where the Final aggregation's median task took 40.4 s against Spark's 8.9 s. ## What changes are included in this PR? - `HllSketchAccumulator` is an enum of a `SparkHllSketch` and a `SparkHllUnion`. Input rows are hashed into the sketch as before. The first `merge_batch` folds the sketch into a union, which merges every partial sketch from then on, as `HllUnionAccumulator` already does. `state()` and `evaluate()` serialize whichever of the two the accumulator holds. - The accumulator holds one or the other rather than both, so it grows by 8 bytes, from 96 to 104. An empty sketch kept next to the union would roughly double what a small group is charged in a Final aggregation. - DataFusion gives an accumulator either `update_batch` or `merge_batch` calls. If one ever gets rows after a merge, they reach the union through a sketch of their own. - `SparkHllSketch::merge_sketch` and the private `SparkHllUnion::to_sketch` are removed, since nothing else used them. `SparkHllUnion::lg_max_k` is added for the case above. - The `hll_agg` bench gets a Final-mode case, `hll_sketch_agg_final_dense`: 8,000 partial states at lgConfigK 18, 20 groups of 400 partials with 200 distinct values each, read one partial per group per round as a reducer would. Estimates can move slightly once a group is dense. Before, every merge into a dense group marked its estimator out of order, which makes the estimate come from the composite estimator instead of the HIP accumulator. Now only merging a dense partial does that, as in datasketches-java, which Spark's `HllSketchAgg.merge` uses. LIST and SET mode results are unchanged. In the dense probe case below, the sum of the 20 estimates went from 1,600,370 to 1,600,533, against a true 1,600,000. ## How are these changes tested? Two new unit tests in `hll_sketch_agg.rs`: - `merge_batch_keeps_one_union_across_calls` merges 40 partial sketches in batches of 7 and checks after every batch that the accumulator holds the same bytes as one `SparkHllUnion` fed the same sketches, in SET mode and after the group turns dense. - `rows_before_and_after_a_merge_all_count` gives one accumulator rows, then a partial sketch, then more rows, and checks that the estimate counts all three. The existing tests cover the update path, the empty group and a Partial/Final plan. Benchmarks, release build on an Apple M5 Max, `cargo bench --bench hll_agg -- --sample-size 10` before and after: | Bench | Before | After | |---|---|---| | `hll_agg/hll_sketch_agg_final_dense` (new) | 1.682 s | 25.8 ms | | `hll_agg/hll_sketch_agg` (Partial) | 1.877 ms | 1.838 ms | | `hll_agg/hll_union_agg` (Partial) | 43.98 ms | 43.86 ms | The Final-mode probe from the issue, per partial row: | Partial states, lgConfigK 18 | Before | After | |---|---|---| | 1.7M rows, 200K groups, 0 to 8 values each | 332 ns | 281 ns | | 200K rows, 2K groups, 50 values each (SET mode) | 2.31 µs | 1.76 µs | | 8K rows, 20 groups, 200 values each (dense) | 207.7 µs | 3.3 µs | -- 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]
