github-actions[bot] commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4377993748
##########
be/src/exec/operator/analytic_source_operator.cpp:
##########
@@ -17,27 +17,531 @@
#include "exec/operator/analytic_source_operator.h"
+#include <algorithm>
#include <cstddef>
+#include <cstdint>
+#include <ranges>
#include <string>
+#include <utility>
+#include "common/config.h"
+#include "core/column/column_const.h"
#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/pod_array.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->ntile_bucket_counts.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 read_bytes = [state](AnalyticSpillBatch::RecordMemory memory,
+ const SpillFile& file) {
+ // The reader retains both a serialized window and PBlock;
deserialization also owns
+ // the decoded Block and may temporarily allocate another copy of its
columns. Local
+ // reads do not coalesce: a large configured spill buffer must not
block a small file.
+ // The footer offsets also outlive the serialized read window. Remote
reads may fetch
+ // a whole small part or coalesce adjacent records up to the
configured read window.
+ const size_t coalesce_bytes =
+ file.is_remote() ? static_cast<size_t>(std::max<int64_t>(
+ 0,
std::min(config::spill_s3_read_coalesce_bytes,
+
state->spill_buffer_size_bytes())))
+ : 0;
+ return std::max({memory.serialized_bytes, memory.footer_bytes,
coalesce_bytes}) +
+ memory.footer_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,
*next_batch->data_file);
+ }
+ if (next_batch->peer_group_file) {
+ reserve_bytes +=
+ read_bytes(next_batch->peer_record_memory,
*next_batch->peer_group_file);
+ }
+ 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,
*_current_batch->data_file);
+ }
+ if (_next_slice_reads_peer_group_record(state)) {
+ reserve_bytes +=
+ read_bytes(_current_batch->peer_record_memory,
*_current_batch->peer_group_file);
+ }
+ 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));
+ const Block* cut_block = nullptr;
+ size_t cut_start = 0;
+ 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()) {
+ cut_block = &_replay_block;
+ cut_start = _replay_block_position;
+ } else if (!batch->data_file && _in_memory_block_index <
batch->blocks.size()) {
+ // The next in-memory Block is moved into _replay_block in this
call. Its first
+ // slice must be admitted too; a previous sort may supply a Block
> batch_size.
+ cut_block = &batch->blocks[_in_memory_block_index];
+ }
+ } else if (!batch->blocks.empty()) {
+ cut_block = &batch->blocks.front();
+ }
+ if (cut_block) {
+ rows = std::min(rows, cut_block->rows() - cut_start);
+ }
+ // A byte-wide PaddedPODArray gives a conservative row capacity for the
fixed-width scalar
+ // buffers, including nullable/struct children. Padding can cross a
power-of-two boundary:
+ // 65535 UInt8 values require 131072 bytes, so even 2 * rows would
under-reserve the null map.
+ const size_t padded_rows =
PaddedPODArray<UInt8>::allocated_bytes_for_elements(rows);
+ size_t bytes = 0;
+ if (cut_block && (cut_start != 0 || rows != cut_block->rows())) {
+ // A retained Block is cut into the next output slice. Count the
actual rows rather
+ // than carrying the previous whole-record read peak into every small
slice.
+ for (const auto& column : *cut_block) {
+ if (!is_column_const(*column.column) &&
column.type->have_maximum_size_of_value()) {
+ // A nullable cut copies both its null map and nested
fixed-width values,
+ // including values whose serialized NULL representation omits
the nested bytes.
+ bytes += padded_rows *
column.type->get_size_of_value_in_memory();
+ } else {
+ for (size_t row = cut_start; row < cut_start + rows; ++row) {
+ bytes += 2 * column.column->serialize_size_at(row);
Review Comment:
[P2] Size array replay cuts from their nested storage
The complex-column fallback still omits storage for NULL array elements.
`DataTypeArray` has no fixed maximum width, so an array of 4096 NULL BIGINTs is
estimated as `2 * (4096 + 8)` bytes per row. But `IColumn::cut()` calls
`ColumnArray::insert_range_from()` and then
`ColumnNullable::insert_range_from()`, which copy every nested Int64 as well as
the null map.
This is reachable with ordinary 512-row input Blocks, a 64 MiB spill buffer,
a 16 MiB analytic sink limit, `minimum_operator_memory_required_kb=4096`, and a
stored nonconstant array payload selected alongside `count(*) over ()`. The
first Block proactively spills; two subsequent roughly 36 MiB allocated Blocks
coalesce into a 1024-row record. On the second 512-row replay slice, no
file-read reservation is added: this fallback plus the configured 4 MiB minimum
and COUNT result admits about 8 MiB, while the copied input alone needs at
least 18 MiB before capacity rounding. Under memory pressure the cut can exceed
admission before another revoke opportunity.
Please size complex cuts recursively from their actual nested ranges/null
maps and allocation capacities, with a retained ARRAY cut budget test. This is
the recursive ARRAY case explicitly left outside the fixed-width-only fix in
discussion 4358224133; that repaired scalar-width branch does not cover this
fallback.
--
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]
