Fix unparse table scan with the projection pushdown (apache#12534)

* unparse the projection base on the source schema

* refactor and enhance the test

datafusion/sql/src/unparser/plan.rs

@@ -574,12 +574,15 @@ impl Unparser<'_> {
                         .iter()
                         .cloned()
                         .map(|i| {
-                            let (qualifier, field) =
-                                table_scan.projected_schema.qualified_field(i);
+                            let schema = table_scan.source.schema();
+                            let field = schema.field(i);
                             if alias.is_some() {
                                 Column::new(alias.clone(), field.name().clone())
                             } else {
-                                Column::new(qualifier.cloned(), field.name().clone())
+                                Column::new(
+                                    Some(table_scan.table_name.clone()),
+                                    field.name().clone(),
+                                )
                             }
                         })
                         .collect::<Vec<_>>();

datafusion/sql/tests/cases/plan_to_sql.rs

@@ -654,6 +654,10 @@ fn test_table_scan_pushdown() -> Result<()> {
         "SELECT t1.id, t1.age FROM t1"
     );
 
+    let scan_with_projection = table_scan(Some("t1"), &schema, Some(vec![1]))?.build()?;
+    let scan_with_projection = plan_to_sql(&scan_with_projection)?;
+    assert_eq!(format!("{}", scan_with_projection), "SELECT t1.age FROM t1");
+
     let scan_with_no_projection = table_scan(Some("t1"), &schema, None)?.build()?;
     let scan_with_no_projection = plan_to_sql(&scan_with_no_projection)?;
     assert_eq!(format!("{}", scan_with_no_projection), "SELECT * FROM t1");
@@ -669,6 +673,17 @@ fn test_table_scan_pushdown() -> Result<()> {
         "SELECT ta.id, ta.age FROM t1 AS ta"
     );
 
+    let table_scan_with_projection_alias =
+        table_scan(Some("t1"), &schema, Some(vec![1]))?
+            .alias("ta")?
+            .build()?;
+    let table_scan_with_projection_alias =
+        plan_to_sql(&table_scan_with_projection_alias)?;
+    assert_eq!(
+        format!("{}", table_scan_with_projection_alias),
+        "SELECT ta.age FROM t1 AS ta"
+    );
+
     let table_scan_with_no_projection_alias = table_scan(Some("t1"), &schema, None)?
         .alias("ta")?
         .build()?;
@@ -745,6 +760,20 @@ fn test_table_scan_pushdown() -> Result<()> {
         "SELECT t1.id, t1.age FROM t1 WHERE (t1.id > t1.age)"
     );
 
+    let table_scan_with_projection_and_filter = table_scan_with_filters(
+        Some("t1"),
+        &schema,
+        Some(vec![1]),
+        vec![col("id").gt(col("age"))],
+    )?
+    .build()?;
+    let table_scan_with_projection_and_filter =
+        plan_to_sql(&table_scan_with_projection_and_filter)?;
+    assert_eq!(
+        format!("{}", table_scan_with_projection_and_filter),
+        "SELECT t1.age FROM t1 WHERE (t1.id > t1.age)"
+    );
+
     let table_scan_with_inline_fetch =
         table_scan_with_filter_and_fetch(Some("t1"), &schema, None, vec![], Some(10))?
             .build()?;