/// Executes the hcomb plan
/// Returns the query id and the result stream
Executes the hcomb plan
Returns the query id and the result stream
Executes the hcomb plan Returns the query id and the result stream
pub async fn execute_query(
        &self,
        provider_desc: HCombTableDesc,
        sql: String,
        source: String,
    ) -> Result<(String, SendableRecordBatchStream)> {
        println!("[hcomb] execute query...");
        let mut execution_context =
            ExecutionContext::with_config(self.execution_config.clone());
        self.last_query
            .store(chrono::Utc::now().timestamp(), Ordering::Relaxed);
        let query_id = provider_desc.query_id().to_owned();
        let batch_stream = self
            .results_service
            .new_query(query_id.clone(), provider_desc.nb_hbee());
        let provider = HCombTable::new(provider_desc, Box::pin(batch_stream));
        let physical_plan;
        {
            // limit scope of df because not send so should not overlab await
            let source_ref: &str = &source;
            execution_context.register_table(source_ref, Arc::new(provider))?;
            let df = execution_context.sql(&sql)?;
            let plan = df.to_logical_plan();
            physical_plan = execution_context.create_physical_plan(&plan)?;
        }
        // if necessary, merge the partitions
        let query_res = match physical_plan.output_partitioning().partition_count() {
            0 => Err(internal_err!("Should have at least one partition")),
            1 => physical_plan.execute(0).await.map_err(|e| e.into()),
            _ => {
                // merge into a single partition
                let physical_plan = MergeExec::new(physical_plan.clone());
                assert_eq!(1, physical_plan.output_partitioning().partition_count());
                physical_plan.execute(0).await.map_err(|e| e.into())
            }
        };
        query_res.map(|res| (query_id.clone(), res))
    }
pub async fn execute_query ( & self , provider_desc : HCombTableDesc , sql : String , source : String , ) -> Result < ( String , SendableRecordBatchStream ) > { println ! ( " " ) ; let mut execution_context = ExecutionContext :: with_config ( self . execution_config . clone ( ) ) ; self . last_query . store ( chrono :: Utc :: now ( ) . timestamp ( ) , Ordering :: Relaxed ) ; let query_id = provider_desc . query_id ( ) . to_owned ( ) ; let batch_stream = self . results_service . new_query ( query_id . clone ( ) , provider_desc . nb_hbee ( ) ) ; let provider = HCombTable :: new ( provider_desc , Box :: pin ( batch_stream ) ) ; let physical_plan ; { // limit scope of df because not send so should not overlab await let source_ref : & str = & source ; execution_context . register_table ( source_ref , Arc :: new ( provider ) ) ? ; let df = execution_context . sql ( & sql ) ? ; let plan = df . to_logical_plan ( ) ; physical_plan = execution_context . create_physical_plan ( & plan ) ? ; } // if necessary, merge the partitions let query_res = match physical_plan . output_partitioning ( ) . partition_count ( ) { 0 => Err ( internal_err ! ( " " ) ) , 1 => physical_plan . execute ( 0 ) . await . map_err ( | e | e . into ( ) ) , _ => { // merge into a single partition let physical_plan = MergeExec :: new ( physical_plan . clone ( ) ) ; assert_eq ! ( 1 physical_plan output_partitioning ( ) partition_count ( ) ) ; physical_plan . execute ( 0 ) . await . map_err ( | e | e . into ( ) ) } } ; query_res . map ( | res | ( query_id . clone ( ) , res ) ) }
Executes the hcomb plan
Returns the query id and the result stream
Executes the hcomb plan Returns the query id and the result stream
&self
provider_desc
HCombTableDesc
sql
String
source
String