github-actions[bot] commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4347489392
##########
be/src/exec/operator/analytic_sink_operator.cpp:
##########
@@ -906,8 +1384,70 @@ void AnalyticSinkLocalState::_remove_unused_rows() {
}
size_t AnalyticSinkOperatorX::get_reserve_mem_size(RuntimeState* state, bool
eos) {
+ return get_reserve_mem_size(state, eos, nullptr);
+}
+
+size_t AnalyticSinkOperatorX::get_reserve_mem_size(RuntimeState* state, bool
eos,
+ const Block* block) {
+ auto& local_state = get_local_state(state);
+ if (!local_state._spill_enabled) {
+ return local_state._reserve_mem_size;
+ }
+ const size_t incoming_bytes = block ? block->allocated_bytes() : 0;
+ const size_t incoming_rows = block ? block->rows() : 0;
+ const size_t staged_bytes =
+ local_state._batch_store ?
local_state._batch_store->pending_write_bytes() : 0;
+ const size_t pending_data_bytes =
+ local_state._batch_store ?
local_state._batch_store->pending_data_bytes() : 0;
+ const size_t pending_peer_bytes =
+ local_state._batch_store ?
local_state._batch_store->pending_peer_bytes() : 0;
+ const size_t spill_limit =
static_cast<size_t>(state->spill_analytic_sink_mem_limit_bytes());
+ const bool data_write =
+ (local_state._batch_store &&
local_state._batch_store->is_spilled() &&
+ (incoming_rows > 0 || pending_data_bytes > 0)) ||
+ ((incoming_rows > 0 || eos) && staged_bytes + incoming_bytes >=
spill_limit);
+ const bool first_forced_write = incoming_rows > 0 &&
state->enable_force_spill();
+ // An input Block may close the preceding peer group at its first row;
empty EOS closes
+ // the final group and can flush a sidecar without writing a data Block.
+ const size_t possible_peer_ends =
+ incoming_rows + (local_state._open_partition_rows > 0 ? 1 : 0);
+ const bool peer_write = local_state._has_peer_group_functions &&
+ ((pending_peer_bytes > 0 &&
local_state._batch_store &&
+
local_state._batch_store->has_spilled_peer_groups()) ||
+ pending_peer_bytes + possible_peer_ends *
sizeof(int64_t) >=
+
static_cast<size_t>(state->spill_buffer_size_bytes()));
+ if (eos && incoming_rows == 0 && !data_write && !peer_write) {
+ // An empty EOS merely publishes an in-memory batch; the previous
write peak is stale.
+ return 0;
+ }
+ if (data_write || first_forced_write || peer_write) {
+ // The writer can hold a PBlock and serialized/compressed bytes at the
same time.
+ // A wide input row is not splittable, so do not cap this at
spill_buffer_size_bytes.
+ // Small records only need a fixed metadata allowance, not a full
spill buffer.
+ return 3 * std::max({incoming_bytes, staged_bytes, size_t {64 *
1024}});
+ }
+ // The spill path materializes comparison/aggregate columns even without a
disk write.
+ return incoming_bytes * 2;
Review Comment:
[P2] Reserve the sink's per-partition result storage
The no-write estimate depends only on input bytes, but
`_finish_spill_partition()` retains one result per function for every partition
ending in the Block. For example, with `batch_size=60000`, unique sorted BIGINT
keys, one DOUBLE value column, and full-partition SUM/AVG/MIN/MAX/COUNT,
roughly 1 MiB of input reserves roughly 2 MiB here, while the five result
columns alone allocate roughly 2.5 MiB, before the copied head Block and
boundary vectors. Under a tight memory limit this can fail during collection
before the next admission/revocation opportunity. Include result/parameter
widths and possible partition boundaries in the sink estimate. This is separate
from the existing source result-column finding: these allocations happen before
batch publication, even without a spill write.
##########
be/test/exec/operator/analytic_sink_operator_test.cpp:
##########
@@ -84,6 +91,27 @@ class CancelAfterChecksRuntimeState : public
MockRuntimeState {
int64_t _cancel_after = -1;
};
+class ScopedSpillFileManager {
+public:
+ Status init() {
+ auto spill_data_dir =
std::make_unique<LocalSpillDataDir>("./ut_dir/analytic_spill_test",
+ 4 * 1024 *
1024);
+ RETURN_IF_ERROR(
+
io::global_local_filesystem()->create_directory(spill_data_dir->path(), false));
+ std::unordered_map<std::string, std::unique_ptr<SpillDataDir>>
data_map;
+ data_map.emplace("test", std::move(spill_data_dir));
+ ExecEnv::GetInstance()->_spill_file_mgr = new
SpillFileManager(std::move(data_map));
+ return ExecEnv::GetInstance()->spill_file_mgr()->init();
+ }
+
+ ~ScopedSpillFileManager() {
+ if (ExecEnv::GetInstance()->spill_file_mgr()) {
+ ExecEnv::GetInstance()->spill_file_mgr()->stop();
+ SAFE_DELETE(ExecEnv::GetInstance()->_spill_file_mgr);
Review Comment:
[P2] Keep the spill manager alive through fixture teardown
`ScopedSpillFileManager` is local to each test body, but `state` and
`shared_state` are fixture members. Tests such as
`SpillAdmissionReservesFirstWriteAndDropsEmptyEosPeak` and
`SpillAdmissionUsesActualOversizeRecord` leave a sealed spilled batch unread,
so this deletion frees its `SpillDataDir` before the batch is destroyed. Later
`SpillFile::~SpillFile()` calls `gc()` and dereferences that freed directory in
`_log_context()` and accounting, producing a heap use-after-free under ASAN.
Keep the manager alive until all fixture-owned stores, readers and batches have
been released, or explicitly close/reset those owners before this helper is
destroyed.
--
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]
