Introduce the View of MDL (#645)

* use session context handle table provider

* use positive match

* introduce the view in mdl

* modify the wren-modeling-py

* remove unused tests

* cargo clippy

* cargo fmt

* revert python change and provide sync fn for sql transform

* cargo fmt

Author: goldmedal

wren-modeling-py/Cargo.lock

@@ -73,24 +73,32 @@ impl ModelAnalyzeRule {
                     &self.analyzed_wren_mdl.wren_mdl(),
                     table_scan.table_name.clone(),
                 ) {
-                    return Ok(Transformed::no(LogicalPlan::TableScan(table_scan)));
-                }
-                let table_name = table_scan.table_name.table();
-                if let Some(model) =
-                    self.analyzed_wren_mdl.wren_mdl().get_model(table_name)
-                {
-                    let field: Vec<Expr> =
-                        used_columns.borrow().iter().cloned().collect();
-                    let model = LogicalPlan::Extension(Extension {
-                        node: Arc::new(ModelPlanNode::new(
-                            model,
-                            field,
-                            Some(LogicalPlan::TableScan(table_scan.clone())),
-                            Arc::clone(&self.analyzed_wren_mdl),
-                        )?),
-                    });
-                    used_columns.borrow_mut().clear();
-                    Ok(Transformed::yes(model))
+                    let table_name = table_scan.table_name.table();
+                    // transform ViewTable to a subquery plan
+                    if let Some(logical_plan) = table_scan.source.get_logical_plan() {
+                        let subquery = LogicalPlanBuilder::from(logical_plan.clone())
+                            .alias(table_name)?
+                            .build()?;
+                        return Ok(Transformed::yes(subquery));
+                    }
+                    if let Some(model) =
+                        self.analyzed_wren_mdl.wren_mdl().get_model(table_name)
+                    {
+                        let field: Vec<Expr> =
+                            used_columns.borrow().iter().cloned().collect();
+                        let model = LogicalPlan::Extension(Extension {
+                            node: Arc::new(ModelPlanNode::new(
+                                model,
+                                field,
+                                Some(LogicalPlan::TableScan(table_scan.clone())),
+                                Arc::clone(&self.analyzed_wren_mdl),
+                            )?),
+                        });
+                        used_columns.borrow_mut().clear();
+                        Ok(Transformed::yes(model))
+                    } else {
+                        Ok(Transformed::no(LogicalPlan::TableScan(table_scan)))
+                    }
                 } else {
                     Ok(Transformed::no(LogicalPlan::TableScan(table_scan)))
                 }
@@ -144,13 +152,15 @@ impl ModelAnalyzeRule {
 }
 
 fn belong_to_mdl(mdl: &WrenMDL, table_reference: TableReference) -> bool {
-    let catalog_mismatch = table_reference
+    let catalog_match = table_reference
         .catalog()
-        .map_or(true, |c| c != mdl.catalog());
+        .map_or(false, |c| c == mdl.catalog());
 
-    let schema_mismatch = table_reference.schema().map_or(true, |s| s != mdl.schema());
+    let schema_match = table_reference
+        .schema()
+        .map_or(false, |s| s == mdl.schema());
 
-    catalog_mismatch || schema_mismatch
+    catalog_match && schema_match
 }
 
 fn analyze_table_scan(
@@ -159,8 +169,6 @@ fn analyze_table_scan(
     required_field: Vec<Expr>,
 ) -> Result<LogicalPlan> {
     if belong_to_mdl(&analyzed_wren_mdl.wren_mdl(), table_scan.table_name.clone()) {
-        Ok(LogicalPlan::TableScan(table_scan))
-    } else {
         let table_name = table_scan.table_name.table();
         if let Some(model) = analyzed_wren_mdl.wren_mdl.get_model(table_name) {
             Ok(LogicalPlan::Extension(Extension {
@@ -174,6 +182,8 @@ fn analyze_table_scan(
         } else {
             Ok(LogicalPlan::TableScan(table_scan))
         }
+    } else {
+        Ok(LogicalPlan::TableScan(table_scan))
     }
 }