Struct StageRunner

struct StageRunner {
    epoch: BatchQueryEpoch,
    state: Arc<RwLock<StageState>>,
    stage: Arc<QueryStage>,
    worker_node_manager: WorkerNodeSelector,
    tasks: Arc<HashMap<u64, TaskStatusHolder>>,
    msg_sender: Sender<QueryMessage>,
    children: Vec<Arc<StageExecution>>,
    compute_client_pool: ComputeClientPoolRef,
    catalog_reader: CatalogReader,
    ctx: Arc<ExecutionContext>,
}

Fields

epoch: BatchQueryEpoch

state: Arc<RwLock<StageState>>

stage: Arc<QueryStage>

worker_node_manager: WorkerNodeSelector

tasks: Arc<HashMap<u64, TaskStatusHolder>>

msg_sender: Sender<QueryMessage>

children: Vec<Arc<StageExecution>>

compute_client_pool: ComputeClientPoolRef

catalog_reader: CatalogReader

ctx: Arc<ExecutionContext>

Implementations

impl StageRunner

async fn run(self, shutdown_rx: ShutdownToken)

async fn send_event(&self, event: QueryMessage)

Send stage event to listener.

async fn schedule_tasks( &mut self, shutdown_rx: ShutdownToken, expr_context: ExprContext, ) -> Result<(), SchedulerError>

Schedule all tasks to CN and wait process all status messages from RPC. Note that when all task is created, it should tell QueryRunner to schedule next.

async fn schedule_tasks_for_root( &mut self, shutdown_rx: ShutdownToken, expr_context: ExprContext, ) -> Result<(), SchedulerError>

async fn schedule_tasks_for_all( &mut self, shutdown_rx: ShutdownToken, ) -> Result<(), SchedulerError>

fn get_fragment_id(&self, table_id: &TableId) -> Result<u32, SchedulerError>

fn get_table_dml_vnode_mapping( &self, table_id: &TableId, ) -> Result<WorkerSlotMapping, SchedulerError>

fn choose_worker( &self, plan_fragment: &PlanFragment, task_id: u32, dml_table_id: Option<TableId>, ) -> Result<Option<WorkerNode>, SchedulerError>

fn find_distributed_lookup_join_node( plan_node: &PlanNode, ) -> Option<&DistributedLookupJoinNode>

async fn notify_stage_scheduled(&self, msg: QueryMessage)

Write message into channel to notify query runner current stage have been scheduled.

async fn notify_stage_completed(&self)

Notify query execution that this stage completed.

async fn notify_stage_state_changed<F>(&self, new_state: F, msg: QueryMessage)

where F: FnOnce(StageState) -> StageState,

async fn cancel_all_scheducancled_tasks(&self) -> Result<(), SchedulerError>

Abort all registered tasks. Note that here we do not care which part of tasks has already failed or completed, cuz the abort task will not fail if the task has already die. See PR (#4560).

async fn schedule_task( &self, task_id: PbTaskId, plan_fragment: PlanFragment, worker: Option<WorkerNode>, expr_context: ExprContext, ) -> Result<Fuse<Streaming<TaskInfoResponse>>, SchedulerError>

pub fn create_plan_fragment( &self, task_id: u64, partition: Option<PartitionInfo>, ) -> PlanFragment

fn convert_plan_node( &self, execution_plan_node: &ExecutionPlanNode, task_id: u64, partition: Option<PartitionInfo>, identity_id: Rc<RefCell<u64>>, ) -> PbPlanNode

fn is_root_stage(&self) -> bool

fn mask_failed_serving_worker(&self, worker: &WorkerNode)