github-actions[bot] commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r3860482267
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -176,8 +195,13 @@ Status
DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
const uint32_t rows = (uint32_t)in_block->rows();
_distinct_row.clear();
- if (_parent->cast<DistinctStreamingAggOperatorX>()._is_streaming_preagg &&
low_memory_mode()) {
- _stop_emplace_flag = true;
+ auto& parent = _parent->cast<DistinctStreamingAggOperatorX>();
+ if (parent._is_streaming_preagg) {
+ const auto memory_limit = parent._memory_limit(state());
+ COUNTER_SET(_memory_use_limit, static_cast<int64_t>(memory_limit));
+ if (low_memory_mode() || (memory_limit > 0 && _memory_usage() >
memory_limit)) {
Review Comment:
[P1] Do not count raw pass-through duplicates toward a distinct limit. A
plain local `LIMIT N` is pushed to this lower distinct aggregation. Once this
cap condition switches it to pass-through, duplicates are emitted raw, but
`push()` still truncates and reaches `_limit` by raw row count and then stops
child input. With limit 2, after key `A` was emitted, a triggering `[A, B]`
block lets duplicate `A` consume the final allowance and prevents `B` from
reaching the global distinct, returning too few rows. Keep deduplicating while
the pushed distinct limit is active or remove/disable that local limit for
pass-through, and add this transition case.
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -289,9 +296,9 @@ bool StreamingAggLocalState::_should_not_do_pre_agg(size_t
rows) {
// But for fixed hash map, it never need to expand
auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
bool ret_flag = false;
- const auto spill_streaming_agg_mem_limit =
p._spill_streaming_agg_mem_limit;
- const bool used_too_much_memory =
- spill_streaming_agg_mem_limit > 0 && _memory_usage() >
spill_streaming_agg_mem_limit;
+ const auto memory_limit = p._memory_limit(state());
Review Comment:
[P1] Protect the regular pure-distinct fallback from the same early-EOS bug.
With `enable_distinct_streaming_aggregation=false`, the lower pure-distinct
stage is still streaming and uses this regular operator. When this dynamic cap
selects pass-through, it copies duplicate group keys verbatim, while generic
`reached_limit()` counts those raw rows against the pushed local distinct limit
and reports EOS. With limit 2, retained `A`, and input `[A, A, B]`, the
duplicates stop the child before `B`, so the global distinct returns too few
keys. Fix the pushed-limit/pass-through interaction in this operator too and
cover the disabled-specialized-operator plan.
##########
be/src/exec/operator/streaming_agg_memory_limit.h:
##########
@@ -0,0 +1,62 @@
+// 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.
+
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <cstdint>
+
+#include "exec/operator/streaming_agg_min_reduction.h"
+
+namespace doris {
+
+// Per-task memory budget of a streaming pre-aggregation (hash table + arenas).
+//
+// 1. The budget is one fifth of the current query memory limit, shared evenly
among the
+// `parallel_tasks` instances of the operator on this BE. The query limit
is read on every
+// call, so a limit lowered or restored by the workload group manager takes
effect at once.
+// 2. It never drops below `min_memory_limit`, which is the last cache tier of
the min-reduction
+// table doubled (the budget also counts the key/state arenas, not only the
bucket array), so a
+// small query limit does not disable pre-aggregation altogether. The floor
itself is capped by
+// the per-task share of the query limit, so the pre-aggregation alone can
never exceed it.
+// 3. `fixed_limit` is an explicit upper bound on top of that; 0 means "no
fixed bound". Callers
+// pass the session variable `spill_streaming_agg_mem_limit` when spilling
is enabled (the
+// downstream agg can spill, the pre-agg cannot, so it must stay small) and
0 otherwise. It is
+// applied last so that a user who sets it explicitly always gets what they
asked for.
+//
+// Returns 0 when neither limit is known, which callers treat as "no cap".
+inline size_t streaming_agg_memory_limit(int64_t query_memory_limit, int
parallel_tasks,
+ int64_t fixed_limit) {
+ if (query_memory_limit <= 0) {
+ return fixed_limit > 0 ? static_cast<size_t>(fixed_limit) : 0;
+ }
+
+ constexpr int64_t memory_limit_divisor = 5;
+ constexpr int64_t min_memory_limit =
+ 2LL * STREAMING_HT_MIN_REDUCTION[STREAMING_HT_MIN_REDUCTION_SIZE -
1].min_ht_mem;
+
+ const int64_t per_task_query_limit = query_memory_limit /
std::max(parallel_tasks, 1);
Review Comment:
[P2] Keep a positive share distinct from the no-cap sentinel. For
`query_memory_limit=1` and `parallel_tasks=2`, integer division makes
`per_task_query_limit` zero, and the helper returns zero even if `fixed_limit`
is positive. Both callers interpret zero as unlimited, so a known positive
query limit disables the new gate. Clamp a known-positive per-task share to at
least one byte or represent unknown/no-cap separately, and add `query_limit <
parallel_tasks` cases.
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -176,8 +195,13 @@ Status
DistinctStreamingAggLocalState::_distinct_pre_agg_with_serialized_key(
const uint32_t rows = (uint32_t)in_block->rows();
_distinct_row.clear();
- if (_parent->cast<DistinctStreamingAggOperatorX>()._is_streaming_preagg &&
low_memory_mode()) {
- _stop_emplace_flag = true;
+ auto& parent = _parent->cast<DistinctStreamingAggOperatorX>();
+ if (parent._is_streaming_preagg) {
+ const auto memory_limit = parent._memory_limit(state());
+ COUNTER_SET(_memory_use_limit, static_cast<int64_t>(memory_limit));
+ if (low_memory_mode() || (memory_limit > 0 && _memory_usage() >
memory_limit)) {
+ _stop_emplace_flag = true;
Review Comment:
[P2] Re-evaluate the cap stop after a restored query limit. This assignment
permanently latches `_stop_emplace_flag`, but workload-group query limits can
fall and later rise, and `_memory_limit()` is intentionally refreshed every
block. After one transient breach, every later block skips distinct emplacement
even when retained usage is below the restored cap, so the remainder of a long
query loses all local deduplication and can flood the exchange. Keep
cap-triggered pass-through separate from the permanent low-reduction/low-memory
stop, and add a lower-then-restore test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -3262,7 +3262,9 @@ public void setDetailShapePlanNodes(String
detailShapePlanNodes) {
@VarAttrDef.VarAttr(name = LOW_MEMORY_MODE_BUFFER_LIMIT, fuzzy = false)
public long lowMemoryModeBufferLimit = 33554432;
- // The memory limit of streaming agg when spilling is enabled
+ // The memory limit of streaming agg when spilling is enabled. It is
applied on top of the
Review Comment:
[P1] Forward this explicit cap to the planning master. The annotation lacks
`needForward=true`, so `getForwardVariables()` omits the setting. A follower
session with `enable_spill=true` and a 1 MiB cap that forwards an insert/select
sends `enable_spill` but reconstructs this value as the master's 256 MiB
default before `toThrift()`; both newly covered streaming operators then use
the wrong upper bound. Mark the variable for forwarding and add an
observer-to-master non-default-value test.
##########
be/test/exec/operator/streaming_agg_operator_test.cpp:
##########
@@ -144,9 +175,26 @@ TEST_F(StreamingAggOperatorTest, test1) {
EXPECT_EQ(local_state->_get_hash_table_size(), 3);
EXPECT_TRUE(op->need_more_input_data(state.get()));
+
EXPECT_EQ(local_state->custom_profile()->get_counter("MemoryUseLimit")->value(),
+ 100 * 1024 * 1024);
+ }
+
+ {
+ // With spilling enabled, spill_streaming_agg_mem_limit caps the
budget.
+ state->set_enable_spill(true);
+ op->_spill_streaming_agg_mem_limit = 16 * 1024 * 1024;
+ Block block {ColumnHelper::create_column_with_name<DataTypeInt64>({1,
2, 3}),
+ ColumnHelper::create_column_with_name<DataTypeInt64>({1,
100, 1000})};
+ auto st = op->push(state.get(), &block, false);
Review Comment:
[P1] Exercise the production cap decision in this test.
`MockStreamingAggLocalState::_should_not_do_pre_agg()` calls the base method
but discards its return value and then returns `should_not_do_pre_agg`, which
remains false in `test1`. Consequently this `push()` never follows the new
cap-triggered pass-through path; these additions only prove the profile value
and hash-table growth, and remain green if production enforcement is broken.
Use a real local state or a fixture that returns the base decision, then lower
the live/fixed cap below measured usage and assert pass-through/no new hash
entries (plus restoration).
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -289,9 +296,9 @@ bool StreamingAggLocalState::_should_not_do_pre_agg(size_t
rows) {
// But for fixed hash map, it never need to expand
auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
bool ret_flag = false;
- const auto spill_streaming_agg_mem_limit =
p._spill_streaming_agg_mem_limit;
- const bool used_too_much_memory =
- spill_streaming_agg_mem_limit > 0 && _memory_usage() >
spill_streaming_agg_mem_limit;
+ const auto memory_limit = p._memory_limit(state());
+ COUNTER_SET(_memory_use_limit, static_cast<int64_t>(memory_limit));
Review Comment:
[P2] Restore TopN pruning after a transient cap pass-through. If a pushed
TopN aggregate hits the cap before its hash table reaches `_sort_limit`, the
pass-through branch changes `need_do_sort_limit` from -1 to 0. After
workload-group limit restoration, aggregation resumes, but heap construction
only runs while state is -1, so state 0 can never activate
`_do_limit_filter()`; the local stage retains and shuffles all later groups.
Keep the state re-eligible or rebuild it on recovery, and test lower/restore
below the TopN threshold.
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -289,9 +296,9 @@ bool StreamingAggLocalState::_should_not_do_pre_agg(size_t
rows) {
// But for fixed hash map, it never need to expand
auto& p = Base::_parent->template cast<StreamingAggOperatorX>();
bool ret_flag = false;
- const auto spill_streaming_agg_mem_limit =
p._spill_streaming_agg_mem_limit;
- const bool used_too_much_memory =
- spill_streaming_agg_mem_limit > 0 && _memory_usage() >
spill_streaming_agg_mem_limit;
+ const auto memory_limit = p._memory_limit(state());
+ COUNTER_SET(_memory_use_limit, static_cast<int64_t>(memory_limit));
+ const bool used_too_much_memory = memory_limit > 0 && _memory_usage() >
memory_limit;
Review Comment:
[P1] Include all retained state in this memory cap. This comparison is not
an upper bound for supported aggregates. `_memory_usage()` counts hash buffers
and arenas but not heap owned by states stored in fixed slots; for example,
`bitmap_union`'s per-group `BitmapValue`/Roaring buffers grow invisibly. After
this condition selects pass-through, generic serializers such as `sum_foreach`
also allocate nested temporary states from the lifetime `_agg_arena_pool`, so
repeated blocks keep increasing retained memory until close. Account
state-owned heap and use a resettable pass-through arena, with tests for both
families.
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -996,6 +997,20 @@ Status StreamingAggOperatorX::init(const TPlanNode& tnode,
RuntimeState* state)
return Status::OK();
}
+size_t StreamingAggOperatorX::_memory_limit(RuntimeState* state) const {
+ constexpr size_t low_memory_mode_limit = 1024 * 1024;
+ if (_low_memory_mode.load(std::memory_order_relaxed)) {
+ return low_memory_mode_limit;
Review Comment:
[P1] Never let low-memory mode raise a tighter cap. This early return
replaces rather than constrains the live query/task and explicit spill bounds.
For `enable_spill=true` and `spill_streaming_agg_mem_limit=512 KiB`, entering
low-memory mode increases the allowed retention to 1 MiB; the live per-task
query share can also legitimately be below 1 MiB. Compute the normal effective
limit first and apply the low-memory value with `min`, then cover both sub-MiB
cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
