add catalog as part of the table path in plan_to_sql (#10612)

* add catalog as part of the table path in plan_to_sql

* Update datafusion/sql/src/unparser/plan.rs

Co-authored-by: Phillip LeBlanc <[email protected]>

---------

Co-authored-by: Phillip LeBlanc <[email protected]>

Author: y-f-u

datafusion/sql/src/unparser/plan.rs

@@ -140,6 +140,9 @@ impl Unparser<'_> {
             LogicalPlan::TableScan(scan) => {
                 let mut builder = TableRelationBuilder::default();
                 let mut table_parts = vec![];
+                if let Some(catalog_name) = scan.table_name.catalog() {
+                    table_parts.push(self.new_ident(catalog_name.to_string()));
+                }
                 if let Some(schema_name) = scan.table_name.schema() {
                     table_parts
                         .push(self.new_ident_quoted_if_needs(schema_name.to_string()));
@@ -505,3 +508,35 @@ impl From<BuilderError> for DataFusionError {
         DataFusionError::External(Box::new(e))
     }
 }
+
+#[cfg(test)]
+mod test {
+    use crate::unparser::plan_to_sql;
+    use arrow::datatypes::{DataType, Field, Schema};
+    use datafusion_expr::{col, logical_plan::table_scan};
+    #[test]
+    fn test_table_references_in_plan_to_sql() {
+        fn test(table_name: &str, expected_sql: &str) {
+            let schema = Schema::new(vec![
+                Field::new("id", DataType::Utf8, false),
+                Field::new("value", DataType::Utf8, false),
+            ]);
+            let plan = table_scan(Some(table_name), &schema, None)
+                .unwrap()
+                .project(vec![col("id"), col("value")])
+                .unwrap()
+                .build()
+                .unwrap();
+            let sql = plan_to_sql(&plan).unwrap();
+
+            assert_eq!(format!("{}", sql), expected_sql)
+        }
+
+        test("catalog.schema.table", "SELECT \"catalog\".\"schema\".\"table\".\"id\", \"catalog\".\"schema\".\"table\".\"value\" FROM \"catalog\".\"schema\".\"table\"");
+        test("schema.table", "SELECT \"schema\".\"table\".\"id\", \"schema\".\"table\".\"value\" FROM \"schema\".\"table\"");
+        test(
+            "table",
+            "SELECT \"table\".\"id\", \"table\".\"value\" FROM \"table\"",
+        );
+    }
+}