github-actions[bot] commented on code in PR #68500:
URL: https://github.com/apache/doris/pull/68500#discussion_r4346826364
##########
be/src/exec/operator/analytic_sink_operator.cpp:
##########
@@ -746,17 +1146,87 @@ Status AnalyticSinkOperatorX::prepare(RuntimeState*
state) {
alignment_of_next_state * alignment_of_next_state;
}
}
+
+ _prepare_spill(state);
return Status::OK();
}
+void AnalyticSinkOperatorX::_prepare_spill(RuntimeState* state) {
+ _window_spill_strategies.resize(_agg_functions_size,
WindowSpillStrategy::UNSUPPORTED);
+ _window_spill_peer_functions.resize(_agg_functions_size,
WindowSpillPeerFunction::NONE);
+ // The planner rewrites an OVER clause without ORDER BY to `RANGE BETWEEN
UNBOUNDED PRECEDING
+ // AND CURRENT ROW`. Every row is then a peer of the current row, so the
frame still covers
+ // the whole partition and the executor has to buffer the partition until
it ends.
+ const bool range_to_current_row_without_order =
+ _has_window && _has_range_window && !_has_window_start &&
_has_window_end &&
+ _window.window_end.type ==
TAnalyticWindowBoundaryType::CURRENT_ROW &&
+ _order_by_eq_expr_ctxs.empty();
+ const bool full_partition_frame = !_has_window || (!_has_window_start &&
!_has_window_end) ||
+ range_to_current_row_without_order;
+ for (size_t i = 0; i < _agg_functions_size; ++i) {
+ _window_spill_strategies[i] =
_agg_functions[i]->window_spill_strategy();
+ _window_spill_peer_functions[i] =
_agg_functions[i]->window_spill_peer_function();
+ }
+ const bool partition_dependent_function =
+ std::ranges::any_of(_window_spill_strategies,
[](WindowSpillStrategy strategy) {
+ return strategy == WindowSpillStrategy::PARTITION_CARDINALITY
||
+ strategy == WindowSpillStrategy::PEER_GROUP;
+ });
+ const bool streaming_frame =
+ !full_partition_frame && _has_window &&
+ (!_has_range_window ||
+ (!_has_window_start && _has_window_end &&
+ _window.window_end.type ==
TAnalyticWindowBoundaryType::CURRENT_ROW));
+ const bool streaming_window = streaming_frame &&
!partition_dependent_function;
+ bool spill_supported = state->enable_spill() && !streaming_window &&
_agg_functions_size > 0;
+ _window_spill_unsupported_reason = "Unsupported";
+ if (!state->enable_spill()) {
+ _window_spill_unsupported_reason = "Disabled";
+ } else if (streaming_window) {
+ _window_spill_unsupported_reason = "Streaming";
+ }
+ for (size_t i = 0; i < _agg_functions_size && spill_supported; ++i) {
+ const auto strategy = _window_spill_strategies[i];
+ bool frame_supported = false;
+ switch (strategy) {
+ case WindowSpillStrategy::PARTITION_REDUCE:
+ frame_supported = full_partition_frame;
+ break;
+ case WindowSpillStrategy::PARTITION_CARDINALITY:
+ frame_supported = _has_window && !_has_range_window &&
!_has_window_start &&
+ _has_window_end &&
+ _window.window_end.type ==
TAnalyticWindowBoundaryType::CURRENT_ROW;
+ break;
+ case WindowSpillStrategy::PEER_GROUP:
+ frame_supported = _has_window && _has_range_window &&
!_has_window_start &&
+ _has_window_end &&
+ _window.window_end.type ==
TAnalyticWindowBoundaryType::CURRENT_ROW;
+ break;
+ case WindowSpillStrategy::UNSUPPORTED:
+ frame_supported = false;
+ break;
+ }
+ if (!frame_supported) {
+ spill_supported = false;
+ _window_spill_unsupported_reason =
+ fmt::format("Unsupported: {}",
_agg_functions[i]->get_name());
+ }
+ }
+ _enable_spill_analytic = spill_supported;
+ _spillable = spill_supported;
+}
+
Status AnalyticSinkOperatorX::sink_impl(doris::RuntimeState* state, Block*
input_block, bool eos) {
auto& local_state = get_local_state(state);
SCOPED_TIMER(local_state.exec_time_counter());
COUNTER_UPDATE(local_state.rows_input_counter(),
(int64_t)input_block->rows());
- local_state._input_eos = eos;
- local_state._remove_unused_rows();
local_state._reserve_mem_size = 0;
SCOPED_PEAK_MEM(&local_state._reserve_mem_size);
+ if (local_state._spill_enabled) {
+ return local_state._sink_spill(state, input_block, eos);
Review Comment:
[P2] Route analytic spill I/O through the blocking scheduler
With an eligible full-partition window and forced spilling or a partition
exceeding the proactive limit, this branch performs synchronous spill writes;
source replay likewise opens and reads files synchronously. The analytic sink
inherits `is_blockable() == false`, and the source never sets a blocking
classification, so a pipeline with ordinary expressions and no other blocking
operator is dispatched to `HybridTaskScheduler`'s simple worker pool. Slow
spill storage or a saturated S3 upload queue can therefore occupy all of its
fixed workers and stall unrelated pipelines until I/O returns. Add spill-aware
blocking classification for both collection and replay, including the first
proactive spill, as the existing spill sort/aggregation paths do.
--
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]
