be/src/exec/operator/operator.h

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#pragma once

#include <fmt/format.h>
#include <glog/logging.h>

#include <atomic>
#include <cstdint>
#include <functional>
#include <limits>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#include "common/be_mock_util.h"
#include "common/exception.h"
#include "common/logging.h"
#include "common/status.h"
#include "core/block/block.h"
#include "exec/exchange/local_exchanger.h"
#include "exec/exchange/vdata_stream_recvr.h"
#include "exec/operator/operator.h"
#include "exec/operator/spill_counters.h"
#include "exec/operator/spill_utils.h"
#include "exec/pipeline/dependency.h"
#include "runtime/memory/mem_tracker.h"
#include "runtime/query_context.h"
#include "runtime/runtime_profile.h"
#include "runtime/runtime_profile_counter_names.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "util/block_budget.h"

namespace doris {
class RowDescriptor;
class RuntimeState;
class TDataSink;
class AsyncResultWriter;
class ScoreRuntime;
class AnnTopNRuntime;
class ParsedPartitionBoundaries;
} // namespace doris

namespace doris {

class OperatorBase;
class OperatorXBase;
class DataSinkOperatorXBase;

using OperatorPtr = std::shared_ptr<OperatorXBase>;
using Operators = std::vector<OperatorPtr>;

using DataSinkOperatorPtr = std::shared_ptr<DataSinkOperatorXBase>;

// This struct is used only for initializing local state.
struct LocalStateInfo {
    RuntimeProfile* parent_profile = nullptr;
    const std::vector<TScanRangeParams>& scan_ranges;
    BasicSharedState* shared_state;
    const std::map<int, std::pair<std::shared_ptr<BasicSharedState>,
                                  std::vector<std::shared_ptr<Dependency>>>>& shared_state_map;
    const int task_idx;
};

// This struct is used only for initializing local sink state.
struct LocalSinkStateInfo {
    const int task_idx;
    RuntimeProfile* parent_profile = nullptr;
    const int sender_id;
    BasicSharedState* shared_state;
    const std::map<int, std::pair<std::shared_ptr<BasicSharedState>,
                                  std::vector<std::shared_ptr<Dependency>>>>& shared_state_map;
    const TDataSink& tsink;
};

class OperatorBase {
public:
    explicit OperatorBase() : _child(nullptr), _is_closed(false) {}
    explicit OperatorBase(bool is_serial_operator)
            : _child(nullptr), _is_closed(false), _is_serial_operator(is_serial_operator) {}
    virtual ~OperatorBase() = default;

    virtual bool is_sink() const { return false; }

    virtual bool is_source() const { return false; }

    [[nodiscard]] virtual Status init(const TDataSink& tsink) { return Status::OK(); }

    [[nodiscard]] virtual std::string get_name() const = 0;
    [[nodiscard]] virtual Status prepare(RuntimeState* state) = 0;
    [[nodiscard]] virtual Status terminate(RuntimeState* state) = 0;
    [[nodiscard]] virtual Status close(RuntimeState* state);
    [[nodiscard]] virtual int node_id() const = 0;
    [[nodiscard]] virtual int parallelism(RuntimeState* state) const {
        return _is_serial_operator ? 1 : state->query_parallel_instance_num();
    }

    [[nodiscard]] virtual Status set_child(OperatorPtr child) {
        if (_child && child != nullptr) {
            return Status::InternalError("Child is already set in node name={}", get_name());
        }
        _child = child;
        return Status::OK();
    }

    // Operators need to be executed serially. (e.g. finalized agg without key)
    [[nodiscard]] virtual bool is_serial_operator() const { return _is_serial_operator; }

    [[nodiscard]] bool is_closed() const { return _is_closed; }

    virtual size_t revocable_mem_size(RuntimeState* state) const { return 0; }

    virtual Status revoke_memory(RuntimeState* state) { return Status::OK(); }

    virtual bool is_hash_join_probe() const { return false; }

    /**
     * Pipeline task is blockable means it will be blocked in the next run. So we should put the
     * pipeline task into the blocking task scheduler.
     */
    virtual bool is_blockable(RuntimeState* state) const = 0;
    virtual void set_low_memory_mode(RuntimeState* state) {}

    OperatorPtr child() { return _child; }
    virtual Status reset(RuntimeState* state) {
        return Status::InternalError("Reset is not implemented in operator: {}", get_name());
    }

    /* -------------- Interfaces to determine the input data properties -------------- */
    /**
     * Return True if this operator relies on the bucket distribution (e.g. COLOCATE join, 1-phase AGG).
     * Data input to this kind of operators must have the same distribution with the table buckets.
     * It is also means `required_data_distribution` should be `BUCKET_HASH_SHUFFLE`.
     * @return
     */
    [[nodiscard]] virtual bool is_colocated_operator() const { return false; }
    /**
     * Return True if this operator relies on the bucket distribution or specific hash data distribution (e.g. SHUFFLED HASH join).
     * Data input to this kind of operators must be HASH distributed according to some rules.
     * All colocated operators are also shuffled operators.
     * It is also means `required_data_distribution` should be `BUCKET_HASH_SHUFFLE` or `HASH_SHUFFLE`.
     * @return
     */
    [[nodiscard]] virtual bool is_shuffled_operator() const { return false; }
    /**
     * For multiple children's operators, return true if this is a shuffled operator or this is followed by a shuffled operator (HASH JOIN and SET OPERATION).
     *
     * For single child's operators, return true if this operator is followed by a shuffled operator.
     * For example, in the plan fragment:
     *   `UNION` -> `SHUFFLED HASH JOIN`
     * The `SHUFFLED HASH JOIN` is a shuffled operator so the UNION operator is followed by a shuffled operator.
     */
    [[nodiscard]] virtual bool followed_by_shuffled_operator() const {
        return _followed_by_shuffled_operator;
    }
    /**
     * Update the operator properties according to the plan node.
     * This is called before `prepare`.
     */
    virtual void update_operator(const TPlanNode& tnode, bool followed_by_shuffled_operator,
                                 bool require_bucket_distribution) {
        _followed_by_shuffled_operator = followed_by_shuffled_operator;
        _require_bucket_distribution = require_bucket_distribution;
    }
    /**
     * Return the required data distribution of this operator.
     */
    [[nodiscard]] virtual DataDistribution required_data_distribution(
            RuntimeState* /*state*/) const;

protected:
    [[nodiscard]] bool child_breaks_local_key_distribution(RuntimeState* state) const;

    OperatorPtr _child = nullptr;

