HappenLee commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4358224133
##########
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);
Review Comment:
**[P2] Estimate fixed-width nullable slices from their in-memory width**
`serialize_size_at()` measures row serialization, which omits the nested
value for NULL. For a NULL `Nullable(BIGINT)` it returns 1 byte, so this
reserves 2 bytes per row. However, `IColumn::cut()` calls
`ColumnNullable::insert_range_from()`, which copies both the null map and the
nested Int64 storage: at least 9 bytes per row, before capacity rounding.
This can under-admit a later slice of an already-loaded coalesced Block,
where no additional file-read reservation is requested. For example, 32
all-NULL Nullable BIGINT input columns and 65,535 output rows give about 4 MiB
of input-slice reservation versus about 18 MiB of copied data. Even adding a 4
MiB minimum and one COUNT result leaves the total reservation at only about 8.5
MiB.
A small targeted change could reuse the existing type-width API:
```cpp
for (const auto& column : *cut_block) {
if (!is_column_const(*column.column) &&
column.type->have_maximum_size_of_value()) {
bytes += 2 * 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);
}
}
}
```
`DataTypeNullable::get_size_of_value_in_memory()` already includes the
null-map byte plus the nested width, regardless of the row's nullness. Keeping
constants on the existing path avoids charging them as materialized full
columns. This also removes the per-row size loop for ordinary fixed-width
columns.
Please add a focused check for non-constant all-NULL and mixed-NULL
fixed-width columns, comparing the slice budget with actual cut allocations,
plus a constant-column case. This suggestion targets fixed-width types; the
variable-length/complex fallback still needs separate assessment. The
arithmetic and call chain above are from static review; the proposed change has
not been compiled or run.
--
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]
