#include <plan-fragment-executor.h>

Collaboration diagram for impala::PlanFragmentExecutor:

Public Types
typedef boost::function< void(const Status &status, RuntimeProfile *profile, bool done)>	ReportStatusCallback

Public Member Functions
	PlanFragmentExecutor (ExecEnv *exec_env, const ReportStatusCallback &report_status_cb)

	~PlanFragmentExecutor ()

Status	Prepare (const TExecPlanFragmentParams &request)

Status	Open ()

Status	GetNext (RowBatch **batch)

void	Close ()

void	Cancel ()
	Initiate cancellation. Must not be called until after Prepare() returned. More...

bool	ReachedLimit ()
	Returns true if this query has a limit and it has been reached. More...

void	ReleaseThreadToken ()
	Releases the thread token for this fragment executor. More...

RuntimeState *	runtime_state ()
	call these only after Prepare() More...

const RowDescriptor &	row_desc ()

RuntimeProfile *	profile ()
	Profile information for plan and output sink. More...

Static Public Attributes
static const std::string	PER_HOST_PEAK_MEM_COUNTER = "PerHostPeakMemUsage"
	Name of the counter that is tracking per query, per host peak mem usage. More...

Private Types
typedef std::map< TPlanNodeId, std::vector< TScanRangeParams > >	PerNodeScanRanges
	typedef for TPlanFragmentExecParams.per_node_scan_ranges More...

Private Member Functions
ObjectPool *	obj_pool ()

void	ReportProfile ()

void	SendReport (bool done)

void	UpdateStatus (const Status &status)

void	FragmentComplete ()
	Called when the fragment execution is complete to finalize counters. More...

void	OptimizeLlvmModule ()

Status	OpenInternal ()

Status	GetNextInternal (RowBatch **batch)

void	StopReportThread ()

void	PrintVolumeIds (const TPlanExecParams &params)
	Print stats about scan ranges for each volumeId in params to info log. More...

void	PrintVolumeIds (const PerNodeScanRanges &per_node_scan_ranges)

const DescriptorTbl &	desc_tbl ()

Private Attributes
ExecEnv *	exec_env_

ExecNode *	plan_

TUniqueId	query_id_

ReportStatusCallback	report_status_cb_
	profile reporting-related More...

boost::scoped_ptr< Thread >	report_thread_

boost::mutex	report_thread_lock_

boost::condition_variable	stop_report_thread_cv_

boost::condition_variable	report_thread_started_cv_

bool	report_thread_active_

bool	done_
	true if plan_->GetNext() indicated that it's done More...

bool	prepared_
	true if Prepare() returned OK More...

bool	closed_
	true if Close() has been called More...

bool	has_thread_token_
	true if this fragment has not returned the thread token to the thread resource mgr More...

Status	status_

boost::mutex	status_lock_

boost::scoped_ptr< DataSink >	sink_

boost::scoped_ptr< RuntimeState >	runtime_state_

boost::scoped_ptr< RowBatch >	row_batch_

boost::scoped_ptr< TRowBatch >	thrift_batch_

RuntimeProfile::Counter *	per_host_mem_usage_

RuntimeProfile::Counter *	rows_produced_counter_
	Number of rows returned by this fragment. More...

RuntimeProfile::Counter *	average_thread_tokens_

MonotonicStopWatch	fragment_sw_
	Stopwatch for this entire fragment. Started in Prepare(), stopped in Close(). More...

AtomicInt< int >	completed_report_sent_

RuntimeProfile::TimeSeriesCounter *	mem_usage_sampled_counter_
	Sampled memory usage at even time intervals. More...

RuntimeProfile::TimeSeriesCounter *	thread_usage_sampled_counter_
	Sampled thread usage (tokens) at even time intervals. More...

Detailed Description