    bool _is_closed;
    bool _followed_by_shuffled_operator = false;
    bool _require_bucket_distribution = false;
    bool _is_serial_operator = false;
};

class PipelineXLocalStateBase {
public:
    PipelineXLocalStateBase(RuntimeState* state, OperatorXBase* parent);
    virtual ~PipelineXLocalStateBase() = default;

    template <class TARGET>
    TARGET& cast() {
        DCHECK(dynamic_cast<TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<TARGET&>(*this);
    }
    template <class TARGET>
    const TARGET& cast() const {
        DCHECK(dynamic_cast<TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<const TARGET&>(*this);
    }

    // Do initialization. This step should be executed only once and in bthread, so we can do some
    // lightweight or non-idempotent operations (e.g. init profile, clone expr ctx from operatorX)
    virtual Status init(RuntimeState* state, LocalStateInfo& info) = 0;
    // Make sure all resources are ready before execution. For example, remote tablets should be
    // loaded to local storage.
    // This is called by execution pthread and different from `Operator::prepare` which is called
    // by bthread.
    virtual Status prepare(RuntimeState* state) = 0;
    // Do initialization. This step can be executed multiple times, so we should make sure it is
    // idempotent (e.g. wait for runtime filters).
    virtual Status open(RuntimeState* state) = 0;
    virtual Status close(RuntimeState* state) = 0;
    virtual Status terminate(RuntimeState* state) = 0;

    // If use projection, we should clear `_origin_block`.
    void clear_origin_block();

    void reached_limit(Block* block, bool* eos);
    RuntimeProfile* operator_profile() { return _operator_profile.get(); }
    RuntimeProfile* common_profile() { return _common_profile.get(); }
    RuntimeProfile* custom_profile() { return _custom_profile.get(); }

    RuntimeProfile::Counter* exec_time_counter() { return _exec_timer; }
    RuntimeProfile::Counter* memory_used_counter() { return _memory_used_counter; }
    OperatorXBase* parent() { return _parent; }
    RuntimeState* state() { return _state; }
    [[nodiscard]] const BlockBudget& block_budget() const { return _budget; }
    VExprContextSPtrs& conjuncts() { return _conjuncts; }
    VExprContextSPtrs& projections() { return _projections; }
    [[nodiscard]] int64_t num_rows_returned() const { return _num_rows_returned; }
    void add_num_rows_returned(int64_t delta) { _num_rows_returned += delta; }
    void set_num_rows_returned(int64_t value) { _num_rows_returned = value; }
    void update_output_block_counters(const Block& block) {
        if (auto rows = block.rows()) {
            COUNTER_UPDATE(_rows_returned_counter, rows);
            COUNTER_UPDATE(_blocks_returned_counter, 1);
            auto block_bytes = static_cast<int64_t>(block.bytes());
            COUNTER_UPDATE(_output_block_bytes_counter, block_bytes);
            if (block_bytes > _max_output_block_bytes) {
                _max_output_block_bytes = block_bytes;
                COUNTER_SET(_max_output_block_bytes_counter, block_bytes);
            }
            if (block_bytes < _min_output_block_bytes) {
                _min_output_block_bytes = block_bytes;
                COUNTER_SET(_min_output_block_bytes_counter, block_bytes);
            }
        }
    }

    [[nodiscard]] virtual std::string debug_string(int indentation_level = 0) const = 0;
    [[nodiscard]] virtual bool is_blockable() const;

    virtual std::vector<Dependency*> dependencies() const { return {nullptr}; }

    // override in Scan
    virtual Dependency* finishdependency() { return nullptr; }
    //  override in Scan  MultiCastSink
    virtual std::vector<Dependency*> execution_dependencies() { return {}; }

    Status filter_block(const VExprContextSPtrs& expr_contexts, Block* block);

    int64_t& estimate_memory_usage() { return _estimate_memory_usage; }

    void reset_estimate_memory_usage() { _estimate_memory_usage = 0; }

    bool low_memory_mode() {
#ifdef BE_TEST
        return false;
#else
        return _state->low_memory_mode();
#endif
    }

protected:
    friend class OperatorXBase;
    template <typename LocalStateType>
    friend class ScanOperatorX;

    ObjectPool* _pool = nullptr;
    int64_t _num_rows_returned {0};
    int64_t _estimate_memory_usage {0};

    /*
    Each operator has its profile like this:
    XXXX_OPERATOR:
        CommonCounters:
            ...
        CustomCounters:
            ...
    */
    // Profile of this operator.
    // Should not modify this profile usually.
    std::unique_ptr<RuntimeProfile> _operator_profile;
    // CommonCounters of this operator.
    // CommonCounters are counters that will be used by all operators.
    std::unique_ptr<RuntimeProfile> _common_profile;
    // CustomCounters of this operator.
    // CustomCounters are counters that will be used by this operator only.
    std::unique_ptr<RuntimeProfile> _custom_profile;

    RuntimeProfile::Counter* _rows_returned_counter = nullptr;
    RuntimeProfile::Counter* _blocks_returned_counter = nullptr;
    RuntimeProfile::Counter* _output_block_bytes_counter = nullptr;
    RuntimeProfile::Counter* _max_output_block_bytes_counter = nullptr;
    RuntimeProfile::Counter* _min_output_block_bytes_counter = nullptr;
    int64_t _max_output_block_bytes = 0;
    int64_t _min_output_block_bytes = std::numeric_limits<int64_t>::max();
    RuntimeProfile::Counter* _wait_for_dependency_timer = nullptr;
    // Account for current memory and peak memory used by this node
    RuntimeProfile::HighWaterMarkCounter* _memory_used_counter = nullptr;
    RuntimeProfile::Counter* _projection_timer = nullptr;
    RuntimeProfile::Counter* _exec_timer = nullptr;
    RuntimeProfile::Counter* _init_timer = nullptr;
    RuntimeProfile::Counter* _open_timer = nullptr;
    RuntimeProfile::Counter* _close_timer = nullptr;

    OperatorXBase* _parent = nullptr;
    RuntimeState* _state = nullptr;
    // Execution-scoped row/byte budget derived from the session batch settings.
    const BlockBudget _budget;
    VExprContextSPtrs _conjuncts;
    VExprContextSPtrs _projections;
    std::shared_ptr<ScoreRuntime> _score_runtime;
    std::shared_ptr<segment_v2::AnnTopNRuntime> _ann_topn_runtime;
    // Used in common subexpression elimination to compute intermediate results.
    std::vector<VExprContextSPtrs> _intermediate_projections;

    bool _closed = false;
    std::atomic<bool> _terminated = false;
    Block _origin_block;
};

template <typename SharedStateArg = FakeSharedState>
class PipelineXLocalState : public PipelineXLocalStateBase {
public:
    using SharedStateType = SharedStateArg;
    PipelineXLocalState(RuntimeState* state, OperatorXBase* parent)
            : PipelineXLocalStateBase(state, parent) {}
    ~PipelineXLocalState() override = default;

    Status init(RuntimeState* state, LocalStateInfo& info) override;
    Status prepare(RuntimeState* state) override { return Status::OK(); }
    Status open(RuntimeState* state) override;

    virtual std::string name_suffix() const;

    Status close(RuntimeState* state) override;
    Status terminate(RuntimeState* state) override;

    [[nodiscard]] std::string debug_string(int indentation_level = 0) const override;

    std::vector<Dependency*> dependencies() const override {
        return _dependency ? std::vector<Dependency*> {_dependency} : std::vector<Dependency*> {};
    }

    virtual bool must_set_shared_state() const {
        return !std::is_same_v<SharedStateArg, FakeSharedState>;
    }

protected:
    Dependency* _dependency = nullptr;
    SharedStateArg* _shared_state = nullptr;
};

template <typename SharedStateArg>
class PipelineXSpillLocalState : public PipelineXLocalState<SharedStateArg> {
public:
    using Base = PipelineXLocalState<SharedStateArg>;
    PipelineXSpillLocalState(RuntimeState* state, OperatorXBase* parent)
            : PipelineXLocalState<SharedStateArg>(state, parent) {}
    ~PipelineXSpillLocalState() override = default;

    Status init(RuntimeState* state, LocalStateInfo& info) override {
        RETURN_IF_ERROR(PipelineXLocalState<SharedStateArg>::init(state, info));

        init_spill_read_counters();

        return Status::OK();
    }

    void init_spill_write_counters() {
        _write_counters.init(Base::custom_profile());

        // Source-only extra write counters
        _spill_write_file_total_size = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_WRITE_FILE_BYTES, TUnit::BYTES, 1);
        _spill_file_total_count = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_WRITE_FILE_TOTAL_COUNT, TUnit::UNIT, 1);
    }

