pub async fn query(
    executor: &Executor,
    data: Arc<QueryableBatch>,
    predicate: Predicate,
    selection: Selection<'_>,
) -> Result<SendableRecordBatchStream> {
    // Build logical plan for filtering data
    // Note that this query will also apply the delete predicates that go with the QueryableBatch

    let indices = match selection {
        Selection::All => None,
        Selection::Some(columns) => {
            let schema = data.schema();
            Some(
                columns
                    .iter()
                    .flat_map(|&column_name| schema.find_index_of(column_name))
                    .collect(),
            )
        }
    };

    let mut expr = vec![];
    if let Some(filter_expr) = predicate.filter_expr() {
        expr.push(filter_expr);
    }

    // TODO: Since we have different type of servers (router, ingester, compactor, and querier),
    // we may want to add more types into the ExecutorType to have better log and resource managment
    let ctx = executor.new_context(ExecutorType::Query);
    let logical_plan = ReorgPlanner::new()
        .scan_single_chunk_plan_with_filter(data.schema(), data, indices, expr)
        .context(LogicalPlanSnafu {})?;

    // Build physical plan
    let physical_plan = ctx
        .prepare_plan(&logical_plan)
        .await
        .context(PhysicalPlanSnafu {})?;

    // Execute the plan and return the filtered stream
    let output_stream = ctx
        .execute_stream(physical_plan)
        .await
        .context(ExecutePlanSnafu {})?;

    Ok(output_stream)
}
/// Query a given Queryable Batch, applying selection and filters as appropriate
/// Return stream of record batches
Query a given Queryable Batch, applying selection and filters as appropriate
Return stream of record batches
Query a given Queryable Batch applying selection and filters as appropriate Return stream of record batches
pub async fn query(
    executor: &Executor,
    data: Arc<QueryableBatch>,
    predicate: Predicate,
    selection: Selection<'_>,
) -> Result<SendableRecordBatchStream> {
    // Build logical plan for filtering data
    // Note that this query will also apply the delete predicates that go with the QueryableBatch
    let indices = match selection {
        Selection::All => None,
        Selection::Some(columns) => {
            let schema = data.schema();
            Some(
                columns
                    .iter()
                    .flat_map(|&column_name| schema.find_index_of(column_name))
                    .collect(),
            )
        }
    };
    let mut expr = vec![];
    if let Some(filter_expr) = predicate.filter_expr() {
        expr.push(filter_expr);
    }
    // TODO: Since we have different type of servers (router, ingester, compactor, and querier),
    // we may want to add more types into the ExecutorType to have better log and resource managment
    let ctx = executor.new_context(ExecutorType::Query);
    let logical_plan = ReorgPlanner::new()
        .scan_single_chunk_plan_with_filter(data.schema(), data, indices, expr)
        .context(LogicalPlanSnafu {})?;
    // Build physical plan
    let physical_plan = ctx
        .prepare_plan(&logical_plan)
        .await
        .context(PhysicalPlanSnafu {})?;
    // Execute the plan and return the filtered stream
    let output_stream = ctx
        .execute_stream(physical_plan)
        .await
        .context(ExecutePlanSnafu {})?;
    Ok(output_stream)
}
pub async fn query ( executor : & Executor , data : Arc < QueryableBatch > , predicate : Predicate , selection : Selection < ' _ > , ) -> Result < SendableRecordBatchStream > { // Build logical plan for filtering data // Note that this query will also apply the delete predicates that go with the QueryableBatch let indices = match selection { Selection :: All => None , Selection :: Some ( columns ) => { let schema = data . schema ( ) ; Some ( columns . iter ( ) . flat_map ( | & column_name | schema . find_index_of ( column_name ) ) . collect ( ) , ) } } ; let mut expr = vec ! [ ] ; if let Some ( filter_expr ) = predicate . filter_expr ( ) { expr . push ( filter_expr ) ; } // TODO: Since we have different type of servers (router, ingester, compactor, and querier), // we may want to add more types into the ExecutorType to have better log and resource managment let ctx = executor . new_context ( ExecutorType :: Query ) ; let logical_plan = ReorgPlanner :: new ( ) . scan_single_chunk_plan_with_filter ( data . schema ( ) , data , indices , expr ) . context ( LogicalPlanSnafu { } ) ? ; // Build physical plan let physical_plan = ctx . prepare_plan ( & logical_plan ) . await . context ( PhysicalPlanSnafu { } ) ? ; // Execute the plan and return the filtered stream let output_stream = ctx . execute_stream ( physical_plan ) . await . context ( ExecutePlanSnafu { } ) ? ; Ok ( output_stream ) }
Query a given Queryable Batch, applying selection and filters as appropriate
Return stream of record batches
Query a given Queryable Batch applying selection and filters as appropriate Return stream of record batches
executor
&Executor
data
Arc<QueryableBatch>
predicate
Predicate
selection
Selection<'_>
executor
&Executor
data
Arc<QueryableBatch>
predicate
Predicate
selection
Selection<'_>