/// Execute the logical plan and collect the results
/// Collecting the results might increase latency and mem consumption but:
/// - reduces connection duration from hbee to hcomb, thus decreasing load on hcomb
/// - allows to collect exec errors at once, effectively choosing between do_put and FAIL action
Execute the logical plan and collect the results
Collecting the results might increase latency and mem consumption but:
reduces connection duration from hbee to hcomb, thus decreasing load on hcomb
allows to collect exec errors at once, effectively choosing between do_put and FAIL action
Execute the logical plan and collect the results Collecting the results might increase latency and mem consumption but : reduces connection duration from hbee to hcomb thus decreasing load on hcomb allows to collect exec errors at once effectively choosing between do_put and FAIL action
async fn query(&self,table: HBeeTableDesc, sql: String, source: String) -> Result<Vec<RecordBatch>> {
        let mut execution_context = ExecutionContext::with_config(self.execution_config.clone());
        let start = Instant::now();
        let provider = HBeeTable::new(Arc::new(table), Arc::clone(&self.range_cache));
        let source_ref: &str = &source;
        execution_context
            .register_table(source_ref, Arc::new(provider))?;
        let physical_plan;
        {
            let df = execution_context.sql(&sql)?;
            let plan = df.to_logical_plan();
            let plan = execution_context.optimize(&plan)?;
            physical_plan = execution_context.create_physical_plan(&plan)?;
        }
        println!(
            "[hbee] planning duration: {}, partitions: {}",
            start.elapsed().as_millis(),
            physical_plan.output_partitioning().partition_count()
        );
        // if necessary, merge the partitions
        let merged_plan = match physical_plan.output_partitioning().partition_count() {
            0 => Err(internal_err!("Should have at least one partition"))?,
            1 => physical_plan,
            _ => {
                // merge into a single partition
                let physical_plan = MergeExec::new(physical_plan.clone());
                assert_eq!(1, physical_plan.output_partitioning().partition_count());
                Arc::new(physical_plan)
            }
        };
        datafusion::physical_plan::collect(merged_plan)
            .await
            .map_err(|e| e.into())
    }
async fn query ( & self , table : HBeeTableDesc , sql : String , source : String ) -> Result < Vec < RecordBatch > > { let mut execution_context = ExecutionContext :: with_config ( self . execution_config . clone ( ) ) ; let start = Instant :: now ( ) ; let provider = HBeeTable :: new ( Arc :: new ( table ) , Arc :: clone ( & self . range_cache ) ) ; let source_ref : & str = & source ; execution_context . register_table ( source_ref , Arc :: new ( provider ) ) ? ; let physical_plan ; { let df = execution_context . sql ( & sql ) ? ; let plan = df . to_logical_plan ( ) ; let plan = execution_context . optimize ( & plan ) ? ; physical_plan = execution_context . create_physical_plan ( & plan ) ? ; } println ! ( " " start elapsed ( ) as_millis ( ) physical_plan output_partitioning ( ) partition_count ( ) ) ; // if necessary, merge the partitions let merged_plan = match physical_plan . output_partitioning ( ) . partition_count ( ) { 0 => Err ( internal_err ! ( " " ) ) ? , 1 => physical_plan , _ => { // merge into a single partition let physical_plan = MergeExec :: new ( physical_plan . clone ( ) ) ; assert_eq ! ( 1 physical_plan output_partitioning ( ) partition_count ( ) ) ; Arc :: new ( physical_plan ) } } ; datafusion :: physical_plan :: collect ( merged_plan ) . await . map_err ( | e | e . into ( ) ) }
Execute the logical plan and collect the results
Collecting the results might increase latency and mem consumption but:
 reduces connection duration from hbee to hcomb, thus decreasing load on hcomb
 allows to collect exec errors at once, effectively choosing between do_put and FAIL action
Execute the logical plan and collect the results Collecting the results might increase latency and mem consumption but : reduces connection duration from hbee to hcomb thus decreasing load on hcomb allows to collect exec errors at once effectively choosing between do_put and FAIL action
&self
table
HBeeTableDesc
sql
String
source
String