    void init_spill_read_counters() {
        _spill_total_timer =
                ADD_TIMER_WITH_LEVEL(Base::custom_profile(), profile::SPILL_TOTAL_TIME, 1);

        _read_counters.init(Base::custom_profile());

        _spill_file_current_size = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_WRITE_FILE_CURRENT_BYTES, TUnit::BYTES, 1);
    }

    // Total time of spill, including spill task scheduling time,
    // serialize block time, write disk file time,
    // and read disk file time, deserialize block time etc.
    RuntimeProfile::Counter* _spill_total_timer = nullptr;

    // Shared spill write counters
    SpillWriteCounters _write_counters;
    // Backward-compatible aliases for commonly accessed write counters
    RuntimeProfile::Counter*& _spill_write_file_timer = _write_counters.spill_write_file_timer;
    RuntimeProfile::Counter*& _spill_write_serialize_block_timer =
            _write_counters.spill_write_serialize_block_timer;
    RuntimeProfile::Counter*& _spill_write_block_count = _write_counters.spill_write_block_count;
    RuntimeProfile::Counter*& _spill_write_block_data_size =
            _write_counters.spill_write_block_data_size;
    RuntimeProfile::Counter*& _spill_write_rows_count = _write_counters.spill_write_rows_count;

    // Source-only write counters (not in SpillWriteCounters)
    // Total bytes of spill data written to disk file(after serialized)
    RuntimeProfile::Counter* _spill_write_file_total_size = nullptr;
    RuntimeProfile::Counter* _spill_file_total_count = nullptr;
    // Current spilled file size
    RuntimeProfile::Counter* _spill_file_current_size = nullptr;

    // Shared spill read counters
    SpillReadCounters _read_counters;
    // Backward-compatible aliases for commonly accessed read counters
    RuntimeProfile::Counter*& _spill_read_file_time = _read_counters.spill_read_file_time;
    RuntimeProfile::Counter*& _spill_read_deserialize_block_timer =
            _read_counters.spill_read_deserialize_block_timer;
    RuntimeProfile::Counter*& _spill_read_block_count = _read_counters.spill_read_block_count;
    RuntimeProfile::Counter*& _spill_read_block_data_size =
            _read_counters.spill_read_block_data_size;
    RuntimeProfile::Counter*& _spill_read_file_size = _read_counters.spill_read_file_size;
    RuntimeProfile::Counter*& _spill_read_rows_count = _read_counters.spill_read_rows_count;
    RuntimeProfile::Counter*& _spill_read_file_count = _read_counters.spill_read_file_count;
};

class DataSinkOperatorXBase;

class PipelineXSinkLocalStateBase {
public:
    PipelineXSinkLocalStateBase(DataSinkOperatorXBase* parent_, RuntimeState* state_);
    virtual ~PipelineXSinkLocalStateBase() = default;

    // Do initialization. This step should be executed only once and in bthread, so we can do some
    // lightweight or non-idempotent operations (e.g. init profile, clone expr ctx from operatorX)
    virtual Status init(RuntimeState* state, LocalSinkStateInfo& info) = 0;

    virtual Status prepare(RuntimeState* state) = 0;
    // Do initialization. This step can be executed multiple times, so we should make sure it is
    // idempotent (e.g. wait for runtime filters).
    virtual Status open(RuntimeState* state) = 0;
    virtual Status terminate(RuntimeState* state) = 0;
    virtual Status close(RuntimeState* state, Status exec_status) = 0;
    [[nodiscard]] virtual bool is_finished() const { return false; }
    [[nodiscard]] virtual bool is_blockable() const { return false; }

    [[nodiscard]] virtual std::string debug_string(int indentation_level) const = 0;

