async fn execute_query<'a>(&'a mut self, query: &'a str) -> Result<QueryResponse, CubeError> {
        let _start = SystemTime::now();

        let query = query.replace("SELECT FROM", "SELECT * FROM");

        let query_lower = query.to_lowercase();
        let query_lower = query_lower.replace("db.`", "");
        let query_lower = query_lower.replace("`", "");

        let ignore = match query_lower.as_str() {
            "rollback" => true,
            "commit" => true,
            _ => false,
        };

        if query_lower.eq("select cast('test plain returns' as char(60)) as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test plain returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if query_lower.eq("select cast('test unicode returns' as char(60)) as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test plain returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if query_lower.eq("select cast('test collated returns' as char character set utf8mb4) collate utf8mb4_bin as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test collated returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if !ignore {
            trace!("query was not detected");

            let ctx = self.server.transport
                .meta(&self.auth_context()?)
                .await?;

            let plan = convert_sql_to_cube_query(&query, Arc::new(ctx),self.state.clone())?;
            match plan {
                crate::compile::QueryPlan::MetaOk(status) => {
                    return Ok(QueryResponse::Ok(status));
                },
                crate::compile::QueryPlan::MetaTabular(status, data_frame) => {
                    return Ok(QueryResponse::ResultSet(status, data_frame));
                },
                crate::compile::QueryPlan::DataFushionSelect(status, plan, ctx) => {
                    let df = DataFrameImpl::new(
                        ctx.state,
                        &plan,
                    );
                    let batches = df.collect().await?;
                    let response =  batch_to_dataframe(&batches)?;

                    return Ok(QueryResponse::ResultSet(status, Arc::new(response)))
                },
                crate::compile::QueryPlan::CubeSelect(status, plan) => {
                    debug!("Request {}", json!(plan.request).to_string());
                    debug!("Meta {:?}", plan.meta);

                    let response = self.server.transport
                        .load(plan.request, &self.auth_context()?)
                        .await?;

                    let mut columns: Vec<dataframe::Column> = vec![];

                    for column_meta in &plan.meta {
                        columns.push(dataframe::Column::new(
                            column_meta.column_to.clone(),
                            column_meta.column_type,
                            ColumnFlags::empty(),
                        ));
                    }

                    let mut rows: Vec<dataframe::Row> = vec![];

                    if let Some(result) = response.results.first() {
                        debug!("Columns {:?}", columns);
                        debug!("Hydration mapping {:?}", plan.meta);
                        trace!("Response from Cube.js {:?}", result.data);

                        for row in result.data.iter() {
                            if let Some(record) = row.as_object() {
                                rows.push(
                                    dataframe::Row::hydrate_from_response(&plan.meta, record)
                                );
                            } else {
                                error!(
                                    "Unable to map row to DataFrame::Row: {:?}, skipping row",
                                    row
                                );
                            }
                        }

                        return Ok(QueryResponse::ResultSet(status, Arc::new(dataframe::DataFrame::new(
                            columns,
                            rows
                        ))));
                    } else {
                        return Ok(QueryResponse::ResultSet(status, Arc::new(dataframe::DataFrame::new(vec![], vec![]))));
                    }
                }
            }
        }

        if ignore {
            Ok(QueryResponse::ResultSet(
                StatusFlags::empty(),
                Arc::new(dataframe::DataFrame::new(vec![], vec![])),
            ))
        } else {
            Err(CubeError::internal("Unsupported query".to_string()))
        }
    }
