HappenLee commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4358544108
##########
be/src/exec/operator/analytic_source_operator.cpp:
##########
@@ -17,27 +17,520 @@
#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_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 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);
+ }
+ 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) {
+ for (size_t row = cut_start; row < cut_start + rows; ++row) {
+ bytes += 2 * column.column->serialize_size_at(row);
+ }
+ }
+ }
+ for (size_t i = 0; i < batch->result_types.size(); ++i) {
+ const auto& type = batch->result_types[i];
+ if (type->have_maximum_size_of_value()) {
+ bytes += rows * type->get_size_of_value_in_memory();
+ } else if (batch->partition_results[i]) {
+ // One partition result may be repeated for every output row.
+ bytes += rows * batch->partition_results[i]->byte_size();
+ }
+ if (batch->change_to_nullable_flags[i]) {
+ bytes += rows;
+ }
+ }
+ return bytes;
+}
+
+bool AnalyticLocalState::_next_slice_reads_peer_group_record(RuntimeState*
state) const {
+ if (!_peer_group_reader) {
+ return false;
+ }
+ // The next call outputs at most batch_size rows starting at
_batch_output_position and needs
+ // a new peer group record once one of them is not covered by the last
group end read so far.
+ DCHECK_GT(_peer_group_block.rows(), 0);
+ const auto& column =
+ assert_cast<const
ColumnInt64&>(*_peer_group_block.get_by_position(0).column);
+ const int64_t last_peer_group_end = column.get_data().back();
+ size_t rows = static_cast<size_t>(state->batch_size());
+ // A retained data Block ends this call even when batch_size extends into
the next Block.
+ // For example, 2 remaining rows and batch_size=8 cannot read a peer
record for row 3.
+ if (_replay_block_position < _replay_block.rows()) {
+ rows = std::min(rows, _replay_block.rows() - _replay_block_position);
+ }
+ const int64_t slice_end =
+ std::min<int64_t>(_batch_output_position + rows,
_current_batch->rows);
+ return last_peer_group_end < slice_end;
+}
+
+Status AnalyticLocalState::_next_peer_group_end(RuntimeState* state) {
+ RETURN_IF_CANCELLED(state);
+ _peer_group_start = _peer_group_end;
+ if (!_peer_group_reader) {
+ DORIS_CHECK_LT(_in_memory_peer_group_index,
_current_batch->peer_group_ends.size());
+ _peer_group_end =
_current_batch->peer_group_ends[_in_memory_peer_group_index++];
+ } else {
+ while (_peer_group_block_position >= _peer_group_block.rows()) {
+ _peer_group_block.clear();
+ RETURN_IF_ERROR(_peer_group_reader->read(&_peer_group_block,
&_peer_group_file_eos));
+ _peer_group_block_position = 0;
+ DORIS_CHECK(!_peer_group_file_eos || !_peer_group_block.empty());
+ }
+ const auto& column =
+ assert_cast<const
ColumnInt64&>(*_peer_group_block.get_by_position(0).column);
+ _peer_group_end = column.get_data()[_peer_group_block_position++];
+ }
+ DORIS_CHECK_GT(_peer_group_end, _peer_group_start);
+ DORIS_CHECK_LE(_peer_group_end, _current_batch->rows);
+ return Status::OK();
+}
+
+void AnalyticLocalState::_next_spill_partition() {
+ ++_partition_index;
+ DORIS_CHECK_LT(_partition_index, _current_batch->partition_ends.size());
+ _partition_start = _partition_end;
+ _partition_end = _current_batch->partition_ends[_partition_index];
+ DORIS_CHECK_GT(_partition_end, _partition_start);
+}
+
+Status AnalyticLocalState::_append_spill_results(RuntimeState* state, Block*
block) {
+ const auto rows = block->rows();
+ DCHECK_GT(rows, 0);
+ DCHECK_LE(_batch_output_position + static_cast<int64_t>(rows),
_current_batch->rows);
+ const auto function_count = _current_batch->strategies.size();
+ std::vector<MutableColumnPtr> results(function_count);
+ std::vector<IColumn*> computed_result_columns(function_count, nullptr);
+ _create_spill_result_columns(rows, results, computed_result_columns);
+
+ size_t offset = 0;
+ while (offset < rows) {
+ while (_batch_output_position >= _partition_end) {
+ _next_spill_partition();
+ }
+ const size_t segment_rows =
+ std::min<size_t>(rows - offset, _partition_end -
_batch_output_position);
+ _append_partition_results(segment_rows, computed_result_columns,
results);
+ if (_has_peer_group_function) {
+ RETURN_IF_ERROR(
+ _append_peer_group_results(state, segment_rows,
computed_result_columns));
+ }
+ _batch_output_position += segment_rows;
+ offset += segment_rows;
+ }
+ _insert_spill_result_columns(block, results);
+ return Status::OK();
+}
+
+void AnalyticLocalState::_create_spill_result_columns(
+ size_t rows, std::vector<MutableColumnPtr>& results,
+ std::vector<IColumn*>& computed_result_columns) const {
+ const auto function_count = _current_batch->strategies.size();
+ for (size_t i = 0; i < function_count; ++i) {
+ results[i] = _current_batch->result_types[i]->create_column();
+ results[i]->reserve(rows);
+ if (_current_batch->strategies[i] ==
WindowSpillStrategy::PARTITION_CARDINALITY ||
+ _current_batch->strategies[i] == WindowSpillStrategy::PEER_GROUP) {
+ if (_current_batch->result_types[i]->is_nullable()) {
+ auto* nullable =
assert_cast<ColumnNullable*>(results[i].get());
+ nullable->get_null_map_data().resize_fill(rows, 0);
+ computed_result_columns[i] = &nullable->get_nested_column();
+ } else {
+ computed_result_columns[i] = results[i].get();
+ }
+ }
+ }
+}
+
+void AnalyticLocalState::_append_partition_results(
+ size_t rows, const std::vector<IColumn*>& computed_result_columns,
+ std::vector<MutableColumnPtr>& results) const {
+ const auto function_count = _current_batch->strategies.size();
+ for (size_t i = 0; i < function_count; ++i) {
+ switch (_current_batch->strategies[i]) {
+ case WindowSpillStrategy::PARTITION_REDUCE:
+ DCHECK(_current_batch->partition_results[i]);
+
results[i]->insert_many_from(*_current_batch->partition_results[i],
_partition_index,
+ rows);
+ break;
+ case WindowSpillStrategy::PARTITION_CARDINALITY:
+ _append_ntile_result(i, rows, computed_result_columns[i]);
+ break;
+ case WindowSpillStrategy::PEER_GROUP:
+ break;
+ case WindowSpillStrategy::UNSUPPORTED:
+ DORIS_CHECK(false);
+ break;
+ }
+ }
+}
+
+void AnalyticLocalState::_append_ntile_result(size_t function_index, size_t
rows,
+ IColumn* result_column) const {
+ const int64_t bucket_num =
+
_current_batch->function_parameters[function_index][_partition_index];
+ DORIS_CHECK_GT(bucket_num, 0);
+ const int64_t partition_size = _partition_end - _partition_start;
+ const int64_t small_bucket_size = partition_size / bucket_num;
+ const int64_t big_bucket_num = partition_size % bucket_num;
+ const int64_t first_small_bucket_row = big_bucket_num * (small_bucket_size
+ 1);
+ const int64_t first_row_index = _batch_output_position - _partition_start;
+ auto& data = assert_cast<ColumnInt64&>(*result_column).get_data();
+ for (size_t row = 0; row < rows; ++row) {
+ const int64_t row_index = first_row_index + row;
+ const int64_t bucket =
+ row_index >= first_small_bucket_row
+ ? big_bucket_num + 1 +
+ (row_index - first_small_bucket_row) /
small_bucket_size
+ : row_index / (small_bucket_size + 1) + 1;
+ data.push_back(bucket);
+ }
+}
+
+Status AnalyticLocalState::_append_peer_group_results(
+ RuntimeState* state, size_t rows, const std::vector<IColumn*>&
computed_result_columns) {
+ const auto function_count = _current_batch->strategies.size();
+ for (size_t row = 0; row < rows; ++row) {
+ if ((row & 4095) == 0) {
+ RETURN_IF_CANCELLED(state);
+ }
+ const int64_t batch_row = _batch_output_position + row;
+ while (batch_row >= _peer_group_end) {
+ RETURN_IF_ERROR(_next_peer_group_end(state));
+ }
+ DCHECK_GE(_peer_group_start, _partition_start);
+ DCHECK_LE(_peer_group_end, _partition_end);
+ for (size_t i = 0; i < function_count; ++i) {
+ if (_current_batch->strategies[i] !=
WindowSpillStrategy::PEER_GROUP) {
+ continue;
+ }
+ assert_cast<ColumnFloat64&>(*computed_result_columns[i])
+ .get_data()
+ .push_back(_current_peer_group_result(i));
Review Comment:
**[P2] Restore peer-range result filling in spill replay**
This replaces the existing peer-range evaluation/fill pattern with per-row
evaluation. In the original path, `_get_next_for_unbounded_range()` passes a
peer/output-Block range to `_insert_result_info()`, and
`WindowFunctionPercentRank::insert_result_into_range()` /
`WindowFunctionCumeDist::insert_result_into_range()` compute the value once
before filling that range. Here `_current_peer_group_result(i)` is called
inside the row loop, repeating the function-kind branch and ratio calculation
for every row, followed by an individual `push_back()`.
Within a partition and peer group, `_peer_group_start`, `_peer_group_end`,
`_partition_start` and `_partition_end` are unchanged, so both results are
constant. With one million rows in one peer, output Blocks of 4096 rows, and
both functions, the old path performs about 490 formula evaluations across 245
Blocks; this path performs two million source-level evaluations. These are
code-level counts, not a measured query-time slowdown. The extra work also
affects eligible queries that enable spill but never actually write to disk.
Please process each intersection of the current peer and output slice as a
range, compute each peer-function result once, and reuse
`ColumnFloat64::insert_many_vals(value, count)`. `_append_spill_results()`
already bounds this call by the partition. Inside this helper the next range
length can be:
```cpp
const int64_t pos = _batch_output_position + offset;
// Advance peer metadata while pos >= _peer_group_end.
const size_t count =
std::min<size_t>(rows - offset, _peer_group_end - pos);
```
Fill that range and advance the local `offset`; keep
`_batch_output_position` advancement in the existing caller. This preserves the
current spill metadata and requires no new cross-call cache. Retain
cancellation checks and validate groups crossing output Blocks and sidecar
records. A focused benchmark with few large peers versus one-row peers would
quantify the CPU/query-time effect.
Original batch-fill implementations:
https://github.com/apache/doris/blob/46e0bd813888309a7faaba3544fe2a98959063f4/be/src/exprs/aggregate/aggregate_function_window.h#L256
--
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]
