fix: aggregation corner case (#15457)

* fix: aggregation corner case

* Update generate_series.rs

* Update physical_planner.rs

* format

* Update generate_series.rs

Author: chenkovsky

datafusion/functions-table/src/generate_series.rs

@@ -138,12 +138,15 @@ impl TableProvider for GenerateSeriesTable {
     async fn scan(
         &self,
         state: &dyn Session,
-        _projection: Option<&Vec<usize>>,
+        projection: Option<&Vec<usize>>,
         _filters: &[Expr],
         _limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let batch_size = state.config_options().execution.batch_size;
-
+        let schema = match projection {
+            Some(projection) => Arc::new(self.schema.project(projection)?),
+            None => self.schema(),
+        };
         let state = match self.args {
             // if args have null, then return 0 row
             GenSeriesArgs::ContainsNull { include_end, name } => GenerateSeriesState {
@@ -175,7 +178,7 @@ impl TableProvider for GenerateSeriesTable {
         };
 
         Ok(Arc::new(LazyMemoryExec::try_new(
-            self.schema(),
+            schema,
             vec![Arc::new(RwLock::new(state))],
         )?))
     }

datafusion/sqllogictest/test_files/aggregate.slt

@@ -6729,3 +6729,26 @@ SELECT a, median(b), arrow_typeof(median(b)) FROM group_median_all_nulls GROUP B
 ----
 group0 NULL Int32
 group1 NULL Int32
+
+query I
+with test AS (SELECT i as c1, i + 1 as c2 FROM generate_series(1, 10) t(i))
+select count(*) from test WHERE 1 = 1;
+----
+10
+
+query I
+with test AS (SELECT i as c1, i + 1 as c2 FROM generate_series(1, 10) t(i))
+select count(c1) from test WHERE 1 = 1;
+----
+10
+
+query II rowsort
+with test AS (SELECT i as c1, i + 1 as c2 FROM generate_series(1, 5) t(i))
+select c2, count(*) from test WHERE 1 = 1 group by c2;
+----
+2 1
+3 1
+4 1
+5 1
+6 1
+