mrhhsg commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r3865089430
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -176,8 +195,17 @@ 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));
+ // A LIMIT pushed down to this stage bounds the hash set by itself and
must keep
+ // deduplicating: pass-through rows would otherwise consume the limit
with duplicates
+ // (see push()).
+ if (low_memory_mode() ||
+ (parent.limit() == -1 && memory_limit > 0 && _memory_usage() >
memory_limit)) {
Review Comment:
Fixed: the `parent.limit() == -1` gate is gone, so the memory budget latches
the operator into pass-through under a pushed-down LIMIT exactly like the
sibling operator. That is safe because `push()` already stops truncating once
`_stop_emplace_flag` is set (the global stage applies the limit again). The
comment now states that. `pushed_limit_with_memory_limit` covers both halves:
within the budget the limit still stops the child after `limit` distinct keys;
once the budget is exceeded the operator latches and appends the block raw
without truncation.
##########
be/src/exec/operator/streaming_aggregation_operator.cpp:
##########
@@ -1106,14 +1130,18 @@ Status StreamingAggOperatorX::pull(RuntimeState* state,
Block* block, bool* eos)
auto& local_state = get_local_state(state);
SCOPED_PEAK_MEM(&local_state._estimate_memory_usage);
if (!local_state._pre_aggregated_block->empty()) {
+ // Pass-through rows are neither aggregated nor deduplicated, so a
limit pushed down to
+ // this local stage must not count them: duplicates would consume the
allowance and stop
+ // the child before enough distinct keys reached the global stage,
which applies the
+ // limit again on the final result.
local_state._pre_aggregated_block->swap(*block);
} else {
RETURN_IF_ERROR(local_state._get_results_with_serialized_key(state,
block, eos));
local_state.make_nullable_output_key(block);
// dispose the having clause, should not be execute in prestreaming agg
RETURN_IF_ERROR(local_state.filter_block(local_state._conjuncts,
block));
+ local_state.reached_limit(block, eos);
Review Comment:
Fixed: the pass-through branch now calls
`update_process_rows(block->rows())` explicitly, so only the local-limit
truncation is skipped. `memory_limit_pass_through_and_recover` asserts
`process_rows` grows by the 3 pass-through rows while `eos` stays false.
##########
be/src/exec/operator/distinct_streaming_aggregation_operator.cpp:
##########
@@ -67,9 +72,23 @@ Status DistinctStreamingAggLocalState::open(RuntimeState*
state) {
RETURN_IF_ERROR(p._probe_expr_ctxs[i]->clone(state,
_probe_expr_ctxs[i]));
}
RETURN_IF_ERROR(_init_hash_method(_probe_expr_ctxs));
+ COUNTER_SET(_memory_use_limit,
static_cast<int64_t>(p._memory_limit(state)));
return Status::OK();
}
+size_t DistinctStreamingAggLocalState::_memory_usage() const {
+ size_t usage = _arena.size();
Review Comment:
Not changed in this PR. The `MethodSerialized` scratch (`Base::arena`,
`stored_keys`, `hash_values`) is reused per block and bounded by one block's
serialized keys; it does not grow with the query.
`StreamingAggLocalState::_memory_usage()` has the same omission at the merge
base, and the distinct helper deliberately uses the same accounting so the two
operators stay comparable. Folding `serialized_keys_size(false)` into both
helpers is a one-liner, but it changes the sibling's existing behaviour and
belongs in its own change; happy to do it as a follow-up.
--
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]
