revert: from_plan keep same schema Agg/Window in #6820

Author: jackwener

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 error DataFusion error: Error during planning: Mismatch between schema and batches
+statement ok
 CREATE TABLE aggregate_test_100_nullable_by_sql AS
 SELECT
     *,
@@ -82,13 +82,15 @@ 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 error DataFusion error: Error during planning: table 'datafusion\.public\.aggregate_test_100_nullable_by_sql' not found
+query III
 SELECT
     COUNT(*), COUNT(n5), COUNT(n9)
 FROM aggregate_test_100_nullable_by_sql
+----
+100 66 72
 
 
 ########
@@ -97,5 +99,5 @@ FROM aggregate_test_100_nullable_by_sql
 statement ok
 DROP TABLE aggregate_test_100_by_sql
 
-statement error DataFusion error: Execution error: Table 'aggregate_test_100_nullable_by_sql' doesn't exist\.
+statement ok
 DROP TABLE aggregate_test_100_nullable_by_sql

datafusion/expr/src/utils.rs

@@ -821,19 +821,23 @@ pub fn from_plan(
                 input: Arc::new(inputs[0].clone()),
             })),
         },
-        LogicalPlan::Window(Window { window_expr, .. }) => {
-            Ok(LogicalPlan::Window(Window::try_new(
-                expr[0..window_expr.len()].to_vec(),
-                Arc::new(inputs[0].clone()),
-            )?))
-        }
-        LogicalPlan::Aggregate(Aggregate { group_expr, .. }) => {
-            Ok(LogicalPlan::Aggregate(Aggregate::try_new(
-                Arc::new(inputs[0].clone()),
-                expr[0..group_expr.len()].to_vec(),
-                expr[group_expr.len()..].to_vec(),
-            )?))
-        }
+        LogicalPlan::Window(Window {
+            window_expr,
+            schema,
+            ..
+        }) => Ok(LogicalPlan::Window(Window {
+            input: Arc::new(inputs[0].clone()),
+            window_expr: expr[0..window_expr.len()].to_vec(),
+            schema: schema.clone(),
+        })),
+        LogicalPlan::Aggregate(Aggregate {
+            group_expr, schema, ..
+        }) => Ok(LogicalPlan::Aggregate(Aggregate::try_new_with_schema(
+            Arc::new(inputs[0].clone()),
+            expr[0..group_expr.len()].to_vec(),
+            expr[group_expr.len()..].to_vec(),
+            schema.clone(),
+        )?)),
         LogicalPlan::Sort(SortPlan { fetch, .. }) => Ok(LogicalPlan::Sort(SortPlan {
             expr: expr.to_vec(),
             input: Arc::new(inputs[0].clone()),