// 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 #include #include #include "common/status.h" #include "exec/filter-context.h" #include "gen-cpp/PlanNodes_types.h" #include "gutil/threading/thread_collision_warner.h" #include "runtime/bufferpool/buffer-pool.h" #include "runtime/bufferpool/reservation-tracker.h" #include "runtime/descriptors.h" // for RowDescriptor #include "runtime/reservation-manager.h" #include "util/runtime-profile-counters.h" #include "util/runtime-profile.h" namespace llvm { class Function; } namespace impala { class DataSink; class FragmentState; class MemPool; class MemTracker; class ObjectPool; class RowBatch; class RuntimeState; class ScalarExpr; class ScalarExprEvaluator; class SubplanNode; class SubplanPlanNode; class TPlan; class TupleRow; class TDebugOptions; class ExecNode; /// PlanNode and ExecNode are the super-classes of all plan nodes and execution nodes /// respectively. The plan nodes contain a subset of the static state of their /// corresponding ExecNode, of which there is one instance per fragment. ExecNode contains /// only runtime state and there can be up to num_instances_per_node() of it per /// fragment. Hence every ExecNode has a corresponding PlanNode which may or may not be at /// the same level of hierarchy as the ExecNode. /// TODO: IMPALA-9216: Move all the static state of ExecNode into PlanNode. /// /// 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 PlanNode { public: PlanNode() = default; /// Initializes this object from the thrift tnode desc. All internal members created and /// initialized here will be owned by state->obj_pool(). /// If overridden in subclass, must first call superclass's Init(). /// Should only be called after all children have been set and Init()-ed. virtual Status Init(const TPlanNode& tnode, FragmentState* state); /// Close() releases all resources that were allocated in Init(). virtual void Close(); /// Create its corresponding exec node. Place exec node in state->obj_pool(). virtual Status CreateExecNode(RuntimeState* state, ExecNode** node) const = 0; /// Creates plan node tree from list of nodes contained in plan via depth-first /// traversal. All nodes are placed in state->obj_pool() and have Init() called on them. /// Returns error if 'plan' is corrupted, otherwise success. static Status CreateTree(FragmentState* state, const TPlan& plan, PlanNode** root); /// Recursively calls Codegen() on all children. /// Expected to be overriden in subclass to generate LLVM IR functions and register /// them with the LlvmCodeGen object. The function pointers of the compiled IR functions /// will be set up when the LlvmCodeGen object is finalized. If overridden in subclass, /// must also call superclass's Codegen() before or after the code generation for this /// plan node. Will only be called once in the node's lifetime. virtual void Codegen(FragmentState* state); virtual ~PlanNode(){} bool IsInSubplan() const { return containing_subplan_ != nullptr; } /// Return true if codegen was disabled by the planner for this ExecNode. Does not /// check to see if codegen was enabled for the enclosing fragment. bool IsNodeCodegenDisabled() const { return tnode_->disable_codegen; } /// Returns true if parents of this node will need to deep copy some output batches. /// Nodes that MarkNeedsDeepCopy must override this method. virtual bool WillNeedDeepCopy(const TQueryOptions& opts) const { for (const PlanNode* child : children_) { if (child->WillNeedDeepCopy(opts)) return true; } return false; } /// TODO: IMPALA-9216: Add accessor methods for these members instead of making /// them public. /// Reference to the thrift node that represents this PlanNode. const TPlanNode* tnode_ = nullptr; /// Conjuncts in this node. 'conjuncts_' live in this exec node's object pool. Note: /// conjunct_evals_ are not created for Aggregation nodes. TODO: move conjuncts to /// query-state's obj pool. std::vector conjuncts_; RowDescriptor* row_descriptor_ = nullptr; // Runtime filter's expressions assigned to this plan node. std::vector runtime_filter_exprs_; std::vector children_; /// Helper function to skip past a TupleCacheNode if present. If the provided PlanNode /// is a TupleCacheNode, it returns the child. Otherwise, it returns the provided /// PlanNode. static PlanNode* LookPastTupleCache(PlanNode* pnode) { if (pnode->tnode_->node_type == TPlanNodeType::TUPLE_CACHE_NODE) { pnode = pnode->children_[0]; } DCHECK(pnode != nullptr); return pnode; } /// Pointer to the containing SubplanPlanNode or NULL if not inside a subplan. /// Set by the containing SubplanPlanNode::Init() before Init() is called on /// 'this' node. Not owned. SubplanPlanNode* containing_subplan_ = nullptr; /// A list of messages that will eventually be added to the exec node's runtime /// profile to convey codegen related information. Populated in Codegen(). std::vector codegen_status_msgs_; bool is_mt_fragment() const { return is_mt_fragment_; } int num_instances_per_node() const { return num_instances_per_node_; } protected: /// Helper method to add codegen messages from status objects. void AddCodegenStatus( const Status& codegen_status, const std::string& extra_label = ""); private: DISALLOW_COPY_AND_ASSIGN(PlanNode); /// Whether multi-threaded fragment strategy is being used. /// ie., either MT_DOP>0 or COMPUTE_PROCESSING_COST=true. bool is_mt_fragment_ = false; /// The effective degree of parallelism of this fragment. int num_instances_per_node_ = 1; /// Create a single exec node derived from thrift node; place exec node in 'pool'. static Status CreatePlanNode(ObjectPool* pool, const TPlanNode& tnode, PlanNode** node); static Status CreateTreeHelper(FragmentState* state, const std::vector& tnodes, PlanNode* parent, int* node_idx, PlanNode** root); }; class ExecNode { public: /// Copies over state from the 'pnode', initializes internal objects from 'pool' that /// cannot fail during initialization and finally initializes references to descriptors /// from 'descs'. ExecNode(ObjectPool* pool, const PlanNode& pnode, const DescriptorTbl& descs); virtual ~ExecNode(); /// 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. /// If overridden in subclass, must first call superclass's Prepare(). virtual Status Prepare(RuntimeState* state) WARN_UNUSED_RESULT; /// Performs any preparatory work prior to calling GetNext(). /// If overridden in subclass, must first call superclass's Open(). /// Open() is called after Prepare() or Reset(), i.e., possibly multiple times /// throughout the lifetime of this node. /// /// Memory resources must be acquired by an ExecNode only during or after the first /// call to Open(). Blocking ExecNodes outside of a subplan must call Open() on their /// child before acquiring their own resources to reduce the peak resource requirement. /// This is particularly important if there are multiple blocking ExecNodes in a /// pipeline because the lower nodes will release resources in Close() before the /// Open() of their parent returns. The resource profile calculation in the frontend /// relies on this when computing the peak resources required for a query. virtual Status Open(RuntimeState* state) WARN_UNUSED_RESULT; /// 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 GetNext() calls, it must *not* be attached to the /// row_batch's tuple_data_pool. virtual Status GetNext( RuntimeState* state, RowBatch* row_batch, bool* eos) WARN_UNUSED_RESULT = 0; /// 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(). 'row_batch' can /// be used to transfer ownership of any such memory. /// 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. virtual Status Reset(RuntimeState* state, RowBatch* row_batch) WARN_UNUSED_RESULT; /// 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()/GetNext() returned with an error). /// Close() releases all resources that were allocated in Open()/GetNext(), even if the /// latter ended with an error. Close() can be called if the node has been prepared or /// the node is closed. /// The default implementation updates runtime profile counters and calls /// Close() on the children. Subclasses should check if the node has already been /// closed (is_closed()), then close themselves, then call the base Close(). /// Nodes that are using tuples returned by a child may call Close() on their children /// before their own Close() if the child node has returned eos. /// It is only safe to call Close() on the child node while the parent node is still /// returning rows if the parent node fully materializes the child's input. virtual void Close(RuntimeState* state); /// Creates exec node tree from list of nodes contained in plan via depth-first /// traversal. All nodes are placed in state->obj_pool(). /// Returns error if 'plan_node' is corrupted, otherwise success. static Status CreateTree(RuntimeState* state, const PlanNode& plan_node, const DescriptorTbl& descs, ExecNode** root) WARN_UNUSED_RESULT; /// Set debug action in 'tree' according to debug_options. static void SetDebugOptions(const TDebugOptions& debug_options, ExecNode* tree); /// Collect all nodes of given 'node_type' that are part of this subtree, and return in /// 'nodes'. void CollectNodes(TPlanNodeType::type node_type, std::vector* nodes); /// Collect all scan node types. void CollectScanNodes(std::vector* nodes); /// Evaluate the conjuncts in 'evaluators' over 'row'. /// Returns true if all exprs return true. static inline bool EvalConjuncts( ScalarExprEvaluator* const* evals, int num_conjuncts, TupleRow* row); /// Codegen EvalConjuncts(). Returns a non-OK status if the function couldn't be /// codegen'd. The codegen'd version uses inlined, codegen'd GetBooleanVal() functions. static Status CodegenEvalConjuncts(LlvmCodeGen* codegen, const std::vector& conjuncts, llvm::Function** fn, const char* name = "EvalConjuncts") WARN_UNUSED_RESULT; /// Returns a string representation in DFS order of the plan rooted at this. std::string DebugString() const; /// Recursive helper method for generating a string for DebugString(). /// Implementations should call DebugString(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 DebugString(int indentation_level, std::stringstream* out) const; const std::vector& conjuncts() const { return conjuncts_; } const std::vector& conjunct_evals() const { return conjunct_evals_; } int id() const { return id_; } TPlanNodeType::type type() const { return type_; } const PlanNode& plan_node() const { return plan_node_; } /// Returns a unique label for this ExecNode of the form "PLAN_NODE_TYPE(id=[int])", /// for example, EXCHANGE_NODE (id=2). std::string label() const; /// Returns the row descriptor for rows produced by this node. The RowDescriptor is /// constant for the lifetime of the fragment instance, and so is shared by reference /// across the plan tree, including in RowBatches. The lifetime of the descriptor is the /// same as the lifetime of this node. const RowDescriptor* row_desc() const { return &row_descriptor_; } ExecNode* child(int i) { return children_[i]; } int num_children() const { return children_.size(); } /// Valid to call in or after Prepare(). SubplanNode* get_containing_subplan() const { return containing_subplan_; } void set_containing_subplan(SubplanNode* sp) { DCHECK(containing_subplan_ == nullptr); containing_subplan_ = sp; } int64_t limit() const { return limit_; } /// Returns the number of rows returned by this Node. int64_t rows_returned() const { DCHECK(getExecutionModel() != NON_TASK_BASED_SYNC); return num_rows_returned_; } /// Returns the number of rows returned by this Node. Thread safe version of /// rows_returned(). /// TODO: The thread safe versions can be removed if we remove the legacy multi-threaded /// scan nodes. int64_t rows_returned_shared() const { // Ideally this function should only be called when num_rows_returned_ is shared by // multiple threads. Both the HdfsScanNodeMt and KuduScanNodeMt call this function // from the scanner code-path for simplicity. DCHECK( getExecutionModel() == NON_TASK_BASED_SYNC || getExecutionModel() == TASK_BASED); return base::subtle::Acquire_Load(&num_rows_returned_); } /// Returns true if a valid limit is set and number of rows returned by this node has /// exceeded the limit. bool ReachedLimit() { DCHECK(getExecutionModel() != NON_TASK_BASED_SYNC); return limit_ != -1 && num_rows_returned_ >= limit_; } /// Returns true if a valid limit is set and number of rows returned by this node has /// exceeded the limit. Thread safe version of ReachedLimit(). bool ReachedLimitShared() { // Ideally this function should only be called when num_rows_returned_ is shared by // multiple threads. Both the HdfsScanNodeMt and KuduScanNodeMt call this function // from the scanner code-path for simplicity. DCHECK( getExecutionModel() == NON_TASK_BASED_SYNC || getExecutionModel() == TASK_BASED); return limit_ != -1 && base::subtle::Acquire_Load(&num_rows_returned_) >= limit_; } RuntimeProfile* runtime_profile() { return runtime_profile_; } MemTracker* mem_tracker() { return mem_tracker_.get(); } MemTracker* expr_mem_tracker() { return expr_mem_tracker_.get(); } MemPool* expr_perm_pool() { return expr_perm_pool_.get(); } MemPool* expr_results_pool() { return expr_results_pool_.get(); } const TBackendResourceProfile& resource_profile() { return resource_profile_; } bool is_closed() const { return is_closed_; } bool last_batch_returned() const { return last_batch_returned_.Load(); } /// Returns true if this node is inside the right-hand side plan tree of a SubplanNode. bool IsInSubplan() const { return plan_node_.IsInSubplan(); } /// Names of counters shared by all exec nodes static const std::string ROW_THROUGHPUT_COUNTER; static Status ParseAndValidateSleepDuration( const string& param, int64_t* sleep_duration_ms); protected: friend class DataSink; friend class ScopedGetNextEventAdder; friend class ScopedOpenEventAdder; enum ExecutionModel { /// Exec nodes with single threaded execution. This is the default execution model /// for majority of the nodes. BlockingJoin node is an exception since it could spawn /// a build thread in some cases. Due to the blocking nature of such join operators, /// it can be considered as not requiring explicit synchronization. NON_TASK_BASED_NO_SYNC, /// Exec nodes which spawn multiple worker threads. Examples are HdfsScanNode and /// and KuduScanNode. Requires syncronization if the main thread and worker threads /// share resources. NON_TASK_BASED_SYNC, /// Task Based multi-threading. Examples are HdfsScanNodeMt and KuduScanNodeMt. TASK_BASED }; virtual ExecutionModel getExecutionModel() const { return NON_TASK_BASED_NO_SYNC; } BufferPool::ClientHandle* buffer_pool_client() { return reservation_manager_.buffer_pool_client(); } Status ClaimBufferReservation(RuntimeState* state) WARN_UNUSED_RESULT { return reservation_manager_.ClaimBufferReservation(state); } Status ReleaseUnusedReservation() WARN_UNUSED_RESULT { return reservation_manager_.ReleaseUnusedReservation(); } /// Reference to the PlanNode shared across fragment instances. /// TODO: Store a specialized reference directly in the child classes when all static /// state is moved there and accessed directly. const PlanNode& plan_node_; /// Unique within a single plan tree. int id_; TPlanNodeType::type type_; ObjectPool* pool_; /// ExecNode lifecycle events for this ExecNode. Initialised for nodes outside subplan. /// Within a subplan, we iterate through the lifecycle multiple times so this would /// have a lot of overhead and not be particularly useful. All times are relative to /// QueryState::fragment_events_start_time(). RuntimeProfile::EventSequence* events_ = nullptr; /// Used to track whether the first GetNext() event was added to 'events_' so that /// ScopedGetNextEventAdder can avoid adding a duplicate event. bool first_getnext_added_ = false; /// Used to track whether the "Last Batch Returned" event was added to 'events_'. AtomicBool last_batch_returned_; /// Conjuncts and their evaluators in this node. 'conjuncts_' live in the /// query-state's object pool while the evaluators live in this exec node's /// object pool. Note: conjunct_evals_ are not created for Aggregation nodes. std::vector conjuncts_; std::vector conjunct_evals_; std::vector children_; RowDescriptor& row_descriptor_; /// Resource information sent from the frontend. const TBackendResourceProfile& resource_profile_; /// debug-only: if debug_action_ is not INVALID, node will perform action in /// debug_phase_ TDebugOptions debug_options_; int64_t limit_; // -1: no limit /// Runtime profile for this node. Owned by the QueryState's ObjectPool. RuntimeProfile* const runtime_profile_; RuntimeProfile::Counter* rows_returned_counter_; RuntimeProfile::Counter* rows_returned_rate_; /// Account for peak memory used by this node boost::scoped_ptr mem_tracker_; /// MemTracker used by 'expr_perm_pool_' and 'expr_results_pool_'. boost::scoped_ptr expr_mem_tracker_; /// MemPool for allocations made by expression evaluators in this node that are /// "permanent" and live until Close() is called. Created in Prepare(). boost::scoped_ptr expr_perm_pool_; /// MemPool for allocations made by expression evaluators in this node that hold /// intermediate or final results of expression evaluation. Should be cleared /// periodically to free accumulated memory. QueryMaintenance() clears this pool, but /// it may be appropriate for ExecNode implementation to clear it at other points in /// execution where the memory is not needed. boost::scoped_ptr expr_results_pool_; /// Pointer to the containing SubplanNode or NULL if not inside a subplan. /// Set by SubplanNode::Prepare() before Prepare() is called on 'this' node. Not owned. SubplanNode* containing_subplan_; virtual bool IsScanNode() const { return false; } /// Executes 'debug_action_' if 'phase' matches 'debug_phase_'. /// 'phase' must not be INVALID. Status ExecDebugAction( TExecNodePhase::type phase, RuntimeState* state) WARN_UNUSED_RESULT { DCHECK_NE(phase, TExecNodePhase::INVALID); // Fast path for the common case when an action is not enabled for this phase. if (LIKELY(debug_options_.phase != phase)) return Status::OK(); return ExecDebugActionImpl(phase, state); } /// Clears 'expr_results_pool_' and returns the result of state->CheckQueryState(). /// Nodes should call this periodically, e.g. once per input row batch. This should /// not be called outside the main execution thread. /// TODO: IMPALA-2399: replace QueryMaintenance() - see JIRA for more details. Status QueryMaintenance(RuntimeState* state) WARN_UNUSED_RESULT; /// Sets the number of rows returned. void SetNumRowsReturned(int64_t value) { DCHECK(getExecutionModel() != NON_TASK_BASED_SYNC); num_rows_returned_ = value; DFAKE_SCOPED_LOCK_THREAD_LOCKED(single_thread_check_); } /// Increment the number of rows returned. void IncrementNumRowsReturned(int64_t value) { DCHECK(getExecutionModel() != NON_TASK_BASED_SYNC); num_rows_returned_ += value; DFAKE_SCOPED_LOCK_THREAD_LOCKED(single_thread_check_); } /// Increment the number of rows returned. Thread safe version of /// IncrementNumRowsReturned(). void IncrementNumRowsReturnedShared(int64_t value) { DCHECK(getExecutionModel() == NON_TASK_BASED_SYNC); base::subtle::Barrier_AtomicIncrement(&num_rows_returned_, value); } /// Decrement the number of rows returned. void DecrementNumRowsReturned(int64_t value) { IncrementNumRowsReturned(-1 * value); } /// Decrement the number of rows returned. Thread safe version of /// DecrementNumRowsReturned(). void DecrementNumRowsReturnedShared(int64_t value) { IncrementNumRowsReturnedShared(-1 * value); } /// Caps the input row batch to ensure that the limit is not exceeded. /// Sets the eos and returns true, if the limit is reached. bool CheckLimitAndTruncateRowBatchIfNeeded(RowBatch* row_batch, bool* eos); /// Caps the input row batch to ensure that the limit is not exceeded. /// Sets the eos and returns true, if the limit is reached. /// Uses thread safe functions. bool CheckLimitAndTruncateRowBatchIfNeededShared(RowBatch* row_batch, bool* eos); /// Waits for all filters in 'filter_ctxs' to arrive (up to runtime_filter_wait_time_ms /// or the query-option override), then records arrival results in the runtime profile. /// Returns true if all filters arrived within the time limit. bool WaitForRuntimeFilters(RuntimeState* state, const std::vector& filter_ctxs); private: DISALLOW_COPY_AND_ASSIGN(ExecNode); /// Keeps track of number of rows returned by an exec node. If this variable is shared /// by multiple threads, it should be accessed using thread-safe functions defined /// above. The single-threaded code-paths should use non-atomic functions defined /// above. The only exceptions are HdfsScanNodeMt and KuduScanNodeMt, which are single /// threaded (task based multi-threading support), but use the thread-safe functions /// in the scanner code-path for code simplicity. This is because both the task based /// MT scan nodes (HdfsScanNodeMt/KuduScanNodeMt) and regular scan nodes /// (HdfsScanNode/KuduScanNode) call common scanner functions. int64_t num_rows_returned_; DFAKE_MUTEX(single_thread_check_); /// Implementation of ExecDebugAction(). This is the slow path we take when there is /// actually a debug action enabled for 'phase'. Status ExecDebugActionImpl( TExecNodePhase::type phase, RuntimeState* state) WARN_UNUSED_RESULT; /// Set in ExecNode::Close(). Used to make Close() idempotent. This is not protected /// by a lock, it assumes all calls to Close() are made by the same thread. bool is_closed_; /// Wraps the buffer pool client for this node. Initialized with the node's minimum /// reservation in ClaimBufferReservation(). After initialization, the client must hold /// onto at least the minimum reservation so that it can be returned to the initial /// reservations pool in Close(). ReservationManager reservation_manager_; }; } // namespace impala