This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new d3e53e20210 Implement streaming distinct leaf stage operator for
multi-stage engine (#19419)
d3e53e20210 is described below
commit d3e53e202107605fb68f6e10aef6f55e25152d29
Author: Yash Mayya <[email protected]>
AuthorDate: Wed Sep 2 16:41:57 2026 -0400
Implement streaming distinct leaf stage operator for multi-stage engine
(#19419)
---
.../MultiStageBrokerRequestHandler.java | 11 +
.../MultiStageBrokerRequestHandlerTest.java | 58 ++
.../common/utils/config/QueryOptionsUtils.java | 6 +
.../common/utils/config/QueryOptionsUtilsTest.java | 7 +-
.../StreamingDistinctCombineOperator.java | 193 ++++++
.../apache/pinot/core/plan/CombinePlanNode.java | 33 +-
.../core/plan/maker/InstancePlanMakerImplV2.java | 10 +
.../core/query/request/context/QueryContext.java | 10 +
.../StreamingDistinctCombineOperatorTest.java | 649 +++++++++++++++++++++
.../tests/MultiStageEngineIntegrationTest.java | 84 +++
.../apache/pinot/spi/utils/CommonConstants.java | 32 +
11 files changed, 1086 insertions(+), 7 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 07a9497b368..08005e3be51 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -147,6 +147,8 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
private final boolean _streamStatsDefault;
@Nullable
protected final String _defaultStreamingGroupByFlushThreshold;
+ @Nullable
+ protected final String _defaultStreamingDistinctFlushThreshold;
protected final PinotMeter _stagesStartedMeter =
BrokerMeter.MSE_STAGES_STARTED.getGlobalMeter();
protected final PinotMeter _stagesFinishedMeter =
BrokerMeter.MSE_STAGES_COMPLETED.getGlobalMeter();
@@ -246,6 +248,11 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
// null indicates "feature disabled", which matches the
broker-config-unset case.
_defaultStreamingGroupByFlushThreshold =
streamingGroupByFlushThreshold > 0 ?
Integer.toString(streamingGroupByFlushThreshold) : null;
+ int streamingDistinctFlushThreshold = _config.getProperty(
+
CommonConstants.Broker.CONFIG_OF_MSE_STREAMING_DISTINCT_FLUSH_THRESHOLD,
+ CommonConstants.Broker.DEFAULT_MSE_STREAMING_DISTINCT_FLUSH_THRESHOLD);
+ _defaultStreamingDistinctFlushThreshold =
+ streamingDistinctFlushThreshold > 0 ?
Integer.toString(streamingDistinctFlushThreshold) : null;
}
@Override
@@ -604,6 +611,10 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
queryOptions.putIfAbsent(CommonConstants.Broker.Request.QueryOptionKey.STREAMING_GROUP_BY_FLUSH_THRESHOLD,
_defaultStreamingGroupByFlushThreshold);
}
+ if (_defaultStreamingDistinctFlushThreshold != null) {
+
queryOptions.putIfAbsent(CommonConstants.Broker.Request.QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD,
+ _defaultStreamingDistinctFlushThreshold);
+ }
}
/// Extension hook for preparing a compiled query's planner-visible options
after compilation, authorization and
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
index 7b3e90c8d4a..e04423ccf31 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandlerTest.java
@@ -167,9 +167,63 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
"No option should be injected when the broker default is unset");
}
+ @Test
+ public void
testApplyBrokerDefaultQueryOptionsInjectsStreamingDistinctFlushThreshold()
+ throws Exception {
+ MultiStageBrokerRequestHandler handler =
newHandlerWithStreamingDistinctFlushThreshold("5000");
+
+ Map<String, String> queryOptions = new HashMap<>();
+ handler.applyBrokerDefaultQueryOptions(queryOptions);
+
Assert.assertEquals(queryOptions.get(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD),
"5000",
+ "Broker default should be injected when query option is absent");
+
Assert.assertFalse(queryOptions.containsKey(QueryOptionKey.STREAMING_GROUP_BY_FLUSH_THRESHOLD),
+ "The distinct default must not also enable streaming group-by");
+ }
+
+ @Test
+ public void
testApplyBrokerDefaultQueryOptionsStreamingDistinctPerQueryOverrideWins()
+ throws Exception {
+ MultiStageBrokerRequestHandler handler =
newHandlerWithStreamingDistinctFlushThreshold("5000");
+
+ Map<String, String> queryOptions = new HashMap<>();
+ queryOptions.put(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD, "0");
+ handler.applyBrokerDefaultQueryOptions(queryOptions);
+
Assert.assertEquals(queryOptions.get(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD),
"0",
+ "Per-query SET = 0 must override the broker default");
+
+ queryOptions.clear();
+ queryOptions.put(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD, "100");
+ handler.applyBrokerDefaultQueryOptions(queryOptions);
+
Assert.assertEquals(queryOptions.get(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD),
"100",
+ "Per-query SET must take precedence over the broker default");
+ }
+
+ @Test
+ public void
testApplyBrokerDefaultQueryOptionsNoStreamingDistinctInjectionWhenConfigUnset()
+ throws Exception {
+ MultiStageBrokerRequestHandler handler =
newHandlerWithStreamingDistinctFlushThreshold(null);
+
+ Map<String, String> queryOptions = new HashMap<>();
+ handler.applyBrokerDefaultQueryOptions(queryOptions);
+
Assert.assertFalse(queryOptions.containsKey(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD),
+ "No option should be injected when the broker default is unset");
+ }
+
private static MultiStageBrokerRequestHandler
newHandlerWithStreamingGroupByFlushThreshold(
@Nullable String streamingGroupByFlushThreshold)
throws Exception {
+ return newHandlerWithFlushThresholds(streamingGroupByFlushThreshold, null);
+ }
+
+ private static MultiStageBrokerRequestHandler
newHandlerWithStreamingDistinctFlushThreshold(
+ @Nullable String streamingDistinctFlushThreshold)
+ throws Exception {
+ return newHandlerWithFlushThresholds(null,
streamingDistinctFlushThreshold);
+ }
+
+ private static MultiStageBrokerRequestHandler newHandlerWithFlushThresholds(
+ @Nullable String streamingGroupByFlushThreshold, @Nullable String
streamingDistinctFlushThreshold)
+ throws Exception {
PinotConfiguration config = new PinotConfiguration();
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_HOSTNAME,
"localhost");
config.setProperty(MultiStageQueryRunner.KEY_OF_QUERY_RUNNER_PORT,
Integer.toString(NetUtils.findOpenPort()));
@@ -177,6 +231,10 @@ public class MultiStageBrokerRequestHandlerTest extends
QueryEnvironmentTestBase
config.setProperty(CommonConstants.Broker.CONFIG_OF_MSE_STREAMING_GROUP_BY_FLUSH_THRESHOLD,
streamingGroupByFlushThreshold);
}
+ if (streamingDistinctFlushThreshold != null) {
+
config.setProperty(CommonConstants.Broker.CONFIG_OF_MSE_STREAMING_DISTINCT_FLUSH_THRESHOLD,
+ streamingDistinctFlushThreshold);
+ }
BrokerQueryEventListenerFactory.init(config);
BrokerMetrics.register(mock(BrokerMetrics.class));
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index 4b0ee0463e2..a48c88cd146 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -549,6 +549,12 @@ public class QueryOptionsUtils {
return
checkedParseIntNonNegative(QueryOptionKey.STREAMING_GROUP_BY_FLUSH_THRESHOLD,
value);
}
+ @Nullable
+ public static Integer getStreamingDistinctFlushThreshold(Map<String, String>
queryOptions) {
+ String value =
queryOptions.get(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD);
+ return
checkedParseIntNonNegative(QueryOptionKey.STREAMING_DISTINCT_FLUSH_THRESHOLD,
value);
+ }
+
public static boolean isNullHandlingEnabled(Map<String, String>
queryOptions) {
return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.ENABLE_NULL_HANDLING));
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
index dceac3385eb..0b8155403aa 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/QueryOptionsUtilsTest.java
@@ -39,7 +39,8 @@ public class QueryOptionsUtilsTest {
private static final List<String> POSITIVE_INT_KEYS =
List.of(NUM_REPLICA_GROUPS_TO_QUERY, MAX_EXECUTION_THREADS,
NUM_GROUPS_LIMIT, MAX_INITIAL_RESULT_HOLDER_CAPACITY,
MAX_STREAMING_PENDING_BLOCKS, MAX_ROWS_IN_JOIN, MAX_ROWS_IN_WINDOW);
- private static final List<String> NON_NEGATIVE_INT_KEYS =
List.of(MULTI_STAGE_LEAF_LIMIT);
+ private static final List<String> NON_NEGATIVE_INT_KEYS =
+ List.of(MULTI_STAGE_LEAF_LIMIT, STREAMING_GROUP_BY_FLUSH_THRESHOLD,
STREAMING_DISTINCT_FLUSH_THRESHOLD);
private static final List<String> UNBOUNDED_INT_KEYS =
List.of(MIN_SEGMENT_GROUP_TRIM_SIZE, MIN_SERVER_GROUP_TRIM_SIZE,
MIN_BROKER_GROUP_TRIM_SIZE,
GROUP_TRIM_THRESHOLD);
@@ -336,6 +337,10 @@ public class QueryOptionsUtilsTest {
// Non-negative ints
case MULTI_STAGE_LEAF_LIMIT:
return QueryOptionsUtils.getMultiStageLeafLimit(map);
+ case STREAMING_GROUP_BY_FLUSH_THRESHOLD:
+ return QueryOptionsUtils.getStreamingGroupByFlushThreshold(map);
+ case STREAMING_DISTINCT_FLUSH_THRESHOLD:
+ return QueryOptionsUtils.getStreamingDistinctFlushThreshold(map);
// Unbounded ints
case MIN_SEGMENT_GROUP_TRIM_SIZE:
return QueryOptionsUtils.getMinSegmentGroupTrimSize(map);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperator.java
new file mode 100644
index 00000000000..0ed892c8101
--- /dev/null
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperator.java
@@ -0,0 +1,193 @@
+/**
+ * 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.operator.streaming;
+
+import com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.core.common.Operator;
+import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock;
+import
org.apache.pinot.core.operator.blocks.results.BaseResultsBlock.EarlyTerminationReason;
+import org.apache.pinot.core.operator.blocks.results.DistinctResultsBlock;
+import org.apache.pinot.core.operator.blocks.results.ExceptionResultsBlock;
+import org.apache.pinot.core.operator.blocks.results.MetadataResultsBlock;
+import
org.apache.pinot.core.operator.combine.merger.DistinctResultsBlockMerger;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.query.QueryThreadContext;
+
+
+/// Streaming combine operator for distinct queries. Instead of accumulating
every distinct value into a single
+/// [org.apache.pinot.core.query.distinct.table.DistinctTable] before
returning (like
+/// [org.apache.pinot.core.operator.combine.DistinctCombineOperator]), this
operator flushes the accumulated values
+/// once the table reaches a configurable threshold.
+///
+/// This bounds server memory usage for high-cardinality distinct queries on
MSE leaf stages, while still performing
+/// partial de-duplication to reduce data volume compared to skipping
leaf-stage de-duplication entirely (the
+/// `is_skip_leaf_stage_group_by` hint).
+///
+/// The downstream FINAL stage de-duplicates partial results from multiple
flushes correctly because:
+///
+/// - Hash exchange routes the same distinct key to the same FINAL worker (the
exchange is keyed on the group keys,
+/// which for a zero-aggregate-call aggregate are exactly the distinct
columns)
+/// - Set union is associative, commutative and idempotent, so a value emitted
in more than one flush window is a
+/// no-op downstream
+///
+/// That downstream stage is a precondition, not an implementation detail: a
flushed block carries only part of the
+/// distinct set, and one value can span several flush windows. Leaves that
must return final results are rejected in
+/// the constructor as a backstop -- the real gate is in
[org.apache.pinot.core.plan.CombinePlanNode], but this
+/// operator is public and nothing else would catch a direct construction,
which would silently emit duplicate rows.
+///
+/// Unlike [StreamingGroupByCombineOperator], no `detachFromWorkerThreadState`
hook is needed: a per-segment
+/// `DistinctTable` is allocated per
[org.apache.pinot.core.operator.query.DistinctOperator] invocation and the
+/// dictionary-based executors materialize actual values in `getResult()`, so
a block handed to the consumer thread
+/// is self-contained and references no reused worker-thread-local state.
+///
+/// Two limits to be aware of when enabling this:
+///
+/// - It bounds only the server-level accumulated table. The per-segment
`DistinctTable`s that feed it are still
+/// built in full (unlike group-by, whose per-segment maps are capped by
`numGroupsLimit`), so peak leaf heap is
+/// roughly `flushThreshold` plus one table per in-flight worker thread. In
particular, when a single segment
+/// already holds more than `flushThreshold` distinct values, its block is
adopted and flushed with nothing merged
+/// into it, so the leaf performs no cross-segment de-duplication at all and
only amplifies rows. Deliberately not
+/// "fixed" by requiring a merge before each flush: that would let the
accumulator grow to two segments' worth and
+/// raise the very ceiling this operator exists to lower. Bounding the
per-segment tables is the real fix.
+/// - The distinct guardrails weaken. All three (`maxRowsInDistinct`,
`maxRowsWithoutChangeInDistinct`,
+/// `maxExecutionTimeMsInDistinct`) are evaluated inside
[DistinctResultsBlockMerger#mergeResultsBlocks], which
+/// only runs for blocks merged into an existing accumulator -- never for
the first block of a flush window, which
+/// is adopted. So in the no-merge regime above, none of them can fire at
all. When merging does happen,
+/// `maxRowsInDistinct` and `maxExecutionTimeMsInDistinct` stay per-query
bounds because [#mergeBlock] deliberately
+/// carries the scanned-doc count across windows, but
`maxRowsWithoutChangeInDistinct` still degrades: its counter
+/// only advances when a merge leaves the table's size unchanged, and every
window restarts from an empty table.
+@SuppressWarnings("rawtypes")
+public class StreamingDistinctCombineOperator extends
BaseStreamingCombineOperator<DistinctResultsBlock> {
+ private static final String EXPLAIN_NAME = "STREAMING_COMBINE_DISTINCT";
+
+ private final int _flushThreshold;
+
+ // Main-thread-only state for accumulating distinct results
+ private DistinctResultsBlock _mergedBlock;
+ private long _numDocsScanned;
+ private EarlyTerminationReason _earlyTerminationReason =
EarlyTerminationReason.NONE;
+
+ public StreamingDistinctCombineOperator(List<Operator> operators,
QueryContext queryContext,
+ ExecutorService executorService, int flushThreshold) {
+ super(new DistinctResultsBlockMerger(queryContext), operators,
queryContext, executorService);
+ Preconditions.checkState(
+ !queryContext.isServerReturnFinalResult() &&
!queryContext.isServerReturnFinalResultKeyUnpartitioned(),
+ "Streaming distinct combine requires a leaf whose results are
de-duplicated by a later stage");
+ _flushThreshold = flushThreshold;
+ }
+
+ @Override
+ public String toExplainString() {
+ return EXPLAIN_NAME;
+ }
+
+ /// Disables the worker-side early-termination check, matching
[StreamingGroupByCombineOperator].
+ ///
+ /// [BaseStreamingCombineOperator#processSegments] calls this on the
PRODUCING thread with a block it has already
+ /// published to the consumer. This operator adopts that same block as its
accumulator and mutates it (merging
+ /// other tables into its `DistinctTable`, and writing `numDocsScanned` /
the early-termination reason), none of
+ /// which is synchronized -- so letting the worker read `isSatisfied()` or
the termination reason off it afterwards
+ /// is a data race on a plain `HashSet` mid-`add`/`rehash`. The sibling
streaming operators are each safe for one
+ /// of two reasons: the group-by one never lets a worker read a published
block, and the selection-only one never
+ /// mutates one. This operator would otherwise be the only one doing both.
+ ///
+ /// Nothing is lost: the consumer still evaluates satisfaction itself in
[#mergeBlock] via the results-block
+ /// merger, and the cross-segment early exit is already given up in this
mode by construction (flushing empties the
+ /// accumulator long before it can reach LIMIT -- see the gate in
+ /// [org.apache.pinot.core.plan.CombinePlanNode]). Returning `false` also
removes the hazard of a worker returning
+ /// early without emitting its `LAST_RESULTS_BLOCK`.
+ @Override
+ protected boolean isQuerySatisfied(DistinctResultsBlock resultsBlock, Object
tracker) {
+ return false;
+ }
+
+ /// Polls per-segment result blocks from worker threads, merges them into
the accumulated distinct table, and
+ /// flushes when the table reaches the flush threshold. Returns one block
per call:
+ /// - A DistinctResultsBlock when flushing accumulated values
+ /// - A MetadataResultsBlock when all operators are done and remaining data
has been flushed
+ /// - An ExceptionResultsBlock on error or timeout
+ @Override
+ protected BaseResultsBlock getNextBlock() {
+ long endTimeMs = _queryContext.getEndTimeMs();
+ try {
+ while (!_querySatisfied && _numOperatorsFinished < _numOperators) {
+ QueryThreadContext.checkTermination(this::getExplainName);
+ BaseResultsBlock resultsBlock =
+ _blockingQueue.poll(endTimeMs - System.currentTimeMillis(),
TimeUnit.MILLISECONDS);
+ if (resultsBlock == null) {
+ throw QueryErrorCode.EXECUTION_TIMEOUT.asException("Timed out while
streaming distinct results");
+ }
+ if (resultsBlock instanceof ExceptionResultsBlock) {
+ return checkTerminateExceptionAndAttachExecutionStats(resultsBlock);
+ }
+ if (resultsBlock == LAST_RESULTS_BLOCK) {
+ _numOperatorsFinished++;
+ continue;
+ }
+ mergeBlock((DistinctResultsBlock) resultsBlock);
+ if (_mergedBlock.getDistinctTable().size() >= _flushThreshold) {
+ return flush();
+ }
+ }
+ } catch (Exception e) {
+ return createExceptionResultsBlockAndAttachExecutionStats(e, "streaming
distinct results");
+ }
+ // All operators done (or the query is satisfied) — flush any remaining
accumulated data
+ if (_mergedBlock != null && _mergedBlock.getDistinctTable().size() > 0) {
+ return flush();
+ }
+ // Return final metadata block. The early-termination reason is carried
over from the accumulated block: the
+ // blocking path reports it on the single results block, but here that
block has already been streamed out, so
+ // without this the reason (and hence BrokerResponse.isPartialResult)
would be silently dropped.
+ MetadataResultsBlock metadataBlock = new MetadataResultsBlock();
+ metadataBlock.setEarlyTerminationReason(_earlyTerminationReason);
+ return attachExecutionStats(metadataBlock);
+ }
+
+ /// Merges a per-segment block into the accumulated result. The first block
of each flush window is adopted as the
+ /// accumulator (mirroring
[org.apache.pinot.core.operator.combine.BaseSingleBlockCombineOperator#mergeResults]),
+ /// which avoids having to construct a `DistinctTable` of the right subtype
from a `DataSchema`.
+ private void mergeBlock(DistinctResultsBlock blockToMerge) {
+ QueryThreadContext.checkTerminationAndSampleUsage(EXPLAIN_NAME);
+ if (_mergedBlock == null) {
+ _mergedBlock = blockToMerge;
+ } else {
+ _resultsBlockMerger.mergeResultsBlocks(_mergedBlock, blockToMerge);
+ }
+ // Carry the running doc count across flush windows so that
maxRowsInDistinct stays a per-query bound instead of
+ // silently becoming a per-flush-window one. NOTE: This count is not
serialized with a data block (the block's
+ // DataTable carries only rows), so overwriting it here cannot double
count in the response metadata.
+ _numDocsScanned += blockToMerge.getNumDocsScanned();
+ _mergedBlock.setNumDocsScanned(_numDocsScanned);
+ if (_mergedBlock.getEarlyTerminationReason() !=
EarlyTerminationReason.NONE) {
+ _earlyTerminationReason = _mergedBlock.getEarlyTerminationReason();
+ }
+ _querySatisfied = _resultsBlockMerger.isQuerySatisfied(_mergedBlock);
+ }
+
+ private DistinctResultsBlock flush() {
+ DistinctResultsBlock block = _mergedBlock;
+ _mergedBlock = null;
+ return block;
+ }
+}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
index dd3a720412b..b753fa74d26 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/plan/CombinePlanNode.java
@@ -34,6 +34,7 @@ import
org.apache.pinot.core.operator.combine.SelectionOnlyCombineOperator;
import org.apache.pinot.core.operator.combine.SelectionOrderByCombineOperator;
import
org.apache.pinot.core.operator.combine.SequentialSortedGroupByCombineOperator;
import org.apache.pinot.core.operator.combine.SortedGroupByCombineOperator;
+import
org.apache.pinot.core.operator.streaming.StreamingDistinctCombineOperator;
import
org.apache.pinot.core.operator.streaming.StreamingGroupByCombineOperator;
import
org.apache.pinot.core.operator.streaming.StreamingSelectionOnlyCombineOperator;
import org.apache.pinot.core.query.executor.ResultsBlockStreamer;
@@ -130,15 +131,35 @@ public class CombinePlanNode implements PlanNode {
// Use streaming operator only for non-empty selection-only query
return new StreamingSelectionOnlyCombineOperator(operators,
_queryContext, _executorService);
}
- // Streaming flushes partial aggregates, so it needs an aggregation
above to merge them back together.
+ // Streaming flushes partial results, so it needs an aggregation above
to merge them back together.
// Leaves that must return final results are excluded, see
StreamingGroupByCombineOperator.
- int flushThreshold = _queryContext.getStreamingGroupByFlushThreshold();
- if (flushThreshold > 0
+ boolean leafReturnsFinalResult =
+ _queryContext.isServerReturnFinalResult() ||
_queryContext.isServerReturnFinalResultKeyUnpartitioned();
+ int groupByFlushThreshold =
_queryContext.getStreamingGroupByFlushThreshold();
+ if (groupByFlushThreshold > 0
&& QueryContextUtils.isAggregationQuery(_queryContext)
&& _queryContext.getGroupByExpressions() != null
- && !_queryContext.isServerReturnFinalResult()
- && !_queryContext.isServerReturnFinalResultKeyUnpartitioned()) {
- return new StreamingGroupByCombineOperator(operators, _queryContext,
_executorService, flushThreshold);
+ && !leafReturnsFinalResult) {
+ return new StreamingGroupByCombineOperator(operators, _queryContext,
_executorService, groupByFlushThreshold);
+ }
+ int distinctFlushThreshold =
_queryContext.getStreamingDistinctFlushThreshold();
+ if (distinctFlushThreshold > 0 &&
QueryContextUtils.isDistinctQuery(_queryContext)
+ // With ORDER BY, the DistinctTable keeps a bounded top-LIMIT heap
instead of accumulating every value, so
+ // it is already memory-bounded and streaming would only ship more
rows.
+ && _queryContext.getOrderByExpressions() == null
+ // NOTE: The planner pushes LIMIT (plus OFFSET) into the leaf
aggregate for any distinct query that has
+ // one -- see
PinotAggregateExchangeNodeInsertRule#isGroupTrimmingEnabled, which is
unconditionally on for
+ // aggregates with no aggregate calls. So the leaf table is normally
bounded at LIMIT already, and this
+ // feature is for the case where that LIMIT is far larger than the
memory we want to spend.
+ //
+ // This branch DOES give up the cross-segment early exit:
DistinctResultsBlockMerger#isQuerySatisfied can
+ // only fire once the accumulated table reaches LIMIT, and flushing
empties it well before that, so every
+ // segment gets scanned. That is the deliberate trade -- a lower
memory ceiling for more scan work. When
+ // LIMIT is at or below the threshold there is no memory to save, so
the short-circuit wins instead.
+ && _queryContext.getLimit() > distinctFlushThreshold
+ && !leafReturnsFinalResult) {
+ return new StreamingDistinctCombineOperator(operators, _queryContext,
_executorService,
+ distinctFlushThreshold);
}
}
if (QueryContextUtils.isAggregationQuery(_queryContext)) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
index dc8021857c1..0668a8aca0f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
@@ -338,6 +338,16 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
queryContext.setStreamingGroupByFlushThreshold(streamingGroupByFlushThreshold);
}
}
+
+ // Set distinct query options. NOTE: This is intentionally outside the
group-by block above, because DISTINCT is a
+ // separate query class from aggregation (see
QueryContextUtils.isDistinctQuery).
+ if (QueryContextUtils.isDistinctQuery(queryContext)) {
+ // Set streamingDistinctFlushThreshold
+ Integer streamingDistinctFlushThreshold =
QueryOptionsUtils.getStreamingDistinctFlushThreshold(queryOptions);
+ if (streamingDistinctFlushThreshold != null) {
+
queryContext.setStreamingDistinctFlushThreshold(streamingDistinctFlushThreshold);
+ }
+ }
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
index 62a7a0e54c1..d97c340bc0d 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
@@ -133,6 +133,8 @@ public class QueryContext {
private int _effectiveSegmentGroupTrimSize;
// Flush threshold for streaming group-by (0 = disabled)
private int _streamingGroupByFlushThreshold;
+ // Flush threshold for streaming distinct (0 = disabled)
+ private int _streamingDistinctFlushThreshold;
// Whether null handling is enabled
private boolean _nullHandlingEnabled;
// Whether server returns the final result
@@ -498,6 +500,14 @@ public class QueryContext {
_streamingGroupByFlushThreshold = streamingGroupByFlushThreshold;
}
+ public int getStreamingDistinctFlushThreshold() {
+ return _streamingDistinctFlushThreshold;
+ }
+
+ public void setStreamingDistinctFlushThreshold(int
streamingDistinctFlushThreshold) {
+ _streamingDistinctFlushThreshold = streamingDistinctFlushThreshold;
+ }
+
public boolean isNullHandlingEnabled() {
return _nullHandlingEnabled;
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperatorTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperatorTest.java
new file mode 100644
index 00000000000..fee7ee70e16
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/streaming/StreamingDistinctCombineOperatorTest.java
@@ -0,0 +1,649 @@
+/**
+ * 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.operator.streaming;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.Operator;
+import org.apache.pinot.core.operator.blocks.results.BaseResultsBlock;
+import
org.apache.pinot.core.operator.blocks.results.BaseResultsBlock.EarlyTerminationReason;
+import org.apache.pinot.core.operator.blocks.results.DistinctResultsBlock;
+import org.apache.pinot.core.operator.blocks.results.MetadataResultsBlock;
+import org.apache.pinot.core.operator.combine.BaseCombineOperator;
+import org.apache.pinot.core.operator.combine.DistinctCombineOperator;
+import org.apache.pinot.core.plan.CombinePlanNode;
+import org.apache.pinot.core.plan.PlanNode;
+import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2;
+import org.apache.pinot.core.plan.maker.PlanMaker;
+import org.apache.pinot.core.query.executor.ResultsBlockStreamer;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.SegmentContext;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/// Test for [StreamingDistinctCombineOperator].
+public class StreamingDistinctCombineOperatorTest {
+ private static final File TEMP_DIR = new File(FileUtils.getTempDirectory(),
"StreamingDistinctCombineOperatorTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME_PREFIX = "testSegment_";
+
+ private static final int NUM_SEGMENTS = 4;
+ // Every segment holds the same 50 distinct values, each appearing twice.
Identical content across segments is
+ // deliberate: it guarantees that a value flushed in one window is
re-emitted in a later one, which is exactly the
+ // idempotence property the downstream FINAL stage relies on.
+ private static final int NUM_DISTINCT_VALUES = 50;
+ private static final int NUM_DOCS_PER_SEGMENT = NUM_DISTINCT_VALUES * 2;
+ private static final long TOTAL_NUM_DOCS = (long) NUM_SEGMENTS *
NUM_DOCS_PER_SEGMENT;
+
+ private static final String INT_COLUMN = "intColumn";
+ private static final String DICT_STRING_COLUMN = "dictStringColumn";
+ private static final String RAW_STRING_COLUMN = "rawStringColumn";
+
+ // The MSE leaf stage pushes no LIMIT down, so the leaf query runs with
Integer.MAX_VALUE. That is what makes
+ // DistinctTable.hasLimit() false and sends every executor down the
unbounded add path, so model it here.
+ private static final String NO_LIMIT = " LIMIT " + Integer.MAX_VALUE;
+ // A filter (any filter) keeps DistinctPlanNode off the
DictionaryBasedDistinctOperator fast path, which reads the
+ // whole dictionary instead of scanning.
+ private static final String MATCH_ALL = " WHERE intColumn >= 0";
+
+ private static final TableConfig TABLE_CONFIG =
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNoDictionaryColumns(List.of(RAW_STRING_COLUMN))
+ // Only used by testIndexBasedDistinctOperatorPath; inert unless
useIndexBasedDistinctOperator is set.
+ .setInvertedIndexColumns(List.of(DICT_STRING_COLUMN)).build();
+ private static final Schema SCHEMA = new Schema.SchemaBuilder()
+ .addSingleValueDimension(INT_COLUMN, FieldSpec.DataType.INT)
+ .addSingleValueDimension(DICT_STRING_COLUMN, FieldSpec.DataType.STRING)
+ .addSingleValueDimension(RAW_STRING_COLUMN, FieldSpec.DataType.STRING)
+ .build();
+
+ private static final PlanMaker PLAN_MAKER = new InstancePlanMakerImplV2();
+ private static final ExecutorService EXECUTOR =
Executors.newCachedThreadPool();
+
+ private List<IndexSegment> _indexSegments;
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(TEMP_DIR);
+ _indexSegments = new ArrayList<>(NUM_SEGMENTS);
+ for (int i = 0; i < NUM_SEGMENTS; i++) {
+ _indexSegments.add(createOfflineSegment(i));
+ }
+ }
+
+ /// Builds the shared fixture: every segment holds the SAME 50 values. See
[#NUM_DISTINCT_VALUES] for why.
+ private IndexSegment createOfflineSegment(int index)
+ throws Exception {
+ List<GenericRow> records = new ArrayList<>(NUM_DOCS_PER_SEGMENT);
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ for (int j = 0; j < 2; j++) {
+ GenericRow record = new GenericRow();
+ record.putValue(INT_COLUMN, i);
+ record.putValue(DICT_STRING_COLUMN, "d" + i);
+ record.putValue(RAW_STRING_COLUMN, "r" + i);
+ records.add(record);
+ }
+ }
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ String segmentName = SEGMENT_NAME_PREFIX + index;
+ segmentGeneratorConfig.setSegmentName(segmentName);
+ segmentGeneratorConfig.setOutDir(TEMP_DIR.getPath());
+
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(records));
+ driver.build();
+
+ return ImmutableSegmentLoader.load(new File(TEMP_DIR, segmentName),
ReadMode.mmap);
+ }
+
+ /// With identical segments the first adopted block already exceeds any
useful threshold, so it is flushed with
+ /// nothing merged into it and the accumulate-across-segments path is never
reached. This fixture gives each
+ /// segment a disjoint range so the accumulator has to merge before it trips
the threshold.
+ @Test
+ public void testAccumulatorMergesAcrossSegmentsBeforeFlushing()
+ throws Exception {
+ File disjointDir = new File(FileUtils.getTempDirectory(),
"StreamingDistinctCombineOperatorTest_disjoint");
+ FileUtils.deleteDirectory(disjointDir);
+ List<IndexSegment> segments = new ArrayList<>(NUM_SEGMENTS);
+ try {
+ for (int index = 0; index < NUM_SEGMENTS; index++) {
+ List<GenericRow> records = new ArrayList<>(NUM_DISTINCT_VALUES);
+ for (int i = index * NUM_DISTINCT_VALUES; i < (index + 1) *
NUM_DISTINCT_VALUES; i++) {
+ GenericRow record = new GenericRow();
+ record.putValue(INT_COLUMN, i);
+ record.putValue(DICT_STRING_COLUMN, "d" + i);
+ record.putValue(RAW_STRING_COLUMN, "r" + i);
+ records.add(record);
+ }
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ String segmentName = "disjointSegment_" + index;
+ segmentGeneratorConfig.setSegmentName(segmentName);
+ segmentGeneratorConfig.setOutDir(disjointDir.getPath());
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new
GenericRowRecordReader(records));
+ driver.build();
+ segments.add(ImmutableSegmentLoader.load(new File(disjointDir,
segmentName), ReadMode.mmap));
+ }
+
+ // Threshold above one segment's cardinality (50), so at least two
segments must merge before a flush.
+ QueryContext queryContext =
+ newQueryContext("SELECT DISTINCT intColumn FROM testTable" +
MATCH_ALL + NO_LIMIT);
+ queryContext.setMaxExecutionThreads(1);
+ List<Operator> operators = new ArrayList<>(segments.size());
+ for (IndexSegment segment : segments) {
+ operators.add(PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(segment), queryContext).run());
+ }
+ StreamingDistinctCombineOperator combineOperator =
+ new StreamingDistinctCombineOperator(operators, queryContext,
EXECUTOR, 75);
+
+ Set<Object> values = new HashSet<>();
+ int numBlocks = 0;
+ combineOperator.start();
+ try {
+ BaseResultsBlock block = combineOperator.nextBlock();
+ while (!(block instanceof MetadataResultsBlock)) {
+ assertNull(block.getErrorMessages(), "Expected no errors but got: "
+ block.getErrorMessages());
+ numBlocks++;
+ for (Object[] row : block.getRows()) {
+ values.add(row[0]);
+ }
+ block = combineOperator.nextBlock();
+ }
+ } finally {
+ combineOperator.stop();
+ }
+
+ // 4 disjoint segments of 50 against a threshold of 75 means each flush
covers two segments.
+ assertEquals(numBlocks, 2, "Expected each flush window to span two
merged segments");
+ Set<Object> expected = new HashSet<>();
+ for (int i = 0; i < NUM_SEGMENTS * NUM_DISTINCT_VALUES; i++) {
+ expected.add(i);
+ }
+ assertEquals(values, expected);
+ } finally {
+ for (IndexSegment segment : segments) {
+ segment.destroy();
+ }
+ FileUtils.deleteDirectory(disjointDir);
+ }
+ }
+
+ /// maxRowsInDistinct bounds scanned docs for the whole query. Each flush
starts from an empty accumulator, so
+ /// the scanned-doc count has to be carried across windows or the bound
silently becomes per-window and never
+ /// trips. Also pins that the reason reaches the terminal metadata block: it
is recorded on the accumulated block,
+ /// which by then has already been streamed out, so without the carry-over
the broker would report a truncated
+ /// result as complete.
+ ///
+ /// The threshold here is above one segment's cardinality (50) on purpose.
Below it every block flushes on
+ /// adoption, `mergeResultsBlocks` never runs, and no distinct guardrail can
fire at all - see the class Javadoc
+ /// on [StreamingDistinctCombineOperator].
+ @Test
+ public void testMaxRowsInDistinctIsPerQueryAcrossFlushWindows() {
+ // 4 segments x 100 docs = 400 scanned; bound well below that.
+ FlushResult bounded = runStreaming(
+ "SET maxRowsInDistinct = 150; SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 75);
+ assertEquals(bounded._earlyTerminationReason,
EarlyTerminationReason.DISTINCT_MAX_ROWS,
+ "maxRowsInDistinct must terminate the query across flush windows, not
per window");
+
+ // Same query without the bound returns the full set and reports no early
termination.
+ FlushResult unbounded = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 75);
+ assertEquals(unbounded._earlyTerminationReason,
EarlyTerminationReason.NONE);
+ assertEquals(distinctValues(unbounded), expectedIntValues());
+ // Asserted on the unbounded run only. The bounded one stops polling as
soon as the merger reports the query
+ // satisfied, so its terminal metadata block is built while the worker
threads are still being cancelled and the
+ // scanned-doc total is whatever they happened to reach. See
testStreamingDistinctProducesMultipleBlocks for the
+ // multi-flush-window version of this assertion.
+ assertEquals(unbounded._numDocsScanned, TOTAL_NUM_DOCS);
+ }
+
+ /// The threshold has to reach the leaf QueryContext through the real
query-option path, and must be left alone
+ /// for non-distinct queries.
+ @Test
+ public void testFlushThresholdQueryOptionWiring() {
+ assertEquals(planned("SET streamingDistinctFlushThreshold = 100; SELECT
DISTINCT intColumn FROM testTable"), 100);
+ assertEquals(planned("SET streamingDistinctFlushThreshold = 0; SELECT
DISTINCT intColumn FROM testTable"), 0);
+ assertEquals(planned("SELECT DISTINCT intColumn FROM testTable"), 0);
+ // A group-by aggregation is not a distinct query, so the distinct
threshold must stay unset.
+ assertEquals(planned(
+ "SET streamingDistinctFlushThreshold = 100; SELECT intColumn, COUNT(*)
FROM testTable GROUP BY intColumn"), 0);
+ }
+
+ /// Goes through the real streaming-plan entry point (which is what applies
the query options on the server)
+ /// rather than setting the QueryContext field directly.
+ private int planned(String query) {
+ QueryContext queryContext = newQueryContext(query);
+ List<SegmentContext> segmentContexts = new ArrayList<>(NUM_SEGMENTS);
+ for (IndexSegment indexSegment : _indexSegments) {
+ segmentContexts.add(new SegmentContext(indexSegment));
+ }
+ PLAN_MAKER.makeStreamingInstancePlan(segmentContexts, queryContext,
EXECUTOR, block -> {
+ });
+ return queryContext.getStreamingDistinctFlushThreshold();
+ }
+
+ /// The routing guard in CombinePlanNode is the real gate, but the operator
is public, so the constructor keeps a
+ /// backstop. Asserted directly because a slipped negation here would leave
the backstop silently inert while every
+ /// routing test still passed.
+ @Test
+ public void testConstructorRejectsLeafReturningFinalResults() {
+ for (boolean keyUnpartitioned : new boolean[]{false, true}) {
+ QueryContext queryContext = newQueryContext("SELECT DISTINCT intColumn
FROM testTable" + NO_LIMIT);
+ queryContext.setServerReturnFinalResult(!keyUnpartitioned);
+
queryContext.setServerReturnFinalResultKeyUnpartitioned(keyUnpartitioned);
+ assertThrows(IllegalStateException.class,
+ () -> new StreamingDistinctCombineOperator(List.of(), queryContext,
EXECUTOR, 10));
+ }
+ // Control: the same construction succeeds with neither flag set, so the
assertions above cannot be passing
+ // because of some unrelated IllegalStateException.
+ QueryContext queryContext = newQueryContext("SELECT DISTINCT intColumn
FROM testTable" + NO_LIMIT);
+ assertNotNull(new StreamingDistinctCombineOperator(List.of(),
queryContext, EXECUTOR, 10));
+ }
+
+ @Test
+ public void testStreamingDistinctProducesMultipleBlocks() {
+ // 50 distinct values with a flush threshold of 10 forces several flushes
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ assertTrue(result._numBlocks > 1, "Expected multiple data blocks but got "
+ result._numBlocks);
+ assertEquals(distinctValues(result), expectedIntValues());
+ // Guards the "carrying the count across flush windows cannot double
count" claim in mergeBlock(): each segment's
+ // docs must be counted exactly once no matter how many windows they were
spread over. Holds because
+ // attachExecutionStats() recomputes the total from the segment operators
rather than from the accumulated block,
+ // which is precisely the coupling a later refactor could break quietly.
+ assertEquals(result._numDocsScanned, TOTAL_NUM_DOCS);
+ }
+
+ /// Pins the correctness argument the design rests on: the leaf may emit the
same value in more than one flush
+ /// window, and de-duplicating the union still yields the exact answer.
+ @Test
+ public void testDuplicateValuesAcrossFlushWindowsAreIdempotent() {
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ assertTrue(result._rows.size() > NUM_DISTINCT_VALUES,
+ "Expected duplicate values across flush windows, but only " +
result._rows.size() + " rows were emitted");
+ assertEquals(distinctValues(result), expectedIntValues());
+ }
+
+ @Test
+ public void testHighThresholdProducesSingleBlock() {
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10_000);
+
+ assertEquals(result._numBlocks, 1, "Expected a single data block when the
threshold exceeds the cardinality");
+ assertEquals(result._rows.size(), NUM_DISTINCT_VALUES);
+ assertEquals(distinctValues(result), expectedIntValues());
+ }
+
+ /// Single raw (non-dictionary-encoded) column — StringDistinctExecutor /
StringDistinctTable.
+ @Test
+ public void testRawStringColumn() {
+ FlushResult result = runStreaming("SELECT DISTINCT rawStringColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ assertTrue(result._numBlocks > 1);
+ Set<Object> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add("r" + i);
+ }
+ assertEquals(distinctValues(result), expected);
+ }
+
+ /// Single dictionary-encoded column reached through the scan path —
DictionaryBasedSingleColumnDistinctExecutor.
+ /// Its getResult() materializes actual values from the segment dictionary,
which is what makes the flushed block
+ /// self-contained (no detachFromWorkerThreadState hook needed).
Segment-local dict ids leaking out here would show
+ /// up as values outside 0..49.
+ @Test
+ public void testDictionaryEncodedColumnMaterializesValues() {
+ FlushResult result = runStreaming("SELECT DISTINCT dictStringColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ Set<Object> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add("d" + i);
+ }
+ assertEquals(distinctValues(result), expected);
+ }
+
+ /// All columns dictionary-encoded —
DictionaryBasedMultiColumnDistinctExecutor.
+ @Test
+ public void testMultiColumnAllDictionaryEncoded() {
+ FlushResult result =
+ runStreaming("SELECT DISTINCT intColumn, dictStringColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ assertTrue(result._numBlocks > 1);
+ Set<List<Object>> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add(List.of(i, "d" + i));
+ }
+ assertEquals(distinctRows(result), expected);
+ }
+
+ /// At least one raw column — RawMultiColumnDistinctExecutor /
MultiColumnDistinctTable.
+ @Test
+ public void testMultiColumnWithRawColumn() {
+ FlushResult result =
+ runStreaming("SELECT DISTINCT intColumn, rawStringColumn FROM
testTable" + MATCH_ALL + NO_LIMIT, 10);
+
+ assertTrue(result._numBlocks > 1);
+ Set<List<Object>> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add(List.of(i, "r" + i));
+ }
+ assertEquals(distinctRows(result), expected);
+ }
+
+ /// DistinctPlanNode can also select the index-based operators when
useIndexBasedDistinctOperator is set. They
+ /// build their DistinctTable differently from the scan path, so make sure
their blocks flow through the streaming
+ /// combine correctly (in particular that real values, not segment-local
dict ids, come out).
+ @Test
+ public void testIndexBasedDistinctOperatorPath() {
+ FlushResult result = runStreaming(
+ "SET useIndexBasedDistinctOperator = true; SELECT DISTINCT
dictStringColumn FROM testTable" + MATCH_ALL
+ + NO_LIMIT, 10);
+
+ Set<Object> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add("d" + i);
+ }
+ assertEquals(distinctValues(result), expected);
+ }
+
+ /// Without a filter, DistinctPlanNode picks
DictionaryBasedDistinctOperator, which reads the dictionary directly
+ /// rather than scanning. The streaming combine must handle those blocks too.
+ @Test
+ public void testDictionaryBasedDistinctOperatorPath() {
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + NO_LIMIT, 10);
+
+ assertEquals(distinctValues(result), expectedIntValues());
+ }
+
+ /// When no rows match, every segment yields an empty table and the operator
must return only the metadata block —
+ /// never an empty data block, and never an EmptyDistinctTable as the
accumulator (it throws on mergeDistinctTable).
+ @Test
+ public void testNoMatchingRowsReturnsOnlyMetadata() {
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable WHERE intColumn < 0" + NO_LIMIT, 10);
+
+ assertEquals(result._numBlocks, 0, "Expected no data blocks when nothing
matches");
+ assertTrue(result._rows.isEmpty());
+ }
+
+ /// A finite LIMIT below the flush threshold still bounds the table, and the
streaming operator must not lose rows
+ /// when the per-segment executors short-circuit on
DistinctTable.isSatisfied().
+ @Test
+ public void testFiniteLimitBelowThreshold() {
+ FlushResult result = runStreaming("SELECT DISTINCT intColumn FROM
testTable" + MATCH_ALL + " LIMIT 20", 10_000);
+
+ Set<Object> values = distinctValues(result);
+ assertEquals(values.size(), 20, "Expected exactly LIMIT distinct values");
+ assertTrue(expectedIntValues().containsAll(values), "Returned values must
be a subset of the real distinct set");
+ }
+
+ /// CombinePlanNode must only pick the streaming operator when every guard
holds.
+ @Test
+ public void testCombinePlanNodeSelection() {
+ // Streaming applies: distinct, no ORDER BY, unbounded limit, not
returning final results
+ assertTrue(buildCombineOperator("SELECT DISTINCT intColumn FROM testTable"
+ NO_LIMIT, 10, false)
+ instanceof StreamingDistinctCombineOperator, "Expected the streaming
operator for an unbounded distinct");
+
+ // Threshold not set (0) — feature off
+ assertTrue(buildCombineOperator("SELECT DISTINCT intColumn FROM testTable"
+ NO_LIMIT, 0, false)
+ instanceof DistinctCombineOperator, "Expected the blocking operator
when the threshold is 0");
+
+ // LIMIT <= threshold already bounds the table and gives the isSatisfied()
short-circuit
+ assertTrue(buildCombineOperator("SELECT DISTINCT intColumn FROM testTable
LIMIT 10", 10, false)
+ instanceof DistinctCombineOperator, "Expected the blocking operator
when LIMIT bounds the table");
+
+ // ORDER BY keeps a bounded top-LIMIT heap
+ assertTrue(
+ buildCombineOperator("SELECT DISTINCT intColumn FROM testTable ORDER
BY intColumn LIMIT 1000", 10, false)
+ instanceof DistinctCombineOperator, "Expected the blocking
operator with ORDER BY");
+
+ // No aggregate above the leaf guaranteed to de-duplicate across flush
windows
+ assertTrue(buildCombineOperator("SELECT DISTINCT intColumn FROM testTable"
+ NO_LIMIT, 10, true, false)
+ instanceof DistinctCombineOperator, "Expected the blocking operator
when the server returns final results");
+ assertTrue(buildCombineOperator("SELECT DISTINCT intColumn FROM testTable"
+ NO_LIMIT, 10, false, true)
+ instanceof DistinctCombineOperator,
+ "Expected the blocking operator when the leaf returns final results
for unpartitioned keys");
+
+ // A group-by aggregation must keep its own operator even with the
distinct threshold set
+ assertFalse(
+ buildCombineOperator("SELECT intColumn, COUNT(*) FROM testTable GROUP
BY intColumn" + NO_LIMIT, 10, false)
+ instanceof StreamingDistinctCombineOperator, "The distinct
threshold must not capture group-by queries");
+ }
+
+ /// A null is tracked by a flag on the DistinctTable rather than as a set
entry, and each flush window starts from a
+ /// fresh accumulator. The null must therefore survive the flush boundary:
it is counted by size(), emitted by
+ /// getRows(), and re-emitted in later windows (harmless, since the
downstream de-duplication is idempotent).
+ @Test
+ public void testNullHandling()
+ throws Exception {
+ File nullDir = new File(FileUtils.getTempDirectory(),
"StreamingDistinctCombineOperatorTest_null");
+ FileUtils.deleteDirectory(nullDir);
+ String nullableColumn = "nullableColumn";
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNullHandlingEnabled(true).build();
+ Schema schema = new Schema.SchemaBuilder()
+ .addSingleValueDimension(INT_COLUMN, FieldSpec.DataType.INT)
+ .addSingleValueDimension(nullableColumn, FieldSpec.DataType.STRING)
+ .build();
+
+ List<IndexSegment> segments = new ArrayList<>(NUM_SEGMENTS);
+ try {
+ for (int index = 0; index < NUM_SEGMENTS; index++) {
+ List<GenericRow> records = new ArrayList<>(NUM_DISTINCT_VALUES);
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ GenericRow record = new GenericRow();
+ record.putValue(INT_COLUMN, i);
+ // Half the rows are null, so every segment contributes both real
values and the null marker.
+ if (i % 2 == 0) {
+ record.putValue(nullableColumn, "n" + i);
+ } else {
+ record.putValue(nullableColumn, null);
+ }
+ records.add(record);
+ }
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(tableConfig, schema);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ String segmentName = "nullSegment_" + index;
+ segmentGeneratorConfig.setSegmentName(segmentName);
+ segmentGeneratorConfig.setOutDir(nullDir.getPath());
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new
GenericRowRecordReader(records));
+ driver.build();
+ segments.add(ImmutableSegmentLoader.load(new File(nullDir,
segmentName), ReadMode.mmap));
+ }
+
+ QueryContext queryContext =
+ newQueryContext("SELECT DISTINCT nullableColumn FROM testTable" +
MATCH_ALL + NO_LIMIT);
+ queryContext.setNullHandlingEnabled(true);
+ List<Operator> operators = new ArrayList<>(segments.size());
+ for (IndexSegment segment : segments) {
+ operators.add(PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(segment), queryContext).run());
+ }
+ StreamingDistinctCombineOperator combineOperator =
+ new StreamingDistinctCombineOperator(operators, queryContext,
EXECUTOR, 5);
+
+ Set<Object> values = new HashSet<>();
+ int numBlocks = 0;
+ combineOperator.start();
+ try {
+ BaseResultsBlock block = combineOperator.nextBlock();
+ while (!(block instanceof MetadataResultsBlock)) {
+ assertNull(block.getErrorMessages(), "Expected no errors but got: "
+ block.getErrorMessages());
+ numBlocks++;
+ for (Object[] row : block.getRows()) {
+ values.add(row[0]);
+ }
+ block = combineOperator.nextBlock();
+ }
+ } finally {
+ combineOperator.stop();
+ }
+
+ assertTrue(numBlocks > 1, "Expected multiple flush windows but got " +
numBlocks);
+ Set<Object> expected = new HashSet<>();
+ expected.add(null);
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i += 2) {
+ expected.add("n" + i);
+ }
+ assertEquals(values, expected, "The null must survive the flush
boundary");
+ } finally {
+ for (IndexSegment segment : segments) {
+ segment.destroy();
+ }
+ FileUtils.deleteDirectory(nullDir);
+ }
+ }
+
+ private BaseCombineOperator<?> buildCombineOperator(String query, int
flushThreshold,
+ boolean serverReturnFinalResult) {
+ return buildCombineOperator(query, flushThreshold,
serverReturnFinalResult, false);
+ }
+
+ private BaseCombineOperator<?> buildCombineOperator(String query, int
flushThreshold,
+ boolean serverReturnFinalResult, boolean
serverReturnFinalResultKeyUnpartitioned) {
+ QueryContext queryContext = newQueryContext(query);
+ queryContext.setStreamingDistinctFlushThreshold(flushThreshold);
+ queryContext.setServerReturnFinalResult(serverReturnFinalResult);
+
queryContext.setServerReturnFinalResultKeyUnpartitioned(serverReturnFinalResultKeyUnpartitioned);
+ List<PlanNode> planNodes = new ArrayList<>(NUM_SEGMENTS);
+ for (IndexSegment indexSegment : _indexSegments) {
+ planNodes.add(PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(indexSegment), queryContext));
+ }
+ ResultsBlockStreamer streamer = block -> {
+ };
+ return new CombinePlanNode(planNodes, queryContext, EXECUTOR,
streamer).run();
+ }
+
+ private FlushResult runStreaming(String query, int flushThreshold) {
+ QueryContext queryContext = newQueryContext(query);
+ List<Operator> operators = new ArrayList<>(NUM_SEGMENTS);
+ for (IndexSegment indexSegment : _indexSegments) {
+ operators.add(PLAN_MAKER.makeSegmentPlanNode(new
SegmentContext(indexSegment), queryContext).run());
+ }
+ StreamingDistinctCombineOperator combineOperator =
+ new StreamingDistinctCombineOperator(operators, queryContext,
EXECUTOR, flushThreshold);
+
+ List<Object[]> rows = new ArrayList<>();
+ int numBlocks = 0;
+ EarlyTerminationReason earlyTerminationReason;
+ long numDocsScanned;
+ combineOperator.start();
+ try {
+ BaseResultsBlock block = combineOperator.nextBlock();
+ while (!(block instanceof MetadataResultsBlock)) {
+ assertNull(block.getErrorMessages(), "Expected no errors but got: " +
block.getErrorMessages());
+ assertTrue(block instanceof DistinctResultsBlock,
+ "Expected DistinctResultsBlock but got: " + block.getClass());
+ numBlocks++;
+ rows.addAll(block.getRows());
+ block = combineOperator.nextBlock();
+ }
+ // The reason is set on the accumulated block, which has already been
streamed out by then, so the terminal
+ // metadata block is the only place the broker can still learn the
results were truncated.
+ earlyTerminationReason = block.getEarlyTerminationReason();
+ numDocsScanned = block.getNumDocsScanned();
+ } finally {
+ combineOperator.stop();
+ }
+ return new FlushResult(numBlocks, rows, earlyTerminationReason,
numDocsScanned);
+ }
+
+ private static QueryContext newQueryContext(String query) {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+ queryContext.setEndTimeMs(System.currentTimeMillis() +
Server.DEFAULT_QUERY_EXECUTOR_TIMEOUT_MS);
+ return queryContext;
+ }
+
+ private static Set<Object> expectedIntValues() {
+ Set<Object> expected = new HashSet<>();
+ for (int i = 0; i < NUM_DISTINCT_VALUES; i++) {
+ expected.add(i);
+ }
+ return expected;
+ }
+
+ private static Set<Object> distinctValues(FlushResult result) {
+ Set<Object> values = new HashSet<>();
+ for (Object[] row : result._rows) {
+ assertEquals(row.length, 1);
+ values.add(row[0]);
+ }
+ return values;
+ }
+
+ private static Set<List<Object>> distinctRows(FlushResult result) {
+ Set<List<Object>> distinct = new HashSet<>();
+ for (Object[] row : result._rows) {
+ distinct.add(Arrays.asList(row));
+ }
+ return distinct;
+ }
+
+ private static class FlushResult {
+ final int _numBlocks;
+ final List<Object[]> _rows;
+ final EarlyTerminationReason _earlyTerminationReason;
+ final long _numDocsScanned;
+
+ FlushResult(int numBlocks, List<Object[]> rows, EarlyTerminationReason
earlyTerminationReason,
+ long numDocsScanned) {
+ _numBlocks = numBlocks;
+ _rows = rows;
+ _earlyTerminationReason = earlyTerminationReason;
+ _numDocsScanned = numDocsScanned;
+ }
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws IOException {
+ for (IndexSegment indexSegment : _indexSegments) {
+ indexSegment.destroy();
+ }
+ FileUtils.deleteDirectory(TEMP_DIR);
+ }
+}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
index ce6db81ee8b..e5f83cb01a3 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java
@@ -30,6 +30,7 @@ import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -2336,6 +2337,89 @@ public class MultiStageEngineIntegrationTest extends
BaseClusterIntegrationTestS
}
}
+ private static final String STREAMING_DISTINCT_OPTION = "SET
streamingDistinctFlushThreshold = 2; ";
+
+ /// Verifies that the streaming distinct feature produces correct results
when the query option
+ /// streamingDistinctFlushThreshold is set. Compares streaming results
against a baseline query without the option.
+ ///
+ /// The threshold is set well below the real cardinality so the leaf flushes
several times and the same value is
+ /// emitted in more than one flush window - the case the intermediate stage
has to de-duplicate.
+ ///
+ /// NOTE: None of these queries may carry an ORDER BY. The planner pushes
collations into the leaf aggregate for
+ /// aggregates with no aggregate calls
(PinotAggregateExchangeNodeInsertRule#isGroupTrimmingEnabled), which makes
+ /// the leaf QueryContext order-by non-null and disqualifies the streaming
operator - the assertions would then
+ /// compare the blocking path against itself and pass no matter what the new
operator does.
+ @Test
+ public void testStreamingDistinct()
+ throws Exception {
+ // Single column
+ assertStreamingDistinctMatchesBaseline("SELECT DISTINCT Carrier FROM
mytable LIMIT 1000");
+
+ // Multiple columns
+ assertStreamingDistinctMatchesBaseline("SELECT DISTINCT Carrier,
OriginState FROM mytable LIMIT 1000");
+
+ // The GROUP BY spelling of the same thing: an aggregate with no aggregate
calls, which the leaf stage rewrites
+ // into a single-stage DISTINCT query
(NonAggregationGroupByToDistinctQueryRewriter).
+ assertStreamingDistinctMatchesBaseline("SELECT Carrier FROM mytable GROUP
BY Carrier LIMIT 1000");
+ }
+
+ /// The streaming operator must be skipped when the leaf-stage LIMIT is at
or below the flush threshold (the limit
+ /// already bounds the table and keeps the early-termination short-circuit)
and when the query has an ORDER BY.
+ @Test
+ public void testStreamingDistinctNotUsedWhenGuardsFail()
+ throws Exception {
+ // LIMIT equal to and below the threshold of 2
+ assertStreamingDistinctOperatorUsed("SELECT DISTINCT Carrier FROM mytable
LIMIT 2", false);
+ assertStreamingDistinctOperatorUsed("SELECT DISTINCT Carrier FROM mytable
LIMIT 1", false);
+ // ORDER BY is pushed into the leaf aggregate, so the leaf sees a bounded
top-LIMIT heap
+ assertStreamingDistinctOperatorUsed("SELECT DISTINCT Carrier FROM mytable
ORDER BY Carrier LIMIT 1000", false);
+ // Control: same shape, no ORDER BY, LIMIT above the threshold
+ assertStreamingDistinctOperatorUsed("SELECT DISTINCT Carrier FROM mytable
LIMIT 1000", true);
+ }
+
+ private void assertStreamingDistinctMatchesBaseline(String query)
+ throws Exception {
+ // Guard against a vacuous comparison: without this, a query that silently
takes the blocking path in both runs
+ // would pass whatever the streaming operator did.
+ assertStreamingDistinctOperatorUsed(query, true);
+
+ JsonNode baselineRows = postQuery(query).get("resultTable").get("rows");
+ assertTrue(baselineRows.size() > 0, "Baseline query should return results:
" + query);
+
+ JsonNode streamingRows =
+ postQuery(STREAMING_DISTINCT_OPTION +
query).get("resultTable").get("rows");
+
+ // Results are unordered, so compare as sorted multisets. This still
catches both duplicates and dropped rows.
+ assertEquals(sortedRows(streamingRows), sortedRows(baselineRows),
+ "Streaming distinct returned a different row multiset for: " + query);
+ }
+
+ /// Asserts which leaf-stage combine operator the server actually picks.
`explainAskingServers` pulls the
+ /// single-stage leaf plan in under `LeafStageCombineOperator`, where
operator names appear in the UpperCamel form
+ /// produced by `BaseOperator#getExplainName()` - so
`STREAMING_COMBINE_DISTINCT` renders as
+ /// `StreamingCombineDistinct`. Note `CombineDistinct` is a substring of it,
so only the streaming name can be
+ /// tested for presence/absence unambiguously.
+ private void assertStreamingDistinctOperatorUsed(String query, boolean
expected)
+ throws Exception {
+ String plan = postQuery("SET explainAskingServers=true; " +
STREAMING_DISTINCT_OPTION + "EXPLAIN PLAN FOR " + query)
+ .toString();
+ assertEquals(plan.contains("StreamingCombineDistinct"), expected,
+ "Unexpected combine operator selection for: " + query + "\nPlan: " +
plan);
+ }
+
+ private static List<List<String>> sortedRows(JsonNode rows) {
+ List<List<String>> result = new ArrayList<>(rows.size());
+ for (JsonNode row : rows) {
+ List<String> values = new ArrayList<>(row.size());
+ for (JsonNode value : row) {
+ values.add(value.asText());
+ }
+ result.add(values);
+ }
+ result.sort(Comparator.comparing(Object::toString));
+ return result;
+ }
+
private JsonNode getQueryResultForDBTest(String column, String tableName,
@Nullable String database,
Map<String, String> headers)
throws Exception {
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 71298ee7cb3..508a838e88f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -546,6 +546,17 @@ public class CommonConstants {
public static final String
CONFIG_OF_MSE_STREAMING_GROUP_BY_FLUSH_THRESHOLD =
"pinot.broker.mse.streaming.group.by.flush.threshold";
public static final int DEFAULT_MSE_STREAMING_GROUP_BY_FLUSH_THRESHOLD =
-1;
+
+ /// Default flush threshold for the streaming distinct leaf-stage operator
on MSE. When positive, the broker
+ /// injects this value as the `streamingDistinctFlushThreshold` query
option for MSE queries that do not already
+ /// specify it, opting the cluster into the streaming distinct behavior by
default. Setting the query option
+ /// explicitly (including to `0` to disable) always wins over the broker
default.
+ ///
+ /// See [Request.QueryOptionKey#STREAMING_DISTINCT_FLUSH_THRESHOLD] for
the conditions a query must meet before
+ /// the threshold takes effect; queries that do not meet them are
unaffected by this default.
+ public static final String
CONFIG_OF_MSE_STREAMING_DISTINCT_FLUSH_THRESHOLD =
+ "pinot.broker.mse.streaming.distinct.flush.threshold";
+ public static final int DEFAULT_MSE_STREAMING_DISTINCT_FLUSH_THRESHOLD =
-1;
// Whether to infer partition hint by default or not.
// This value can always be overridden by INFER_PARTITION_HINT query option
public static final String CONFIG_OF_INFER_PARTITION_HINT =
"pinot.broker.multistage.infer.partition.hint";
@@ -797,6 +808,27 @@ public class CommonConstants {
/// Flush threshold for streaming group-by on MSE leaf stages.
public static final String STREAMING_GROUP_BY_FLUSH_THRESHOLD =
"streamingGroupByFlushThreshold";
+ /// Flush threshold for streaming distinct on MSE leaf stages. When
positive, the leaf flushes its
+ /// accumulated distinct values downstream once they reach this count
and starts a fresh table, bounding
+ /// server memory and pushing the residual de-duplication into the
partitioned intermediate stage.
+ ///
+ /// The value is also a feature gate, so it is a silent no-op unless
ALL of the following hold. Setting it
+ /// produces no error and no diagnostic when they do not:
+ ///
+ /// - the query is a DISTINCT query (an MSE aggregate with no
aggregate calls; a group-by with real
+ /// aggregations uses [#STREAMING_GROUP_BY_FLUSH_THRESHOLD] instead)
+ /// - it has no ORDER BY — an ordered distinct already keeps a bounded
top-LIMIT heap
+ /// - the leaf-stage LIMIT is strictly greater than this threshold — a
smaller LIMIT already bounds the
+ /// table and gives the early-termination short-circuit, which
streaming would throw away
+ /// - the leaf is not returning final results (the
`is_partitioned_by_group_by_keys` and
+ /// `is_leaf_return_final_result` hints), because then no stage
above the leaf is guaranteed to
+ /// de-duplicate across flush windows
+ ///
+ /// NOTE: This relies on a downstream stage de-duplicating the partial
flushes, which is what the MSE hash
+ /// exchange over the distinct columns provides. Do not set it on the
gRPC streaming query path, where
+ /// there is no such stage and the client would observe duplicate rows
across flush windows.
+ public static final String STREAMING_DISTINCT_FLUSH_THRESHOLD =
"streamingDistinctFlushThreshold";
+
public static final String NUM_REPLICA_GROUPS_TO_QUERY =
"numReplicaGroupsToQuery";
public static final String ORDERED_PREFERRED_POOLS =
"orderedPreferredPools";
public static final String USE_FIXED_REPLICA = "useFixedReplica";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
