[Enhancement] Eventize the spillable sort

be/src/common/config.h

@@ -1603,6 +1603,9 @@ CONF_mDouble(spill_max_dir_bytes_ratio, "0.8"); // 80%
 CONF_Int64(spill_read_buffer_min_bytes, "1048576");
 CONF_mInt64(mem_limited_chunk_queue_block_size, "8388608");
 
+// Route the spillable sort (ORDER BY / TOP-N) operator onto the pipeline event scheduler instead of the busy-poller.
+CONF_mBool(enable_spill_sort_events, "false");
+
 // The max number of threads for exec_state_report thread pool.
 CONF_mInt32(exec_state_report_max_threads, "2");
 // The max number of threads for priority_exec_state_report thread pool.

be/src/common/config_exec_flow_fwd.h

@@ -164,6 +164,9 @@ CONF_Int64(spill_read_buffer_min_bytes, "1048576");
 
 CONF_mInt64(mem_limited_chunk_queue_block_size, "8388608");
 
+// Route the spillable sort (ORDER BY / TOP-N) operator onto the pipeline event scheduler instead of the busy-poller.
+CONF_mBool(enable_spill_sort_events, "false");
+
 // The max number of threads for exec_state_report thread pool.
 CONF_mInt32(exec_state_report_max_threads, "2");
 

be/src/common/config_fwd_headers_manifest.json

@@ -174,6 +174,7 @@
         "spill_max_dir_bytes_ratio",
         "spill_read_buffer_min_bytes",
         "mem_limited_chunk_queue_block_size",
+        "enable_spill_sort_events",
         "exec_state_report_max_threads",
         "priority_exec_state_report_max_threads",
         "report_exec_rpc_request_retry_num",

be/src/exec/pipeline/sort/local_merge_sort_source_operator.cpp

@@ -24,6 +24,12 @@ Status LocalMergeSortSourceOperator::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(Operator::prepare(state));
     _sort_context->ref();
     _sort_context->attach_source_observer(state, observer());