// This method executes query and return it as DataFrame
This method executes query and return it as DataFrame
This method executes query and return it as DataFrame
async fn execute_query<'a>(&'a mut self, query: &'a str) -> Result<QueryResponse, CubeError> {
        let _start = SystemTime::now();
        let query = query.replace("SELECT FROM", "SELECT * FROM");
        let query_lower = query.to_lowercase();
        let query_lower = query_lower.replace("db.`", "");
        let query_lower = query_lower.replace("`", "");
        let ignore = match query_lower.as_str() {
            "rollback" => true,
            "commit" => true,
            _ => false,
        };
        if query_lower.eq("select cast('test plain returns' as char(60)) as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test plain returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if query_lower.eq("select cast('test unicode returns' as char(60)) as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test plain returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if query_lower.eq("select cast('test collated returns' as char character set utf8mb4) collate utf8mb4_bin as anon_1") {
            return Ok(
                QueryResponse::ResultSet(StatusFlags::empty(), Arc::new(
                    dataframe::DataFrame::new(
                        vec![dataframe::Column::new(
                            "anon_1".to_string(),
                            ColumnType::MYSQL_TYPE_STRING,
                            ColumnFlags::empty(),
                        )],
                        vec![dataframe::Row::new(vec![
                            dataframe::TableValue::String("test collated returns".to_string())
                        ])]
                    )
                ),)
            )
        } else if !ignore {
            trace!("query was not detected");
            let ctx = self.server.transport
                .meta(&self.auth_context()?)
                .await?;
            let plan = convert_sql_to_cube_query(&query, Arc::new(ctx),self.state.clone())?;
            match plan {
                crate::compile::QueryPlan::MetaOk(status) => {
                    return Ok(QueryResponse::Ok(status));
                },
                crate::compile::QueryPlan::MetaTabular(status, data_frame) => {
                    return Ok(QueryResponse::ResultSet(status, data_frame));
                },
                crate::compile::QueryPlan::DataFushionSelect(status, plan, ctx) => {
                    let df = DataFrameImpl::new(
                        ctx.state,
                        &plan,
                    );
                    let batches = df.collect().await?;
                    let response =  batch_to_dataframe(&batches)?;
                    return Ok(QueryResponse::ResultSet(status, Arc::new(response)))
                },
                crate::compile::QueryPlan::CubeSelect(status, plan) => {
                    debug!("Request {}", json!(plan.request).to_string());
                    debug!("Meta {:?}", plan.meta);
                    let response = self.server.transport
                        .load(plan.request, &self.auth_context()?)
                        .await?;
                    let mut columns: Vec<dataframe::Column> = vec![];
                    for column_meta in &plan.meta {
                        columns.push(dataframe::Column::new(
                            column_meta.column_to.clone(),
                            column_meta.column_type,
                            ColumnFlags::empty(),
                        ));
                    }
                    let mut rows: Vec<dataframe::Row> = vec![];
                    if let Some(result) = response.results.first() {
                        debug!("Columns {:?}", columns);
                        debug!("Hydration mapping {:?}", plan.meta);
                        trace!("Response from Cube.js {:?}", result.data);
                        for row in result.data.iter() {
                            if let Some(record) = row.as_object() {
                                rows.push(
                                    dataframe::Row::hydrate_from_response(&plan.meta, record)
                                );
                            } else {
                                error!(
                                    "Unable to map row to DataFrame::Row: {:?}, skipping row",
                                    row
                                );
                            }
                        }
                        return Ok(QueryResponse::ResultSet(status, Arc::new(dataframe::DataFrame::new(
                            columns,
                            rows
                        ))));
                    } else {
                        return Ok(QueryResponse::ResultSet(status, Arc::new(dataframe::DataFrame::new(vec![], vec![]))));
                    }
                }
            }
        }
        if ignore {
            Ok(QueryResponse::ResultSet(
                StatusFlags::empty(),
                Arc::new(dataframe::DataFrame::new(vec![], vec![])),
            ))
        } else {
            Err(CubeError::internal("Unsupported query".to_string()))
        }
    }