PlanFragmentExecutor handles all aspects of the execution of a single plan fragment, including setup and tear-down, both in the success and error case. Tear-down frees all memory allocated for this plan fragment and closes all data streams; it happens automatically in the d'tor. The executor makes an aggregated profile for the entire fragment available, which includes profile information for the plan itself as well as the output sink, if any. The ReportStatusCallback passed into the c'tor is invoked periodically to report the execution status. The frequency of those reports is controlled by the flag status_report_interval; setting that flag to 0 disables periodic reporting altogether Regardless of the value of that flag, if a report callback is specified, it is invoked at least once at the end of execution with an overall status and profile (and 'done' indicator). The only exception is when execution is cancelled, in which case the callback is not invoked (the coordinator already knows that execution stopped, because it initiated the cancellation). Aside from Cancel(), which may be called asynchronously, this class is not thread-safe.

Definition at line 64 of file plan-fragment-executor.h.

Member Typedef Documentation

typedef std::map<TPlanNodeId, std::vector<TScanRangeParams> > impala::PlanFragmentExecutor::PerNodeScanRanges

private

typedef for TPlanFragmentExecParams.per_node_scan_ranges

Definition at line 221 of file plan-fragment-executor.h.

typedef boost::function< void (const Status& status, RuntimeProfile* profile, bool done)> impala::PlanFragmentExecutor::ReportStatusCallback

Callback to report execution status of plan fragment. 'profile' is the cumulative profile, 'done' indicates whether the execution is done or still continuing. Note: this does not take a const RuntimeProfile&, because it might need to call functions like PrettyPrint() or ToThrift(), neither of which is const because they take locks.

Definition at line 74 of file plan-fragment-executor.h.

Constructor & Destructor Documentation

impala::PlanFragmentExecutor::PlanFragmentExecutor	(	ExecEnv *	exec_env,
		const ReportStatusCallback &	report_status_cb
	)

report_status_cb, if !empty(), is used to report the accumulated profile information periodically during execution (Open() or GetNext()).

Definition at line 63 of file plan-fragment-executor.cc.

impala::PlanFragmentExecutor::~PlanFragmentExecutor ( )

Closes the underlying plan fragment and frees up all resources allocated in Open()/GetNext(). It is an error to delete a PlanFragmentExecutor with a report callback before Open()/GetNext() (depending on whether the fragment has a sink) indicated that execution is finished.

Definition at line 71 of file plan-fragment-executor.cc.

References Close(), exec_env_, query_id_, report_thread_active_, impala::ExecEnv::resource_broker(), runtime_state_, and impala::ResourceBroker::UnregisterQueryResourceMgr().

Member Function Documentation

void impala::PlanFragmentExecutor::Cancel ( )

Initiate cancellation. Must not be called until after Prepare() returned.

Definition at line 531 of file plan-fragment-executor.cc.

References prepared_, runtime_state_, and VLOG_QUERY.

void impala::PlanFragmentExecutor::Close ( )

Closes the underlying plan fragment and frees up all resources allocated in Open()/GetNext().

Definition at line 564 of file plan-fragment-executor.cc.

References impala::ExecEnv::cgroups_mgr(), impala::ExecNode::Close(), closed_, exec_env_, mem_usage_sampled_counter_, plan_, row_batch_, runtime_state(), runtime_state_, sink_, impala::PeriodicCounterUpdater::StopTimeSeriesCounter(), impala::ExecEnv::thread_mgr(), impala::CgroupsMgr::UnregisterFragment(), and impala::ThreadResourceMgr::UnregisterPool().

Referenced by ~PlanFragmentExecutor().

const DescriptorTbl& impala::PlanFragmentExecutor::desc_tbl ( )

inlineprivate

Definition at line 269 of file plan-fragment-executor.h.

References runtime_state_.

Referenced by Prepare().

void impala::PlanFragmentExecutor::FragmentComplete ( )

private

Called when the fragment execution is complete to finalize counters.

Definition at line 492 of file plan-fragment-executor.cc.

