optimizations: Call apply_summarize on Scan -> Summarize

src/optimizations/mod.rs

@@ -8,6 +8,7 @@ use push_count_into_scan::PushCountIntoScan;
 use push_filter_into_scan::PushFilterIntoScan;
 use push_limit_into_limit::PushLimitIntoLimit;
 use push_limit_into_scan::PushLimitIntoScan;
+use push_summarize_into_scan::PushSummarizeIntoScan;
 use remove_redundant_sorts_before_count::RemoveRedundantSortsBeforeCount;
 
 mod convert_sort_limit_to_topn;
@@ -16,6 +17,7 @@ mod push_count_into_scan;
 mod push_filter_into_scan;
 mod push_limit_into_limit;
 mod push_limit_into_scan;
+mod push_summarize_into_scan;
 mod push_topn_into_limit;
 mod push_topn_into_scan;
 mod remove_redundant_sorts_before_count;
@@ -55,6 +57,8 @@ impl Default for Optimizer {
             // Count.
             opt!(PushCountIntoScan),
             opt!(RemoveRedundantSortsBeforeCount),
+            // Summarize.
+            opt!(PushSummarizeIntoScan),
         ];
         let patterns = optimizations.iter().map(|o| o.pattern()).collect();
         Self {

src/optimizations/push_summarize_into_scan.rs

@@ -0,0 +1,27 @@
+use crate::{pattern, workflow::WorkflowStep};
+
+use super::{Group, Optimization, Pattern};
+
+pub struct PushSummarizeIntoScan;
+
+impl Optimization for PushSummarizeIntoScan {
+    fn pattern(&self) -> Pattern {
+        pattern!(Scan Summarize)
+    }
+
+    fn apply(&self, steps: &[WorkflowStep], _groups: &[Group]) -> Option<Vec<WorkflowStep>> {
+        let WorkflowStep::Scan(mut scan) = steps[0].clone() else {
+            return None;
+        };
+        let WorkflowStep::Summarize(config) = &steps[1] else {
+            return None;
+        };
+
+        scan.handle = scan
+            .connector
+            .apply_summarize(config, scan.handle.as_ref())?
+            .into();
+
+        Some(vec![WorkflowStep::Scan(scan)])
+    }
+}