    template <class TARGET>
    TARGET& cast() {
        DCHECK(dynamic_cast<TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<TARGET&>(*this);
    }
    template <class TARGET>
    const TARGET& cast() const {
        DCHECK(dynamic_cast<const TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<const TARGET&>(*this);
    }

    DataSinkOperatorXBase* parent() { return _parent; }
    RuntimeState* state() { return _state; }
    RuntimeProfile* operator_profile() { return _operator_profile; }
    RuntimeProfile* common_profile() { return _common_profile; }
    RuntimeProfile* custom_profile() { return _custom_profile; }

    [[nodiscard]] RuntimeProfile* faker_runtime_profile() const {
        return _faker_runtime_profile.get();
    }

    RuntimeProfile::Counter* rows_input_counter() { return _rows_input_counter; }
    RuntimeProfile::Counter* exec_time_counter() { return _exec_timer; }
    RuntimeProfile::Counter* memory_used_counter() { return _memory_used_counter; }

    virtual std::vector<Dependency*> dependencies() const { return {nullptr}; }

    // override in exchange sink , AsyncWriterSink
    virtual Dependency* finishdependency() { return nullptr; }

    bool low_memory_mode() { return _state->low_memory_mode(); }

protected:
    DataSinkOperatorXBase* _parent = nullptr;
    RuntimeState* _state = nullptr;
    RuntimeProfile* _operator_profile = nullptr;
    RuntimeProfile* _common_profile = nullptr;
    RuntimeProfile* _custom_profile = nullptr;
    // Set to true after close() has been called. subclasses should check and set this in
    // close().
    bool _closed = false;
    bool _terminated = false;
    //NOTICE: now add a faker profile, because sometimes the profile record is useless
    //so we want remove some counters and timers, eg: in join node, if it's broadcast_join
    //and shared hash table, some counter/timer about build hash table is useless,
    //so we could add those counter/timer in faker profile, and those will not display in web profile.
    std::unique_ptr<RuntimeProfile> _faker_runtime_profile =
            std::make_unique<RuntimeProfile>(profile::FAKER_PROFILE);

    RuntimeProfile::Counter* _rows_input_counter = nullptr;
    RuntimeProfile::Counter* _init_timer = nullptr;
    RuntimeProfile::Counter* _open_timer = nullptr;
    RuntimeProfile::Counter* _close_timer = nullptr;
    RuntimeProfile::Counter* _wait_for_dependency_timer = nullptr;
    RuntimeProfile::Counter* _wait_for_finish_dependency_timer = nullptr;
    RuntimeProfile::Counter* _exec_timer = nullptr;
    RuntimeProfile::HighWaterMarkCounter* _memory_used_counter = nullptr;
};

template <typename SharedStateArg = FakeSharedState>
class PipelineXSinkLocalState : public PipelineXSinkLocalStateBase {
public:
    using SharedStateType = SharedStateArg;
    PipelineXSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
            : PipelineXSinkLocalStateBase(parent, state) {}
    ~PipelineXSinkLocalState() override = default;

    Status init(RuntimeState* state, LocalSinkStateInfo& info) override;

    Status prepare(RuntimeState* state) override { return Status::OK(); }
    Status open(RuntimeState* state) override { return Status::OK(); }

    Status terminate(RuntimeState* state) override;
    Status close(RuntimeState* state, Status exec_status) override;

    [[nodiscard]] std::string debug_string(int indentation_level) const override;

    virtual std::string name_suffix();

    std::vector<Dependency*> dependencies() const override {
        return _dependency ? std::vector<Dependency*> {_dependency} : std::vector<Dependency*> {};
    }

    virtual bool must_set_shared_state() const {
        return !std::is_same_v<SharedStateArg, FakeSharedState>;
    }

protected:
    Dependency* _dependency = nullptr;
    SharedStateType* _shared_state = nullptr;
};

class DataSinkOperatorXBase : public OperatorBase {
public:
    DataSinkOperatorXBase(const int operator_id, const int node_id, const int dest_id)
            : _operator_id(operator_id), _node_id(node_id), _dests_id({dest_id}) {}
    DataSinkOperatorXBase(const int operator_id, const TPlanNode& tnode, const int dest_id)
            : OperatorBase(tnode.__isset.is_serial_operator && tnode.is_serial_operator),
              _operator_id(operator_id),
              _node_id(tnode.node_id),
              _dests_id({dest_id}) {}

    DataSinkOperatorXBase(const int operator_id, const int node_id, std::vector<int>& dests)
            : _operator_id(operator_id), _node_id(node_id), _dests_id(dests) {}

#ifdef BE_TEST
    DataSinkOperatorXBase() : _operator_id(-1), _node_id(0), _dests_id({-1}) {};
#endif

    ~DataSinkOperatorXBase() override = default;

    // For agg/sort/join sink.
    virtual Status init(const TPlanNode& tnode, RuntimeState* state);

    virtual bool reset_to_rerun(RuntimeState* state, OperatorXBase* root) const { return false; }

    Status init(const TDataSink& tsink) override;
    [[nodiscard]] virtual Status init(RuntimeState* state, TLocalPartitionType::type type,
                                      const int num_buckets,
                                      const std::map<int, int>& shuffle_idx_to_instance_idx) {
        return Status::InternalError("init() is only implemented in local exchange!");
    }

    Status prepare(RuntimeState* state) override { return Status::OK(); }
    Status terminate(RuntimeState* state) override;
    [[nodiscard]] bool is_finished(RuntimeState* state) const {
        auto result = state->get_sink_local_state_result();
        if (!result) {
            return result.error();
        }
        return result.value()->is_finished();
    }

    [[nodiscard]] Status sink(RuntimeState* state, Block* block, bool eos) {
        RETURN_IF_ERROR(block->check_column_and_type_not_null());
        RETURN_IF_ERROR(block->check_no_column_string64());
        RETURN_IF_ERROR(block->check_type_and_column());
        return sink_impl(state, block, eos);
    }

    [[nodiscard]] virtual Status sink_impl(RuntimeState* state, Block* block, bool eos) = 0;

    [[nodiscard]] virtual Status setup_local_state(RuntimeState* state,
                                                   LocalSinkStateInfo& info) = 0;

    // Returns the memory this sink operator expects to allocate in the next
    // execution round (sink only — pipeline task sums all operators + sink).
    [[nodiscard]] virtual size_t get_reserve_mem_size(RuntimeState* state, bool eos) {
        return state->minimum_operator_memory_required_bytes();
    }
    [[nodiscard]] virtual size_t get_reserve_mem_size(RuntimeState* state, bool eos,
                                                      const Block* block) {
        return get_reserve_mem_size(state, eos);
    }
    bool is_blockable(RuntimeState* state) const override {
        return state->get_sink_local_state()->is_blockable();
    }

    [[nodiscard]] bool is_spillable() const { return _spillable; }

    template <class TARGET>
    TARGET& cast() {
        DCHECK(dynamic_cast<TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<TARGET&>(*this);
    }
    template <class TARGET>
    const TARGET& cast() const {
        DCHECK(dynamic_cast<const TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<const TARGET&>(*this);
    }

    [[nodiscard]] virtual std::shared_ptr<BasicSharedState> create_shared_state() const = 0;

    Status close(RuntimeState* state) override {
        return Status::InternalError("Should not reach here!");
    }

    [[nodiscard]] virtual std::string debug_string(int indentation_level) const;

    [[nodiscard]] virtual std::string debug_string(RuntimeState* state,
                                                   int indentation_level) const;

    [[nodiscard]] bool is_sink() const override { return true; }

    static Status close(RuntimeState* state, Status exec_status) {
        auto result = state->get_sink_local_state_result();
        if (!result) {
            return result.error();
        }
        return result.value()->close(state, exec_status);
    }

    [[nodiscard]] int operator_id() const { return _operator_id; }

    [[nodiscard]] const std::vector<int>& dests_id() const { return _dests_id; }

    [[nodiscard]] int nereids_id() const { return _nereids_id; }

    [[nodiscard]] int node_id() const override { return _node_id; }

    [[nodiscard]] std::string get_name() const override { return _name; }

    virtual bool should_dry_run(RuntimeState* state) { return false; }

    [[nodiscard]] virtual bool count_down_destination() { return true; }

protected:
    template <typename Writer, typename Parent>
        requires(std::is_base_of_v<AsyncResultWriter, Writer>)
    friend class AsyncWriterSink;
    // _operator_id : the current Operator's ID, which is not visible to the user.
    // _node_id : the plan node ID corresponding to the Operator, which is visible on the profile.
    // _dests_id : the target _operator_id of the sink, for example, in the case of a multi-sink, there are multiple targets.
    const int _operator_id;
    const int _node_id;
    int _nereids_id = -1;
    bool _spillable = false;
    std::vector<int> _dests_id;
    std::string _name;
};

template <typename LocalStateType>
class DataSinkOperatorX : public DataSinkOperatorXBase {
public:
    DataSinkOperatorX(const int id, const int node_id, const int dest_id)
            : DataSinkOperatorXBase(id, node_id, dest_id) {}
    DataSinkOperatorX(const int id, const TPlanNode& tnode, const int dest_id)
            : DataSinkOperatorXBase(id, tnode, dest_id) {}

    DataSinkOperatorX(const int id, const int node_id, std::vector<int> dest_ids)
            : DataSinkOperatorXBase(id, node_id, dest_ids) {}
#ifdef BE_TEST
    DataSinkOperatorX() = default;
#endif
    ~DataSinkOperatorX() override = default;

    Status setup_local_state(RuntimeState* state, LocalSinkStateInfo& info) override;
    std::shared_ptr<BasicSharedState> create_shared_state() const override;

    using LocalState = LocalStateType;
    [[nodiscard]] LocalState& get_local_state(RuntimeState* state) const {
        return state->get_sink_local_state()->template cast<LocalState>();
    }
};

template <typename SharedStateArg>
class PipelineXSpillSinkLocalState : public PipelineXSinkLocalState<SharedStateArg> {
public:
    using Base = PipelineXSinkLocalState<SharedStateArg>;
    PipelineXSpillSinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state)
            : Base(parent, state) {}
    ~PipelineXSpillSinkLocalState() override = default;

    Status init(RuntimeState* state, LocalSinkStateInfo& info) override {
        RETURN_IF_ERROR(Base::init(state, info));
        init_spill_counters();
        return Status::OK();
    }

    void init_spill_counters() {
        _spill_total_timer =
                ADD_TIMER_WITH_LEVEL(Base::custom_profile(), profile::SPILL_TOTAL_TIME, 1);

        _write_counters.init(Base::custom_profile());

        // SpillFileWriter looks up these counters via get_counter() in its
        // constructor. They must be registered on the CustomCounters profile
        // before any SpillFileWriter is created, otherwise the lookups return
        // nullptr and COUNTER_UPDATE will SEGV.
        _spill_write_file_total_size = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_WRITE_FILE_BYTES, TUnit::BYTES, 1);
        _spill_file_total_count = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_WRITE_FILE_TOTAL_COUNT, TUnit::UNIT, 1);

        _spill_max_rows_of_partition = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_MAX_ROWS_OF_PARTITION, TUnit::UNIT, 1);
        _spill_min_rows_of_partition = ADD_COUNTER_WITH_LEVEL(
                Base::custom_profile(), profile::SPILL_MIN_ROWS_OF_PARTITION, TUnit::UNIT, 1);
    }

