arunkumarucet commented on code in PR #19056: URL: https://github.com/apache/pinot/pull/19056#discussion_r3642986758
########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/utils/SortedLongDistinctSet.java: ########## @@ -0,0 +1,373 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.query.aggregation.utils; + +import it.unimi.dsi.fastutil.longs.AbstractLongSet; +import it.unimi.dsi.fastutil.longs.LongCollection; +import it.unimi.dsi.fastutil.longs.LongIterator; +import it.unimi.dsi.fastutil.longs.LongSet; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import org.apache.pinot.spi.query.QueryThreadContext; + + +/** + * A {@link LongSet} for exact DISTINCT aggregation over LONG columns that stores distinct values as sorted runs and + * unions them lazily. + * + * <p>Distinct LONG values read from a value-sorted dictionary (the whole dictionary, as the unfiltered no-scan + * aggregation path does; a filtered path iterating matched dictionary ids in ascending order could adopt the same + * approach) are already sorted and duplicate-free, so they are kept as a {@code long[]} run without hashing. Merging + * two of these sets during the (serial) combine phase appends the other set's runs instead of inserting every element + * into a hash table; the accumulated runs are unioned into one sorted, duplicate-free array on demand, the first time + * {@link #size()}, {@link #iterator()} or {@link #contains(long)} is called (i.e. when the result is serialized or + * its distinct count / sum / average is extracted). The union runs on the calling query thread as a multi-pass merge + * between two pre-sized buffers -- a single up-front allocation instead of one per pairwise merge, no shared + * ForkJoin/common-pool usage -- and checks query termination between passes. To bound retained memory when many + * segments carry overlapping values, pending runs are eagerly compacted (deduplicated) once their total length + * crosses an internal threshold, so pending memory does not grow unboundedly with segment count. + * + * <p>This replaces per-element hashing (which dominated the wall-clock cost of high-cardinality DISTINCT_COUNT + * queries) with sorted merging. Serialization as a {@code LongSet} and DISTINCT_SUM / DISTINCT_AVG value iteration + * work unchanged. Inputs that are not already sorted-and-distinct (e.g. a {@code LongOpenHashSet} arriving via a + * mixed merge) are sorted and de-duplicated before being added. One documented contract deviation: + * {@link #addAll(LongCollection)} returns {@code true} for any non-empty operand, even if every element was already + * present -- computing the exact "changed" answer would force the union eagerly. No caller on the aggregation path + * reads the return value. + * + * <p>Not thread-safe, and more strongly so than a typical mutable collection: the read accessors {@link #size()}, + * {@link #contains(long)} and {@link #iterator()} are <em>not</em> pure reads -- the first such call triggers + * {@code materialize()}, which reassigns internal state to union the pending runs. Instances are therefore built per + * segment and both mutated and first-read only by the single-threaded combine phase; a finished result must be safely + * published before any other thread reads it, and the first read must not race a mutation. + * + * <p>Element removal is unsupported: {@code remove}/{@code rem} and the iterator's {@code remove()} are not + * implemented, so the inherited {@code removeAll}/{@code retainAll} throw {@link UnsupportedOperationException}. The + * DISTINCT aggregation path never removes elements (it only builds, merges, counts, sums, iterates and serializes). + */ +public final class SortedLongDistinctSet extends AbstractLongSet { + private static final long[] EMPTY = new long[0]; + private static final String SCOPE = "SortedLongDistinctSet"; + + // Hard cap on the total pending values: beyond this the exact distinct result cannot be represented in one array. + private static final long MAX_PENDING_TOTAL = Integer.MAX_VALUE - 8; + + // When the pending runs' total length crosses this, they are eagerly compacted into one deduplicated run inside + // addAll(). This bounds peak retained memory to roughly the threshold plus the incoming run, instead of the sum of + // all per-segment distinct counts -- which, when many segments carry overlapping values, can far exceed the global + // distinct count. Large enough that it never triggers for typical segment counts and cardinalities. + private static final long COMPACT_THRESHOLD = 8L << 20; + + // Pending sorted, duplicate-free, exact-length runs awaiting union. Non-null exactly when not yet materialized. + private List<long[]> _runs; + private long _pendingTotal; + // Next pending total that triggers eager compaction. Doubles with the compacted size (geometric backoff) so that + // when the true distinct count exceeds COMPACT_THRESHOLD, appends do not each trigger a full re-merge (which would + // be quadratic in segment count); amortized compaction work stays O(n log n) and retained memory stays proportional + // to the actual distinct count rather than the number of segments. + private long _compactMin = COMPACT_THRESHOLD; + + // Materialized sorted, duplicate-free values in [0, _size). Non-null exactly when materialized. The array may be + // longer than _size when the dedupe waste was immaterial (< 25%); every consumer bounds accesses by _size. + private long[] _values; + private int _size; + + /** + * Wraps a single run that is already sorted ascending and duplicate-free. Private so the checked {@link #fromValues} + * factory is the only entry point: an unsorted or duplicate-bearing array here would silently corrupt results + * ({@link #size()} over-counts, {@link #contains(long)} misreports) rather than fail loudly. + */ + private SortedLongDistinctSet(long[] sortedDistinct) { + _runs = new ArrayList<>(4); + if (sortedDistinct.length > 0) { + _runs.add(sortedDistinct); + _pendingTotal = sortedDistinct.length; + } + } + + /** + * Creates a set from the prefix {@code [0, size)} of {@code values}. When {@code sortedDistinct} is false the prefix + * is sorted and de-duplicated first (which reorders the caller's {@code values} array in place) so the sorted-run + * invariant always holds; when true the caller guarantees the prefix is already sorted ascending and duplicate-free. + * The set takes ownership of {@code values} (it is retained without copying when {@code size == values.length}), so + * the caller must not modify the array after this call. + */ + public static SortedLongDistinctSet fromValues(long[] values, int size, boolean sortedDistinct) { + if (!sortedDistinct) { + Arrays.sort(values, 0, size); + size = dedupeSorted(values, size); Review 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. ########## pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/utils/SortedLongDistinctSet.java: ########## @@ -0,0 +1,373 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.query.aggregation.utils; + +import it.unimi.dsi.fastutil.longs.AbstractLongSet; +import it.unimi.dsi.fastutil.longs.LongCollection; +import it.unimi.dsi.fastutil.longs.LongIterator; +import it.unimi.dsi.fastutil.longs.LongSet; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import org.apache.pinot.spi.query.QueryThreadContext; + + +/** Review Comment: Done, converted the doc comments to markdown style. ########## pinot-core/src/main/java/org/apache/pinot/core/operator/query/NonScanBasedAggregationOperator.java: ########## @@ -254,12 +254,26 @@ private static Set getDistinctValueSet(Dictionary dictionary) { } return intSet; case LONG: - LongOpenHashSet longSet = new LongOpenHashSet(dictionarySize); + // A numeric dictionary is value-sorted and duplicate-free, so its values are already the sorted distinct set: + // read them straight into a sorted run instead of hashing each one. SortedLongDistinctSet then unions the + // per-segment runs during combine without hashing (the dominant cost for high-cardinality DISTINCT_COUNT). + // Only LONG is optimized here -- it is the common high-cardinality distinct case (IDs, timestamps); INT / + // FLOAT / DOUBLE keep the hash-set path. The same sorted-run approach generalizes to them if profiling shows + // it is worthwhile. + // Sortedness is verified per value during the copy (near-zero cost) rather than trusting + // Dictionary.isSorted(), because a mis-reporting dictionary implementation would silently corrupt DISTINCT + // results instead of failing loudly. + long[] longValues = new long[dictionarySize]; + boolean longSorted = true; for (int dictId = 0; dictId < dictionarySize; dictId++) { QueryThreadContext.checkTerminationAndSampleUsagePeriodically(dictId, EXPLAIN_NAME); - longSet.add(dictionary.getLongValue(dictId)); + long value = dictionary.getLongValue(dictId); + if (dictId > 0 && value <= longValues[dictId - 1]) { + longSorted = false; Review Comment: Done — dropped the scan and now pass `dictionary.isSorted()` through. Also updated the operator test to stub `isSorted()` for the sorted/unsorted cases. -- 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]