async fn execute_query < ' a > ( & ' a mut self , query : & ' a str ) -> Result < QueryResponse , CubeError > { let _start = SystemTime :: now ( ) ; let query = query . replace ( " " , " " ) ; let query_lower = query . to_lowercase ( ) ; let query_lower = query_lower . replace ( " " , " " ) ; let query_lower = query_lower . replace ( " " , " " ) ; let ignore = match query_lower . as_str ( ) { " " => true , " " => true , _ => false , } ; if query_lower . eq ( " " ) { return Ok ( QueryResponse :: ResultSet ( StatusFlags :: empty ( ) , Arc :: new ( dataframe :: DataFrame :: new ( vec ! [ dataframe Column new ( " " to_string ( ) ColumnType MYSQL_TYPE_STRING ColumnFlags empty ( ) ) ] , vec ! [ dataframe Row new ( vec [ dataframe TableValue String ( " " to_string ( ) ) ] ) ] ) ) , ) ) } else if query_lower . eq ( " " ) { return Ok ( QueryResponse :: ResultSet ( StatusFlags :: empty ( ) , Arc :: new ( dataframe :: DataFrame :: new ( vec ! [ dataframe Column new ( " " to_string ( ) ColumnType MYSQL_TYPE_STRING ColumnFlags empty ( ) ) ] , vec ! [ dataframe Row new ( vec [ dataframe TableValue String ( " " to_string ( ) ) ] ) ] ) ) , ) ) } else if query_lower . eq ( " " ) { return Ok ( QueryResponse :: ResultSet ( StatusFlags :: empty ( ) , Arc :: new ( dataframe :: DataFrame :: new ( vec ! [ dataframe Column new ( " " to_string ( ) ColumnType MYSQL_TYPE_STRING ColumnFlags empty ( ) ) ] , vec ! [ dataframe Row new ( vec [ dataframe TableValue String ( " " to_string ( ) ) ] ) ] ) ) , ) ) } else if ! ignore { trace ! ( " " ) ; let ctx = self . server . transport . meta ( & self . auth_context ( ) ? ) . await ? ; let plan = convert_sql_to_cube_query ( & query , Arc :: new ( ctx ) , self . state . clone ( ) ) ? ; match plan { crate :: compile :: QueryPlan :: MetaOk ( status ) => { return Ok ( QueryResponse :: Ok ( status ) ) ; } , crate :: compile :: QueryPlan :: MetaTabular ( status , data_frame ) => { return Ok ( QueryResponse :: ResultSet ( status , data_frame ) ) ; } , crate :: compile :: QueryPlan :: DataFushionSelect ( status , plan , ctx ) => { let df = DataFrameImpl :: new ( ctx . state , & plan , ) ; let batches = df . collect ( ) . await ? ; let response = batch_to_dataframe ( & batches ) ? ; return Ok ( QueryResponse :: ResultSet ( status , Arc :: new ( response ) ) ) } , crate :: compile :: QueryPlan :: CubeSelect ( status , plan ) => { debug ! ( " " json ( plan request ) to_string ( ) ) ; debug ! ( " " plan meta ) ; let response = self . server . transport . load ( plan . request , & self . auth_context ( ) ? ) . await ? ; let mut columns : Vec < dataframe :: Column > = vec ! [ ] ; for column_meta in & plan . meta { columns . push ( dataframe :: Column :: new ( column_meta . column_to . clone ( ) , column_meta . column_type , ColumnFlags :: empty ( ) , ) ) ; } let mut rows : Vec < dataframe :: Row > = vec ! [ ] ; if let Some ( result ) = response . results . first ( ) { debug ! ( " " columns ) ; debug ! ( " " plan meta ) ; trace ! ( " " result data ) ; for row in result . data . iter ( ) { if let Some ( record ) = row . as_object ( ) { rows . push ( dataframe :: Row :: hydrate_from_response ( & plan . meta , record ) ) ; } else { error ! ( " " row ) ; } } return Ok ( QueryResponse :: ResultSet ( status , Arc :: new ( dataframe :: DataFrame :: new ( columns , rows ) ) ) ) ; } else { return Ok ( QueryResponse :: ResultSet ( status , Arc :: new ( dataframe :: DataFrame :: new ( vec ! [ ] , vec ! [ ] ) ) ) ) ; } } } } if ignore { Ok ( QueryResponse :: ResultSet ( StatusFlags :: empty ( ) , Arc :: new ( dataframe :: DataFrame :: new ( vec ! [ ] , vec ! [ ] ) ) , ) ) } else { Err ( CubeError :: internal ( " " . to_string ( ) ) ) } }
This method executes query and return it as DataFrame
This method executes query and return it as DataFrame
&'a mut self
query
&'a str
&'a mut self
query
&'a str