refactor: query config options (#6781)

* feat: refactor columnar and vector conversion

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* feat: initialize config options from query context

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix: failure tests

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* chore: revert ColumnarValue::try_from_vector

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

src/flow/src/batching_mode/task.rs

@@ -714,7 +714,7 @@ impl BatchingTask {
             })?
             .data;
         // only apply optimize after complex rewrite is done
-        let new_plan = apply_df_optimizer(rewrite).await?;
+        let new_plan = apply_df_optimizer(rewrite, &query_ctx).await?;
 
         let info = PlanInfo {
             plan: new_plan.clone(),

src/flow/src/batching_mode/utils.rs

@@ -122,13 +122,13 @@ pub async fn sql_to_df_plan(
     };
     let plan = engine
         .planner()
-        .plan(&query_stmt, query_ctx)
+        .plan(&query_stmt, query_ctx.clone())
         .await
         .map_err(BoxedError::new)
         .context(ExternalSnafu)?;
 
     let plan = if optimize {
-        apply_df_optimizer(plan).await?
+        apply_df_optimizer(plan, &query_ctx).await?
     } else {
         plan
     };

src/flow/src/df_optimizer.rs

@@ -44,6 +44,7 @@ use query::optimizer::count_wildcard::CountWildcardToTimeIndexRule;
 use query::parser::QueryLanguageParser;
 use query::query_engine::DefaultSerializer;
 use query::QueryEngine;
+use session::context::QueryContextRef;
 use snafu::ResultExt;
 /// note here we are using the `substrait_proto_df` crate from the `substrait` module and
 /// rename it to `substrait_proto`
@@ -57,8 +58,9 @@ use crate::plan::TypedPlan;
 // TODO(discord9): use `Analyzer` to manage rules if more `AnalyzerRule` is needed
 pub async fn apply_df_optimizer(
     plan: datafusion_expr::LogicalPlan,
+    query_ctx: &QueryContextRef,
 ) -> Result<datafusion_expr::LogicalPlan, Error> {
-    let cfg = ConfigOptions::new();
+    let cfg = query_ctx.create_config_options();
     let analyzer = Analyzer::with_rules(vec![
         Arc::new(CountWildcardToTimeIndexRule),
         Arc::new(AvgExpandRule),
@@ -107,12 +109,12 @@ pub async fn sql_to_flow_plan(
         .context(ExternalSnafu)?;
     let plan = engine
         .planner()
-        .plan(&stmt, query_ctx)
+        .plan(&stmt, query_ctx.clone())
         .await
         .map_err(BoxedError::new)
         .context(ExternalSnafu)?;
 
-    let opted_plan = apply_df_optimizer(plan).await?;
+    let opted_plan = apply_df_optimizer(plan, &query_ctx).await?;
 
     // TODO(discord9): add df optimization
     let sub_plan = DFLogicalSubstraitConvertor {}

src/flow/src/test_utils.rs

@@ -172,7 +172,9 @@ pub async fn sql_to_substrait(engine: Arc<dyn QueryEngine>, sql: &str) -> proto:
         .plan(&stmt, QueryContext::arc())
         .await
         .unwrap();
-    let plan = apply_df_optimizer(plan).await.unwrap();
+    let plan = apply_df_optimizer(plan, &QueryContext::arc())
+        .await
+        .unwrap();
 
     // encode then decode so to rely on the impl of conversion from logical plan to substrait plan
     let bytes = DFLogicalSubstraitConvertor {}

src/flow/src/transform.rs

@@ -293,7 +293,9 @@ mod test {
             .plan(&stmt, QueryContext::arc())
             .await
             .unwrap();
-        let plan = apply_df_optimizer(plan).await.unwrap();
+        let plan = apply_df_optimizer(plan, &QueryContext::arc())
+            .await
+            .unwrap();
 
         // encode then decode so to rely on the impl of conversion from logical plan to substrait plan
         let bytes = DFLogicalSubstraitConvertor {}
@@ -315,7 +317,7 @@ mod test {
             .plan(&stmt, QueryContext::arc())
             .await
             .unwrap();
-        let plan = apply_df_optimizer(plan).await;
+        let plan = apply_df_optimizer(plan, &QueryContext::arc()).await;
 
         assert!(plan.is_err());
     }