// 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.
// This file is copied from
// https://github.com/apache/impala/blob/branch-2.9.0/be/src/exec/exec-node.h
// and modified by Doris

#pragma once

#include <gen_cpp/PlanNodes_types.h>

#include <atomic>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <vector>

#include "common/global_types.h"
#include "common/status.h"
#include "runtime/descriptors.h"
#include "util/runtime_profile.h"
#include "vec/core/block.h"
#include "vec/exprs/vexpr_fwd.h"

namespace doris {
class ObjectPool;
class RuntimeState;
class MemTracker;
class QueryStatistics;

namespace pipeline {
class OperatorBase;
} // namespace pipeline

using std::string;
using std::stringstream;
using std::vector;

// Superclass of all executor nodes.
// All subclasses need to make sure to check RuntimeState::is_cancelled()
// periodically in order to ensure timely termination after the cancellation
// flag gets set.
class ExecNode {
public:
    // Init conjuncts.
    ExecNode(ObjectPool* pool, const TPlanNode& tnode, const DescriptorTbl& descs);

    virtual ~ExecNode();

    /// Initializes this object from the thrift tnode desc. The subclass should
    /// do any initialization that can fail in Init() rather than the ctor.
    /// If overridden in subclass, must first call superclass's Init().
    [[nodiscard]] virtual Status init(const TPlanNode& tnode, RuntimeState* state);

    // Sets up internal structures, etc., without doing any actual work.
    // Must be called prior to open(). Will only be called once in this
    // node's lifetime.
    // All code generation (adding functions to the LlvmCodeGen object) must happen
    // in prepare().  Retrieving the jit compiled function pointer must happen in
    // open().
    // If overridden in subclass, must first call superclass's prepare().
    [[nodiscard]] virtual Status prepare(RuntimeState* state);

    /*
     * For open and alloc_resource:
     *  Base class ExecNode's `open` only calls `alloc_resource`, which opens some public projections.
     *  If was overrided, `open` must call corresponding `alloc_resource` since it's a (early) part of opening.
     *  Or just call `ExecNode::open` is alternative way.
     *  Then `alloc_resource` call father's after it's own business to make the progress completed, including the projections.
     *  In Pipeline engine: 
     *      PipeContext::prepare -> node::prepare
     *      Task::open -> StreamingOp::open -> node::alloc_resource, for sink+source splits, only open in SinkOperator.
     *  So in pipeline, the things directly done by open(like call child's) wouldn't be done in `open`.
    */
    // Performs any preparatory work prior to calling get_next().
    // Can be called repeatedly (after calls to close()).
    // Caller must not be holding any io buffers. This will cause deadlock.
    [[nodiscard]] virtual Status open(RuntimeState* state);

    // Alloc and open resource for the node
    // Only pipeline operator use exec node need to impl the virtual function
    // so only vectorized exec node need to impl
    [[nodiscard]] virtual Status alloc_resource(RuntimeState* state);

    // Retrieves rows and returns them via row_batch. Sets eos to true
    // if subsequent calls will not retrieve any more rows.
    // Data referenced by any tuples returned in row_batch must not be overwritten
    // by the callee until close() is called. The memory holding that data
    // can be returned via row_batch's tuple_data_pool (in which case it may be deleted
    // by the caller) or held on to by the callee. The row_batch, including its
    // tuple_data_pool, will be destroyed by the caller at some point prior to the final
    // close() call.
    // In other words, if the memory holding the tuple data will be referenced
    // by the callee in subsequent get_next() calls, it must *not* be attached to the
    // row_batch's tuple_data_pool.
    // Caller must not be holding any io buffers. This will cause deadlock.
    // TODO: AggregationNode and HashJoinNode cannot be "re-opened" yet.
    [[nodiscard]] virtual Status get_next(RuntimeState* state, vectorized::Block* block, bool* eos);
    // new interface to compatible new optimizers in FE
    [[nodiscard]] virtual Status get_next_after_projects(
            RuntimeState* state, vectorized::Block* block, bool* eos,
            const std::function<Status(RuntimeState*, vectorized::Block*, bool*)>& fn,
            bool clear_data = true);

    // Used by pipeline streaming operators.
    vectorized::Block* get_clear_input_block() {
        clear_origin_block();
        return &_origin_block;
    }
    bool has_output_row_descriptor() const { return _output_row_descriptor != nullptr; }
    // If use projection, we should clear `_origin_block`.
    void clear_origin_block() {
        _origin_block.clear_column_data(intermediate_row_desc().num_materialized_slots());
    }

    // Emit data, both need impl with method: sink
    // Eg: Aggregation, Sort, Scan
    [[nodiscard]] virtual Status pull(RuntimeState* state, vectorized::Block* output_block,
                                      bool* eos) {
        return get_next(state, output_block, eos);
    }