    std::vector<Dependency*> dependencies() const override {
        auto dependencies = Base::dependencies();
        return dependencies;
    }

    void update_max_min_rows_counter() {
        int64_t max_rows = 0;
        int64_t min_rows = std::numeric_limits<int64_t>::max();

        for (auto rows : _rows_in_partitions) {
            if (rows > max_rows) {
                max_rows = rows;
            }
            if (rows < min_rows) {
                min_rows = rows;
            }
        }

        COUNTER_SET(_spill_max_rows_of_partition, max_rows);
        COUNTER_SET(_spill_min_rows_of_partition, min_rows);
    }

    std::vector<int64_t> _rows_in_partitions;

    // Total time of spill, including spill task scheduling time,
    // serialize block time, write disk file time,
    // and read disk file time, deserialize block time etc.
    RuntimeProfile::Counter* _spill_total_timer = nullptr;

    // Shared spill write counters
    SpillWriteCounters _write_counters;
    // Backward-compatible aliases for commonly accessed write counters
    RuntimeProfile::Counter*& _spill_write_file_timer = _write_counters.spill_write_file_timer;
    RuntimeProfile::Counter*& _spill_write_serialize_block_timer =
            _write_counters.spill_write_serialize_block_timer;
    RuntimeProfile::Counter*& _spill_write_block_count = _write_counters.spill_write_block_count;
    RuntimeProfile::Counter*& _spill_write_block_data_size =
            _write_counters.spill_write_block_data_size;
    RuntimeProfile::Counter*& _spill_write_rows_count = _write_counters.spill_write_rows_count;

    // Sink-only counters
    // Total bytes written to spill files (required by SpillFileWriter)
    RuntimeProfile::Counter* _spill_write_file_total_size = nullptr;
    // Total number of spill files created (required by SpillFileWriter)
    RuntimeProfile::Counter* _spill_file_total_count = nullptr;
    RuntimeProfile::Counter* _spill_max_rows_of_partition = nullptr;
    RuntimeProfile::Counter* _spill_min_rows_of_partition = nullptr;
};

// The projection belongs to this operator and transforms its original output into the final
// output consumed by downstream operators. Most operators execute it after get_block(), while
// Scan executes it in Scanner before the block enters ScannerContext's queue.
struct OperatorProjection {
    OperatorProjection(const DescriptorTbl& descs, TTupleId output_tuple_id)
            : output_row_descriptor(descs, std::vector {output_tuple_id}) {}
    explicit OperatorProjection(RowDescriptor output_row_desc)
            : output_row_descriptor(std::move(output_row_desc)) {}

    VExprContextSPtrs projections;
    std::vector<VExprContextSPtrs> intermediate_projections;
    RowDescriptor output_row_descriptor;
    std::vector<RowDescriptor> intermediate_output_row_descriptors;
};

