fix(datafusion): preserve quoted names in temporary views

crates/integrations/datafusion/src/sql_context.rs

@@ -63,7 +63,7 @@ use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::logical_expr::{Expr as LogicalExpr, LogicalPlan, Volatility};
 use datafusion::prelude::{DataFrame, SessionContext};
-use datafusion::sql::planner::IdentNormalizer;
+use datafusion::sql::planner::{object_name_to_table_reference, IdentNormalizer};
 use datafusion::sql::sqlparser::ast::{
     AlterColumnOperation, AlterTableOperation, BinaryLength, CharacterLength, ColumnDef,
     ColumnOption, CommentDef, CommentObject, CreateFunction, CreateFunctionBody, CreateTable,
@@ -2162,19 +2162,18 @@ impl SQLContext {
         let query_sql = create_view.query.to_string();
 
         if create_view.temporary {
-            let view_name = create_view.name.to_string();
-            let table_ref: TableReference = view_name.as_str().into();
-            let (catalog, database, name) = self.resolve_temp_table_name(table_ref)?;
+            let table_ref = object_name_to_table_reference(
+                create_view.name.clone(),
+                self.ctx.enable_ident_normalization(),
+            )?;
             let df = self.ctx.sql_with_options(&query_sql, sql_options).await?;
             let logical_plan = df.logical_plan().clone();
-            if create_view.if_not_exists
-                && self.temp_table_exist(format!("{catalog}.{database}.{name}"))?
-            {
+            if create_view.if_not_exists && self.temp_table_exist(table_ref.clone())? {
                 return ok_result(&self.ctx);
             }
             // Create a ViewTable and register it as a temp table
             let view_table = datafusion::datasource::ViewTable::new(logical_plan, Some(query_sql));
-            self.register_temp_table(format!("{catalog}.{database}.{name}"), Arc::new(view_table))?;
+            self.register_temp_table(table_ref, Arc::new(view_table))?;
             ok_result(&self.ctx)
         } else {
             validate_persistent_create_view(create_view)?;

crates/integrations/datafusion/tests/sql_context_tests.rs

@@ -2451,6 +2451,46 @@ async fn test_create_temporary_view_fully_qualified() {
     assert_eq!(total_rows, 2);
 }
 
+#[tokio::test]
+async fn test_create_temporary_view_preserves_quoted_names() {
+    let (_tmp, catalog) = create_test_env();
+    let ctx = create_sql_context(catalog).await;
+
+    for qualifier in ["", "default.", "paimon.default."] {
+        for name in [
+            "\"MixedView\"",
+            "\"view.dot\"",
+            "\"view name\"",
+            "\"view\"\"quote\"",
+        ] {
+            let view = format!("{qualifier}{name}");
+            ctx.sql(&format!(
+                "CREATE TEMPORARY VIEW {view} AS SELECT CAST(1 AS INT) AS id"
+            ))
+            .await
+            .unwrap();
+            assert_eq!(
+                collect_ids(&ctx, &format!("SELECT id FROM {view}")).await,
+                vec![1]
+            );
+
+            ctx.sql(&format!(
+                "CREATE TEMPORARY VIEW IF NOT EXISTS {view} AS SELECT CAST(2 AS INT) AS id"
+            ))
+            .await
+            .unwrap();
+            assert_eq!(
+                collect_ids(&ctx, &format!("SELECT id FROM {view}")).await,
+                vec![1]
+            );
+
+            ctx.sql(&format!("DROP TEMPORARY VIEW {view}"))
+                .await
+                .unwrap();
+        }
+    }
+}
+
 #[tokio::test]
 async fn test_create_temporary_view_database_qualified() {
     let (_tmp, catalog) = create_test_env();