    [[nodiscard]] virtual Status push(RuntimeState* state, vectorized::Block* input_block,
                                      bool eos) {
        return Status::OK();
    }

    bool can_read() const { return _can_read; }

    [[nodiscard]] virtual bool can_terminate_early() { return false; }

    // Sink Data to ExecNode to do some stock work, both need impl with method: get_result
    // `eos` means source is exhausted, exec node should do some finalize work
    // Eg: Aggregation, Sort
    [[nodiscard]] virtual Status sink(RuntimeState* state, vectorized::Block* input_block,
                                      bool eos);

    // Resets the stream of row batches to be retrieved by subsequent GetNext() calls.
    // Clears all internal state, returning this node to the state it was in after calling
    // Prepare() and before calling Open(). This function must not clear memory
    // still owned by this node that is backing rows returned in GetNext().
    // Prepare() and Open() must have already been called before calling Reset().
    // GetNext() may have optionally been called (not necessarily until eos).
    // Close() must not have been called.
    // Reset() is not idempotent. Calling it multiple times in a row without a preceding
    // call to Open() is invalid.
    // If overridden in a subclass, must call superclass's Reset() at the end. The default
    // implementation calls Reset() on children.
    // Note that this function may be called many times (proportional to the input data),
    // so should be fast.
    [[nodiscard]] virtual Status reset(RuntimeState* state);

    // close() will get called for every exec node, regardless of what else is called and
    // the status of these calls (i.e. prepare() may never have been called, or
    // prepare()/open()/get_next() returned with an error).
    // close() releases all resources that were allocated in open()/get_next(), even if the
    // latter ended with an error. close() can be called if the node has been prepared or
    // the node is closed.
    // After calling close(), the caller calls open() again prior to subsequent calls to
    // get_next(). The default implementation updates runtime profile counters and calls
    // close() on the children. To ensure that close() is called on the entire plan tree,
    // each implementation should start out by calling the default implementation.
    virtual Status close(RuntimeState* state);

    void increase_ref() { ++_ref; }
    int decrease_ref() { return --_ref; }

    // Release and close resource for the node
    // Only pipeline operator use exec node need to impl the virtual function
    // so only vectorized exec node need to impl
    virtual void release_resource(RuntimeState* state);

    // Creates exec node tree from list of nodes contained in plan via depth-first
    // traversal. All nodes are placed in pool.
    // Returns error if 'plan' is corrupted, otherwise success.
    [[nodiscard]] static Status create_tree(RuntimeState* state, ObjectPool* pool,
                                            const TPlan& plan, const DescriptorTbl& descs,
                                            ExecNode** root);

    // Collect all nodes of given 'node_type' that are part of this subtree, and return in
    // 'nodes'.
    void collect_nodes(TPlanNodeType::type node_type, std::vector<ExecNode*>* nodes);

    // Collect all scan node types.
    void collect_scan_nodes(std::vector<ExecNode*>* nodes);

    virtual void prepare_for_next() {}

    // Returns a string representation in DFS order of the plan rooted at this.
    std::string debug_string() const;

    // recursive helper method for generating a string for Debug_string().
    // implementations should call debug_string(int, std::stringstream) on their children.
    // Input parameters:
    //   indentation_level: Current level in plan tree.
    // Output parameters:
    //   out: Stream to accumulate debug string.
    virtual void debug_string(int indentation_level, std::stringstream* out) const;

    int id() const { return _id; }
    TPlanNodeType::type type() const { return _type; }
    virtual const RowDescriptor& row_desc() const {
        return _output_row_descriptor ? *_output_row_descriptor : _row_descriptor;
    }
    virtual const RowDescriptor& intermediate_row_desc() const { return _row_descriptor; }

    //  input expr -> intermediate_projections[0] -> intermediate_projections[1] -> intermediate_projections[2]    ... ->     final projections         ->         output expr
    //  prepare        _row_descriptor          intermediate_row_desc[0]             intermediate_row_desc[1]            intermediate_row_desc.end()          _output_row_descriptor

    [[nodiscard]] const RowDescriptor& intermediate_row_desc(int idx) {
        if (idx == 0) {
            return intermediate_row_desc();
        }
        DCHECK((idx - 1) < _intermediate_output_row_descriptor.size());
        return _intermediate_output_row_descriptor[idx - 1];
    }

    [[nodiscard]] const RowDescriptor& projections_row_desc() const {
        if (_intermediate_output_row_descriptor.empty()) {
            return intermediate_row_desc();
        } else {
            return _intermediate_output_row_descriptor.back();
        }
    }

