Preserve recursive CTE work-table identifier case

datafusion/proto/src/logical_plan/mod.rs

@@ -1256,7 +1256,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                 let schema = convert_required!(*schema)?;
                 let cte_work_table = CteWorkTable::new(name.as_str(), Arc::new(schema));
                 LogicalPlanBuilder::scan(
-                    name.as_str(),
+                    TableReference::bare(name.as_str()),
                     provider_as_source(Arc::new(cte_work_table)),
                     None,
                 )?

datafusion/proto/tests/cases/roundtrip_logical_plan.rs

@@ -3528,6 +3528,45 @@ async fn roundtrip_recursive_query() {
     );
 }
 
+#[tokio::test]
+async fn roundtrip_recursive_query_preserves_normalized_name() -> Result<()> {
+    for query in [
+        "WITH RECURSIVE current_layer (component_id) AS (
+            SELECT 1 UNION ALL
+            SELECT current_layer.component_id + 1 FROM current_layer
+            WHERE current_layer.component_id < 3
+        ) SELECT component_id FROM current_layer",
+        "WITH RECURSIVE \"Mixed\" (\"n\") AS (
+            SELECT 1 UNION ALL
+            SELECT \"Mixed\".\"n\" + 1 FROM \"Mixed\"
+            WHERE \"Mixed\".\"n\" < 3
+        ) SELECT \"n\" FROM \"Mixed\"",
+    ] {
+        let mut config = SessionConfig::new();
+        config
+            .options_mut()
+            .sql_parser
+            .uppercase_unquoted_identifiers = true;
+        let ctx = SessionContext::new_with_config(config);
+        let plan = ctx.sql(query).await?.logical_plan().clone();
+        let output = ctx
+            .execute_logical_plan(plan.clone())
+            .await?
+            .collect()
+            .await?;
+        let bytes = logical_plan_to_bytes(&plan)?;
+        let decoded = logical_plan_from_bytes(&bytes, &ctx.task_ctx())?;
+
+        assert_eq!(format!("{plan:?}"), format!("{decoded:?}"));
+        let decoded_output = ctx.execute_logical_plan(decoded).await?.collect().await?;
+        assert_eq!(
+            pretty_format_batches(&output)?.to_string(),
+            pretty_format_batches(&decoded_output)?.to_string()
+        );
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn roundtrip_union_query() -> Result<()> {
     let query = "SELECT a FROM t1

datafusion/sql/src/cte.rs

@@ -162,7 +162,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         // Step 2.2: Create a temporary relation logical plan that will be used
         // as the input to the recursive term
         let work_table_plan = LogicalPlanBuilder::scan(
-            cte_name.to_string(),
+            TableReference::bare(cte_name.to_string()),
             Arc::clone(&work_table_source),
             None,
         )?

datafusion/sql/tests/sql_integration.rs

@@ -3755,6 +3755,40 @@ fn uppercase_unquoted_identifiers_apply_to_values_columns() {
         .unwrap();
 }
 
+#[test]
+fn recursive_cte_preserves_normalized_and_quoted_names() {
+    let mut config = datafusion_common::config::ConfigOptions::default();
+    config.sql_parser.uppercase_unquoted_identifiers = true;
+    config.execution.enable_recursive_ctes = true;
+
+    for (sql, table_name) in [
+        (
+            "WITH RECURSIVE current_layer (component_id) AS (\
+             SELECT 1 UNION ALL \
+             SELECT current_layer.component_id + 1 FROM current_layer \
+             WHERE current_layer.component_id < 3) \
+             SELECT component_id FROM current_layer",
+            "CURRENT_LAYER",
+        ),
+        (
+            "WITH RECURSIVE \"Mixed\" (\"n\") AS (\
+             SELECT 1 UNION ALL \
+             SELECT \"Mixed\".\"n\" + 1 FROM \"Mixed\" \
+             WHERE \"Mixed\".\"n\" < 3) \
+             SELECT \"n\" FROM \"Mixed\"",
+            "Mixed",
+        ),
+    ] {
+        let plan = logical_plan_with_config(sql, config.clone()).unwrap();
+        let display = plan.display_indent().to_string();
+        assert!(display.contains("RecursiveQuery"), "{display}");
+        assert!(
+            display.contains(&format!("TableScan: {table_name}")),
+            "{display}"
+        );
+    }
+}
+
 #[test]
 fn uppercase_unquoted_prepared_statement_names() {
     use datafusion_expr::logical_plan::Statement as PlanStatement;