Merge remote-tracking branch 'upstream/main' into internal_column

Author: lichuang

src/query/sql/src/planner/binder/bind_context.rs

@@ -15,6 +15,7 @@
 use std::collections::BTreeMap;
 use std::hash::Hash;
 
+use common_ast::ast::Query;
 use common_ast::ast::TableAlias;
 use common_catalog::plan::InternalColumn;
 use common_exception::ErrorCode;
@@ -30,7 +31,6 @@ use dashmap::DashMap;
 use super::AggregateInfo;
 use super::INTERNAL_COLUMN_FACTORY;
 use crate::normalize_identifier;
-use crate::optimizer::SExpr;
 use crate::plans::ScalarExpr;
 use crate::ColumnSet;
 use crate::IndexType;
@@ -137,8 +137,7 @@ pub struct BindContext {
 #[derive(Clone, Debug)]
 pub struct CteInfo {
     pub columns_alias: Vec<String>,
-    pub s_expr: SExpr,
-    pub bind_context: BindContext,
+    pub query: Query,
 }
 
 impl BindContext {

src/query/sql/src/planner/binder/select.rs

@@ -206,11 +206,9 @@ impl Binder {
                         "duplicate cte {table_name}"
                     )));
                 }
-                let (s_expr, cte_bind_context) = self.bind_query(bind_context, &cte.query).await?;
                 let cte_info = CteInfo {
                     columns_alias: cte.alias.columns.iter().map(|c| c.name.clone()).collect(),
-                    s_expr,
-                    bind_context: cte_bind_context.clone(),
+                    query: cte.query.clone(),
                 };
                 bind_context.ctes_map.insert(table_name, cte_info);
             }

src/query/sql/src/planner/binder/table.rs

@@ -56,6 +56,7 @@ use common_storages_result_cache::ResultCacheReader;
 use common_storages_result_cache::ResultScan;
 use common_storages_view::view_table::QUERY;
 use common_users::UserApiProvider;
+use dashmap::DashMap;
 
 use crate::binder::copy::parse_stage_location_v2;
 use crate::binder::location::parse_uri_location;
@@ -139,7 +140,9 @@ impl Binder {
                 };
                 // Check and bind common table expression
                 if let Some(cte_info) = bind_context.ctes_map.get(&table_name) {
-                    return self.bind_cte(bind_context, &table_name, alias, &cte_info);
+                    return self
+                        .bind_cte(bind_context, &table_name, alias, &cte_info)
+                        .await;
                 }
 
                 if database == "system" {
@@ -154,15 +157,36 @@ impl Binder {
                 };
 
                 // Resolve table with catalog
-                let table_meta: Arc<dyn Table> = self
+                let table_meta = match self
                     .resolve_data_source(
                         tenant.as_str(),
                         catalog.as_str(),
                         database.as_str(),
                         table_name.as_str(),
                         &navigation_point,
                     )
-                    .await?;
+                    .await
+                {
+                    Ok(table) => table,
+                    Err(_) => {
+                        let mut parent = bind_context.parent.as_ref();
+                        loop {
+                            if parent.is_none() {
+                                break;
+                            }
+                            if let Some(cte_info) = parent.unwrap().ctes_map.get(&table_name) {
+                                return self
+                                    .bind_cte(bind_context, &table_name, alias, &cte_info)
+                                    .await;
+                            }
+                            parent = parent.unwrap().parent.as_ref();
+                        }
+                        return Err(ErrorCode::UnknownTable(format!(
+                            "Unknown table '{table_name}'"
+                        )));
+                    }
+                };
+
                 match table_meta.engine() {
                     "VIEW" => {
                         let query = table_meta
@@ -418,15 +442,23 @@ impl Binder {
         }
     }
 
-    fn bind_cte(
+    async fn bind_cte(
         &mut self,
         bind_context: &BindContext,
         table_name: &str,
         alias: &Option<TableAlias>,
         cte_info: &CteInfo,
     ) -> Result<(SExpr, BindContext)> {
-        let mut new_bind_context = bind_context.clone();
-        new_bind_context.columns = cte_info.bind_context.columns.clone();
+        let new_bind_context = BindContext {
+            parent: Some(Box::new(bind_context.clone())),
+            columns: vec![],
+            aggregate_info: Default::default(),
+            in_grouping: false,
+            ctes_map: Box::new(DashMap::new()),
+            is_view: false,
+        };
+        let (s_expr, mut new_bind_context) =
+            self.bind_query(&new_bind_context, &cte_info.query).await?;
         let mut cols_alias = cte_info.columns_alias.clone();
         if let Some(alias) = alias {
             for (idx, col_alias) in alias.columns.iter().enumerate() {
@@ -456,7 +488,7 @@ impl Binder {
         for (index, column_name) in cols_alias.iter().enumerate() {
             new_bind_context.columns[index].column_name = column_name.clone();
         }
-        Ok((cte_info.s_expr.clone(), new_bind_context))
+        Ok((s_expr, new_bind_context))
     }
 
     async fn bind_base_table(

tests/sqllogictests/suites/query/cte.test

@@ -365,3 +365,9 @@ query II
 select * from  (WITH source AS (select 1 as e) SELECT * FROM source) A,   (WITH source AS (select 2 as e) SELECT * FROM source) B
 ----
 1 2
+
+query II
+with v as (select * from numbers(2)) select * from v t1, (select * from v where number = 0) t2 where t1.number = 1 and t2.number = 1
+----
+
+