make temp table name unique

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

@@ -212,6 +212,7 @@ pub struct CteInfo {
     pub cte_idx: IndexType,
     // If cte is materialized, save its columns
     pub columns: Vec<ColumnBinding>,
+    pub m_cte_name_to_temp_table: HashMap<String, String>,
 }
 
 impl BindContext {

src/query/sql/src/planner/binder/bind_query/bind.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use databend_common_ast::ast::CreateOption;
@@ -88,22 +89,26 @@ impl Binder {
                 .iter()
                 .map(|ident| self.normalize_identifier(ident).name)
                 .collect();
-            let cte_info = CteInfo {
+            let mut cte_info = CteInfo {
                 columns_alias: column_name,
                 query: *cte.query.clone(),
                 recursive: with.recursive,
                 cte_idx: idx,
                 columns: vec![],
                 materialized: cte.materialized,
+                m_cte_name_to_temp_table: HashMap::new(),
             };
+            // If the CTE is materialized, we'll construct a temp table for it.
+            if cte.materialized {
+                let temp_table_name = self.m_cte_to_temp_table(cte)?;
+                cte_info
+                    .m_cte_name_to_temp_table
+                    .insert(cte.alias.name.name.clone(), temp_table_name);
+            }
             bind_context
                 .cte_context
                 .cte_map
                 .insert(table_name, cte_info);
-            // If the CTE is materialized, we'll construct a temp table for it.
-            if cte.materialized {
-                self.m_cte_to_temp_table(cte)?;
-            }
         }
 
         Ok(())
@@ -177,21 +182,28 @@ impl Binder {
         ))
     }
 
-    fn m_cte_to_temp_table(&self, cte: &CTE) -> Result<()> {
-        let engine = if self.ctx.get_settings().get_persist_materialized_cte()?
-            || !self.ctx.get_cluster().is_empty()
-        {
+    // The return value is temp_table name`
+    fn m_cte_to_temp_table(&self, cte: &CTE) -> Result<String> {
+        let engine = if self.ctx.get_settings().get_persist_materialized_cte()? {
             Engine::Fuse
         } else {
             Engine::Memory
         };
         let database = self.ctx.get_current_database();
         let catalog = self.ctx.get_current_catalog();
+        let query_id = self.ctx.get_id();
+        // Navigate the temp table for cte to `cte_name + query_id`
+        // Avoid the conflict of the temp table name with the same name of user's temp table.
+        let table_name = format!(
+            "{}_{}",
+            cte.alias.name.name,
+            query_id.split('-').next().unwrap_or(&query_id)
+        );
         let create_table_stmt = CreateTableStmt {
             create_option: CreateOption::CreateOrReplace,
             catalog: Some(Identifier::from_name(Span::None, catalog.clone())),
             database: Some(Identifier::from_name(Span::None, database.clone())),
-            table: cte.alias.name.clone(),
+            table: Identifier::from_name(cte.alias.name.span, table_name.clone()),
             source: None,
             engine: Some(engine),
             uri_location: None,
@@ -212,9 +224,8 @@ impl Binder {
             return Err(ErrorCode::Internal("Binder's Subquery executor is not set"));
         };
 
-        // Todo: clear the table with the same name in table cache
         self.ctx
             .remove_table_from_cache(&catalog, &database, &cte.alias.name.name);
-        Ok(())
+        Ok(table_name)
     }
 }

src/query/sql/src/planner/binder/bind_table_reference/bind_table.rs

@@ -97,14 +97,31 @@ impl Binder {
             };
         }
 
+        let mut temp_table_name = None;
+        if let Some(cte_info) = cte_map.get(&table_name)
+            && cte_info.materialized
+        {
+            temp_table_name = Some(
+                cte_info
+                    .m_cte_name_to_temp_table
+                    .get(&table_name)
+                    .unwrap()
+                    .to_string(),
+            );
+        }
+
         let navigation = self.resolve_temporal_clause(bind_context, temporal)?;
 
         // Resolve table with catalog
         let table_meta = {
             match self.resolve_data_source(
                 catalog.as_str(),
                 database.as_str(),
-                table_name.as_str(),
+                if let Some(temp_table_name) = temp_table_name.as_ref() {
+                    temp_table_name.as_str()
+                } else {
+                    table_name.as_str()
+                },
                 navigation.as_ref(),
                 max_batch_size,
                 self.ctx.clone().get_abort_checker(),
@@ -150,6 +167,7 @@ impl Binder {
                     database.clone(),
                     table_meta,
                     table_name_alias,
+                    None,
                     bind_context.view_info.is_some(),
                     bind_context.planning_agg_index,
                     false,
@@ -228,6 +246,7 @@ impl Binder {
                         database.clone(),
                         table_meta,
                         table_name_alias,
+                        None,
                         false,
                         false,
                         false,
@@ -259,6 +278,7 @@ impl Binder {
                     catalog,
                     database.clone(),
                     table_meta,
+                    Some(table_name.clone()),
                     table_name_alias,
                     bind_context.view_info.is_some(),
                     bind_context.planning_agg_index,

src/query/sql/src/planner/binder/bind_table_reference/bind_table_function.rs

@@ -145,6 +145,7 @@ impl Binder {
                 "system".to_string(),
                 table.clone(),
                 table_alias_name,
+                None,
                 false,
                 false,
                 false,
@@ -207,6 +208,7 @@ impl Binder {
                 "system".to_string(),
                 table.clone(),
                 table_alias_name,
+                None,
                 false,
                 false,
                 false,

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

@@ -130,6 +130,7 @@ impl Binder {
             "system".to_string(),
             table.clone(),
             table_alias_name,
+            None,
             false,
             false,
             true,

src/query/sql/src/planner/dataframe.rs

@@ -98,6 +98,7 @@ impl Dataframe {
                 database.to_string(),
                 table_meta,
                 None,
+                None,
                 false,
                 false,
                 false,

src/query/sql/src/planner/expression_parser.rs

@@ -68,6 +68,7 @@ pub fn bind_table(table_meta: Arc<dyn Table>) -> Result<(BindContext, MetadataRe
         "default".to_string(),
         table_meta,
         None,
+        None,
         false,
         false,
         false,

src/query/sql/src/planner/metadata.rs

@@ -349,13 +349,18 @@ impl Metadata {
         catalog: String,
         database: String,
         table_meta: Arc<dyn Table>,
+        table_name: Option<String>,
         table_alias_name: Option<String>,
         source_of_view: bool,
         source_of_index: bool,
         source_of_stage: bool,
         consume: bool,
     ) -> IndexType {
-        let table_name = table_meta.name().to_string();
+        let table_name = if let Some(table_name) = table_name {
+            table_name
+        } else {
+            table_meta.name().to_string()
+        };
 
         let table_index = self.tables.len();
         // If exists table alias name, use it instead of origin name