mrhhsg commented on code in PR #67140:
URL: https://github.com/apache/doris/pull/67140#discussion_r3860654459
##########
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:
Fixed. The pass-through branch only latches `need_do_sort_limit = 1` (and
builds the heap) once the hash table has reached `limit`; a smaller table
leaves the state at `-1`, so the aggregating branch can still build the heap
after aggregation resumes. There is no TopN fixture for this operator in the
unit tests, so this one is covered by inspection only; the change is confined
to not assigning `0`.
##########
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:
Partially addressed. The arena part is fixed: pass-through serialization now
uses a per-block `Arena` instead of the operator-lifetime `_agg_arena_pool`, so
`collect`/`map`/`foreach`/... no longer accumulate dead bytes for the rest of
the query (that growth also kept `_memory_usage()` above the budget forever).
Heap owned by aggregate states (e.g. `BitmapValue` inside fixed slots) is not
counted — that is the existing accounting model shared with the blocking agg
sink and the previous spill-only limit, and is out of scope here; happy to
track it in a separate issue.
##########
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:
Fixed: the normal budget is computed first and low-memory mode applies
`min(1MB, budget)`. `test1` now sets `spill_streaming_agg_mem_limit = 512KB`
before entering low-memory mode and expects 512KB.
##########
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:
Fixed. `MockStreamingAggLocalState` gained `use_real_decision`; the new
`memory_limit_pass_through_and_recover` test drives `push()` with the
production `_should_not_do_pre_agg()` using a real `sum` evaluator: a lowered
query limit yields pass-through (no hash-table growth, 3 raw rows pulled, `eos
== false` despite `_limit = 2`), and restoring the limit resumes aggregation
(hash table grows to 4).
##########
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:
Fixed: `needForward = true` added, consistent with `enable_spill` /
`enable_force_spill`.
--
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]