References impala::AtomicInt< T >::CompareAndSwap(), completed_report_sent_, impala::MonotonicStopWatch::ElapsedTime(), fragment_sw_, ReleaseThreadToken(), runtime_state_, SendReport(), impala::MonotonicStopWatch::Stop(), and StopReportThread().

Referenced by GetNext(), and OpenInternal().

Status impala::PlanFragmentExecutor::GetNext ( RowBatch ** batch )

Return results through 'batch'. Sets '*batch' to NULL if no more results. '*batch' is owned by PlanFragmentExecutor and must not be deleted. When *batch == NULL, GetNext() should not be called anymore. Also, report_status_cb will have been called for the final time and the status-reporting thread will have been stopped.

Definition at line 455 of file plan-fragment-executor.cc.

References done_, FragmentComplete(), GetNextInternal(), impala::RowBatch::num_rows(), impala::PrintId(), query_id_, runtime_state_, UpdateStatus(), VLOG_FILE, and VLOG_QUERY.

Status impala::PlanFragmentExecutor::GetNextInternal ( RowBatch ** batch )

private

Executes GetNext() logic and returns resulting status. sets done_ to true if the last row batch was returned.

Definition at line 471 of file plan-fragment-executor.cc.

References COUNTER_ADD, done_, impala::ExecNode::GetNext(), impala::Status::OK, plan_, profile(), RETURN_IF_ERROR, row_batch_, rows_produced_counter_, runtime_state_, and SCOPED_TIMER.

Referenced by GetNext(), and OpenInternal().

ObjectPool* impala::PlanFragmentExecutor::obj_pool ( )

inlineprivate

Definition at line 218 of file plan-fragment-executor.h.

References runtime_state_.

Referenced by Prepare().

Status impala::PlanFragmentExecutor::Open ( )

Start execution. Call this prior to GetNext(). If this fragment has a sink, Open() will send all rows produced by the fragment to that sink. Therefore, Open() may block until all rows are produced (and a subsequent call to GetNext() will not return any rows). This also starts the status-reporting thread, if the interval flag is > 0 and a callback was specified in the c'tor. If this fragment has a sink, report_status_cb will have been called for the final time when Open() returns, and the status-reporting thread will have been stopped.

Definition at line 302 of file plan-fragment-executor.cc.

References impala::Status::IsCancelled(), impala::Status::IsMemLimitExceeded(), impala::Status::msg(), impala::Status::ok(), OpenInternal(), OptimizeLlvmModule(), report_status_cb_, report_thread_, report_thread_active_, report_thread_lock_, report_thread_started_cv_, ReportProfile(), runtime_state_, UpdateStatus(), and VLOG_QUERY.

Status impala::PlanFragmentExecutor::OpenInternal ( )

private

Executes Open() logic and returns resulting status. Does not set status_. If this plan fragment has no sink, OpenInternal() does nothing. If this plan fragment has a sink and OpenInternal() returns without an error condition, all rows will have been sent to the sink, the sink will have been closed, a final report will have been sent and the report thread will have been stopped. sink_ will be set to NULL after successful execution.

Definition at line 331 of file plan-fragment-executor.cc.

References done_, FragmentComplete(), GetNextInternal(), impala::RowBatch::GetRow(), impala::RowBatch::num_rows(), impala::Status::OK, impala::ExecNode::Open(), plan_, impala::PrintRow(), profile(), RETURN_IF_ERROR, row_desc(), runtime_state(), runtime_state_, SCOPED_TIMER, sink_, VLOG_ROW, and VLOG_ROW_IS_ON.

Referenced by Open().

void impala::PlanFragmentExecutor::OptimizeLlvmModule ( )

private

Optimizes the code-generated functions in runtime_state_->llvm_codegen(). Must be called between plan_->Prepare() and plan_->Open(). This is somewhat time consuming so we don't want it to do it in PlanFragmentExecutor()::Prepare() to allow starting plan fragments more quickly and in parallel (in a deep plan tree, the fragments are started in level order).