class OperatorXBase : public OperatorBase {
public:
    OperatorXBase(ObjectPool* pool, const TPlanNode& tnode, const int operator_id,
                  const DescriptorTbl& descs)
            : OperatorBase(tnode.__isset.is_serial_operator && tnode.is_serial_operator),
              _operator_id(operator_id),
              _node_id(tnode.node_id),
              _type(tnode.node_type),
              _pool(pool),
              _row_descriptor(descs, tnode.row_tuples),
              _resource_profile(tnode.resource_profile),
              _limit(tnode.limit) {
        if (tnode.__isset.projections) {
            DCHECK(tnode.__isset.output_tuple_id);
            _projection.emplace(descs, tnode.output_tuple_id);
        }
        if (!tnode.intermediate_output_tuple_id_list.empty()) {
            // common subexpression elimination
            DCHECK(_projection.has_value());
            _projection->intermediate_output_row_descriptors.reserve(
                    tnode.intermediate_output_tuple_id_list.size());
            for (auto output_tuple_id : tnode.intermediate_output_tuple_id_list) {
                _projection->intermediate_output_row_descriptors.push_back(
                        RowDescriptor(descs, std::vector {output_tuple_id}));
            }
        }
    }

    OperatorXBase(ObjectPool* pool, int node_id, int operator_id)
            : OperatorBase(),
              _operator_id(operator_id),
              _node_id(node_id),
              _pool(pool),
              _limit(-1) {}

#ifdef BE_TEST
    OperatorXBase() : _operator_id(-1), _node_id(0), _limit(-1) {};
#endif
    virtual Status init(const TPlanNode& tnode, RuntimeState* state);
    Status init(const TDataSink& tsink) override {
        throw Exception(Status::FatalError("should not reach here!"));
    }
    virtual Status init(TLocalPartitionType::type type) {
        throw Exception(Status::FatalError("should not reach here!"));
    }
    [[noreturn]] virtual const std::vector<TRuntimeFilterDesc>& runtime_filter_descs() {
        throw doris::Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, _op_name);
    }

    // Per-fragment shared partition-boundary parse result, used for
    // runtime-filter partition pruning. Returns nullptr for operators that
    // don't support this feature (default). Scan operators override to expose
    // their parsed boundaries; the per-instance pruning state lives on the
    // ScanLocalState. This sits on the generic OperatorXBase so non-templated
    // ScanLocalStateBase methods can fetch it without down-casting `_parent`
    // to a specific scan type.
    virtual const ParsedPartitionBoundaries* parsed_partition_boundaries() const { return nullptr; }
    [[nodiscard]] std::string get_name() const override { return _op_name; }
    [[nodiscard]] virtual bool need_more_input_data(RuntimeState* state) const { return true; }
    bool is_blockable(RuntimeState* state) const override {
        return state->get_sink_local_state()->is_blockable() || _blockable;
    }

    Status prepare(RuntimeState* state) override;

    Status terminate(RuntimeState* state) override;
    [[nodiscard]] Status get_block(RuntimeState* state, Block* block, bool* eos) {
        RETURN_IF_ERROR(get_block_impl(state, block, eos));
        RETURN_IF_ERROR(block->check_column_and_type_not_null());
        RETURN_IF_ERROR(block->check_no_column_string64());
        RETURN_IF_ERROR(block->check_type_and_column());
        return Status::OK();
    }

    [[nodiscard]] virtual Status get_block_impl(RuntimeState* state, Block* block, bool* eos) = 0;

    Status close(RuntimeState* state) override;

    // Returns the memory this single operator expects to allocate in the next
    // execution round.  Each operator reports only its OWN requirement — the
    // pipeline task is responsible for summing all operators + sink.
    // After the value is consumed the caller should invoke
    // reset_reserve_mem_size() so the next round starts from zero.
    // If this method is not overridden by a subclass, its default value is the
    // minimum operator memory (typically 1 MB).
    [[nodiscard]] virtual size_t get_reserve_mem_size(RuntimeState* state) {
        return state->minimum_operator_memory_required_bytes();
    }

    virtual std::string debug_string(int indentation_level = 0) const;

    virtual std::string debug_string(RuntimeState* state, int indentation_level = 0) const;

    virtual Status setup_local_state(RuntimeState* state, LocalStateInfo& info) = 0;

    template <class TARGET>
    TARGET& cast() {
        DCHECK(dynamic_cast<TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<TARGET&>(*this);
    }
    template <class TARGET>
    const TARGET& cast() const {
        DCHECK(dynamic_cast<const TARGET*>(this))
                << " Mismatch type! Current type is " << typeid(*this).name()
                << " and expect type is" << typeid(TARGET).name();
        return reinterpret_cast<const TARGET&>(*this);
    }

    [[nodiscard]] OperatorPtr get_child() { return _child; }

    [[nodiscard]] VExprContextSPtrs& conjuncts() { return _conjuncts; }
    // Describes the operator's original output before its optional projection. For most operators
    // this is the block returned by get_block(); Scan uses it to describe Scanner's unprojected
    // block because Scan executes the projection before ScanOperatorX::get_block_impl().
    [[nodiscard]] const RowDescriptor& operator_row_desc_before_projection() const {
        return _row_descriptor;
    }

    [[nodiscard]] int operator_id() const { return _operator_id; }
    [[nodiscard]] int node_id() const override { return _node_id; }
    [[nodiscard]] int nereids_id() const { return _nereids_id; }

    [[nodiscard]] int64_t limit() const { return _limit; }

    // Describes the final block returned by get_block_after_projects() to downstream operators.
    // Without a projection, the original and final row descriptors are the same.
    [[nodiscard]] const RowDescriptor& operator_row_desc_after_projection() const {
        return has_projection() ? _projection->output_row_descriptor
                                : operator_row_desc_before_projection();
    }

    [[nodiscard]] bool has_projection() const { return _projection.has_value(); }

    [[nodiscard]] bool has_intermediate_projection() const {
        return has_projection() && !_projection->intermediate_projections.empty();
    }

#ifdef BE_TEST
    OperatorProjection& set_projection_for_test(RowDescriptor output_row_desc) {
        return _projection.emplace(std::move(output_row_desc));
    }
#endif

    [[nodiscard]] virtual Status get_block_after_projects(RuntimeState* state, Block* block,
                                                          bool* eos);

    // Apply the optional projection to the block produced by the operator implementation.
    Status do_projections(RuntimeState* state, Block* origin_block, Block* output_block) const;
    void set_parallel_tasks(int parallel_tasks) { _parallel_tasks = parallel_tasks; }
    int parallel_tasks() const { return _parallel_tasks; }

    // To keep compatibility with older FE
    void set_serial_operator() { _is_serial_operator = true; }

    // Resets this operator's estimated memory usage to zero so that the next
    // call to get_reserve_mem_size() starts fresh.  The pipeline task calls
    // this after consuming the reserve size for all operators in a round.
    virtual void reset_reserve_mem_size(RuntimeState* state) {}

protected:
    template <typename Dependency>
    friend class PipelineXLocalState;
    friend class PipelineXLocalStateBase;
    friend class Scanner;
    const int _operator_id;
    const int _node_id; // unique w/in single plan tree
    int _nereids_id = -1;
    TPlanNodeType::type _type;
    ObjectPool* _pool = nullptr;

private:
    // The expr of operator set to private permissions, as cannot be executed concurrently,
    // should use local state's expr.
    VExprContextSPtrs _conjuncts;
    std::optional<OperatorProjection> _projection;

protected:
    RowDescriptor _row_descriptor;

    /// Resource information sent from the frontend.
    const TBackendResourceProfile _resource_profile;

    int64_t _limit; // -1: no limit

    uint32_t _debug_point_count = 0;
    std::atomic_uint32_t _bytes_per_row = 0;

    std::string _op_name;
    int _parallel_tasks = 0;

    // _blockable is true if the operator contains expressions that may block execution
    bool _blockable = false;
};

