Speed up exact DISTINCT aggregations on dictionary-encoded LONG columns#19056
Speed up exact DISTINCT aggregations on dictionary-encoded LONG columns#19056arunkumarucet wants to merge 3 commits into
Conversation
For unfiltered DISTINCTCOUNT/DISTINCTSUM/DISTINCTAVG on a dictionary-encoded LONG column, the no-scan path was hashing every dictionary value into a per-segment LongOpenHashSet and hash-merging the sets during combine, even though immutable dictionaries are already sorted and duplicate-free. Profiling a high-cardinality DISTINCTCOUNT (10M rows, 1.53M distinct) showed ~80% of CPU in hash set add/rehash/merge. Keep the dictionary values as sorted runs instead (SortedLongDistinctSet): segment merge appends runs, and the union happens once at result extraction as a multi-pass merge between two pre-allocated buffers on the query thread, with termination checks between passes and eager compaction to bound memory when many segments overlap. Mutable (realtime) dictionaries are detected via a per-value sortedness check and fall back to sort+dedupe. Wire format is unchanged (same LongSet object type and byte layout). Measured on ClickBench hits (10M rows, 10 segments): DISTINCTCOUNT(UserID) 37ms -> 19ms warm, identical results; group-by and approximate variants unchanged.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19056 +/- ##
============================================
+ Coverage 65.44% 65.46% +0.02%
Complexity 1421 1421
============================================
Files 3425 3426 +1
Lines 216146 216465 +319
Branches 34231 34296 +65
============================================
+ Hits 141452 141712 +260
- Misses 63312 63345 +33
- Partials 11382 11408 +26
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
I recently found a lot of similar perf optimization by rewrite the data structure wrapper, shall we consider rewrite a new wrapper class similar to fastutils? cc: @gortiz |
Jackie-Jiang
added
enhancement
Jackie-Jiang
left a comment
Jackie-Jiang
left a comment
There was a problem hiding this comment.
This can backfire when there are mixed NonScanBasedAggregationOperator and regular AggregationOperator across different segments. We need to find a way to prevent that
| public static SortedLongDistinctSet fromValues(long[] values, int size, boolean sortedDistinct) { | ||
| if (!sortedDistinct) { | ||
| Arrays.sort(values, 0, size); | ||
| size = dedupeSorted(values, size); |
There was a problem hiding this comment.
Do we need to dedup here if it is always constructed from a dictionary?
There was a problem hiding this comment.
Good point — dictionary values are unique by construction, so the dedupe was redundant. Removed it; fromValues now only sorts when the dictionary is insertion-ordered, and the doc states the input must be duplicate-free. The generic addAll(LongCollection) path keeps its dedupe since that input can be any collection.
| import org.apache.pinot.spi.query.QueryThreadContext; | ||
|
|
||
|
|
||
| /** |
There was a problem hiding this comment.
(minor) Switch to markdown style
There was a problem hiding this comment.
Done, converted the doc comments to markdown style.
| longSet.add(dictionary.getLongValue(dictId)); | ||
| long value = dictionary.getLongValue(dictId); | ||
| if (dictId > 0 && value <= longValues[dictId - 1]) { | ||
| longSorted = false; |
There was a problem hiding this comment.
No need to scan. Dictionary.isSorted() will return this info. It should never be mis-reported (or never false positive)
There was a problem hiding this comment.
Done — dropped the scan and now pass dictionary.isSorted() through. Also updated the operator test to stub isSorted() for the sorted/unsorted cases.
- Use Dictionary.isSorted() instead of scanning for sortedness - Drop the dedupe in fromValues: dictionary values are unique by construction, so only the sort is needed for unsorted (mutable) dictionaries - Switch doc comments to markdown style
Within one query, some segments can take the no-scan path (producing SortedLongDistinctSet) while others take the scan path (producing LongOpenHashSet), e.g. a filter that matches all documents in only some segments. Block merge order is nondeterministic, so a hash set could end up absorbing a sorted set and re-hash every dictionary-sourced value, losing the sorted-run optimization unpredictably. Add SortedLongDistinctSet.union(set1, set2) which always makes the sorted set the absorbing side regardless of operand order, and use it at all merge sites (BaseDistinctAggregateAggregationFunction and the SmartHLL/SmartHLLPlus/SmartULL set merges).
|
Regarding the mixed Added |
4 participants
Motivation
While benchmarking
DISTINCTCOUNTon ClickBench data (10M rows,UserIDLONG with ~1.53M distinct values) I profiled the server with async-profiler and found ~80% of CPU inLongOpenHashSetadd/rehash/addAll, even though the query runs on the no-scan path where values come straight out of the dictionary — which is already sorted and duplicate-free for immutable segments.NonScanBasedAggregationOperator.getDistinctValueSet()hashes every dictionary value into a per-segmentLongOpenHashSet, and the combine phase then hash-merges the per-segment sets. For high-cardinality LONG columns basically the whole query is spent re-deduplicating values the dictionary already deduplicated.Change
Added
SortedLongDistinctSet(aLongSetbacked by sorted runs) and used it in the LONG branch ofgetDistinctValueSet():merge()during the combine phase just appends the other set's runs.Only LONG is changed for now — it's the common high-cardinality case (IDs, timestamps). INT/FLOAT/DOUBLE keep the hash-set path; the same approach applies if profiling justifies it.
The intermediate result still serializes through the existing
ObjectType.LongSetlayout (size+ values), so nothing changes on the wire and mixed-version brokers/servers are unaffected.DISTINCTSUM/DISTINCTAVGand the SMARTHLL/SMARTULL under-threshold paths consume the same set through theSet/LongSetinterfaces and are covered by tests.Numbers
ClickBench
hits, 10M rows / 10 segments, single server (M-series laptop), warm best-of-4:DISTINCTCOUNT(UserID)(1.53M distinct)COUNT(DISTINCT UserID)DISTINCTCOUNT(RegionID)(3.6K distinct)DISTINCTCOUNT(UserID)group by RegionIDDISTINCTCOUNTHLL/SMARTHLL/HLLPLUSResults are identical to the hash-set implementation (verified value-for-value, not just counts).
An isolated microbenchmark of the union itself (10 runs matching the real per-segment cardinalities, ~1.59M values): balanced pairwise merge 2.8ms, k-way heap merge 7.3ms, concat+sort 2.3ms — pairwise multi-pass merging with reused buffers was chosen; the allocation reduction mattered more than the merge strategy (the old per-level allocations produced ~50MB of garbage per query and GC showed up at ~12% of profile samples, ~2% after).
Testing
SortedLongDistinctSetTest(15 cases): single/multi runs, overlapping and disjoint merges, mixed merges withLongOpenHashSetin both directions, unsorted input fallback, merge-after-materialization, dedupe-slack handling, eager compaction (3x3M runs with an arithmetically known union of 7M), serde round-trip, and the unsupported-removal contract.NonScanBasedAggregationOperatorTest(4 cases): the sortedness-detection loop against mocked sorted, unsorted, and duplicate-bearing dictionaries (the realtime/mutable case).DistinctCountQueriesTest,DistinctQueriesTest,DistinctSum/AvgAggregationFunctionTest,SegmentPartitionedDistinctCountQueriesTestall pass.LongOpenHashSetground truth.