Definition at line 270 of file plan-fragment-executor.cc.

References impala::LlvmCodeGen::FinalizeModule(), impala::Status::GetDetail(), impala::Status::ok(), and runtime_state_.

Referenced by Open().

Status impala::PlanFragmentExecutor::Prepare ( const TExecPlanFragmentParams & request )

Prepare for execution. Call this prior to Open(). This call won't block. runtime_state() and row_desc() will not be valid until Prepare() is called. If request.query_options.mem_limit > 0, it is used as an approximate limit on the number of bytes this query can consume at runtime. The query will be aborted (MEM_LIMIT_EXCEEDED) if it goes over that limit.

Definition at line 80 of file plan-fragment-executor.cc.

void impala::PlanFragmentExecutor::PrintVolumeIds ( const TPlanExecParams & params )

private

Print stats about scan ranges for each volumeId in params to info log.

Referenced by Prepare().

void impala::PlanFragmentExecutor::PrintVolumeIds ( const PerNodeScanRanges & per_node_scan_ranges )

private

Definition at line 284 of file plan-fragment-executor.cc.

References impala::RuntimeProfile::AddInfoString(), impala::HdfsScanNode::HDFS_SPLIT_STATS_DESC, impala::HdfsScanNode::PrintHdfsSplitStats(), profile(), query_id_, impala::HdfsScanNode::UpdateHdfsSplitStats(), and VLOG_FILE.

RuntimeProfile * impala::PlanFragmentExecutor::profile ( )

Profile information for plan and output sink.

Definition at line 543 of file plan-fragment-executor.cc.

References runtime_state_.

Referenced by GetNextInternal(), OpenInternal(), Prepare(), PrintVolumeIds(), ReportProfile(), and SendReport().

bool impala::PlanFragmentExecutor::ReachedLimit ( )

Returns true if this query has a limit and it has been reached.

Definition at line 547 of file plan-fragment-executor.cc.

References plan_, and impala::ExecNode::ReachedLimit().

void impala::PlanFragmentExecutor::ReleaseThreadToken ( )

Releases the thread token for this fragment executor.

Definition at line 551 of file plan-fragment-executor.cc.

References average_thread_tokens_, has_thread_token_, runtime_state_, impala::PeriodicCounterUpdater::StopSamplingCounter(), impala::PeriodicCounterUpdater::StopTimeSeriesCounter(), and thread_usage_sampled_counter_.

Referenced by FragmentComplete().

void impala::PlanFragmentExecutor::ReportProfile ( )

private

Main loop of profile reporting thread. Exits when notified on done_cv_. On exit, no report is sent, ie, this will not send the final report.

Definition at line 374 of file plan-fragment-executor.cc.

References completed_report_sent_, impala::RuntimeProfile::PrettyPrint(), profile(), impala::AtomicInt< T >::Read(), report_status_cb_, report_thread_active_, report_thread_lock_, report_thread_started_cv_, runtime_state_, SendReport(), stop_report_thread_cv_, VLOG_FILE, and VLOG_FILE_IS_ON.

Referenced by Open().

const RowDescriptor & impala::PlanFragmentExecutor::row_desc ( )

Definition at line 539 of file plan-fragment-executor.cc.

References plan_, and impala::ExecNode::row_desc().

Referenced by OpenInternal(), and Prepare().

RuntimeState* impala::PlanFragmentExecutor::runtime_state ( )

inline

call these only after Prepare()

Definition at line 127 of file plan-fragment-executor.h.

References runtime_state_.

Referenced by Close(), OpenInternal(), Prepare(), and SendReport().

void impala::PlanFragmentExecutor::SendReport ( bool done )

private

Invoked the report callback if there is a report callback and the current status isn't CANCELLED. Sets 'done' to true in the callback invocation if done == true or we have an error status.

Definition at line 423 of file plan-fragment-executor.cc.