template <typename LocalStateType>
class OperatorX : public OperatorXBase {
public:
    OperatorX(ObjectPool* pool, const TPlanNode& tnode, const int operator_id,
              const DescriptorTbl& descs)
            : OperatorXBase(pool, tnode, operator_id, descs) {}
    OperatorX(ObjectPool* pool, int node_id, int operator_id)
            : OperatorXBase(pool, node_id, operator_id) {};

#ifdef BE_TEST
    OperatorX() = default;
#endif

    ~OperatorX() override = default;

    Status setup_local_state(RuntimeState* state, LocalStateInfo& info) override;
    using LocalState = LocalStateType;
    [[nodiscard]] LocalState& get_local_state(RuntimeState* state) const {
        return state->get_local_state(operator_id())->template cast<LocalState>();
    }

    // Returns memory this single operator expects to allocate in the next round.
    // Does NOT include child operators — the pipeline task iterates all
    // operators itself.
    size_t get_reserve_mem_size(RuntimeState* state) override {
        auto& local_state = get_local_state(state);
        auto estimated_size = local_state.estimate_memory_usage();
        if (estimated_size < state->minimum_operator_memory_required_bytes()) {
            estimated_size = state->minimum_operator_memory_required_bytes();
        }
        return estimated_size;
    }

    void reset_reserve_mem_size(RuntimeState* state) override {
        auto& local_state = get_local_state(state);
        local_state.reset_estimate_memory_usage();
    }
};

/**
 * StreamingOperatorX indicates operators which always processes block in streaming way (one-in-one-out).
 */
template <typename LocalStateType>
class StreamingOperatorX : public OperatorX<LocalStateType> {
public:
    StreamingOperatorX(ObjectPool* pool, const TPlanNode& tnode, int operator_id,
                       const DescriptorTbl& descs)
            : OperatorX<LocalStateType>(pool, tnode, operator_id, descs) {}

#ifdef BE_TEST
    StreamingOperatorX() = default;
#endif

    virtual ~StreamingOperatorX() = default;

    Status get_block_impl(RuntimeState* state, Block* block, bool* eos) override;

    virtual Status pull(RuntimeState* state, Block* block, bool* eos) = 0;
};

/**
 * StatefulOperatorX indicates the operators with some states inside.
 *
 * Specifically, we called an operator stateful if an operator can determine its output by itself.
 * For example, hash join probe operator is a typical StatefulOperator. When it gets a block from probe side, it will hold this block inside (e.g. _child_block).
 * If there are still remain rows in probe block, we can get output block by calling `get_block` without any data from its child.
 * In a nutshell, it is a one-to-many relation between input blocks and output blocks for StatefulOperator.
 */
template <typename LocalStateType>
class StatefulOperatorX : public OperatorX<LocalStateType> {
public:
    StatefulOperatorX(ObjectPool* pool, const TPlanNode& tnode, const int operator_id,
                      const DescriptorTbl& descs)
            : OperatorX<LocalStateType>(pool, tnode, operator_id, descs) {}
#ifdef BE_TEST
    StatefulOperatorX() = default;
#endif
    virtual ~StatefulOperatorX() = default;

    using OperatorX<LocalStateType>::get_local_state;

    [[nodiscard]] Status get_block_impl(RuntimeState* state, Block* block, bool* eos) override;

    [[nodiscard]] virtual Status pull(RuntimeState* state, Block* block, bool* eos) const = 0;
    [[nodiscard]] virtual Status push(RuntimeState* state, Block* input_block, bool eos) const = 0;
    bool need_more_input_data(RuntimeState* state) const override { return true; }
};

template <typename Writer, typename Parent>
    requires(std::is_base_of_v<AsyncResultWriter, Writer>)
class AsyncWriterSink : public PipelineXSinkLocalState<BasicSharedState> {
public:
    using Base = PipelineXSinkLocalState<BasicSharedState>;
    AsyncWriterSink(DataSinkOperatorXBase* parent, RuntimeState* state)
            : Base(parent, state), _async_writer_dependency(nullptr) {
        _finish_dependency =
                std::make_shared<Dependency>(parent->operator_id(), parent->node_id(),
                                             parent->get_name() + "_FINISH_DEPENDENCY", true);
    }

    Status init(RuntimeState* state, LocalSinkStateInfo& info) override;

    Status open(RuntimeState* state) override;

    Status sink(RuntimeState* state, Block* block, bool eos);

    std::vector<Dependency*> dependencies() const override {
        return {_async_writer_dependency.get()};
    }
    Status close(RuntimeState* state, Status exec_status) override;

    Dependency* finishdependency() override { return _finish_dependency.get(); }

protected:
    VExprContextSPtrs _output_vexpr_ctxs;
    std::unique_ptr<Writer> _writer;

    std::shared_ptr<Dependency> _async_writer_dependency;
    std::shared_ptr<Dependency> _finish_dependency;
};

#ifdef BE_TEST
class DummyOperatorLocalState final : public PipelineXLocalState<FakeSharedState> {
public:
    ENABLE_FACTORY_CREATOR(DummyOperatorLocalState);

    DummyOperatorLocalState(RuntimeState* state, OperatorXBase* parent)
            : PipelineXLocalState<FakeSharedState>(state, parent) {
        _tmp_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
                                                    "DummyOperatorDependency", true);
        _finish_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
                                                       "DummyOperatorDependency", true);
        _filter_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
                                                       "DummyOperatorDependency", true);
    }
    Dependency* finishdependency() override { return _finish_dependency.get(); }
    ~DummyOperatorLocalState() = default;

    std::vector<Dependency*> dependencies() const override { return {_tmp_dependency.get()}; }
    std::vector<Dependency*> execution_dependencies() override {
        return {_filter_dependency.get()};
    }

