From cab4a6ad9cbc1be23c364795ac8b62c07060b445 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Thu, 30 May 2024 15:07:14 -0600
Subject: [PATCH] coalesce batches after scan

---
 core/src/execution/datafusion/planner.rs | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/core/src/execution/datafusion/planner.rs b/core/src/execution/datafusion/planner.rs
index 7a37e3aae..3bf81b386 100644
--- a/core/src/execution/datafusion/planner.rs
+++ b/core/src/execution/datafusion/planner.rs
@@ -39,6 +39,7 @@ use datafusion::{
     },
     physical_plan::{
         aggregates::{AggregateMode as DFAggregateMode, PhysicalGroupBy},
+        coalesce_batches::CoalesceBatchesExec,
         filter::FilterExec,
         joins::{utils::JoinFilter, HashJoinExec, PartitionMode, SortMergeJoinExec},
         limit::LocalLimitExec,
@@ -725,7 +726,11 @@ impl PhysicalPlanner {
                             .map(|r| (r, format!("col_{}", idx)))
                     })
                     .collect();
-                Ok((scans, Arc::new(ProjectionExec::try_new(exprs?, child)?)))
+
+                let projection = Arc::new(ProjectionExec::try_new(exprs?, child)?);
+                let coalesced: Arc<dyn ExecutionPlan> =
+                    Arc::new(CoalesceBatchesExec::new(projection, 4196));
+                Ok((scans, coalesced))
             }
             OpStruct::Filter(filter) => {
                 assert!(children.len() == 1);