References impala::Status::ok(), per_host_mem_usage_, profile(), impala::query_mem_tracker(), report_status_cb_, runtime_state(), impala::RuntimeProfile::Counter::Set(), status_, and status_lock_.

Referenced by FragmentComplete(), ReportProfile(), and UpdateStatus().

void impala::PlanFragmentExecutor::StopReportThread ( )

private

Stops report thread, if one is running. Blocks until report thread terminates. Idempotent.

Definition at line 441 of file plan-fragment-executor.cc.

References report_thread_, report_thread_active_, report_thread_lock_, and stop_report_thread_cv_.

Referenced by FragmentComplete(), and UpdateStatus().

void impala::PlanFragmentExecutor::UpdateStatus ( const Status & status )

private

If status_.ok(), sets status_ to status. If we're transitioning to an error status, stops report thread and sends a final report.

Definition at line 514 of file plan-fragment-executor.cc.

References impala::AtomicInt< T >::CompareAndSwap(), completed_report_sent_, impala::Status::IsMemLimitExceeded(), impala::Status::ok(), runtime_state_, SendReport(), status_, status_lock_, and StopReportThread().

Referenced by GetNext(), and Open().

Member Data Documentation

RuntimeProfile::Counter* impala::PlanFragmentExecutor::average_thread_tokens_

private

Average number of thread tokens for the duration of the plan fragment execution. Fragments that do a lot of cpu work (non-coordinator fragment) will have at least 1 token. Fragments that contain a hdfs scan node will have 1+ tokens depending on system load. Other nodes (e.g. hash join node) can also reserve additional tokens. This is a measure of how much CPU resources this fragment used during the course of the execution.

Definition at line 201 of file plan-fragment-executor.h.

Referenced by Prepare(), and ReleaseThreadToken().

bool impala::PlanFragmentExecutor::closed_

private

true if Close() has been called

Definition at line 162 of file plan-fragment-executor.h.

Referenced by Close().

AtomicInt<int> impala::PlanFragmentExecutor::completed_report_sent_

private

(Atomic) Flag that indicates whether a completed fragment report has been or will be fired. It is initialized to 0 and atomically swapped to 1 when a completed fragment report is about to be fired. Used for reducing the probability that a report is sent twice at the end of the fragment.

Definition at line 210 of file plan-fragment-executor.h.

Referenced by FragmentComplete(), ReportProfile(), and UpdateStatus().

bool impala::PlanFragmentExecutor::done_

private

true if plan_->GetNext() indicated that it's done

Definition at line 156 of file plan-fragment-executor.h.

Referenced by GetNext(), GetNextInternal(), and OpenInternal().

ExecEnv* impala::PlanFragmentExecutor::exec_env_

private

Definition at line 137 of file plan-fragment-executor.h.

Referenced by Close(), Prepare(), and ~PlanFragmentExecutor().

MonotonicStopWatch impala::PlanFragmentExecutor::fragment_sw_

private

Stopwatch for this entire fragment. Started in Prepare(), stopped in Close().

Definition at line 204 of file plan-fragment-executor.h.

Referenced by FragmentComplete(), and Prepare().

bool impala::PlanFragmentExecutor::has_thread_token_

private

true if this fragment has not returned the thread token to the thread resource mgr

Definition at line 165 of file plan-fragment-executor.h.

Referenced by Prepare(), and ReleaseThreadToken().

RuntimeProfile::TimeSeriesCounter* impala::PlanFragmentExecutor::mem_usage_sampled_counter_

private

Sampled memory usage at even time intervals.

Definition at line 213 of file plan-fragment-executor.h.

Referenced by Close(), and Prepare().

RuntimeProfile::Counter* impala::PlanFragmentExecutor::per_host_mem_usage_

private

A counter for the per query, per host peak mem usage. Note that this is not the max of the peak memory of all fragments running on a host since it needs to take into account when they are running concurrently. All fragments for a single query on a single host will have the same value for this counter.