private:
    std::shared_ptr<Dependency> _tmp_dependency;
    std::shared_ptr<Dependency> _finish_dependency;
    std::shared_ptr<Dependency> _filter_dependency;
};

class DummyOperator final : public OperatorX<DummyOperatorLocalState> {
public:
    DummyOperator() : OperatorX<DummyOperatorLocalState>(nullptr, 0, 0) {}

    [[nodiscard]] bool is_source() const override { return true; }

    Status get_block_impl(RuntimeState* state, Block* block, bool* eos) override {
        *eos = _eos;
        return Status::OK();
    }
    void set_low_memory_mode(RuntimeState* state) override { _low_memory_mode = true; }
    Status terminate(RuntimeState* state) override {
        _terminated = true;
        return Status::OK();
    }
    size_t revocable_mem_size(RuntimeState* state) const override { return _revocable_mem_size; }
    size_t get_reserve_mem_size(RuntimeState* state) override {
        return _disable_reserve_mem
                       ? 0
                       : OperatorX<DummyOperatorLocalState>::get_reserve_mem_size(state);
    }
    Status revoke_memory(RuntimeState* state) override {
        _revoke_called = true;
        return Status::OK();
    }

private:
    friend class AssertNumRowsLocalState;
    bool _eos = false;
    bool _low_memory_mode = false;
    bool _terminated = false;
    size_t _revocable_mem_size = 0;
    bool _disable_reserve_mem = false;
    bool _revoke_called = false;
};

class DummySinkLocalState final : public PipelineXSinkLocalState<BasicSharedState> {
public:
    using Base = PipelineXSinkLocalState<BasicSharedState>;
    ENABLE_FACTORY_CREATOR(DummySinkLocalState);
    DummySinkLocalState(DataSinkOperatorXBase* parent, RuntimeState* state) : Base(parent, state) {
        _tmp_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
                                                    "DummyOperatorDependency", true);
        _finish_dependency = Dependency::create_shared(_parent->operator_id(), _parent->node_id(),
                                                       "DummyOperatorDependency", true);
    }

    std::vector<Dependency*> dependencies() const override { return {_tmp_dependency.get()}; }
    Dependency* finishdependency() override { return _finish_dependency.get(); }
    bool is_finished() const override { return _is_finished; }

private:
    std::shared_ptr<Dependency> _tmp_dependency;
    std::shared_ptr<Dependency> _finish_dependency;
    std::atomic_bool _is_finished = false;
};

class DummySinkOperatorX final : public DataSinkOperatorX<DummySinkLocalState> {
public:
    DummySinkOperatorX(int op_id, int node_id, int dest_id)
            : DataSinkOperatorX<DummySinkLocalState>(op_id, node_id, dest_id) {}
    Status sink_impl(RuntimeState* state, Block* in_block, bool eos) override {
        return _return_eof ? Status::Error<ErrorCode::END_OF_FILE>("source have closed")
                           : Status::OK();
    }
    void set_low_memory_mode(RuntimeState* state) override { _low_memory_mode = true; }
    Status terminate(RuntimeState* state) override {
        _terminated = true;
        return Status::OK();
    }
    size_t revocable_mem_size(RuntimeState* state) const override { return _revocable_mem_size; }
    size_t get_reserve_mem_size(RuntimeState* state, bool eos) override {
        return _disable_reserve_mem
                       ? 0
                       : DataSinkOperatorX<DummySinkLocalState>::get_reserve_mem_size(state, eos);
    }
    Status revoke_memory(RuntimeState* state) override {
        _revoke_called = true;
        return Status::OK();
    }

private:
    bool _low_memory_mode = false;
    bool _terminated = false;
    std::atomic_bool _return_eof = false;
    size_t _revocable_mem_size = 0;
    bool _disable_reserve_mem = false;
    bool _revoke_called = false;
};
#endif

/// Instantiated once in operator.cpp; suppresses per-TU implicit instantiation.
extern template class PipelineXSinkLocalState<HashJoinSharedState>;
extern template class PipelineXSinkLocalState<PartitionedHashJoinSharedState>;
extern template class PipelineXSinkLocalState<SortSharedState>;
extern template class PipelineXSinkLocalState<SpillSortSharedState>;
extern template class PipelineXSinkLocalState<NestedLoopJoinSharedState>;
extern template class PipelineXSinkLocalState<AnalyticSharedState>;
extern template class PipelineXSinkLocalState<AggSharedState>;
extern template class PipelineXSinkLocalState<BucketedAggSharedState>;
extern template class PipelineXSinkLocalState<PartitionedAggSharedState>;
extern template class PipelineXSinkLocalState<FakeSharedState>;
extern template class PipelineXSinkLocalState<UnionSharedState>;
extern template class PipelineXSinkLocalState<PartitionSortNodeSharedState>;
extern template class PipelineXSinkLocalState<MultiCastSharedState>;
extern template class PipelineXSinkLocalState<SetSharedState>;
extern template class PipelineXSinkLocalState<LocalExchangeSharedState>;
extern template class PipelineXSinkLocalState<BasicSharedState>;
extern template class PipelineXSinkLocalState<DataQueueSharedState>;
extern template class PipelineXLocalState<HashJoinSharedState>;
extern template class PipelineXLocalState<PartitionedHashJoinSharedState>;
extern template class PipelineXLocalState<SortSharedState>;
extern template class PipelineXLocalState<SpillSortSharedState>;
extern template class PipelineXLocalState<NestedLoopJoinSharedState>;
extern template class PipelineXLocalState<AnalyticSharedState>;
extern template class PipelineXLocalState<AggSharedState>;
extern template class PipelineXLocalState<BucketedAggSharedState>;
extern template class PipelineXLocalState<PartitionedAggSharedState>;
extern template class PipelineXLocalState<FakeSharedState>;
extern template class PipelineXLocalState<UnionSharedState>;
extern template class PipelineXLocalState<DataQueueSharedState>;
extern template class PipelineXLocalState<MultiCastSharedState>;
extern template class PipelineXLocalState<PartitionSortNodeSharedState>;
extern template class PipelineXLocalState<SetSharedState>;
extern template class PipelineXLocalState<LocalExchangeSharedState>;
extern template class PipelineXLocalState<BasicSharedState>;
#ifdef BE_TEST
extern template class OperatorX<DummyOperatorLocalState>;
extern template class DataSinkOperatorX<DummySinkLocalState>;
#endif

} // namespace doris