github-actions[bot] commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4348010167
##########
be/src/exec/operator/analytic_sink_operator.cpp:
##########
@@ -905,9 +1383,102 @@ void AnalyticSinkLocalState::_remove_unused_rows() {
DCHECK_GE(_partition_by_pose.end, 0);
}
+bool AnalyticSinkOperatorX::is_blockable(RuntimeState* state) const {
+ // The scheduler can classify the first task before local-state open
initializes its flag.
+ return _enable_spill_analytic ||
DataSinkOperatorX<AnalyticSinkLocalState>::is_blockable(state);
+}
+
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);
- return local_state._reserve_mem_size;
+ 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;
+ }
+ // In the worst case every input row ends a partition. Its boundary and
each reduction
+ // result remain in the batch while the input Block is collected, even
without a write.
+ size_t partition_result_bytes = incoming_rows * (sizeof(size_t) +
sizeof(int64_t));
+ for (size_t i = 0; i < local_state._spill_strategies.size(); ++i) {
+ switch (local_state._spill_strategies[i]) {
+ case WindowSpillStrategy::PARTITION_REDUCE: {
+ const auto& type = local_state._agg_functions[i]->data_type();
+ partition_result_bytes += type->have_maximum_size_of_value()
+ ? incoming_rows *
type->get_size_of_value_in_memory()
+ : incoming_bytes;
+ if (type->is_nullable()) {
+ partition_result_bytes += incoming_rows;
+ }
+ break;
+ }
+ case WindowSpillStrategy::PARTITION_CARDINALITY:
+ partition_result_bytes += incoming_rows * sizeof(int64_t);
+ break;
+ case WindowSpillStrategy::PEER_GROUP:
+ partition_result_bytes += incoming_rows * sizeof(int64_t);
+ break;
+ case WindowSpillStrategy::UNSUPPORTED:
+ DORIS_CHECK(false);
+ }
+ }
+ 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}}) +
+ partition_result_bytes;
+ }
+ // The spill path materializes comparison/aggregate columns even without a
disk write.
+ return incoming_bytes * 2 + partition_result_bytes;
Review Comment:
[P2] Reserve the materialized constant argument columns
Constant window arguments stay inline in Nereids, and
`_materialize_spill_columns()` expands a full column for each function. With
`batch_size=65535`, distinct BIGINT keys, and 32 `NTILE` calls using BIGINT
bucket literals 2147483648 through 2147483679 over `(PARTITION BY k ORDER BY
k)`, those columns use about 16 MiB while the per-partition parameter vectors
use another 16 MiB in the same sink call. This estimate is only about 18 MiB:
`partition_result_bytes` covers the vectors, while `incoming_bytes * 2` sees
only the narrow key Block. Under a tight memory limit collection can exceed
admission before another revoke opportunity. Include the argument
materializations separately, or keep scalar NTILE parameters unexpanded.
##########
be/src/exec/operator/analytic_source_operator.cpp:
##########
@@ -17,27 +17,502 @@
#include "exec/operator/analytic_source_operator.h"
+#include <algorithm>
#include <cstddef>
+#include <cstdint>
+#include <ranges>
#include <string>
+#include <utility>
#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
#include "exec/operator/operator.h"
+#include "exec/spill/spill_file.h"
+#include "exec/spill/spill_file_reader.h"
#include "exprs/vectorized_agg_fn.h"
namespace doris {
AnalyticLocalState::AnalyticLocalState(RuntimeState* state, OperatorXBase*
parent)
- : PipelineXLocalState<AnalyticSharedState>(state, parent) {}
+ : Base(state, parent) {}
Status AnalyticLocalState::init(RuntimeState* state, LocalStateInfo& info) {
- RETURN_IF_ERROR(PipelineXLocalState<AnalyticSharedState>::init(state,
info));
+ RETURN_IF_ERROR(Base::init(state, info));
SCOPED_TIMER(exec_time_counter());
SCOPED_TIMER(_init_timer);
_get_next_timer = ADD_TIMER(custom_profile(), "GetNextTime");
_filtered_rows_counter = ADD_COUNTER(custom_profile(), "FilteredRows",
TUnit::UNIT);
+ _partition_replay_timer = ADD_TIMER(custom_profile(),
"PartitionReplayTime");
return Status::OK();
}
+Status AnalyticLocalState::close(RuntimeState* state) {
+ if (_closed) {
+ return Status::OK();
+ }
+ _finish_spill_batch();
+ return Base::close(state);
+}
+
+void AnalyticLocalState::_finish_spill_batch() {
+ if (_batch_reader) {
+ auto st = _batch_reader->close();
+ LOG_IF(WARNING, !st.ok()) << "close analytic spill batch reader
failed: " << st;
+ _batch_reader.reset();
+ }
+ if (_peer_group_reader) {
+ auto st = _peer_group_reader->close();
+ LOG_IF(WARNING, !st.ok()) << "close analytic spill peer group reader
failed: " << st;
+ _peer_group_reader.reset();
+ }
+ COUNTER_UPDATE(_memory_used_counter, -_in_memory_batch_bytes);
+ _in_memory_batch_bytes = 0;
+ _current_batch.reset();
+ _replay_block.clear();
+ _replay_block_position = 0;
+ _peer_group_block.clear();
+}
+
+Status AnalyticLocalState::_open_spill_batch(RuntimeState* state,
+
std::shared_ptr<AnalyticSpillBatch> batch) {
+ DCHECK(batch != nullptr);
+ DCHECK_GT(batch->rows, 0);
+ DORIS_CHECK(!batch->partition_ends.empty());
+ DORIS_CHECK_EQ(batch->partition_ends.back(), batch->rows);
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->result_types.size());
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->peer_functions.size());
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->partition_results.size());
+ DORIS_CHECK_EQ(batch->strategies.size(),
batch->function_parameters.size());
+ DORIS_CHECK_EQ(batch->strategies.size(),
batch->change_to_nullable_flags.size());
+ _current_batch = std::move(batch);
+ _in_memory_block_index = 0;
+ _batch_output_position = 0;
+ _partition_index = 0;
+ _partition_start = 0;
+ _partition_end = _current_batch->partition_ends[0];
+ _peer_group_start = 0;
+ _peer_group_end = 0;
+ _peer_group_block_position = 0;
+ _in_memory_peer_group_index = 0;
+ _peer_group_file_eos = false;
+ for (const auto& block : _current_batch->blocks) {
+ _in_memory_batch_bytes += block.allocated_bytes();
+ }
+ COUNTER_UPDATE(_memory_used_counter, _in_memory_batch_bytes);
+
+ if (_current_batch->data_file) {
+ _batch_reader = _current_batch->data_file->create_reader(state,
operator_profile());
+ RETURN_IF_ERROR(_batch_reader->open());
+ }
+ if (_current_batch->peer_group_file) {
+ _peer_group_reader =
+ _current_batch->peer_group_file->create_reader(state,
operator_profile());
+ RETURN_IF_ERROR(_peer_group_reader->open());
+ }
+
+ _has_peer_group_function =
+ std::ranges::any_of(_current_batch->strategies,
[](WindowSpillStrategy strategy) {
+ return strategy == WindowSpillStrategy::PEER_GROUP;
+ });
+ if (_has_peer_group_function) {
+ RETURN_IF_ERROR(_next_peer_group_end(state));
+ }
+ return Status::OK();
+}
+
+Status AnalyticLocalState::_read_batch_block(RuntimeState* state, Block*
block, bool* batch_eos) {
+ RETURN_IF_CANCELLED(state);
+ if (_batch_reader) {
+ return _batch_reader->read(block, batch_eos);
+ }
+ if (_in_memory_block_index >= _current_batch->blocks.size()) {
+ *batch_eos = true;
+ block->clear();
+ return Status::OK();
+ }
+ auto& next_block = _current_batch->blocks[_in_memory_block_index++];
+ const auto block_bytes =
static_cast<int64_t>(next_block.allocated_bytes());
+ COUNTER_UPDATE(_memory_used_counter, -block_bytes);
+ _in_memory_batch_bytes -= block_bytes;
+ block->swap(std::move(next_block));
+ *batch_eos = false;
+ return Status::OK();
+}
+
+Status AnalyticLocalState::_next_replay_rows(RuntimeState* state, Block*
block, bool* batch_eos) {
+ while (_replay_block_position >= _replay_block.rows()) {
+ _replay_block.clear();
+ _replay_block_position = 0;
+ RETURN_IF_ERROR(_read_batch_block(state, &_replay_block, batch_eos));
+ if (*batch_eos) {
+ return Status::OK();
+ }
+ }
+ *batch_eos = false;
+ // Spilled Blocks are coalesced up to the spill buffer size, so a replayed
Block can be much
+ // larger than the batch size expected by downstream operators.
+ DCHECK_GT(state->batch_size(), 0);
+ const auto batch_size = static_cast<size_t>(state->batch_size());
+ const size_t rows = std::min(batch_size, _replay_block.rows() -
_replay_block_position);
+ if (_replay_block_position == 0 && rows == _replay_block.rows()) {
+ block->swap(_replay_block);
+ _replay_block.clear();
+ return Status::OK();
+ }
+ Block slice;
+ for (const auto& column : _replay_block) {
+ slice.insert({column.column->cut(_replay_block_position, rows),
column.type, column.name});
+ }
+ block->swap(slice);
+ _replay_block_position += rows;
+ return Status::OK();
+}
+
+size_t AnalyticLocalState::_spill_replay_reserve_bytes(RuntimeState* state)
const {
+ if (!_shared_state->spill_enabled.load()) {
+ return 0;
+ }
+ const auto spill_buffer_bytes =
static_cast<size_t>(state->spill_buffer_size_bytes());
+ const auto read_bytes =
[spill_buffer_bytes](AnalyticSpillBatch::RecordMemory memory,
+ bool opening) {
+ // The reader retains both a serialized window and PBlock;
deserialization also owns
+ // the decoded Block and may temporarily allocate another copy of its
columns.
+ return (opening ? std::max(spill_buffer_bytes, memory.serialized_bytes)
+ : memory.serialized_bytes) +
+ memory.serialized_bytes + 2 * memory.decoded_bytes;
+ };
+ if (!_current_batch || _batch_output_position >= _current_batch->rows) {
+ // The next call opens the batch at the head of the queue: it opens
the readers of its
+ // spilled files, reads the first peer group record and then the first
data record.
+ LockGuard lock(_shared_state->buffer_mutex);
+ if (_replay_admission_checked && !_admitted_batch) {
+ return 0;
+ }
+ if (_shared_state->spill_batches.empty() && !_admitted_batch) {
+ return 0;
+ }
+ const auto& next_batch =
+ _admitted_batch ? _admitted_batch :
_shared_state->spill_batches.front();
+ size_t reserve_bytes = 0;
+ if (next_batch->data_file) {
+ reserve_bytes += read_bytes(next_batch->data_record_memory, true);
+ }
+ if (next_batch->peer_group_file) {
+ reserve_bytes += read_bytes(next_batch->peer_record_memory, true);
+ }
+ return reserve_bytes;
+ }
+ size_t reserve_bytes = 0;
+ if (_batch_reader && _replay_block_position >= _replay_block.rows()) {
+ reserve_bytes += read_bytes(_current_batch->data_record_memory, false);
+ }
+ if (_next_slice_reads_peer_group_record(state)) {
+ reserve_bytes += read_bytes(_current_batch->peer_record_memory, false);
+ }
+ return reserve_bytes;
+}
+
+size_t AnalyticLocalState::_next_output_reserve_bytes(RuntimeState* state)
const {
+ const AnalyticSpillBatch* batch = _current_batch.get();
+ if (!batch || _batch_output_position >= batch->rows) {
+ LockGuard lock(_shared_state->buffer_mutex);
+ if (_replay_admission_checked && !_admitted_batch) {
+ return 0;
+ }
+ if (!_admitted_batch && _shared_state->spill_batches.empty()) {
+ return 0;
+ }
+ batch = (_admitted_batch ? _admitted_batch :
_shared_state->spill_batches.front()).get();
+ }
+ size_t rows = static_cast<size_t>(std::min<int64_t>(state->batch_size(),
batch->rows));
+ if (batch == _current_batch.get()) {
+ rows = static_cast<size_t>(std::min<int64_t>(rows, batch->rows -
_batch_output_position));
+ if (_replay_block_position < _replay_block.rows()) {
+ rows = std::min(rows, _replay_block.rows() -
_replay_block_position);
+ }
+ } else if (!batch->blocks.empty()) {
+ rows = std::min(rows, batch->blocks.front().rows());
+ }
+ size_t bytes = 0;
+ if (batch == _current_batch.get() && _replay_block_position <
_replay_block.rows() &&
Review Comment:
[P2] Reserve the first slice of each in-memory replay Block
This charges input cuts only when `_replay_block` was already loaded before
admission. Oversized in-memory Blocks are reachable: a revoked upstream sort
writes 4096-row spill records, and its single-cursor merger returns the whole
record even with `batch_size=1024`. With a selected 8 KiB payload per row, 64
MiB analytic/spill-buffer thresholds, force spill off, and
`minimum_operator_memory_required_kb=4096`, that roughly 32 MiB Block stays in
analytic memory. The first replay reserves only 4 MiB plus result columns, then
`_next_replay_rows()` copies at least 8 MiB for the first slice. Under memory
pressure this exceeds admission before another scheduling opportunity. Include
the upcoming in-memory Block's first cut in the estimate, for both new batches
and later Block transitions.
##########
be/src/exec/operator/analytic_source_operator.cpp:
##########
@@ -17,27 +17,502 @@
#include "exec/operator/analytic_source_operator.h"
+#include <algorithm>
#include <cstddef>
+#include <cstdint>
+#include <ranges>
#include <string>
+#include <utility>
#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
#include "exec/operator/operator.h"
+#include "exec/spill/spill_file.h"
+#include "exec/spill/spill_file_reader.h"
#include "exprs/vectorized_agg_fn.h"
namespace doris {
AnalyticLocalState::AnalyticLocalState(RuntimeState* state, OperatorXBase*
parent)
- : PipelineXLocalState<AnalyticSharedState>(state, parent) {}
+ : Base(state, parent) {}
Status AnalyticLocalState::init(RuntimeState* state, LocalStateInfo& info) {
- RETURN_IF_ERROR(PipelineXLocalState<AnalyticSharedState>::init(state,
info));
+ RETURN_IF_ERROR(Base::init(state, info));
SCOPED_TIMER(exec_time_counter());
SCOPED_TIMER(_init_timer);
_get_next_timer = ADD_TIMER(custom_profile(), "GetNextTime");
_filtered_rows_counter = ADD_COUNTER(custom_profile(), "FilteredRows",
TUnit::UNIT);
+ _partition_replay_timer = ADD_TIMER(custom_profile(),
"PartitionReplayTime");
return Status::OK();
}
+Status AnalyticLocalState::close(RuntimeState* state) {
+ if (_closed) {
+ return Status::OK();
+ }
+ _finish_spill_batch();
+ return Base::close(state);
+}
+
+void AnalyticLocalState::_finish_spill_batch() {
+ if (_batch_reader) {
+ auto st = _batch_reader->close();
+ LOG_IF(WARNING, !st.ok()) << "close analytic spill batch reader
failed: " << st;
+ _batch_reader.reset();
+ }
+ if (_peer_group_reader) {
+ auto st = _peer_group_reader->close();
+ LOG_IF(WARNING, !st.ok()) << "close analytic spill peer group reader
failed: " << st;
+ _peer_group_reader.reset();
+ }
+ COUNTER_UPDATE(_memory_used_counter, -_in_memory_batch_bytes);
+ _in_memory_batch_bytes = 0;
+ _current_batch.reset();
+ _replay_block.clear();
+ _replay_block_position = 0;
+ _peer_group_block.clear();
+}
+
+Status AnalyticLocalState::_open_spill_batch(RuntimeState* state,
+
std::shared_ptr<AnalyticSpillBatch> batch) {
+ DCHECK(batch != nullptr);
+ DCHECK_GT(batch->rows, 0);
+ DORIS_CHECK(!batch->partition_ends.empty());
+ DORIS_CHECK_EQ(batch->partition_ends.back(), batch->rows);
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->result_types.size());
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->peer_functions.size());
+ DORIS_CHECK_EQ(batch->strategies.size(), batch->partition_results.size());
+ DORIS_CHECK_EQ(batch->strategies.size(),
batch->function_parameters.size());
+ DORIS_CHECK_EQ(batch->strategies.size(),
batch->change_to_nullable_flags.size());
+ _current_batch = std::move(batch);
+ _in_memory_block_index = 0;
+ _batch_output_position = 0;
+ _partition_index = 0;
+ _partition_start = 0;
+ _partition_end = _current_batch->partition_ends[0];
+ _peer_group_start = 0;
+ _peer_group_end = 0;
+ _peer_group_block_position = 0;
+ _in_memory_peer_group_index = 0;
+ _peer_group_file_eos = false;
+ for (const auto& block : _current_batch->blocks) {
+ _in_memory_batch_bytes += block.allocated_bytes();
+ }
+ COUNTER_UPDATE(_memory_used_counter, _in_memory_batch_bytes);
+
+ if (_current_batch->data_file) {
+ _batch_reader = _current_batch->data_file->create_reader(state,
operator_profile());
+ RETURN_IF_ERROR(_batch_reader->open());
+ }
+ if (_current_batch->peer_group_file) {
+ _peer_group_reader =
+ _current_batch->peer_group_file->create_reader(state,
operator_profile());
+ RETURN_IF_ERROR(_peer_group_reader->open());
+ }
+
+ _has_peer_group_function =
+ std::ranges::any_of(_current_batch->strategies,
[](WindowSpillStrategy strategy) {
+ return strategy == WindowSpillStrategy::PEER_GROUP;
+ });
+ if (_has_peer_group_function) {
+ RETURN_IF_ERROR(_next_peer_group_end(state));
+ }
+ return Status::OK();
+}
+
+Status AnalyticLocalState::_read_batch_block(RuntimeState* state, Block*
block, bool* batch_eos) {
+ RETURN_IF_CANCELLED(state);
+ if (_batch_reader) {
+ return _batch_reader->read(block, batch_eos);
+ }
+ if (_in_memory_block_index >= _current_batch->blocks.size()) {
+ *batch_eos = true;
+ block->clear();
+ return Status::OK();
+ }
+ auto& next_block = _current_batch->blocks[_in_memory_block_index++];
+ const auto block_bytes =
static_cast<int64_t>(next_block.allocated_bytes());
+ COUNTER_UPDATE(_memory_used_counter, -block_bytes);
+ _in_memory_batch_bytes -= block_bytes;
+ block->swap(std::move(next_block));
+ *batch_eos = false;
+ return Status::OK();
+}
+
+Status AnalyticLocalState::_next_replay_rows(RuntimeState* state, Block*
block, bool* batch_eos) {
+ while (_replay_block_position >= _replay_block.rows()) {
+ _replay_block.clear();
+ _replay_block_position = 0;
+ RETURN_IF_ERROR(_read_batch_block(state, &_replay_block, batch_eos));
+ if (*batch_eos) {
+ return Status::OK();
+ }
+ }
+ *batch_eos = false;
+ // Spilled Blocks are coalesced up to the spill buffer size, so a replayed
Block can be much
+ // larger than the batch size expected by downstream operators.
+ DCHECK_GT(state->batch_size(), 0);
+ const auto batch_size = static_cast<size_t>(state->batch_size());
+ const size_t rows = std::min(batch_size, _replay_block.rows() -
_replay_block_position);
+ if (_replay_block_position == 0 && rows == _replay_block.rows()) {
+ block->swap(_replay_block);
+ _replay_block.clear();
+ return Status::OK();
+ }
+ Block slice;
+ for (const auto& column : _replay_block) {
+ slice.insert({column.column->cut(_replay_block_position, rows),
column.type, column.name});
+ }
+ block->swap(slice);
+ _replay_block_position += rows;
+ return Status::OK();
+}
+
+size_t AnalyticLocalState::_spill_replay_reserve_bytes(RuntimeState* state)
const {
+ if (!_shared_state->spill_enabled.load()) {
+ return 0;
+ }
+ const auto spill_buffer_bytes =
static_cast<size_t>(state->spill_buffer_size_bytes());
+ const auto read_bytes =
[spill_buffer_bytes](AnalyticSpillBatch::RecordMemory memory,
+ bool opening) {
+ // The reader retains both a serialized window and PBlock;
deserialization also owns
+ // the decoded Block and may temporarily allocate another copy of its
columns.
+ return (opening ? std::max(spill_buffer_bytes, memory.serialized_bytes)
Review Comment:
[P2] Size local reader admission from the actual record
Opening a local spill reader does not allocate `spill_buffer_size_bytes`:
`_read_footer()` reads its footer/offsets and `_block_slice()` grows the buffer
to the actual record because local coalescing is disabled. With the valid 256
MiB buffer setting and 1 MiB analytic sink limit, a roughly 2 MiB
full-partition batch spills, but this floor requests at least 256 MiB before
replay. After EOS the sink has no revocable store, so a workload group with 64
MiB available can pause/cancel its only consumer even though replay needs only
a few MiB. Reserve from the actual record/footer requirements and apply a
remote coalescing allowance only where it is used.
--
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]
