fn execute(plan: Arc<dyn ExecutionPlan>, executors: Vec<ExecutorMeta>) -> SendableExecutionPlan {
    Box::pin(async move {
        debug!("execute() {}", &format!("{:?}", plan)[0..60]);
        // execute children first
        let mut children: Vec<Arc<dyn ExecutionPlan>> = vec![];
        for child in plan.children() {
            let executed_child = execute(child.clone(), executors.clone()).await?;
            children.push(executed_child);
        }
        let plan = plan.with_new_children(children)?;

        let new_plan: Arc<dyn ExecutionPlan> = if plan.as_any().is::<QueryStageExec>() {
            let stage = plan.as_any().downcast_ref::<QueryStageExec>().unwrap();
            let partition_locations = execute_query_stage(
                &stage.job_uuid.clone(),
                stage.stage_id,
                stage.children()[0].clone(),
                executors.clone(),
            )
            .await?;

            // replace the query stage with a ShuffleReaderExec that can read the partitions
            // produced by the executed query stage
            let shuffle_reader = ShuffleReaderExec::try_new(partition_locations, stage.schema())?;
            Arc::new(shuffle_reader)
        } else {
            plan
        };

        debug!("execute is returning:");
        pretty_print(new_plan.clone(), 0);

        Ok(new_plan)
    })
}
/// Visitor pattern to walk the plan, depth-first, and then execute query stages when walking
/// up the tree
Visitor pattern to walk the plan, depth-first, and then execute query stages when walking
up the tree
Visitor pattern to walk the plan depth - first and then execute query stages when walking up the tree
fn execute(plan: Arc<dyn ExecutionPlan>, executors: Vec<ExecutorMeta>) -> SendableExecutionPlan {
    Box::pin(async move {
        debug!("execute() {}", &format!("{:?}", plan)[0..60]);
        // execute children first
        let mut children: Vec<Arc<dyn ExecutionPlan>> = vec![];
        for child in plan.children() {
            let executed_child = execute(child.clone(), executors.clone()).await?;
            children.push(executed_child);
        }
        let plan = plan.with_new_children(children)?;
        let new_plan: Arc<dyn ExecutionPlan> = if plan.as_any().is::<QueryStageExec>() {
            let stage = plan.as_any().downcast_ref::<QueryStageExec>().unwrap();
            let partition_locations = execute_query_stage(
                &stage.job_uuid.clone(),
                stage.stage_id,
                stage.children()[0].clone(),
                executors.clone(),
            )
            .await?;
            // replace the query stage with a ShuffleReaderExec that can read the partitions
            // produced by the executed query stage
            let shuffle_reader = ShuffleReaderExec::try_new(partition_locations, stage.schema())?;
            Arc::new(shuffle_reader)
        } else {
            plan
        };
        debug!("execute is returning:");
        pretty_print(new_plan.clone(), 0);
        Ok(new_plan)
    })
}
fn execute ( plan : Arc < dyn ExecutionPlan > , executors : Vec < ExecutorMeta > ) -> SendableExecutionPlan { Box :: pin ( async move { debug ! ( " " format ( " " plan ) [ 0 60 ] ) ; // execute children first let mut children : Vec < Arc < dyn ExecutionPlan > > = vec ! [ ] ; for child in plan . children ( ) { let executed_child = execute ( child . clone ( ) , executors . clone ( ) ) . await ? ; children . push ( executed_child ) ; } let plan = plan . with_new_children ( children ) ? ; let new_plan : Arc < dyn ExecutionPlan > = if plan . as_any ( ) . is :: < QueryStageExec > ( ) { let stage = plan . as_any ( ) . downcast_ref :: < QueryStageExec > ( ) . unwrap ( ) ; let partition_locations = execute_query_stage ( & stage . job_uuid . clone ( ) , stage . stage_id , stage . children ( ) [ 0 ] . clone ( ) , executors . clone ( ) , ) . await ? ; // replace the query stage with a ShuffleReaderExec that can read the partitions // produced by the executed query stage let shuffle_reader = ShuffleReaderExec :: try_new ( partition_locations , stage . schema ( ) ) ? ; Arc :: new ( shuffle_reader ) } else { plan } ; debug ! ( " " ) ; pretty_print ( new_plan . clone ( ) , 0 ) ; Ok ( new_plan ) } ) }
Visitor pattern to walk the plan, depth-first, and then execute query stages when walking
up the tree
Visitor pattern to walk the plan depth - first and then execute query stages when walking up the tree
plan
Arc<dyn ExecutionPlan>
executors
Vec<ExecutorMeta>
plan
Arc<dyn ExecutionPlan>
executors
Vec<ExecutorMeta>