async fn execute_query_stage(
    job_uuid: &Uuid,
    stage_id: usize,
    plan: Arc<dyn ExecutionPlan>,
    executors: Vec<ExecutorMeta>,
) -> Result<Vec<PartitionLocation>> {
    info!("execute_query_stage() stage_id={}", stage_id);
    pretty_print(plan.clone(), 0);

    let partition_count = plan.output_partitioning().partition_count();
    let mut meta = Vec::with_capacity(partition_count);

    // TODO make this concurrent by executing all partitions at once instead of one at a time

    for child_partition in 0..partition_count {
        debug!(
            "execute_query_stage() stage_id={}, partition_id={}",
            stage_id, child_partition
        );
        let executor_meta = &executors[child_partition % executors.len()];

        let mut client = BallistaClient::try_new(&executor_meta.host, executor_meta.port).await?;

        let _partition_metadata = client
            .execute_partition(*job_uuid, stage_id, child_partition, plan.clone())
            .await?;
        meta.push(PartitionLocation {
            partition_id: PartitionId::new(*job_uuid, stage_id, child_partition),
            executor_meta: executor_meta.clone(),
        });
    }

    debug!(
        "execute_query_stage() stage_id={} produced {:?}",
        stage_id, meta
    );

    Ok(meta)
}
/// Execute a query stage by sending each partition to an executor
Execute a query stage by sending each partition to an executor
Execute a query stage by sending each partition to an executor
async fn execute_query_stage(
    job_uuid: &Uuid,
    stage_id: usize,
    plan: Arc<dyn ExecutionPlan>,
    executors: Vec<ExecutorMeta>,
) -> Result<Vec<PartitionLocation>> {
    info!("execute_query_stage() stage_id={}", stage_id);
    pretty_print(plan.clone(), 0);
    let partition_count = plan.output_partitioning().partition_count();
    let mut meta = Vec::with_capacity(partition_count);
    // TODO make this concurrent by executing all partitions at once instead of one at a time
    for child_partition in 0..partition_count {
        debug!(
            "execute_query_stage() stage_id={}, partition_id={}",
            stage_id, child_partition
        );
        let executor_meta = &executors[child_partition % executors.len()];
        let mut client = BallistaClient::try_new(&executor_meta.host, executor_meta.port).await?;
        let _partition_metadata = client
            .execute_partition(*job_uuid, stage_id, child_partition, plan.clone())
            .await?;
        meta.push(PartitionLocation {
            partition_id: PartitionId::new(*job_uuid, stage_id, child_partition),
            executor_meta: executor_meta.clone(),
        });
    }
    debug!(
        "execute_query_stage() stage_id={} produced {:?}",
        stage_id, meta
    );
    Ok(meta)
}
async fn execute_query_stage ( job_uuid : & Uuid , stage_id : usize , plan : Arc < dyn ExecutionPlan > , executors : Vec < ExecutorMeta > , ) -> Result < Vec < PartitionLocation > > { info ! ( " " stage_id ) ; pretty_print ( plan . clone ( ) , 0 ) ; let partition_count = plan . output_partitioning ( ) . partition_count ( ) ; let mut meta = Vec :: with_capacity ( partition_count ) ; // TODO make this concurrent by executing all partitions at once instead of one at a time for child_partition in 0 .. partition_count { debug ! ( " " stage_id child_partition ) ; let executor_meta = & executors [ child_partition % executors . len ( ) ] ; let mut client = BallistaClient :: try_new ( & executor_meta . host , executor_meta . port ) . await ? ; let _partition_metadata = client . execute_partition ( * job_uuid , stage_id , child_partition , plan . clone ( ) ) . await ? ; meta . push ( PartitionLocation { partition_id : PartitionId :: new ( * job_uuid , stage_id , child_partition ) , executor_meta : executor_meta . clone ( ) , } ) ; } debug ! ( " " stage_id meta ) ; Ok ( meta ) }
Execute a query stage by sending each partition to an executor
Execute a query stage by sending each partition to an executor
job_uuid
&Uuid
stage_id
usize
plan
Arc<dyn ExecutionPlan>
executors
Vec<ExecutorMeta>
job_uuid
&Uuid
stage_id
usize
plan
Arc<dyn ExecutionPlan>
executors
Vec<ExecutorMeta>