revert #6595

Author: jackwener

datafusion/common/src/dfschema.rs

@@ -384,12 +384,8 @@ impl DFSchema {
         let self_fields = self.fields().iter();
         let other_fields = other.fields().iter();
         self_fields.zip(other_fields).all(|(f1, f2)| {
-            // TODO: resolve field when exist alias
-            // f1.qualifier() == f2.qualifier()
-            //     && f1.name() == f2.name()
-            // column(t1.a) field is "t1"."a"
-            // column(x) as t1.a field is ""."t1.a"
-            f1.qualified_name() == f2.qualified_name()
+            f1.qualifier() == f2.qualifier()
+                && f1.name() == f2.name()
                 && Self::datatype_is_semantically_equal(f1.data_type(), f2.data_type())
         })
     }

datafusion/core/tests/sqllogictests/test_files/pg_compat/pg_compat_null.slt

@@ -70,7 +70,7 @@ WITH HEADER ROW
 LOCATION '../../testing/data/csv/aggregate_test_100.csv'
 
 
-statement ok
+statement error DataFusion error: Error during planning: Mismatch between schema and batches
 CREATE TABLE aggregate_test_100_nullable_by_sql AS
 SELECT
     *,
@@ -82,15 +82,13 @@ SELECT
         WHEN c3 % 3 != 0 THEN c9
         ELSE NULL
     END AS n9
-FROM aggregate_test_100_by_sql
+FROM aggregate_test_100_by_sql;
 
 
-query III
+query error DataFusion error: Error during planning: table 'datafusion\.public\.aggregate_test_100_nullable_by_sql' not found
 SELECT
     COUNT(*), COUNT(n5), COUNT(n9)
 FROM aggregate_test_100_nullable_by_sql
-----
-100 66 72
 
 
 ########
@@ -99,5 +97,5 @@ FROM aggregate_test_100_nullable_by_sql
 statement ok
 DROP TABLE aggregate_test_100_by_sql
 
-statement ok
+statement error DataFusion error: Execution error: Table 'aggregate_test_100_nullable_by_sql' doesn't exist\.
 DROP TABLE aggregate_test_100_nullable_by_sql

datafusion/core/tests/sqllogictests/test_files/union.slt

@@ -432,13 +432,13 @@ logical_plan
 Sort: t1.c1 ASC NULLS LAST
 --Union
 ----TableScan: t1 projection=[c1]
-----Projection: t2.c1a AS c1
+----Projection: t2.c1a AS t1.c1
 ------TableScan: t2 projection=[c1a]
 physical_plan
 SortPreservingMergeExec: [c1@0 ASC NULLS LAST]
 --UnionExec
 ----CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1], output_ordering=[c1@0 ASC NULLS LAST], has_header=true
-----ProjectionExec: expr=[c1a@0 as c1]
+----ProjectionExec: expr=[c1a@0 as t1.c1]
 ------CsvExec: file_groups={1 group: [[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1a], output_ordering=[c1a@0 ASC NULLS LAST], has_header=true
 
 statement ok

datafusion/expr/src/utils.rs

@@ -730,10 +730,13 @@ pub fn from_plan(
     inputs: &[LogicalPlan],
 ) -> Result<LogicalPlan> {
     match plan {
-        LogicalPlan::Projection(_) => Ok(LogicalPlan::Projection(Projection::try_new(
-            expr.to_vec(),
-            Arc::new(inputs[0].clone()),
-        )?)),
+        LogicalPlan::Projection(Projection { schema, .. }) => {
+            Ok(LogicalPlan::Projection(Projection::try_new_with_schema(
+                expr.to_vec(),
+                Arc::new(inputs[0].clone()),
+                schema.clone(),
+            )?))
+        }
         LogicalPlan::Dml(DmlStatement {
             table_name,
             table_schema,