From ace4eb475bf45f424dd792269eb1a61c24277821 Mon Sep 17 00:00:00 2001
From: Tony Solomonik <tony.solomonik@gmail.com>
Date: Sat, 30 Nov 2024 18:49:00 +0200
Subject: [PATCH] optimizations: Call `apply_summarize` on Scan -> Summarize

---
 src/optimizations/mod.rs                      |  4 +++
 src/optimizations/push_summarize_into_scan.rs | 27 +++++++++++++++++++
 2 files changed, 31 insertions(+)
 create mode 100644 src/optimizations/push_summarize_into_scan.rs

diff --git a/src/optimizations/mod.rs b/src/optimizations/mod.rs
index 54c4617..a291ce7 100644
--- a/src/optimizations/mod.rs
+++ b/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 {
diff --git a/src/optimizations/push_summarize_into_scan.rs b/src/optimizations/push_summarize_into_scan.rs
new file mode 100644
index 0000000..e040533
--- /dev/null
+++ b/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)])
+    }
+}