gortiz commented on code in PR #19469:
URL: https://github.com/apache/pinot/pull/19469#discussion_r4080364158
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -268,12 +367,87 @@ private MseBlock.Eos consumeGroupBy() {
MseBlock block = _input.nextBlock();
while (block.isData()) {
_groupByExecutor.processBlock((MseBlock.Data) block);
+ if (_spillThreshold > 0 && _groupByExecutor.getNumGroups() >=
_spillThreshold) {
+ spillCurrentGroups();
+ }
checkTerminationAndSampleUsage();
block = _input.nextBlock();
}
return (MseBlock.Eos) block;
}
+ private void spillCurrentGroups() {
+ assert _groupByExecutor != null;
+ if (_groupByExecutor.getNumGroups() == 0) {
+ return;
+ }
+ if (_spillManager == null) {
+ _spillManager =
+ new AggregationSpillManager(_spillPartitions, _groupKeyIds.length,
getSpillSchema(), _aggFunctions);
+ _spillDirectory = _spillManager.getSpillDirectory();
+ }
+ AggregationSpillManager.SpillResult spillResult =
+ _spillManager.spill(_groupByExecutor.getIntermediateResultIterator());
+ _statMap.merge(StatKey.SPILL_COUNT, 1L);
+ _statMap.merge(StatKey.SPILLED_ROWS, spillResult.getRows());
+ _statMap.merge(StatKey.SPILLED_BYTES, spillResult.getBytes());
+ _groupByExecutor = newInputGroupByExecutor();
+ }
+
+ private DataSchema getSpillSchema() {
+ String[] columnNames = _resultSchema.getColumnNames().clone();
+ ColumnDataType[] columnDataTypes =
_resultSchema.getColumnDataTypes().clone();
+ int numKeys = _groupKeyIds.length;
+ for (int i = 0; i < _aggFunctions.length; i++) {
+ columnDataTypes[numKeys + i] = _aggFunctions[i].getType() ==
AggregationFunctionType.ANYVALUE
+ ? ColumnDataType.OBJECT :
_aggFunctions[i].getIntermediateResultColumnType();
+ }
+ return new DataSchema(columnNames, columnDataTypes);
+ }
+
+ @Nullable
+ private MseBlock.Data restoreSpillPartition(int partitionId) {
+ assert _spillManager != null;
+ if (!_spillManager.hasPartition(partitionId)) {
+ return null;
+ }
+ MultistageGroupByExecutor executor = newSpillMergeGroupByExecutor();
+ _spillManager.consumePartition(partitionId, block -> {
+ executor.processSpillBlock(block);
+ if (executor.getNumGroups() > _spillThreshold) {
+ throw QueryErrorCode.SERVER_RESOURCE_LIMIT_EXCEEDED.asException(
Review Comment:
**CORRECTNESS/UX: enabling spill can turn a query that used to succeed into
a hard SERVER_RESOURCE_LIMIT_EXCEEDED**
Setting mseAggregationSpillThreshold has two coupled effects: (a) the input
executor's numGroupsLimit becomes Integer.MAX_VALUE, and (b) any restored
partition above the threshold throws.
So for a skewed key distribution — which is exactly the shape that motivates
spilling — the outcome flips from 'degrade to numGroupsLimit / trim' to 'query
fails'. A user who turns the option on to make a big GROUP BY work can get a
new failure mode on a query that previously returned (truncated) results.
The PR explicitly scopes out recursive repartitioning, which is the real
fix, and that is a reasonable v1 call. But the intermediate fallback is cheap:
instead of throwing, fall back to the pre-existing numGroupsLimit behaviour for
that partition (set NUM_GROUPS_LIMIT_REACHED, stop admitting groups) and only
throw when _errorOnNumGroupsLimit is set. That keeps the failure semantics
identical to non-spill mode. At minimum this behaviour change needs to be in
the release notes, not only in the PR body.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregationSpillManager.java:
##########
@@ -0,0 +1,428 @@
+/**
+ * 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.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.pinot.common.datablock.DataBlock;
+import org.apache.pinot.common.datablock.DataBlockUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SerializedDataBlock;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/// Manages hash-partitioned aggregation spill files for one operator. The
caller must finish reading before calling
+/// [#close()], which recursively removes the operator-scoped directory. This
class is not thread-safe.
+@SuppressWarnings("rawtypes")
+class AggregationSpillManager implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AggregationSpillManager.class);
+ private static final String SPILL_FILE_PREFIX = "partition-";
+ private static final String SPILL_FILE_SUFFIX = ".spill";
+ private static final String SPILL_SCOPE = "AggregationSpillManager#spill";
+ private static final String RESTORE_SCOPE =
"AggregationSpillManager#consumePartition";
+ private static final int MAX_BUFFERED_ROWS = 1024;
+ private static final int MAX_BUFFERED_PARTITIONS = 8;
+
+ private final int _numPartitions;
+ private final int _numGroupKeys;
+ private final DataSchema _spillSchema;
+ private final AggregationFunction[] _aggFunctions;
+ private final Path _spillDirectory;
+ private final Map<Integer, FileChannel> _spillWriters = new HashMap<>();
+ private final ByteBuffer _recordLengthBuffer =
ByteBuffer.allocate(Integer.BYTES);
+
+ AggregationSpillManager(int numPartitions, int numGroupKeys, DataSchema
spillSchema,
+ AggregationFunction[] aggFunctions) {
+ if (numPartitions <= 0 || numPartitions >
Server.MAX_MSE_AGGREGATION_SPILL_PARTITIONS) {
+ throw new IllegalArgumentException(
+ "Number of spill partitions must be between 1 and " +
Server.MAX_MSE_AGGREGATION_SPILL_PARTITIONS);
+ }
+ if (numGroupKeys < 0 || numGroupKeys > spillSchema.size()) {
+ throw new IllegalArgumentException("Invalid number of group keys: " +
numGroupKeys);
+ }
+ _numPartitions = numPartitions;
+ _numGroupKeys = numGroupKeys;
+ _spillSchema = spillSchema;
+ _aggFunctions = aggFunctions;
+ try {
+ _spillDirectory = Files.createTempDirectory("pinot-aggregation-spill-");
Review Comment:
**RISK: spill location is hard-coded to java.io.tmpdir with no byte budget
and no orphan cleanup**
Three separate operational gaps in one line.
1. Location is not configurable. Every other on-disk Pinot feature (segment
dir, minion task dir, consumer dir) has a config key. In containerized
deployments java.io.tmpdir is very often a small tmpfs (i.e. RAM), which means
'spill to disk' silently becomes 'spill to memory' and defeats the whole point.
Add a `pinot.query.mse.aggregation.spill.dir` with a sensible default derived
from the instance data dir.
2. No cap on bytes written, per query or per server. Before this PR a
runaway GROUP BY was bounded by numGroupsLimit. With a spill threshold set, the
input executor's numGroupsLimit is raised to Integer.MAX_VALUE (see
MultistageGroupByExecutor.forSpillInput), so the only bound left is local disk.
N concurrent spilling queries can fill the volume and take down unrelated
ingestion/query paths on that server. SPILLED_BYTES is tracked but never
checked. Please add a per-query max spill bytes and ideally a server-wide
budget.
3. Orphans. Directories are only removed on close/cancel/error paths; a JVM
crash or kill -9 leaves `pinot-aggregation-spill-*` behind forever. Worth a
startup sweep of the spill root (easier once #1 gives you a dedicated
directory).
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -268,12 +367,87 @@ private MseBlock.Eos consumeGroupBy() {
MseBlock block = _input.nextBlock();
while (block.isData()) {
_groupByExecutor.processBlock((MseBlock.Data) block);
+ if (_spillThreshold > 0 && _groupByExecutor.getNumGroups() >=
_spillThreshold) {
+ spillCurrentGroups();
+ }
checkTerminationAndSampleUsage();
block = _input.nextBlock();
}
return (MseBlock.Eos) block;
}
+ private void spillCurrentGroups() {
+ assert _groupByExecutor != null;
+ if (_groupByExecutor.getNumGroups() == 0) {
+ return;
+ }
+ if (_spillManager == null) {
+ _spillManager =
+ new AggregationSpillManager(_spillPartitions, _groupKeyIds.length,
getSpillSchema(), _aggFunctions);
+ _spillDirectory = _spillManager.getSpillDirectory();
+ }
+ AggregationSpillManager.SpillResult spillResult =
+ _spillManager.spill(_groupByExecutor.getIntermediateResultIterator());
+ _statMap.merge(StatKey.SPILL_COUNT, 1L);
+ _statMap.merge(StatKey.SPILLED_ROWS, spillResult.getRows());
+ _statMap.merge(StatKey.SPILLED_BYTES, spillResult.getBytes());
+ _groupByExecutor = newInputGroupByExecutor();
+ }
+
+ private DataSchema getSpillSchema() {
+ String[] columnNames = _resultSchema.getColumnNames().clone();
+ ColumnDataType[] columnDataTypes =
_resultSchema.getColumnDataTypes().clone();
+ int numKeys = _groupKeyIds.length;
+ for (int i = 0; i < _aggFunctions.length; i++) {
+ columnDataTypes[numKeys + i] = _aggFunctions[i].getType() ==
AggregationFunctionType.ANYVALUE
Review Comment:
**The ANYVALUE special case is a workaround for a bug in
AnyValueAggregationFunction, and it does not generalize**
This works around AnyValueAggregationFunction._resultType being resolved
lazily in ensureResultType(BlockValSet): on a merge/intermediate executor that
never sees raw data, getIntermediateResultColumnType() falls back to STRING,
which would then mis-describe the spilled column.
Two problems with fixing it here:
- It is a denylist of exactly one function, matched by
AggregationFunctionType. Any other function whose intermediate column type is
lazily resolved or otherwise not a faithful description of what
extractGroupByResult() returns will silently produce a wrong spill schema — and
the failure will surface as a deserialization error or a wrong result, far from
here. Nothing prevents the next such function from being added.
- The actual contract being violated belongs to AggregationFunction, not to
this operator.
Preferred: fix AnyValue so getIntermediateResultColumnType() is honest (e.g.
return OBJECT when _resultType is unknown), and drop the special case. If that
is too invasive for this PR, add a TODO with an issue link explaining the
invariant that any new function must satisfy.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java:
##########
@@ -309,7 +312,7 @@ private Object deserializeValue(ByteBuffer buffer) {
case BIG_DECIMAL:
return new BigDecimal(new String(deserializeVariableBytes(buffer),
StandardCharsets.UTF_8));
case BYTES:
- return deserializeVariableBytes(buffer);
+ return new ByteArray(deserializeVariableBytes(buffer));
Review Comment:
**SCOPE + TESTS: this is a real pre-existing bug fix for ANY_VALUE(BYTES)
bundled into a spill PR, with no test of its own**
Before this change, deserializeValue() returned a raw byte[] for the BYTES
case, and extractFinalResult() does `(Comparable<?>) intermediateResult` —
byte[] is not Comparable, so ANY_VALUE over a BYTES column would
ClassCastException on any path that deserializes an intermediate result (i.e.
every leaf -> intermediate stage hop, not just the new spill path). Switching
to ByteArray fixes that and is the right call.
But:
- It is an independent, user-visible bug fix living inside a 1.9k-line
feature PR. It deserves its own PR and its own changelog line so it is
backportable without the spill feature.
- AnyValueAggregationFunctionTest is untouched. The only coverage is
testSpillRoundTripsBinaryAnyValue in AggregateOperatorTest, which exercises it
through the spill path. Add a direct serialize/deserialize round-trip test for
the BYTES case, and one that asserts extractFinalResult() no longer throws —
that is the regression that actually shipped.
- Please double-check callers that consume the deserialized value and may
have been coded against byte[] (the in-memory type changes; the wire format
does not).
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -131,23 +157,58 @@ public AggregateOperator(OpChainExecutionContext context,
MultiStageOperator inp
_comparator = comparator;
_errorOnNumGroupsLimit = getErrorOnNumGroupsLimit(node.getNodeHint(),
context.getOpChainMetadata());
+ _numGroupsLimit =
MultistageGroupByExecutor.getNumGroupsLimit(_opChainMetadata, _nodeHint);
+ int spillThreshold = 0;
+ int spillPartitions = Server.DEFAULT_MSE_AGGREGATION_SPILL_PARTITIONS;
+ boolean spillEligible = !groupKeys.isEmpty() && !_leafReturnFinalResult
Review Comment:
**Spill is silently disabled whenever group trim is active — which includes
any GROUP BY with a pushed-down LIMIT**
groupTrimSize is set to something other than Integer.MAX_VALUE whenever
`node.getLimit() > 0 && minGroupTrimSize > 0`, and minGroupTrimSize defaults to
5000. So for a plan where the planner pushes a LIMIT into the aggregate,
spillEligible is false and the user's mseAggregationSpillThreshold is ignored
without a word.
That is a large fraction of real high-cardinality GROUP BY queries.
Excluding trim mode from v1 is a fine scoping decision — the problem is that
the exclusion is invisible. Combine with the QueryRunner gate comment: there
should be exactly one place that reports 'spill was requested but not applied,
because X'.
Also: this reads a bit awkwardly as a nested if. If configuredSpillThreshold
is null the partition option is never even validated, so `SET
mseAggregationSpillPartitions = 999` with no threshold set is accepted silently
instead of throwing. Consider validating options independently of eligibility.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/SendStatsPredicate.java:
##########
@@ -153,6 +165,11 @@ public boolean isSendStats() {
return _sendStats;
}
+ @Override
+ public boolean isClusterVersionCompatible() {
Review Comment:
**Graceful degradation for undecodable stats has shipped since 1.4.0 on the
mailbox leg. Only the (unreleased) stream leg is unprotected — close that gap
instead of adding this flag**
Before adding _clusterVersionCompatible, check what already ships. Two
independent mechanisms already degrade rather than fail when stats cannot be
handled, and both are in stable releases:
Make stats system more resilient in MSE (#15312, 2025-03-25)
-> BlockingMultiStreamConsumer.mergeStats: catch (Exception) ->
LOGGER.warn + _stats = null
Substitute TransferableBlock with MseBlock (#15245, 2025-04-14)
-> MailboxSendOperator.sendEos: catch (Exception) around
stats.serialize() -> List.of()
Both are ancestors of release-1.4.0, release-1.5.0 and release-1.5.1
(verified against the shipped sources, not just master).
Why that matters here: an unknown StatKey ordinal on an older node raises
ArrayIndexOutOfBoundsException at StatMap.java:267 (K key = keys[ordinal],
unchecked). On the mailbox leg that AIOOBE is caught by mergeStats, the stage's
stats are dropped, and the query completes. That is precisely the guarantee
this PR's gate is trying to add, and it has been true of every release since
1.4.0. This class's own javadoc agrees: the predicate was written for 1.3.0 and
lower, and the Mode enum says ALWAYS is strictly better than SAFE once everyone
is >= 1.4. Tying a brand-new feature to a compatibility mode whose stated
purpose is a version two majors back needs a stronger justification than is
given.
The one leg where it genuinely does not hold is stream-mode reporting, and
note that leg is itself unreleased — SubmitWithStream stats (#18458) and the
plugin-type follow-up (#18736) are master-only, first stable release 1.6.0.
There:
MultiStageStatsTreeDecoder.decodeNode (line 79) wraps deserializeStatMap
in catch (IOException) only.
AIOOBE is unchecked, so it escapes decode(), and StreamingQuerySession
only catches DecodeFailedException.
So the accurate statement is: the spill stat keys are already safe on the
mailbox path in every shipped release, and unsafe only on a path that has not
shipped yet. That is a two-line fix in the stats layer, not a compatibility
subsystem in the query layer:
1. StatMap.merge(DataInput) (line 267, and line 370 for the sibling):
bounds-check the ordinal and throw IllegalArgumentException rather than letting
AIOOBE escape. MultiStageQueryStats.mergeUpstream ALREADY catches
IllegalArgumentException | IllegalStateException, so this handles the mailbox
case at the intended layer instead of via an outer catch-all.
2. MultiStageStatsTreeDecoder.decodeNode: wrap RuntimeException into
DecodeFailedException so the stream leg degrades the same way the mailbox leg
already does.
With those, appending StatKey constants stops being a hazard for every
future stat, this flag and its plumbing are unnecessary, spill stops forcing
operators onto SAFE mode, and SendStatsPredicate goes back to covering only the
<= 1.3 case it was written for. Fixing it in the stats layer also pays forward:
the next rolling upgrade that adds a StatKey (1.6 -> 1.7 in stream mode) hits
exactly this gap.
Caveat, for completeness: entries carry no length prefix, so a reader
hitting an unknown ordinal cannot resume mid-buffer and must discard that whole
stat buffer. Discarding the buffer is already what both call sites do, so it
costs nothing. Per-entry type tags (see the note on the StatKey enum) would
additionally allow skipping one unknown key, but that is an optimisation on
top, not a prerequisite.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -808,6 +808,20 @@ public static class QueryOptionKey {
/// Flush threshold for streaming group-by on MSE leaf stages.
public static final String STREAMING_GROUP_BY_FLUSH_THRESHOLD =
"streamingGroupByFlushThreshold";
+ /// Maximum number of groups retained by a keyed MSE aggregation
executor before its intermediate states are
+ /// spilled to local disk. This option is honored only when the
server-level aggregation spill gate is enabled.
+ /// An absent value disables spilling. The first spill version does
not apply to global aggregation,
+ /// leaf-final-result, or group-trim modes.
+ public static final String MSE_AGGREGATION_SPILL_THRESHOLD =
"mseAggregationSpillThreshold";
Review Comment:
**EXTENSIBILITY 5/5: state in the javadoc that groups is a v1 proxy for
memory, and link the follow-up issue**
This javadoc defines the option as 'maximum number of groups retained ...
before its intermediate states are spilled'. Accurate, but it reads as the
intended long-term model rather than a v1 approximation, and that framing is
what future contributors will build on.
Group count is a proxy for the resource actually being protected, and a poor
one: bytes-per-group ranges from ~8 (SUM) to a whole HashSet (DISTINCTCOUNT) to
KBs (HLL/Theta) to unbounded (ANY_VALUE over STRING). That is why no user can
pick a safe value from first principles — the right number depends on which
aggregations the query uses.
Ask, purely documentary: say here that the unit is groups, that it is a
stand-in for a memory budget, and that a byte-based trigger is the intended
direction. Link a tracking issue that records what it needs — an
estimated-bytes-per-group method on AggregationFunction, which most functions
can answer statically (8 for SUM, log2m-derived for HLL, nominalEntries-derived
for Theta) at no per-row cost.
This costs nothing in this PR and is what stops the next person from
hardening the group-count model further, or from adding a second count-based
knob, in the belief that counting groups was the design rather than the
starting point. Together with the option rename (QueryOptionsUtils), the
numGroupsLimit backstop, the split of the three overloaded meanings, and the
shouldSpill() predicate, the byte-based version becomes an additive change
instead of a breaking one.
Also worth correcting here while you are in this javadoc: 'The first spill
version does not apply to ... group-trim modes' understates it. Trim is active
whenever the planner pushes a LIMIT into the aggregate and minGroupTrimSize > 0
(default 5000), so spill is silently off for a large share of real GROUP BY
queries. See the separate note on the eligibility check.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregationSpillManager.java:
##########
@@ -0,0 +1,428 @@
+/**
+ * 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.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.pinot.common.datablock.DataBlock;
+import org.apache.pinot.common.datablock.DataBlockUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SerializedDataBlock;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/// Manages hash-partitioned aggregation spill files for one operator. The
caller must finish reading before calling
+/// [#close()], which recursively removes the operator-scoped directory. This
class is not thread-safe.
+@SuppressWarnings("rawtypes")
+class AggregationSpillManager implements AutoCloseable {
Review Comment:
**SCOPE: #12080 is an SSE issue asking for two features; this PR delivers
MSE spill only, so `Close: #12080` over-claims twice**
Issue #12080 is 'Support parallel combine and disk spill for groupBy
execution'. Its text is explicitly about the SSE path — the reporter was
reading GroupByExecutor (org.apache.pinot.core.query.aggregation.groupby),
compared it to Druid's GroupByV2Engine, and asked for two things: (1) spill to
disk for the merging buffer (Druid SpillingGrouper) and (2) parallel combine
when merging sorted aggregation results (Druid ParallelCombiner, with the
combining-thread-tree diagram).
This PR implements neither of those literally: it adds spill to
MultistageGroupByExecutor / AggregateOperator (MSE only, pinot-query-runtime),
and does no parallel combine at all. Please downgrade `Close: #12080` to
'Partially addresses #12080' or link a narrower MSE-scoped issue, so the SSE
request and the parallel-combine request are not silently closed by an
unrelated change.
On whether this could later serve SSE, since that is the natural follow-up
question — three obstacles, increasing in difficulty:
1. Module direction (mechanical). This class is package-private in
org.apache.pinot.query.runtime.operator and its I/O is expressed in MseBlock /
RowHeapDataBlock / SerializedDataBlock. pinot-core does not depend on
pinot-query-runtime and cannot (the dependency runs the other way), so SSE
cannot reach it. The serialization core is already portable — DataBlock lives
in pinot-common — so sharing it means moving the class down and swapping the
MseBlock wrappers for a neutral row sink. If SSE reuse is plausibly on the
roadmap, factoring the file I/O and partitioning away from the MseBlock types
now is cheap; retrofitting after it ships is not.
2. Different attachment point. SSE has no single hash table per operator.
Per-segment GroupByExecutors feed GroupByCombineOperator, which merges into a
server-level IndexedTable — that is the structure that actually grows, so spill
would attach there, not to the per-segment executors.
3. Concurrency, which is the real blocker. ConcurrentIndexedTable is a
ConcurrentHashMap written by many segment-processing threads at once, whereas
this class documents 'This class is not thread-safe' and the MSE design leans
on AggregateOperator being single-threaded per worker: it can drain the table,
swap in a fresh one, and carry on. Under concurrent writers that needs either a
stop-the-world barrier across all combine threads or per-thread tables that
spill independently — which is in fact what Druid does, SpillingGrouper per
processing thread plus ParallelCombiner to merge them. So an SSE version is a
different design, not a port.
Worth noting too that SSE already answers 'too many groups' by trimming
(numGroupsLimit, minSegmentGroupTrimSize, minServerGroupTrimSize, TableResizer,
and ConcurrentIndexedTable's own _trimThreshold) and its group-by results above
the limit are documented as approximate. Spill in SSE would therefore be about
making approximate results exact, which is a larger product decision than the
memory-safety framing used here. None of that blocks this PR — it is an
argument for scoping the issue reference accurately rather than for widening
the change.
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java:
##########
@@ -187,6 +190,28 @@ public void testSelfStatsAreNotNegative() {
Assert.assertTrue(checked > 0, "expected some self stats to check, got: "
+ statsTree);
}
+ @Test
+ public void testMSEAggregationSpill() {
Review Comment:
**TESTS: unit coverage is genuinely strong; the end-to-end gap is
object/sketch intermediates and MV keys**
Credit where due — AggregateOperatorTest covers intermediate/final agg
modes, composite null keys, filters, hash collisions, oversized restore
partitions, group limits, and cleanup on all four terminal paths. That is more
than most feature PRs land with.
Remaining gaps, roughly in priority order:
1. No end-to-end coverage for DISTINCTCOUNT / HLL / Theta-sketch style
aggregations. These are the aggregations whose intermediate state is large —
i.e. the actual reason a user enables spill — and they are the ones most likely
to expose a serialization problem in the OBJECT column path.
testDistinctGroupBySpill is a bare DISTINCT with no agg functions, so it does
not cover this. AVG (AvgPair) is the only object intermediate exercised here.
2. No multi-value / array group key anywhere, despite deepHashCode going to
some trouble to handle arrays.
3. No test that the restored output is correct when a single spill file
contains records written across several spill rounds for the same key
(testGroupBySpillWritesMultipleBoundedRecords is about record count, not
cross-round merge of one key).
4. Nothing asserts the spill directory is gone after a normal successful
drain in the e2e path — only the unit tests check that.
Also note this test asserts spillCount/spilledRows/spilledBytes are positive
but never asserts an upper bound, so a regression that spills on every block
would still pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregationSpillManager.java:
##########
@@ -0,0 +1,428 @@
+/**
+ * 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.query.runtime.operator;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.FileVisitResult;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.pinot.common.datablock.DataBlock;
+import org.apache.pinot.common.datablock.DataBlockUtils;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.query.aggregation.function.AggregationFunction;
+import org.apache.pinot.query.runtime.blocks.MseBlock;
+import org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
+import org.apache.pinot.query.runtime.blocks.SerializedDataBlock;
+import org.apache.pinot.spi.query.QueryThreadContext;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/// Manages hash-partitioned aggregation spill files for one operator. The
caller must finish reading before calling
+/// [#close()], which recursively removes the operator-scoped directory. This
class is not thread-safe.
+@SuppressWarnings("rawtypes")
+class AggregationSpillManager implements AutoCloseable {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AggregationSpillManager.class);
+ private static final String SPILL_FILE_PREFIX = "partition-";
+ private static final String SPILL_FILE_SUFFIX = ".spill";
+ private static final String SPILL_SCOPE = "AggregationSpillManager#spill";
+ private static final String RESTORE_SCOPE =
"AggregationSpillManager#consumePartition";
+ private static final int MAX_BUFFERED_ROWS = 1024;
+ private static final int MAX_BUFFERED_PARTITIONS = 8;
+
+ private final int _numPartitions;
+ private final int _numGroupKeys;
+ private final DataSchema _spillSchema;
+ private final AggregationFunction[] _aggFunctions;
+ private final Path _spillDirectory;
+ private final Map<Integer, FileChannel> _spillWriters = new HashMap<>();
+ private final ByteBuffer _recordLengthBuffer =
ByteBuffer.allocate(Integer.BYTES);
+
+ AggregationSpillManager(int numPartitions, int numGroupKeys, DataSchema
spillSchema,
+ AggregationFunction[] aggFunctions) {
+ if (numPartitions <= 0 || numPartitions >
Server.MAX_MSE_AGGREGATION_SPILL_PARTITIONS) {
+ throw new IllegalArgumentException(
+ "Number of spill partitions must be between 1 and " +
Server.MAX_MSE_AGGREGATION_SPILL_PARTITIONS);
+ }
+ if (numGroupKeys < 0 || numGroupKeys > spillSchema.size()) {
+ throw new IllegalArgumentException("Invalid number of group keys: " +
numGroupKeys);
+ }
+ _numPartitions = numPartitions;
+ _numGroupKeys = numGroupKeys;
+ _spillSchema = spillSchema;
+ _aggFunctions = aggFunctions;
+ try {
+ _spillDirectory = Files.createTempDirectory("pinot-aggregation-spill-");
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to create aggregation spill
directory", e);
+ }
+ }
+
+ SpillResult spill(Iterator<Object[]> rows) {
+ List<Object[]>[] partitions = createPartitions();
+ boolean[] touched = new boolean[_numPartitions];
+ int[] touchedPartitions = new int[_numPartitions];
+ int numTouchedPartitions = 0;
+ int numRows = 0;
+ int numBufferedRows = 0;
+ long serializedBytes = 0;
+ int numRowsProcessed = 0;
+ int maxBufferedRows = MAX_BUFFERED_ROWS * Math.min(_numPartitions,
MAX_BUFFERED_PARTITIONS);
+ while (rows.hasNext()) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numRowsProcessed++,
SPILL_SCOPE);
+ Object[] row = rows.next();
+ int partitionId = getPartition(row);
+ List<Object[]> partition = partitions[partitionId];
+ if (partition == null) {
+ partition = new ArrayList<>();
+ partitions[partitionId] = partition;
+ }
+ if (!touched[partitionId]) {
+ touched[partitionId] = true;
+ touchedPartitions[numTouchedPartitions++] = partitionId;
+ }
+ partition.add(row);
+ numRows++;
+ numBufferedRows++;
+ if (partition.size() == MAX_BUFFERED_ROWS) {
+ serializedBytes += appendPartition(partitionId, partition);
+ partition.clear();
+ numBufferedRows -= MAX_BUFFERED_ROWS;
+ }
+ if (numBufferedRows >= maxBufferedRows) {
+ serializedBytes += flushPartitions(partitions, touched,
touchedPartitions, numTouchedPartitions);
+ numTouchedPartitions = 0;
+ numBufferedRows = 0;
+ }
+ }
+ if (numTouchedPartitions > 0) {
+ serializedBytes += flushPartitions(partitions, touched,
touchedPartitions, numTouchedPartitions);
+ }
+ return new SpillResult(numRows, serializedBytes);
+ }
+
+ boolean hasPartition(int partitionId) {
+ checkPartitionId(partitionId);
+ return Files.exists(getSpillFile(partitionId));
+ }
+
+ /// Consumes all records from a partition and deletes its file after the
attempt, including when reading or
+ /// processing fails.
+ void consumePartition(int partitionId, Consumer<MseBlock.Data> consumer) {
+ checkPartitionId(partitionId);
+ Path spillFile = getSpillFile(partitionId);
+ if (!Files.exists(spillFile)) {
+ return;
+ }
+
+ RuntimeException failure = null;
+ closeSpillWriter(partitionId);
+ try (DataInputStream input =
+ new DataInputStream(new
BufferedInputStream(Files.newInputStream(spillFile)))) {
+ long remainingBytes = Files.size(spillFile);
+ int numRecordsRead = 0;
+ while (remainingBytes > 0) {
+
QueryThreadContext.checkTerminationAndSampleUsagePeriodically(numRecordsRead++,
RESTORE_SCOPE);
+ if (remainingBytes < Integer.BYTES) {
+ throw new IOException("Truncated spill record length in: " +
spillFile);
+ }
+ int recordLength = input.readInt();
+ remainingBytes -= Integer.BYTES;
+ if (recordLength < 0 || recordLength > remainingBytes) {
+ throw new IOException("Invalid spill record length " + recordLength
+ " in: " + spillFile);
+ }
+ byte[] bytes = input.readNBytes(recordLength);
+ if (bytes.length != recordLength) {
+ throw new IOException("Truncated spill record in: " + spillFile);
+ }
+ remainingBytes -= recordLength;
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ DataBlock dataBlock = DataBlockUtils.readFrom(buffer);
+ if (buffer.hasRemaining()) {
+ throw new IOException("Trailing bytes in aggregation spill record: "
+ spillFile);
+ }
+ consumer.accept(new SerializedDataBlock(dataBlock));
+ }
+ } catch (IOException e) {
+ failure = new UncheckedIOException("Failed to read aggregation spill
partition: " + partitionId, e);
+ throw failure;
+ } catch (RuntimeException e) {
+ failure = e;
+ throw e;
+ } finally {
+ try {
+ deleteSpillFile(spillFile);
+ } catch (RuntimeException e) {
+ if (failure != null) {
+ failure.addSuppressed(e);
+ } else {
+ LOGGER.warn("Failed to delete consumed aggregation spill partition;
close will retry: {}", spillFile, e);
+ }
+ }
+ }
+ }
+
+ int getNumPartitions() {
+ return _numPartitions;
+ }
+
+ Path getSpillDirectory() {
+ return _spillDirectory;
+ }
+
+ @VisibleForTesting
+ int getNumOpenSpillWriters() {
+ return _spillWriters.size();
+ }
+
+ @Override
+ public void close() {
+ if (!Files.exists(_spillDirectory)) {
+ return;
+ }
+ RuntimeException failure = null;
+ try {
+ closeSpillWriters();
+ } catch (RuntimeException e) {
+ failure = e;
+ }
+ try {
+ Files.walkFileTree(_spillDirectory, new SimpleFileVisitor<>() {
+ @Override
+ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
+ throws IOException {
+ Files.delete(file);
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult postVisitDirectory(Path directory, IOException
exception)
+ throws IOException {
+ if (exception != null) {
+ throw exception;
+ }
+ Files.delete(directory);
+ return FileVisitResult.CONTINUE;
+ }
+ });
+ } catch (IOException e) {
+ RuntimeException cleanupFailure =
+ new UncheckedIOException("Failed to delete aggregation spill
directory", e);
+ if (failure != null) {
+ failure.addSuppressed(cleanupFailure);
+ } else {
+ failure = cleanupFailure;
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<Object[]>[] createPartitions() {
+ return new List[_numPartitions];
+ }
+
+ private long flushPartitions(List<Object[]>[] partitions, boolean[] touched,
int[] touchedPartitions,
+ int numTouchedPartitions) {
+ long serializedBytes = 0;
+ for (int i = 0; i < numTouchedPartitions; i++) {
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(i,
SPILL_SCOPE);
+ int partitionId = touchedPartitions[i];
+ List<Object[]> partition = partitions[partitionId];
+ if (!partition.isEmpty()) {
+ serializedBytes += appendPartition(partitionId, partition);
+ partition.clear();
+ }
+ touched[partitionId] = false;
+ }
+ return serializedBytes;
+ }
+
+ private int getPartition(Object[] row) {
Review Comment:
**Document the invariant this partitioner depends on: hash agreement across
the serialization round trip**
Correctness of the whole scheme rests on one unstated invariant: two rows
that the merge-side GroupIdGenerator will consider the same group must land in
the same partition here. Partitioning is computed pre-serialization on the
in-memory representation; grouping is decided post-serialization on the
deserialized representation.
I walked the cases and did not find a break (over-collision is harmless, and
deepHashCode is content-based for the array types where the group map is also
content-based), but this is the kind of invariant that a future change to
DataBlock type coercion, to BOOLEAN/TIMESTAMP handling, or to the group-key
generators can silently violate — and the symptom is duplicate group rows in
the output, not an exception.
Please state the invariant in the javadoc of getPartition/deepHashCode, and
add a test with a multi-value (array) group key end-to-end through spill.
AggregationSpillManagerTest.testDeepHashCode covers the hash function in
isolation but nothing covers an array-valued group key surviving the round trip.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -131,23 +157,58 @@ public AggregateOperator(OpChainExecutionContext context,
MultiStageOperator inp
_comparator = comparator;
_errorOnNumGroupsLimit = getErrorOnNumGroupsLimit(node.getNodeHint(),
context.getOpChainMetadata());
+ _numGroupsLimit =
MultistageGroupByExecutor.getNumGroupsLimit(_opChainMetadata, _nodeHint);
+ int spillThreshold = 0;
+ int spillPartitions = Server.DEFAULT_MSE_AGGREGATION_SPILL_PARTITIONS;
+ boolean spillEligible = !groupKeys.isEmpty() && !_leafReturnFinalResult
+ && groupTrimSize == Integer.MAX_VALUE &&
QueryOptionsUtils.isMSEAggregationSpillEnabled(_opChainMetadata);
+ if (spillEligible) {
+ Integer configuredSpillThreshold =
QueryOptionsUtils.getMSEAggregationSpillThreshold(_opChainMetadata);
+ if (configuredSpillThreshold != null) {
+ spillThreshold = configuredSpillThreshold;
+ Integer configuredSpillPartitions =
QueryOptionsUtils.getMSEAggregationSpillPartitions(_opChainMetadata);
+ if (configuredSpillPartitions != null) {
+ spillPartitions = configuredSpillPartitions;
+ }
+ }
+ }
+ _spillThreshold = spillThreshold;
+ _spillPartitions = spillPartitions;
Review Comment:
**EXTENSIBILITY 2/5: keep numGroupsLimit as a hard ceiling instead of
replacing it — it is also a real regression today**
Not asking to change the trigger unit in this PR. Asking that it not become
the ONLY memory bound, because that is both a regression now and the thing that
makes a later byte-based trigger awkward.
Today, without spill, in-memory groups are bounded by numGroupsLimit:
MultistageGroupByExecutor hands it to the GroupIdGenerator, which returns
INVALID_ID once reached and stops creating groups.
With spill on, forSpillInput passes Integer.MAX_VALUE as numGroupsLimit, so
the input executor is unbounded and the only bound left is _spillThreshold.
Nothing validates the threshold against numGroupsLimit —
getMSEAggregationSpillThreshold only checks positive. So `SET
mseAggregationSpillThreshold = 100000000` is accepted and that operator now
holds up to 100M groups where it previously stopped at numGroupsLimit. For SUM
that is merely large; for DISTINCTCOUNT or an HLL/Theta sketch it is an OOM on
a query that used to return truncated-but-safe results. The OOM-prevention
feature can cause an OOM through a knob whose safe value the user cannot
derive, because it depends on per-group state size.
Ask: keep numGroupsLimit enforced on the input executor as a backstop rather
than replacing it, so spill triggers at min(spillThreshold, numGroupsLimit).
Why this is the key door-opener: numGroupsLimit then keeps meaning 'the most
groups any one hash table may hold', independent of what unit the spill trigger
uses. A byte-based trigger can be added later as an additional, earlier trigger
without renegotiating the memory ceiling or changing what numGroupsLimit means.
If instead this PR establishes 'spillThreshold replaces numGroupsLimit', every
future change to the trigger is also a change to the operator's safety
semantics.
Second-best if you disagree: reject spillThreshold > numGroupsLimit at parse
time with a clear message. That fixes the regression but still leaves the two
concepts fused.
##########
pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java:
##########
@@ -549,6 +549,29 @@ public static Integer
getStreamingGroupByFlushThreshold(Map<String, String> quer
return
checkedParseIntNonNegative(QueryOptionKey.STREAMING_GROUP_BY_FLUSH_THRESHOLD,
value);
}
+ @Nullable
+ public static Integer getMSEAggregationSpillThreshold(Map<String, String>
queryOptions) {
Review Comment:
**EXTENSIBILITY 1/5: name the option for its unit now — a query option is
the one surface you cannot migrate later**
Not asking to change the trigger in this PR. A group count is a defensible
v1 and it matches numGroupsLimit. The ask is only that it not block a
byte-based trigger later, and the option name is where that gets decided,
because query options are the hardest surface to change once released.
mseAggregationSpillThreshold reads unit-free but means groups. If the
trigger later becomes bytes you get two bad choices: silently redefine the unit
— every existing SET breaks, and breaks quietly, since the same number means
something roughly 1000x different — or add mseAggregationSpillMemoryThreshold
and live with two knobs and undefined precedence.
Rename now to say what it counts: mseAggregationSpillMaxGroups (or
mseAggregationSpillGroupThreshold), leaving mseAggregationSpillMaxBytes free.
The two can then coexist with obvious semantics — spill when either is exceeded
— and the migration needs no deprecation cycle because nothing has shipped yet.
Mirror it in the CommonConstants javadoc.
While here: getMSEAggregationSpillPartitions is validated (1..64) but the
threshold only goes through checkedParseIntPositive, so it has no upper bound
at all. That is what lets it silently exceed numGroupsLimit — see EXTENSIBILITY
2/5 on AggregateOperator.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -268,12 +367,87 @@ private MseBlock.Eos consumeGroupBy() {
MseBlock block = _input.nextBlock();
while (block.isData()) {
_groupByExecutor.processBlock((MseBlock.Data) block);
+ if (_spillThreshold > 0 && _groupByExecutor.getNumGroups() >=
_spillThreshold) {
Review Comment:
**EXTENSIBILITY 4/5: put the spill decision behind one predicate — a byte
budget cannot be checked once per input block**
The trigger is inlined here as `getNumGroups() >= _spillThreshold`,
evaluated once per input block after processBlock returns. Two things are fused
in that one line: the POLICY (what counts as too big) and the CHECK FREQUENCY
(once per block).
A group count tolerates block-granularity checks — overshoot is bounded by
rows-per-block. A byte budget does not: one DISTINCTCOUNT or sketch block can
add tens of MB of per-group state that a group count would barely register, so
a byte-based trigger needs to be evaluated inside the block loop, not after it.
Changing that later means touching this operator, MultistageGroupByExecutor,
and the overshoot assumptions in restoreSpillPartition together.
Ask: move the decision to a single predicate owned by the executor,
something like
if (_spillEnabled && _groupByExecutor.shouldSpill()) {
spillCurrentGroups(); }
The executor already owns the hash table, the group count, and the result
holders, so it is the only place that could ever cheaply estimate retained
bytes. With the predicate in place, the policy can change and the check can
move inside processBlock without the operator noticing.
This is a small refactor now and the difference between a contained change
and a cross-cutting one later.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/AggregateOperator.java:
##########
@@ -131,23 +157,58 @@ public AggregateOperator(OpChainExecutionContext context,
MultiStageOperator inp
_comparator = comparator;
_errorOnNumGroupsLimit = getErrorOnNumGroupsLimit(node.getNodeHint(),
context.getOpChainMetadata());
+ _numGroupsLimit =
MultistageGroupByExecutor.getNumGroupsLimit(_opChainMetadata, _nodeHint);
+ int spillThreshold = 0;
+ int spillPartitions = Server.DEFAULT_MSE_AGGREGATION_SPILL_PARTITIONS;
+ boolean spillEligible = !groupKeys.isEmpty() && !_leafReturnFinalResult
+ && groupTrimSize == Integer.MAX_VALUE &&
QueryOptionsUtils.isMSEAggregationSpillEnabled(_opChainMetadata);
+ if (spillEligible) {
+ Integer configuredSpillThreshold =
QueryOptionsUtils.getMSEAggregationSpillThreshold(_opChainMetadata);
+ if (configuredSpillThreshold != null) {
+ spillThreshold = configuredSpillThreshold;
+ Integer configuredSpillPartitions =
QueryOptionsUtils.getMSEAggregationSpillPartitions(_opChainMetadata);
+ if (configuredSpillPartitions != null) {
+ spillPartitions = configuredSpillPartitions;
+ }
+ }
+ }
+ _spillThreshold = spillThreshold;
+ _spillPartitions = spillPartitions;
// Initialize the appropriate executor.
- AggregateNode.AggType aggType = node.getAggType();
// TODO: Allow leaf return final result for non-group-by queries
- boolean leafReturnFinalResult = node.isLeafReturnFinalResult();
if (groupKeys.isEmpty()) {
_aggregationExecutor =
- new MultistageAggregationExecutor(_aggFunctions, filterArgIds,
maxFilterArgId, aggType, _resultSchema);
+ new MultistageAggregationExecutor(_aggFunctions, _filterArgIds,
_maxFilterArgId, _aggType, _resultSchema);
_groupByExecutor = null;
} else {
- _groupByExecutor =
- new MultistageGroupByExecutor(getGroupKeyIds(groupKeys),
_aggFunctions, filterArgIds, maxFilterArgId, aggType,
- leafReturnFinalResult, _resultSchema,
context.getOpChainMetadata(), node.getNodeHint());
+ _groupByExecutor = newInputGroupByExecutor();
_aggregationExecutor = null;
}
}
+ private MultistageGroupByExecutor newInputGroupByExecutor() {
+ if (_spillThreshold > 0) {
+ return MultistageGroupByExecutor.forSpillInput(_groupKeyIds,
_aggFunctions, _filterArgIds, _maxFilterArgId,
+ _aggType, _resultSchema, _opChainMetadata, _nodeHint,
_spillThreshold);
+ }
+ return new MultistageGroupByExecutor(_groupKeyIds, _aggFunctions,
_filterArgIds, _maxFilterArgId, _aggType,
+ _leafReturnFinalResult, _resultSchema, _opChainMetadata, _nodeHint);
+ }
+
+ private MultistageGroupByExecutor newSpillMergeGroupByExecutor() {
+ AggregateNode.AggType mergeAggType =
+ _aggType.isOutputIntermediateFormat() ?
AggregateNode.AggType.INTERMEDIATE : AggregateNode.AggType.FINAL;
+ int[] spillGroupKeyIds = new int[_groupKeyIds.length];
+ for (int i = 0; i < spillGroupKeyIds.length; i++) {
+ spillGroupKeyIds[i] = i;
+ }
+ int expectedPartitionGroups =
Review Comment:
**EXTENSIBILITY 3/5: _spillThreshold means three different things — split
them before one of them changes unit**
_spillThreshold is currently load-bearing for three unrelated decisions:
1. spill trigger line 370 getNumGroups() >= _spillThreshold
2. restore failure limit line 418 executor.getNumGroups() >
_spillThreshold -> SERVER_RESOURCE_LIMIT_EXCEEDED
3. merge-table sizing hint line 206 expectedPartitionGroups =
ceil(threshold / partitions)
Only (1) is a natural fit for a byte budget. (2) is a correctness-visible
cap on a restored partition and (3) is an initial-capacity hint for a hash
table — both want group counts, and (3) specifically wants a count because that
is what maxInitialResultHolderCapacity takes.
So the day the trigger becomes bytes, all three have to be untangled at
once, in code that by then has shipped and has semantics users depend on. Doing
it now is nearly free: keep one user-facing knob, derive three named internal
fields from it, and let them diverge later.
_spillTriggerMaxGroups // (1) the knob; a byte budget can be added
alongside it
_restorePartitionMaxGroups // (2) correctness cap, stays a count
_mergeTableInitialCapacity // (3) sizing hint, stays a count
Separately on (3): expectedPartitionGroups assumes a uniform hash
distribution. That is fine for an initial-capacity hint, but the comment should
say it is a hint and not a bound, so nobody later mistakes it for one.
--
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]
