mrhhsg commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r3860653286
##########
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:
Fixed. Two changes: (1) the memory budget never switches this operator to
pass-through while a LIMIT is pushed down to it (`parent.limit() != -1`) — the
limit already bounds the hash set; (2) `push()` no longer truncates against
`_limit` once `_stop_emplace_flag` is set (low reduction / low-memory mode), so
raw rows cannot consume the allowance; the global stage applies the limit.
Covered by `DistinctStreamingAggOperatorTest.pushed_limit_keeps_deduplicating`
and `pass_through_does_not_consume_pushed_limit`. Note that the truncation
itself pre-dates this PR (the low-reduction path passed rows through before as
well).
##########
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:
Fixed. The budget decision is now per block (`pass_through` local), and only
the pre-existing switches (low reduction rate, `low_memory_mode()`) latch
`_stop_emplace_flag`. When the query limit is restored, the next block
deduplicates against the retained hash set. `refresh_memory_limit` now asserts
the lower-then-restore sequence.
##########
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:
Fixed: the per-task share is clamped to at least one byte when the query
limit is positive, so a known limit can no longer collapse to the "no cap"
sentinel. `budget_floor_and_fixed_limit` covers `(1, 2, 0)` and `(1, 2, 256MB)`.
##########
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:
Fixed. `pull()` only applies `reached_limit()` to the aggregated output
(`_get_results_with_serialized_key`); pass-through rows from
`_pre_aggregated_block` are neither aggregated nor deduplicated and must not
count toward a limit pushed to this local stage.
`StreamingAggOperatorTest.memory_limit_pass_through_and_recover` pushes `[2, 3,
3]` through with `_limit = 2` and asserts the pulled block keeps its 3 rows
with `eos == false`.
--
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]