Definition at line 189 of file plan-fragment-executor.h.

Referenced by Prepare(), and SendReport().

const string impala::PlanFragmentExecutor::PER_HOST_PEAK_MEM_COUNTER = "PerHostPeakMemUsage"

static

Name of the counter that is tracking per query, per host peak mem usage.

Definition at line 134 of file plan-fragment-executor.h.

Referenced by Prepare().

ExecNode* impala::PlanFragmentExecutor::plan_

private

Definition at line 138 of file plan-fragment-executor.h.

Referenced by Close(), GetNextInternal(), OpenInternal(), Prepare(), ReachedLimit(), and row_desc().

bool impala::PlanFragmentExecutor::prepared_

private

true if Prepare() returned OK

Definition at line 159 of file plan-fragment-executor.h.

Referenced by Cancel(), and Prepare().

TUniqueId impala::PlanFragmentExecutor::query_id_

private

Definition at line 139 of file plan-fragment-executor.h.

Referenced by GetNext(), Prepare(), PrintVolumeIds(), and ~PlanFragmentExecutor().

ReportStatusCallback impala::PlanFragmentExecutor::report_status_cb_

private

profile reporting-related

Definition at line 142 of file plan-fragment-executor.h.

Referenced by Open(), ReportProfile(), and SendReport().

boost::scoped_ptr<Thread> impala::PlanFragmentExecutor::report_thread_

private

Definition at line 143 of file plan-fragment-executor.h.

Referenced by Open(), and StopReportThread().

bool impala::PlanFragmentExecutor::report_thread_active_

private

Definition at line 153 of file plan-fragment-executor.h.

Referenced by Open(), ReportProfile(), StopReportThread(), and ~PlanFragmentExecutor().

boost::mutex impala::PlanFragmentExecutor::report_thread_lock_

private

Definition at line 144 of file plan-fragment-executor.h.

Referenced by Open(), ReportProfile(), and StopReportThread().

boost::condition_variable impala::PlanFragmentExecutor::report_thread_started_cv_

private

Indicates that profile reporting thread started. Tied to report_thread_lock_.

Definition at line 152 of file plan-fragment-executor.h.

Referenced by Open(), and ReportProfile().

boost::scoped_ptr<RowBatch> impala::PlanFragmentExecutor::row_batch_

private

Definition at line 182 of file plan-fragment-executor.h.

Referenced by Close(), GetNextInternal(), and Prepare().

RuntimeProfile::Counter* impala::PlanFragmentExecutor::rows_produced_counter_

private

Number of rows returned by this fragment.

Definition at line 192 of file plan-fragment-executor.h.

Referenced by GetNextInternal(), and Prepare().

boost::scoped_ptr<RuntimeState> impala::PlanFragmentExecutor::runtime_state_

private

Definition at line 181 of file plan-fragment-executor.h.

Referenced by Cancel(), Close(), desc_tbl(), FragmentComplete(), GetNext(), GetNextInternal(), obj_pool(), Open(), OpenInternal(), OptimizeLlvmModule(), Prepare(), profile(), ReleaseThreadToken(), ReportProfile(), runtime_state(), UpdateStatus(), and ~PlanFragmentExecutor().

boost::scoped_ptr<DataSink> impala::PlanFragmentExecutor::sink_

private

Output sink for rows sent to this fragment. May not be set, in which case rows are returned via GetNext's row batch Created in Prepare (if required), owned by this object.

Definition at line 180 of file plan-fragment-executor.h.

Referenced by Close(), OpenInternal(), and Prepare().

Status impala::PlanFragmentExecutor::status_

private

Overall execution status. Either ok() or set to the first error status that was encountered.

Definition at line 169 of file plan-fragment-executor.h.

Referenced by SendReport(), and UpdateStatus().

boost::mutex impala::PlanFragmentExecutor::status_lock_

private

Protects status_ lock ordering: