perf: hold the filter on a scan as its residual predicate

src/db.rs

@@ -543,6 +543,11 @@ fn optimizer_pipeline() -> HepOptimizerPipeline {
             HepBatchStrategy::fix_point_topdown(10),
             &[NormalizationRuleImpl::PushLimitIntoTableScan],
         )
+        .after_batch(
+            "Skip Residual Filter Columns",
+            HepBatchStrategy::once_topdown(),
+            &[NormalizationRuleImpl::SkipResidualFilterColumns],
+        )
         .implementations(&[
             // DQL
             ImplementationRuleImpl::SimpleAggregate,
@@ -1526,6 +1531,7 @@ pub(crate) mod test {
     use crate::optimizer::core::statistics_meta::StatisticMetaLoader;
     use crate::planner::operator::join::{JoinCondition, JoinOperator, JoinType};
     use crate::planner::operator::project::ProjectOperator;
+    use crate::planner::operator::table_scan::TableScanOperator;
     use crate::planner::operator::values::ValuesOperator;
     use crate::planner::operator::Operator;
     use crate::planner::{Childrens, ExprRef, LogicalPlan, PlanArena, TableArenaCell};
@@ -2155,14 +2161,18 @@ pub(crate) mod test {
         if matches!(right_child.operator, Operator::Project(_)) {
             right_child = right_child.childrens.pop_only();
         }
-        let Operator::Filter(filter_op) = right_child.operator else {
+        let Operator::TableScan(TableScanOperator {
+            predicate: Some(predicate),
+            ..
+        }) = right_child.operator
+        else {
             unreachable!("expected pushed-down filter on right child");
         };
         let ScalarExpression::Binary {
             left_expr,
             right_expr,
             ..
-        } = best_plan_arena.expression(filter_op.predicate)
+        } = best_plan_arena.expression(predicate)
         else {
             unreachable!("expected binary filter predicate");
         };
@@ -2209,7 +2219,7 @@ pub(crate) mod test {
             let plan = row[0].utf8().unwrap();
             assert_eq!(
                 plan,
-                "Projection [t1.a, t1.b] [Project => (Sort Option: Follow)] Filter (t1.b > 0), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b] [SeqScan => (Sort Option: None)]"
+                "Projection [t1.a, t1.b] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b], Filter: (t1.b > 0) [SeqScan => (Sort Option: None)]"
             );
         }
         // Aggregate
@@ -2236,7 +2246,7 @@ pub(crate) mod test {
             let plan = row[0].utf8().unwrap();
             assert_eq!(
                 plan,
-                "Projection [(t1.a + $1), Max((t1.b + $2))] [Project => (Sort Option: Follow)] Aggregate [Max((t1.b + 0))] -> Group By [(t1.a + 0)] [HashAggregate => (Sort Option: None)] Filter (t1.b > 1), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b] [SeqScan => (Sort Option: None)]"
+                "Projection [(t1.a + $1), Max((t1.b + $2))] [Project => (Sort Option: Follow)] Aggregate [Max((t1.b + 0))] -> Group By [(t1.a + 0)] [HashAggregate => (Sort Option: None)] TableScan t1 -> [t1.a, t1.b], Filter: (t1.b > 1) [SeqScan => (Sort Option: None)]"
             );
         }
         {
@@ -2255,7 +2265,7 @@ pub(crate) mod test {
             let plan = row[0].utf8().unwrap();
             assert_eq!(
                 plan,
-                "Projection [#0.a, #0.b, t1.a, t1.b, 9] [Project => (Sort Option: Follow)] LeftOuter Join Where (#0.a > 0) [NestLoopJoin => (Sort Option: None)] Projection [(t1.a) as (#0.a), (t1.b) as (#0.b)] [Project => (Sort Option: Follow)] Filter (t1.b > 0), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b] [SeqScan => (Sort Option: None)] Projection [t1.a, t1.b] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b] [IndexScan By pk_index => ($3, +inf) => (Sort Option: OrderBy: (t1.a Asc Nulls Last) ignore_prefix_len: 0)]"
+                "Projection [#0.a, #0.b, t1.a, t1.b, 9] [Project => (Sort Option: Follow)] LeftOuter Join Where (#0.a > 0) [NestLoopJoin => (Sort Option: None)] Projection [(t1.a) as (#0.a), (t1.b) as (#0.b)] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b], Filter: (t1.b > 0) [SeqScan => (Sort Option: None)] Projection [t1.a, t1.b] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.a, t1.b] [IndexScan By pk_index => ($3, +inf) => (Sort Option: OrderBy: (t1.a Asc Nulls Last) ignore_prefix_len: 0)]"
             );
         }
 

src/db/program.rs

@@ -371,6 +371,7 @@ mod tests {
                     limit: (None, None),
                     index_infos: vec![index.clone(), other.clone()],
                     with_pk: false,
+                    predicate: None,
                 }),
                 Childrens::None,
             );
@@ -454,7 +455,14 @@ mod tests {
     #[test]
     fn parameter_order_dependent_predicates_remain_correct() -> Result<(), DatabaseError> {
         fn has_filter(plan: &LogicalPlan) -> bool {
-            matches!(plan.operator, Operator::Filter(_)) || plan.childrens.iter().any(has_filter)
+            matches!(
+                &plan.operator,
+                Operator::Filter(_)
+                    | Operator::TableScan(TableScanOperator {
+                        predicate: Some(_),
+                        ..
+                    })
+            ) || plan.childrens.iter().any(has_filter)
         }
 
         let mut db = DataBaseBuilder::path(".")
@@ -736,7 +744,14 @@ mod tests {
         };
         // `$3 - 20` is computed when the range is bound, so the whole predicate is the range.
         fn has_filter(plan: &LogicalPlan) -> bool {
-            matches!(plan.operator, Operator::Filter(_)) || plan.childrens.iter().any(has_filter)
+            matches!(
+                &plan.operator,
+                Operator::Filter(_)
+                    | Operator::TableScan(TableScanOperator {
+                        predicate: Some(_),
+                        ..
+                    })
+            ) || plan.childrens.iter().any(has_filter)
         }
         assert!(matches!(
             &info.lookup,

src/execution/dql/aggregate/hash_agg.rs

@@ -26,14 +26,10 @@ use crate::planner::LogicalPlan;
 use crate::planner::{ExecMetaArena, MetaArena};
 use crate::storage::StorageTransaction;
 use crate::types::value::DataValue;
-use hashbrown::hash_table::IntoIter as HashTableIntoIter;
 use hashbrown::HashTable;
 use std::hash::BuildHasher;
 use std::mem;
 
-/// Group indexes, which locate each group's keys and accumulators.
-type HashAggOutput = HashTableIntoIter<usize>;
-
 pub struct HashAggExecutor<'a> {
     agg_calls: &'a [ExprRef],
     groupby_exprs: &'a [ExprRef],
@@ -42,7 +38,11 @@ pub struct HashAggExecutor<'a> {
     // `groupby_exprs.len()` keys and `agg_calls.len()` accumulators per group.
     keys: Vec<DataValue>,
     accs: Vec<Box<dyn Accumulator>>,
-    output: Option<HashAggOutput>,
+    // Group indexes by key hash, which locate each group's keys and accumulators. Kept across
+    // executions for its capacity.
+    groups: HashTable<usize>,
+    // The next group to output, once every input row is folded.
+    output: Option<usize>,
 }
 
 impl<'a> ReadExecutor<'a> for HashAggExecutor<'a> {
@@ -67,6 +67,7 @@ impl<'a> ReadExecutor<'a> for HashAggExecutor<'a> {
             input,
             keys: Vec::new(),
             accs: Vec::new(),
+            groups: HashTable::new(),
             output: None,
         }))
     }
@@ -80,6 +81,7 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for HashAggExecutor<'a>
     ) -> Result<(), DatabaseError> {
         self.keys.clear();
         self.accs.clear();
+        self.groups.clear();
         self.output = None;
         self.input.reset(arena, plan_arena)
     }
@@ -93,8 +95,7 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for HashAggExecutor<'a>
         let acc_len = self.agg_calls.len();
         if self.output.is_none() {
             let hasher = FxBuildHasher::default();
-            let mut groups: HashTable<usize> = HashTable::new();
-            let mut group_count = 0;
+            let groups = &mut self.groups;
 
             while self.input.next_tuple(arena, plan_arena)? {
                 let tuple = arena.result_tuple();
@@ -120,8 +121,7 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for HashAggExecutor<'a>
                         group
                     }
                     None => {
-                        let group = group_count;
-                        group_count += 1;
+                        let group = groups.len();
                         groups.insert_unique(hash, group, |&group| {
                             hasher.hash_one(&keys[group * len..][..len])
                         });
@@ -137,13 +137,15 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for HashAggExecutor<'a>
                 )?;
             }
 
-            self.output = Some(groups.into_iter());
+            self.output = Some(0);
         }
 
-        let Some(group) = self.output.as_mut().and_then(Iterator::next) else {
+        // Groups are numbered as they appear, so they are output in that order.
+        let Some(group) = self.output.filter(|group| *group < self.groups.len()) else {
             arena.finish();
             return Ok(());
         };
+        self.output = Some(group + 1);
         let group_keys = self.keys[group * len..][..len]
             .iter_mut()
             .map(|value| mem::replace(value, DataValue::Null));

src/execution/dql/index_scan.rs

@@ -13,14 +13,15 @@
 // limitations under the License.
 
 use crate::errors::DatabaseError;
+use crate::execution::dql::seq_scan::next_scan_tuple;
 use crate::execution::{
     ExecBuilder, ExecContext, ExecNode, ExecutorNode, IterSlot, ReadExecutor, StorageIter,
 };
 use crate::expression::range_detacher::{IndexRangeColumn, RangeDetacher};
 use crate::planner::operator::table_scan::TableScanOperator;
 use crate::planner::operator::SortOption;
 use crate::planner::{ExecMetaArena, MetaArena};
-use crate::storage::{IndexRanges, Iter, StorageTransaction};
+use crate::storage::{IndexRanges, StorageTransaction};
 use crate::types::index::{IndexInfo, IndexLookup};
 use std::borrow::Cow;
 
@@ -157,7 +158,8 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for IndexScan<'a> {
 
         let state = arena.local_state(plan_arena);
         let iter = state.iters[self.iter].as_index();
-        if iter.next_tuple_into(state.table_codec, state.result.output_mut())? {
+        let output = state.result.output_mut();
+        if next_scan_tuple(self.op, iter, state.table_codec, output, plan_arena)? {
             arena.resume();
         } else {
             arena.finish();
@@ -222,6 +224,7 @@ mod tests {
             limit: (None, None),
             index_infos: vec![],
             with_pk: true,
+            predicate: None,
         };
         let info = IndexInfo {
             meta: *table.indexes().next().unwrap(),

src/execution/dql/seq_scan.rs

@@ -18,7 +18,9 @@ use crate::execution::{
 };
 use crate::planner::operator::table_scan::TableScanOperator;
 use crate::planner::{ExecMetaArena, MetaArena};
+use crate::storage::table_codec::TableCodec;
 use crate::storage::{Iter, StorageTransaction};
+use crate::types::tuple::Tuple;
 
 pub(crate) struct SeqScan<'a> {
     op: &'a TableScanOperator,
@@ -74,8 +76,8 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for SeqScan<'a> {
             state.iters[self.iter] = StorageIter::Tuple(iter);
         }
         let iter = state.iters[self.iter].as_tuple();
-
-        if iter.next_tuple_into(state.table_codec, state.result.output_mut())? {
+        let output = state.result.output_mut();
+        if next_scan_tuple(self.op, iter, state.table_codec, output, plan_arena)? {
             arena.resume();
         } else {
             arena.finish();
@@ -84,6 +86,29 @@ impl<'a, T: StorageTransaction + 'a> ExecutorNode<'a, T> for SeqScan<'a> {
     }
 }
 
+#[inline]
+pub(crate) fn next_scan_tuple<I: Iter + ?Sized, A: MetaArena>(
+    op: &TableScanOperator,
+    iter: &mut I,
+    table_codec: &mut TableCodec,
+    tuple: &mut Tuple,
+    arena: &A,
+) -> Result<bool, DatabaseError> {
+    while iter.next_tuple_into(table_codec, tuple)? {
+        let holds = match op.predicate {
+            Some(predicate) => arena
+                .expression(predicate)
+                .eval(arena, Some(&*tuple))?
+                .is_true()?,
+            None => true,
+        };
+        if holds {
+            return Ok(true);
+        }
+    }
+    Ok(false)
+}
+
 // GRCOV_EXCL_START
 #[cfg(all(test, not(target_arch = "wasm32")))]
 mod tests {
@@ -140,6 +165,7 @@ mod tests {
             limit: (Some(1), Some(2)),
             index_infos: vec![],
             with_pk: true,
+            predicate: None,
         };
         let mut exec_buffer = crate::execution::ExecBuffer::default();
         let mut exec_arena = crate::execution::ExecArena::new(&tables, &tx);

src/optimizer/plan_utils.rs

@@ -101,7 +101,9 @@ pub(crate) fn predicate_scan_mut<'plan>(
     &'plan mut Option<PhysicalOption>,
 )> {
     match (&mut plan.operator, plan.childrens.as_mut()) {
-        (Operator::TableScan(scan), _) if scan.limit == (None, None) => {
+        (Operator::TableScan(scan), _)
+            if scan.limit == (None, None) && scan.predicate.is_none() =>
+        {
             Some((scan, &mut plan.physical_option))
         }
         (Operator::Sort(_), Childrens::Only(input)) => predicate_scan_mut(input, arena),

src/optimizer/rule/implementation/dql/filter.rs

@@ -177,13 +177,24 @@ fn estimate_filter_rows(
     ) {
         return Ok(None);
     }
+    estimate_predicate_rows(filter.predicate, input, rows, arena, loader)
+}
+
+/// Rows of the `rows` from `input` that `predicate` keeps, when it is a point filter.
+pub(crate) fn estimate_predicate_rows(
+    predicate: ExprRef,
+    input: &LogicalPlan,
+    rows: usize,
+    arena: &PlanArena,
+    loader: &StatisticMetaLoader<'_>,
+) -> Result<Option<usize>, DatabaseError> {
     let mut visitor = PointFilterVisitor {
         arena,
         column: None,
         values: HashSet::new(),
         supported: true,
     };
-    visitor.visit(filter.predicate, arena)?;
+    visitor.visit(predicate, arena)?;
     let Some(column) = visitor.column.filter(|_| visitor.supported) else {
         return Ok(None);
     };
@@ -241,6 +252,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: vec![],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );

src/optimizer/rule/implementation/dql/table_scan.rs

@@ -19,6 +19,7 @@ use crate::optimizer::core::rule::{
     keep_best_physical_option, BestPhysicalOption, ImplementationRule, MatchPattern,
 };
 use crate::optimizer::core::statistics_meta::StatisticMetaLoader;
+use crate::optimizer::rule::implementation::dql::filter::estimate_predicate_rows;
 use crate::planner::operator::{Operator, PhysicalOption, PlanImpl, RowCountOption, SortOption};
 use crate::planner::{LogicalPlan, PlanArena};
 use crate::types::index::{IndexLookup, IndexType};
@@ -61,12 +62,23 @@ impl ImplementationRule for SeqScanImplementation {
                 .flatten()
                 .map(|index_statistics| index_statistics.histogram().values_len());
 
-            let option = PhysicalOption::new(
+            let mut rows = cost;
+            let mut filtered_rows = None;
+            if let (Some(predicate), Some(scanned)) = (scan_op.predicate, cost) {
+                if let Some(kept) =
+                    estimate_predicate_rows(predicate, plan, scanned, arena, loader)?
+                {
+                    rows = Some(kept);
+                    filtered_rows = Some(scanned - kept);
+                }
+            }
+            let mut option = PhysicalOption::new(
                 PlanImpl::SeqScan,
                 SortOption::None,
-                cost.map(|rows| RowCountOption::Estimated { rows, ndv: None })
+                rows.map(|rows| RowCountOption::Estimated { rows, ndv: None })
                     .unwrap_or(RowCountOption::Unknown),
             );
+            option.filtered_rows = filtered_rows;
             keep_best_physical_option(best_physical_option, option, cost);
             Ok(())
         } else {