    int64_t rows_returned() const { return _num_rows_returned; }
    int64_t limit() const { return _limit; }
    bool reached_limit() const { return _limit != -1 && _num_rows_returned >= _limit; }
    /// Only use in vectorized exec engine to check whether reach limit and cut num row for block
    // and add block rows for profile
    void reached_limit(vectorized::Block* block, bool* eos);
    const std::vector<TupleId>& get_tuple_ids() const { return _tuple_ids; }

    RuntimeProfile* faker_runtime_profile() const { return _faker_runtime_profile.get(); }
    RuntimeProfile* runtime_profile() const { return _runtime_profile.get(); }
    RuntimeProfile::Counter* memory_used_counter() const { return _memory_used_counter; }

    MemTracker* mem_tracker() const { return _mem_tracker.get(); }

    virtual std::string get_name();

    // Names of counters shared by all exec nodes
    static const std::string ROW_THROUGHPUT_COUNTER;

    ExecNode* child(int i) { return _children[i]; }

    size_t children_count() const { return _children.size(); }

    std::shared_ptr<QueryStatistics> get_query_statistics() { return _query_statistics; }

protected:
    friend class DataSink;

    /// Release all memory of block which got from child. The block
    // 1. clear mem of valid column get from child, make sure child can reuse the mem
    // 2. delete and release the column which create by function all and other reason
    void release_block_memory(vectorized::Block& block, uint16_t child_idx = 0);

    /// Only use in vectorized exec engine try to do projections to trans _row_desc -> _output_row_desc
    Status do_projections(vectorized::Block* origin_block, vectorized::Block* output_block);

    int _id; // unique w/in single plan tree
    TPlanNodeType::type _type;
    ObjectPool* _pool = nullptr;
    std::vector<TupleId> _tuple_ids;

    vectorized::VExprContextSPtrs _conjuncts;

    std::vector<ExecNode*> _children;
    RowDescriptor _row_descriptor;
    vectorized::Block _origin_block;

    std::unique_ptr<RowDescriptor> _output_row_descriptor;
    vectorized::VExprContextSPtrs _projections;

    std::vector<RowDescriptor> _intermediate_output_row_descriptor;
    // Used in common subexpression elimination to compute intermediate results.
    std::vector<vectorized::VExprContextSPtrs> _intermediate_projections;

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

    int64_t _limit; // -1: no limit
    int64_t _num_rows_returned = 0;

    std::unique_ptr<RuntimeProfile> _runtime_profile;

    // Record this node memory size. it is expected that artificial guarantees are accurate,
    // which will providea reference for operator memory.
    std::unique_ptr<MemTracker> _mem_tracker;

    RuntimeProfile::Counter* _exec_timer = nullptr;
    RuntimeProfile::Counter* _rows_returned_counter = nullptr;
    RuntimeProfile::Counter* _output_bytes_counter = nullptr;
    RuntimeProfile::Counter* _block_count_counter = nullptr;
    RuntimeProfile::Counter* _rows_returned_rate = nullptr;
    RuntimeProfile::Counter* _memory_used_counter = nullptr;
    RuntimeProfile::Counter* _projection_timer = nullptr;
    // Account for peak memory used by this node
    RuntimeProfile::Counter* _peak_memory_usage_counter = nullptr;

    //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>("faker profile");

    // Execution options that are determined at runtime.  This is added to the
    // runtime profile at close().  Examples for options logged here would be
    // "Codegen Enabled"
    std::mutex _exec_options_lock;
    std::string _runtime_exec_options;

    // Set to true if this is a vectorized exec node.
    bool _is_vec = false;

    bool is_closed() const { return _is_closed; }

    // Create a single exec node derived from thrift node; place exec node in 'pool'.
    static Status create_node(RuntimeState* state, ObjectPool* pool, const TPlanNode& tnode,
                              const DescriptorTbl& descs, ExecNode** node);

    virtual bool is_scan_node() const { return false; }

    void init_runtime_profile(const std::string& name);

    // Appends option to '_runtime_exec_options'
    void add_runtime_exec_option(const std::string& option);

    std::atomic<bool> _can_read = false;

    std::shared_ptr<QueryStatistics> _query_statistics = nullptr;

    //_keep_origin is used to avoid copying during projection,
    // currently set to false only in the nestloop join.
    bool _keep_origin = true;

private:
    static Status create_tree_helper(RuntimeState* state, ObjectPool* pool,
                                     const std::vector<TPlanNode>& tnodes,
                                     const DescriptorTbl& descs, ExecNode* parent, int* node_idx,
                                     ExecNode** root);

    friend class pipeline::OperatorBase;
    bool _is_closed = false;
    bool _is_resource_released = false;
    std::atomic_int _ref = 0; // used by pipeline operator to release resource.
};

} // namespace doris