+    // Subscribe to every partition spiller's source list so flush-all ("partition ready") and restore
+    // completions of any partition wake this INPUT_EMPTY sleeper. The pip-list subscription above only
+    // carries the non-spill "partition ready" axis; on the spill path that wakeup is emitted by the
+    // partition's spiller, which the source is otherwise not subscribed to. N subscriptions, one per spilled
+    // partition.
+    _sort_context->subscribe_source_to_spillers(state, observer());
     return Status::OK();
 }
 void LocalMergeSortSourceOperator::close(RuntimeState* state) {
@@ -46,6 +52,21 @@ Status LocalMergeSortSourceOperator::set_finished(RuntimeState* state) {
     return _sort_context->set_finished();
 }
 
+BlockReason LocalMergeSortSourceOperator::block_reason() const {
+    // Parked INPUT_EMPTY exactly when has_output() == false and not finished. The only non-terminal gates of
+    // has_output() (is_partition_sort_finished / is_partition_ready) are both woken by the partition
+    // spillers' flush-all / restore completions, which prepare()'s subscribe_source_to_spillers covers ->
+    // WAIT_RESTORE. Runnable or finished -> NONE. Both returns go through named<R, kCoveredWakeups>(), so a
+    // reason outside this source's coverage mask would not compile.
+    if (_is_finished || is_finished()) {
+        return named<BlockReason::NONE, kCoveredWakeups>();
+    }
+    if (has_output()) {
+        return named<BlockReason::NONE, kCoveredWakeups>();
+    }
+    return named<BlockReason::WAIT_RESTORE, kCoveredWakeups>();
+}
+
 bool LocalMergeSortSourceOperator::has_output() const {
     return _sort_context->is_partition_sort_finished() && !_sort_context->is_output_finished() &&
            _sort_context->is_partition_ready();

be/src/exec/pipeline/sort/local_merge_sort_source_operator.h

@@ -17,6 +17,7 @@
 #include <utility>
 
 #include "column/vectorized_fwd.h"
+#include "exec/pipeline/primitives/block_reason.h"
 #include "exec/pipeline/sort/sort_context.h"
 #include "exec/pipeline/source_operator.h"
 #include "exprs/sort_exec_exprs.h"
@@ -47,6 +48,14 @@ class LocalMergeSortSourceOperator final : public SourceOperator {
 
     bool is_finished() const override;
 
+    // The source sleeps INPUT_EMPTY on has_output(): both its non-terminal gates
+    // (is_partition_sort_finished / is_partition_ready) are spiller axes, woken by the partition spillers'
+    // source list that prepare()'s subscribe_source_to_spillers covers -> WAIT_RESTORE. covered_wakeups()
+    // declares the one reason that subscription covers, so a park on an uncovered reason is caught at park time.
+    BlockReason block_reason() const override;
+    static constexpr uint32_t kCoveredWakeups = block_reason_bit(BlockReason::WAIT_RESTORE);
+    uint32_t covered_wakeups() const override { return kCoveredWakeups; }
+
     StatusOr<ChunkPtr> pull_chunk(RuntimeState* state) override;
 
     void add_morsel(Morsel* morsel) {}
@@ -59,6 +68,8 @@ class LocalMergeSortSourceOperator final : public SourceOperator {
     SortContext* _sort_context;
 };
 
+static_assert(LocalMergeSortSourceOperator::kCoveredWakeups & block_reason_bit(BlockReason::WAIT_RESTORE));
+
 class LocalMergeSortSourceOperatorFactory final : public SourceOperatorFactory {
 public:
     LocalMergeSortSourceOperatorFactory(int32_t id, int32_t plan_node_id,

be/src/exec/pipeline/sort/local_parallel_merge_sort_source_operator.cpp

@@ -30,6 +30,11 @@ Status LocalParallelMergeSortSourceOperator::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(Operator::prepare(state));
     _sort_context->ref();
     _sort_context->attach_source_observer(state, observer());
+    // Subscribe to every partition spiller's source list so flush-all ("partition ready") and restore
+    // completions of any partition wake this INPUT_EMPTY sleeper. In the gathered branch all M source
+    // drivers share one context and each subscribes, so a partition emits M notifies; the observer's atomic
+    // event bitmask coalesces them. N subscriptions per driver, one per spilled partition.
+    _sort_context->subscribe_source_to_spillers(state, observer());
     _merger->bind_profile(_merge_parallel_id, _unique_metrics.get());
     _merger->attach_observer(state, observer());
     return Status::OK();
@@ -76,6 +81,33 @@ Status LocalParallelMergeSortSourceOperator::set_finished(RuntimeState* state) {
     return _sort_context->set_finished();
 }
 
+BlockReason LocalParallelMergeSortSourceOperator::block_reason() const {
+    // Mixed park model. The restore axis is woken by the partition spillers' source list (covered by
+    // prepare()'s subscribe_source_to_spillers) and has two sub-cases: still building -- a sink's flush-all
+    // completion advances is_partition_sort_finished, the callback running on the spiller IO thread -- or the
+    // steady state, where all sinks are done but a spilled partition is not yet restored, so a leaf chunk
+    // provider stays pending (chunks_sorter->has_output() == false) until that partition's restore completion
+    // fires on its spiller source list. Both are exactly !is_partition_sort_finished() || !is_partition_ready().
+    // The merge-path stage-coordination axis -- all sinks done, every partition restored, the merger still
+    // cascading stages -- is woken by the merger's own observable (attach_observer), not a spiller, and the
+    // enum has no WAIT_MERGE, so it stays generic NONE (the merger pip wakes it, as the non-spill parallel
+    // sort already runs). A false WAIT_RESTORE on a drained merge-park is harmless (still in the covered mask);
+    // a false NONE on a restore-park would silently leave the wakeup net unarmed, so name restore whenever
+    // any partition could still send the source a spiller wakeup.
+    // Every return passes through named<R, kCoveredWakeups>(): WAIT_RESTORE must sit in this source's
+    // coverage mask or it does not compile; NONE is the always-legal "not parked" answer.
+    if (_is_finished || is_finished()) {
+        return named<BlockReason::NONE, kCoveredWakeups>();
+    }
+    if (has_output()) {
+        return named<BlockReason::NONE, kCoveredWakeups>();
+    }
+    if (!_sort_context->is_partition_sort_finished() || !_sort_context->is_partition_ready()) {
+        return named<BlockReason::WAIT_RESTORE, kCoveredWakeups>();
+    }
+    return named<BlockReason::NONE, kCoveredWakeups>();
+}
+
 Status LocalParallelMergeSortSourceOperatorFactory::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(SourceOperatorFactory::prepare(state));
     _state = state;

be/src/exec/pipeline/sort/local_parallel_merge_sort_source_operator.h

@@ -19,6 +19,7 @@
 
 #include "column/vectorized_fwd.h"
 #include "compute_env/sorting/merge_path.h"
+#include "exec/pipeline/primitives/block_reason.h"
 #include "exec/pipeline/sort/sort_context.h"
 #include "exec/pipeline/source_operator.h"
 #include "exprs/sort_exec_exprs.h"
@@ -68,6 +69,18 @@ class LocalParallelMergeSortSourceOperator final : public SourceOperator {
 
     bool is_finished() const override;
 
+    // Mixed park model: the restore axis (still building, or a spilled partition not yet restored so a leaf
+    // chunk provider is pending) is woken by the partition spillers' source list that
+    // subscribe_source_to_spillers covers -> WAIT_RESTORE; the merge-path stage-coordination axis is woken by
+    // the merger's own observable, not a spiller (no WAIT_MERGE in the enum) -> NONE. covered_wakeups()
+    // declares only the spiller-covered reason; block_reason() names restore whenever a partition could still
+    // send the source a spiller wakeup. Note: this source is is_mutable(), so it parks LOCAL_WAITING (busy
+    // local-queue re-poll) and is promoted to INPUT_EMPTY -- where the park-time check reads this net -- only
+    // after the local-wait timeout; the net therefore arms on restore-parks that outlast the spin.
+    BlockReason block_reason() const override;
+    static constexpr uint32_t kCoveredWakeups = block_reason_bit(BlockReason::WAIT_RESTORE);
+    uint32_t covered_wakeups() const override { return kCoveredWakeups; }
+
     StatusOr<ChunkPtr> pull_chunk(RuntimeState* state) override;
 
     void add_morsel(Morsel* morsel) {}
@@ -81,6 +94,8 @@ class LocalParallelMergeSortSourceOperator final : public SourceOperator {
     bool _is_finished = false;
 };
 
+static_assert(LocalParallelMergeSortSourceOperator::kCoveredWakeups & block_reason_bit(BlockReason::WAIT_RESTORE));
+
 class LocalParallelMergeSortSourceOperatorFactory final : public SourceOperatorFactory {
 public:
     LocalParallelMergeSortSourceOperatorFactory(int32_t id, int32_t plan_node_id,

be/src/exec/pipeline/sort/sort_context.cpp

@@ -60,6 +60,19 @@ bool SortContext::is_partition_ready() const {
     });
 }
 
+void SortContext::subscribe_source_to_spillers(RuntimeState* state, PipelineObserver* observer) {
+    for (auto& sorter : _chunks_sorter_partitions) {
+        if (sorter->spiller() != nullptr) {
+            sorter->spiller()->observable().subscribe_source(state, observer);
+        }
+    }
+}
+
+// Cancel is intentionally minimal. Each partition spiller's cancel is already driven from the sink side
+// (SpillablePartitionSortSinkOperator::set_finishing/set_finished call _chunks_sorter->cancel() ->
+// _spiller->cancel() when cancelled), and any in-flight restore/flush IO holds its own query-lifetime pin
+// for the duration of its completion, so an explicit per-spiller cancel() here would be redundant (and
+// could only race the IO task's own guard); none is issued.
 void SortContext::cancel() {}
 
 StatusOr<ChunkPtr> SortContext::pull_chunk() {

be/src/exec/pipeline/sort/sort_context.h

@@ -72,6 +72,12 @@ class SortContext final : public ContextWithDependency {
     bool is_partition_ready() const;
     void cancel();
 
+    // Subscribe a source driver's observer to every non-empty partition spiller's source list, so that a
+    // flush-all ("partition ready") or restore completion of any partition wakes the source. The sort source
+    // merges all partitions, so unlike the single-spiller agg source it creates one subscription per spilled
+    // partition. Unconditional per spiller: the poller gate lives inside SpillEventObservable::subscribe_source.
+    void subscribe_source_to_spillers(RuntimeState* state, PipelineObserver* observer);
+
     StatusOr<ChunkPtr> pull_chunk();
 
     void set_runtime_filter_collector(RuntimeFilterHub* hub, int32_t plan_node_id,

be/src/exec/pipeline/sort/spillable_partition_sort_sink_operator.cpp

@@ -37,6 +37,12 @@ Status SpillablePartitionSortSinkOperator::prepare(RuntimeState* state) {
     }
     _peak_revocable_mem_bytes = _unique_metrics->AddHighWaterMarkCounter(
             "PeakRevocableMemoryBytes", TUnit::BYTES, RuntimeProfile::Counter::create_strategy(TUnit::BYTES));
+
+    // Subscribe this sink driver to the spiller's sink list so flush/channel completions wake the OUTPUT_FULL
+    // sleeper directly. Unconditional: the gate for the poller mode lives inside subscribe_sink (no-op when
+    // the event scheduler is disabled). observer() is valid here (assigned before prepare).
+    _chunks_sorter->spiller()->observable().subscribe_sink(state, observer());
+
     return Status::OK();
 }
 

be/src/exec/pipeline/sort/spillable_partition_sort_sink_operator.h

@@ -14,11 +14,20 @@
 
 #pragma once
 
+#include <cstdint>
+
+#include "common/config_exec_flow_fwd.h"
 #include "compute_env/spill/spiller.h"
+#include "exec/pipeline/primitives/block_reason.h"
+#include "exec/pipeline/primitives/spillable_flat_sink_mixin.h"
 #include "exec/pipeline/sort/partition_sort_sink_operator.h"
+#include "exec/pipeline/spill_process_channel.h"
 
 namespace starrocks::pipeline {
-class SpillablePartitionSortSinkOperator final : public PartitionSortSinkOperator {
+class SpillablePartitionSortSinkOperator final : public PartitionSortSinkOperator,
+                                                 public SpillableFlatSinkMixin<SpillablePartitionSortSinkOperator> {
+    friend class SpillableFlatSinkMixin<SpillablePartitionSortSinkOperator>;
+
 public:
     template <class... Args>
     SpillablePartitionSortSinkOperator(Args&&... args)
@@ -31,7 +40,14 @@ class SpillablePartitionSortSinkOperator final : public PartitionSortSinkOperato
 
     void close(RuntimeState* state) override;
 
-    bool need_input() const override { return !is_finished() && !_chunks_sorter->is_full(); }
+    // The sink sleeps OUTPUT_FULL on need_input(): a full writer (spiller is_full(), woken by the
+    // flush-completion defer on the sink list) or the spill-process channel still holding a task (woken by the
+    // channel handshake). The two reasons are read separately -- ChunksSorter::is_full() collapses them into a
+    // single bool, which the SpillableFlatSinkMixin predicate keeps distinct for block_reason().
+    bool need_input() const override { return spill_sink_need_input(); }
+    BlockReason block_reason() const override { return spill_sink_block_reason(); }
+    static constexpr uint32_t kCoveredWakeups = kSpillSinkCoveredWakeups;
+    uint32_t covered_wakeups() const override { return kCoveredWakeups; }
 
     bool is_finished() const override { return _is_finished || _sort_context->is_finished(); }
 
@@ -53,9 +69,20 @@ class SpillablePartitionSortSinkOperator final : public PartitionSortSinkOperato
     Status set_finished(RuntimeState* state) override;
 
 private:
+    // Raw spiller/spill-channel pair that the SpillableFlatSinkMixin reads to compute
+    // need_input()/block_reason(). These return the underlying spiller and channel directly
+    // (not the collapsed ChunksSorter::is_full(), which ORs spiller-full with channel-has-task into one bool)
+    // so the mixin reads WAIT_FLUSH (spiller->is_full()) and WAIT_CHANNEL (channel->has_task()) as distinct
+    // named reasons.
+    const std::shared_ptr<spill::Spiller>& _spiller() const { return _chunks_sorter->spiller(); }
+    SpillProcessChannelPtr _spill_channel() const { return _chunks_sorter->spill_channel(); }
+
     DECLARE_ONCE_DETECTOR(_set_finishing_once);
 };
 
+static_assert(SpillablePartitionSortSinkOperator::kCoveredWakeups & block_reason_bit(BlockReason::WAIT_FLUSH));
+static_assert(SpillablePartitionSortSinkOperator::kCoveredWakeups & block_reason_bit(BlockReason::WAIT_CHANNEL));
+
 class SpillablePartitionSortSinkOperatorFactory final : public PartitionSortSinkOperatorFactory {
 public:
     template <class... Args>
@@ -69,7 +96,10 @@ class SpillablePartitionSortSinkOperatorFactory final : public PartitionSortSink
     Status prepare(RuntimeState* state) override;
     void close(RuntimeState* state) override;
 
-    bool support_event_scheduler() const override { return false; }
+    // Event-scheduler opt-in, gated on enable_spill_sort_events (default false). This is the only false edge
+    // among the sort fragment's factories (the source and spill-process factories already report true), so
+    // the kill-switch demotes the whole sort fragment back to the poller without a rebuild.
+    bool support_event_scheduler() const override { return config::enable_spill_sort_events; }
 
 private:
     std::shared_ptr<spill::SpilledOptions> _spill_options;

be/test/CMakeLists.txt

@@ -56,6 +56,8 @@ set(EXEC_FILES
         ./compute_env/spill/spill_observable_test.cpp
         ./exec/pipeline/spill_process_channel_test.cpp
         ./exec/pipeline/block_reason_test.cpp
+        ./exec/pipeline/spillable_sort_source_lockstep_test.cpp
+        ./exec/pipeline/spillable_sort_observer_wakeup_test.cpp
         ./exec/pipeline/schedule/pipeline_timer_test.cpp
         ./exec/pipeline/pipeline_control_flow_test.cpp
         ./exec/pipeline/pipeline_driver_queue_test.cpp