fix(core): implement the scope analysis to support the correlated sub…

…query syntax (#787)

* introduce scope analysis

* refactor the analysis rule

* expand view before analyze model

* implement the scope analysis and refactor model analysis

* enhance the doc

* cache the catalog_schema prefix

* add comment and missing rule

* rollback unused change

* cargo fmt

wren-modeling-rs/core/src/logical_plan/analyze/expand_view.rs

@@ -0,0 +1,62 @@
+use crate::logical_plan::utils::belong_to_mdl;
+use crate::mdl::utils::quoted;
+use crate::mdl::{AnalyzedWrenMDL, SessionStateRef};
+use datafusion::common::tree_node::Transformed;
+use datafusion::common::Result;
+use datafusion::config::ConfigOptions;
+use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder};
+use datafusion::optimizer::AnalyzerRule;
+use std::sync::Arc;
+
+pub struct ExpandWrenViewRule {
+    analyzed_wren_mdl: Arc<AnalyzedWrenMDL>,
+    session_state: SessionStateRef,
+}
+
+impl ExpandWrenViewRule {
+    pub fn new(
+        analyzed_wren_mdl: Arc<AnalyzedWrenMDL>,
+        session_state: SessionStateRef,
+    ) -> Self {
+        Self {
+            analyzed_wren_mdl,
+            session_state,
+        }
+    }
+}
+
+impl AnalyzerRule for ExpandWrenViewRule {
+    fn analyze(&self, plan: LogicalPlan, _: &ConfigOptions) -> Result<LogicalPlan> {
+        let plan = plan
+            .transform_up_with_subqueries(|plan| match &plan {
+                LogicalPlan::TableScan(table_scan) => {
+                    if belong_to_mdl(
+                        &self.analyzed_wren_mdl.wren_mdl(),
+                        table_scan.table_name.clone(),
+                        Arc::clone(&self.session_state),
+                    ) && self
+                        .analyzed_wren_mdl
+                        .wren_mdl()
+                        .get_view(table_scan.table_name.table())
+                        .is_some()
+                    {
+                        if let Some(logical_plan) = table_scan.source.get_logical_plan() {
+                            let subquery = LogicalPlanBuilder::from(logical_plan.clone())
+                                .alias(quoted(table_scan.table_name.table()))?
+                                .build()?;
+                            return Ok(Transformed::yes(subquery));
+                        }
+                    }
+                    Ok(Transformed::no(plan))
+                }
+                _ => Ok(Transformed::no(plan)),
+            })?
+            .map_data(|plan| plan.recompute_schema())?
+            .data;
+        Ok(plan)
+    }
+
+    fn name(&self) -> &str {
+        "ExpandWrenViewRule"
+    }
+}

wren-modeling-rs/core/src/logical_plan/analyze/mod.rs

@@ -1,3 +1,4 @@
+pub mod expand_view;
 pub mod model_anlayze;
 pub mod model_generation;
 pub mod plan;