Struct BatchPlanFragmenter

pub struct BatchPlanFragmenter {
    query_id: QueryId,
    next_stage_id: u32,
    worker_node_manager: WorkerNodeSelector,
    catalog_reader: CatalogReader,
    batch_parallelism: usize,
    stage_graph_builder: Option<StageGraphBuilder>,
    stage_graph: Option<StageGraph>,
}

BatchPlanFragmenter splits a query plan into fragments.

Fields

query_id: QueryId

next_stage_id: u32

worker_node_manager: WorkerNodeSelector

catalog_reader: CatalogReader

batch_parallelism: usize

stage_graph_builder: Option<StageGraphBuilder>

stage_graph: Option<StageGraph>

Implementations

impl BatchPlanFragmenter

pub fn new( worker_node_manager: WorkerNodeSelector, catalog_reader: CatalogReader, batch_parallelism: Option<NonZeroU64>, batch_node: PlanRef, ) -> Result<Self, SchedulerError>

fn split_into_stage( &mut self, batch_node: PlanRef, ) -> Result<(), SchedulerError>

Split the plan node into each stages, based on exchange node.

impl BatchPlanFragmenter

pub async fn generate_complete_query(self) -> Result<Query, SchedulerError>

After split, the stage_graph in the framenter may has the stage with incomplete source info, we need to fetch the source info to complete the stage in this function. Why separate this two step(split() and generate_complete_query())? The step of fetching source info is a async operation so that we can’t do it in the split step.

fn new_stage( &mut self, root: PlanRef, exchange_info: Option<ExchangeInfo>, ) -> Result<Arc<QueryStage>, SchedulerError>

fn visit_node( &mut self, node: PlanRef, builder: &mut QueryStageBuilder, parent_exec_node: Option<&mut ExecutionPlanNode>, ) -> Result<(), SchedulerError>

fn visit_exchange( &mut self, node: PlanRef, builder: &mut QueryStageBuilder, parent_exec_node: Option<&mut ExecutionPlanNode>, ) -> Result<(), SchedulerError>

fn collect_stage_source( node: PlanRef, ) -> Result<Option<SourceScanInfo>, SchedulerError>

Check whether this stage contains a source node. If so, use SplitEnumeratorImpl to get the split info from exteneral source.

For current implementation, we can guarantee that each stage has only one source.

fn collect_stage_file_scan( node: PlanRef, ) -> Result<Option<FileScanInfo>, SchedulerError>

fn collect_stage_table_scan( &self, node: PlanRef, ) -> Result<Option<TableScanInfo>, SchedulerError>

Check whether this stage contains a table scan node and the table’s information if so.

If there are multiple scan nodes in this stage, they must have the same distribution, but maybe different vnodes partition. We just use the same partition for all the scan nodes.

fn collect_dml_table_id(node: &PlanRef) -> Option<TableId>

Returns the dml table id if any.

fn collect_stage_lookup_join_parallelism( &self, node: PlanRef, ) -> Result<Option<usize>, SchedulerError>