diff --git a/src/db.rs b/src/db.rs
index 47e49870..85f5caa0 100644
--- a/src/db.rs
+++ b/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)]"
             );
         }
 
diff --git a/src/db/program.rs b/src/db/program.rs
index 9f9ebc3c..16b4ac10 100644
--- a/src/db/program.rs
+++ b/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,
diff --git a/src/execution/dql/aggregate/hash_agg.rs b/src/execution/dql/aggregate/hash_agg.rs
index cf08dd19..4c94a252 100644
--- a/src/execution/dql/aggregate/hash_agg.rs
+++ b/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));
diff --git a/src/execution/dql/index_scan.rs b/src/execution/dql/index_scan.rs
index b768ce96..308d8563 100644
--- a/src/execution/dql/index_scan.rs
+++ b/src/execution/dql/index_scan.rs
@@ -13,6 +13,7 @@
 // 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,
 };
@@ -20,7 +21,7 @@ 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(),
diff --git a/src/execution/dql/seq_scan.rs b/src/execution/dql/seq_scan.rs
index 97cf14aa..98957f27 100644
--- a/src/execution/dql/seq_scan.rs
+++ b/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);
diff --git a/src/optimizer/plan_utils.rs b/src/optimizer/plan_utils.rs
index db057bcd..d5bc0a32 100644
--- a/src/optimizer/plan_utils.rs
+++ b/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),
diff --git a/src/optimizer/rule/implementation/dql/filter.rs b/src/optimizer/rule/implementation/dql/filter.rs
index c73612d1..c69040ef 100644
--- a/src/optimizer/rule/implementation/dql/filter.rs
+++ b/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,
         );
diff --git a/src/optimizer/rule/implementation/dql/table_scan.rs b/src/optimizer/rule/implementation/dql/table_scan.rs
index bdc59c8c..42240783 100644
--- a/src/optimizer/rule/implementation/dql/table_scan.rs
+++ b/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 {
diff --git a/src/optimizer/rule/normalization/column_pruning.rs b/src/optimizer/rule/normalization/column_pruning.rs
index aa023d09..7fd84e3f 100644
--- a/src/optimizer/rule/normalization/column_pruning.rs
+++ b/src/optimizer/rule/normalization/column_pruning.rs
@@ -27,9 +27,10 @@ use crate::planner::operator::visitor::{ChildrenVisit, OperatorExprVisitor, Oper
 use crate::planner::operator::visitor_mut::{
     ChildrenVisitMut, OperatorExprVisitorMut, OperatorVisitorMut,
 };
-use crate::planner::operator::Operator;
+use crate::planner::operator::{Operator, PhysicalOption, PlanImpl};
 use crate::planner::MetaArena;
 use crate::planner::{Childrens, ExprRef, LogicalPlan, PlanArena};
+use crate::types::index::IndexType;
 use crate::types::value::{DataValue, Utf8Type};
 use crate::types::CharLengthUnits;
 use crate::types::LogicalType;
@@ -489,6 +490,8 @@ impl ColumnPruning {
                         }
                         keep
                     });
+                    // Read by column above; an all-referenced scan stays pinned.
+                    op.deserializers.unpin();
                     if outcome.removed_positions.len() > output_start {
                         changed = true;
                     }
@@ -818,6 +821,89 @@ impl NormalizationRule for ColumnPruning {
     }
 }
 
+/// A covered index read leaves NULL the unpinned scan columns nothing reads, skipping their
+/// index fields: no operator above references them, nor does the scan's predicate, which a
+/// chosen index range has left the residual.
+#[derive(Clone)]
+pub struct SkipResidualFilterColumns;
+
+impl SkipResidualFilterColumns {
+    fn collect(
+        plan: &LogicalPlan,
+        referenced: &mut ReferencedColumns,
+        arena: &mut PlanArena,
+    ) -> Result<(), DatabaseError> {
+        match &plan.operator {
+            Operator::TableScan(scan) => ColumnPruning::extend_expr_referenced_columns(
+                scan.predicate.iter(),
+                referenced,
+                arena,
+            )?,
+            operator => {
+                ColumnPruning::extend_operator_referenced_columns(operator, referenced, arena)?
+            }
+        }
+        for child in plan.childrens.iter() {
+            Self::collect(child, referenced, arena)?;
+        }
+        Ok(())
+    }
+
+    fn skip(plan: &mut LogicalPlan, referenced: &ReferencedColumns, arena: &PlanArena) -> bool {
+        let mut changed = false;
+        match plan.childrens.as_mut() {
+            Childrens::None => (),
+            Childrens::Only(child) => changed |= Self::skip(child, referenced, arena),
+            Childrens::Twins { left, right } => {
+                changed |= Self::skip(left, referenced, arena);
+                changed |= Self::skip(right, referenced, arena);
+            }
+        }
+        let (
+            Operator::TableScan(scan),
+            Some(PhysicalOption {
+                plan: PlanImpl::IndexScan(info),
+                ..
+            }),
+        ) = (&plan.operator, plan.physical_option.as_mut())
+        else {
+            return changed;
+        };
+        // A primary key read decodes the table row rather than index fields.
+        if !info.covered || matches!(arena.index(info.meta).ty, IndexType::PrimaryKey { .. }) {
+            return changed;
+        }
+        let len = scan.columns().count();
+        for (position, column) in scan.deserializers.read_columns().enumerate() {
+            let skipped = info
+                .cover_mapping
+                .as_ref()
+                .is_some_and(|slots| slots[position] == usize::MAX);
+            if !column.pinned && !skipped && !referenced.contains(column.column, arena) {
+                info.cover_mapping.get_or_insert_with(|| (0..len).collect())[position] = usize::MAX;
+                changed = true;
+            }
+        }
+        changed
+    }
+}
+
+impl NormalizationRule for SkipResidualFilterColumns {
+    fn apply(
+        &self,
+        plan: &mut LogicalPlan,
+        arena: &mut PlanArena,
+        _loader: Option<&StatisticMetaLoader<'_>>,
+    ) -> Result<bool, DatabaseError> {
+        let mut referenced = ReferencedColumns::with_arena_capacity(arena);
+        Self::collect(plan, &mut referenced, arena)?;
+
+        let output = plan.output_schema(arena);
+        referenced.extend(output.iter().copied(), arena);
+        Ok(Self::skip(plan, &referenced, arena))
+    }
+}
+
 // GRCOV_EXCL_START
 #[cfg(all(test, not(target_arch = "wasm32")))]
 mod tests {
@@ -1087,5 +1173,93 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn skip_residual_filter_columns_in_covered_index_reads() -> Result<(), DatabaseError> {
+        use crate::db::DataBaseBuilder;
+        use crate::planner::operator::{PhysicalOption, PlanImpl};
+        use crate::types::tuple::TupleLike;
+        use crate::types::value::DataValue;
+
+        fn find_scan(plan: &LogicalPlan) -> Option<&LogicalPlan> {
+            matches!(plan.operator, Operator::TableScan(_))
+                .then_some(plan)
+                .or_else(|| plan.childrens.iter().find_map(find_scan))
+        }
+
+        let mut db = DataBaseBuilder::path(".")
+            .histogram_buckets(2)
+            .build_in_memory()?;
+        db.ddl("create table skip_t(id int primary key, a int, b int)")?;
+        db.ddl("create index skip_ab on skip_t(a, b)")?;
+        db.run("insert into skip_t values(1, 1, 10), (2, 1, 20), (3, 2, 30)")?
+            .done()?;
+        db.analyze("skip_t")?;
+        let mut tx = db.new_transaction(false)?;
+
+        for (sql, mapping, expected) in [
+            // The range eliminates `a = 1`, so nothing reads `a`.
+            (
+                "select b from skip_t where a = 1",
+                Some(vec![usize::MAX, 1]),
+                vec![10, 20],
+            ),
+            // The whole predicate is the range, so `b` goes unread.
+            (
+                "select a from skip_t where a = 1 and b > 10",
+                Some(vec![0, usize::MAX]),
+                vec![1],
+            ),
+            // The residual filter still reads `b`.
+            (
+                "select a from skip_t where a = 1 and b % 7 = 6",
+                None,
+                vec![1],
+            ),
+            // The query output reads every column.
+            ("select a, b from skip_t where a = 1", None, vec![1, 1]),
+        ] {
+            let mut program = db.program_sql(sql, &[])?;
+            let scan = find_scan(program.executor.plan()).expect("table scan");
+            let Some(PhysicalOption {
+                plan: PlanImpl::IndexScan(info),
+                ..
+            }) = &scan.physical_option
+            else {
+                panic!("expected index scan: {sql}")
+            };
+            assert!(info.covered, "{sql}");
+            assert_eq!(info.cover_mapping, mapping, "{sql}");
+
+            let mut iter = program.execute(&mut tx, [])?;
+            let mut rows = Vec::new();
+            while iter
+                .next_tuple(|_, row| rows.push(row.value_at(0).clone()))?
+                .is_some()
+            {}
+            iter.done()?;
+            let expected: Vec<_> = expected.into_iter().map(DataValue::Int32).collect();
+            assert_eq!(rows, expected, "{sql}");
+        }
+
+        // An update writes back the whole row it scans, so its scan stays pinned.
+        drop(tx);
+        db.run("update skip_t set b = b + 1 where a = 1")?.done()?;
+        let mut iter = db.run("select id, a, b from skip_t order by id")?;
+        let mut rows = Vec::new();
+        while iter
+            .next_tuple(|_, row| {
+                rows.push((0..3).map(|i| row.value_at(i).clone()).collect::<Vec<_>>())
+            })?
+            .is_some()
+        {}
+        iter.done()?;
+        let expected: Vec<Vec<_>> = [[1, 1, 11], [2, 1, 21], [3, 2, 30]]
+            .into_iter()
+            .map(|row| row.into_iter().map(DataValue::Int32).collect())
+            .collect();
+        assert_eq!(rows, expected);
+        Ok(())
+    }
 }
 // GRCOV_EXCL_STOP
diff --git a/src/optimizer/rule/normalization/elimination.rs b/src/optimizer/rule/normalization/elimination.rs
index 3bb75ba0..232b4ef7 100644
--- a/src/optimizer/rule/normalization/elimination.rs
+++ b/src/optimizer/rule/normalization/elimination.rs
@@ -330,6 +330,23 @@ impl NormalizationRule for EliminateIndexFilter {
         arena: &mut PlanArena,
         _loader: Option<&StatisticMetaLoader<'_>>,
     ) -> Result<bool, DatabaseError> {
+        if let (
+            Operator::TableScan(scan),
+            Some(PhysicalOption {
+                plan: PlanImpl::IndexScan(index_info),
+                ..
+            }),
+        ) = (&mut plan.operator, &plan.physical_option)
+        {
+            if scan.predicate.is_none()
+                || !matches!(index_info.lookup, Some(IndexLookup::Static(_)))
+                || scan.predicate == index_info.residual_predicate
+            {
+                return Ok(false);
+            }
+            scan.predicate = index_info.residual_predicate;
+            return Ok(true);
+        }
         if !matches!(plan.operator, Operator::Filter(_)) {
             return Ok(false);
         }
@@ -1370,6 +1387,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: vec![index_info.clone()],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
@@ -1434,6 +1452,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: vec![index_info],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
@@ -1752,6 +1771,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: vec![index_info],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
@@ -2000,6 +2020,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: vec![index_info],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
diff --git a/src/optimizer/rule/normalization/join_reorder.rs b/src/optimizer/rule/normalization/join_reorder.rs
index fd50a3c2..756c4260 100644
--- a/src/optimizer/rule/normalization/join_reorder.rs
+++ b/src/optimizer/rule/normalization/join_reorder.rs
@@ -757,6 +757,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: Vec::new(),
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
diff --git a/src/optimizer/rule/normalization/mod.rs b/src/optimizer/rule/normalization/mod.rs
index 6e80de89..d253dc1a 100644
--- a/src/optimizer/rule/normalization/mod.rs
+++ b/src/optimizer/rule/normalization/mod.rs
@@ -16,7 +16,9 @@ use crate::errors::DatabaseError;
 use crate::expression::visitor_mut::ExprVisitorMut;
 use crate::optimizer::core::rule::NormalizationRule;
 use crate::optimizer::core::statistics_meta::StatisticMetaLoader;
-use crate::optimizer::rule::normalization::column_pruning::ColumnPruning;
+use crate::optimizer::rule::normalization::column_pruning::{
+    ColumnPruning, SkipResidualFilterColumns,
+};
 use crate::optimizer::rule::normalization::combine_operators::{CollapseGroupByAgg, CombineFilter};
 use crate::optimizer::rule::normalization::compilation_in_advance::EvaluatorBind;
 use crate::optimizer::rule::normalization::constant_propagation::{
@@ -65,6 +67,7 @@ pub(crate) use simplification::constant_calculation_current;
 #[derive(Debug, Copy, Clone)]
 pub enum NormalizationRuleImpl {
     ColumnPruning,
+    SkipResidualFilterColumns,
     // Combine operators
     EliminateProject,
     CollapseGroupByAgg,
@@ -171,7 +174,8 @@ pub enum NormalizationPassKind {
 impl NormalizationRuleImpl {
     pub fn pass_kind(&self) -> NormalizationPassKind {
         match self {
-            NormalizationRuleImpl::ColumnPruning => {
+            NormalizationRuleImpl::ColumnPruning
+            | NormalizationRuleImpl::SkipResidualFilterColumns => {
                 NormalizationPassKind::WholeTreePass(WholeTreePassKind::ColumnPruning)
             }
             NormalizationRuleImpl::ConstantCalculation | NormalizationRuleImpl::EvaluatorBind => {
@@ -183,7 +187,8 @@ impl NormalizationRuleImpl {
 
     pub fn root_tag(&self) -> NormalizationRuleRootTag {
         match self {
-            NormalizationRuleImpl::ColumnPruning => NormalizationRuleRootTag::Any,
+            NormalizationRuleImpl::ColumnPruning
+            | NormalizationRuleImpl::SkipResidualFilterColumns => NormalizationRuleRootTag::Any,
             NormalizationRuleImpl::EliminateProject => NormalizationRuleRootTag::Any,
             NormalizationRuleImpl::CollapseGroupByAgg => NormalizationRuleRootTag::Aggregate,
             NormalizationRuleImpl::GuardMarkApply => NormalizationRuleRootTag::Project,
@@ -201,7 +206,8 @@ impl NormalizationRuleImpl {
             NormalizationRuleImpl::ConstantCalculation => NormalizationRuleRootTag::Any,
             NormalizationRuleImpl::EvaluatorBind => NormalizationRuleRootTag::Any,
             NormalizationRuleImpl::MinMaxToTopK => NormalizationRuleRootTag::Aggregate,
-            NormalizationRuleImpl::EliminateIndexFilter => NormalizationRuleRootTag::Filter,
+            // Also rewrites the predicates scans hold.
+            NormalizationRuleImpl::EliminateIndexFilter => NormalizationRuleRootTag::Any,
             NormalizationRuleImpl::ParameterizeInnerJoin
             | NormalizationRuleImpl::GreedyJoinReorder => NormalizationRuleRootTag::Join,
             NormalizationRuleImpl::ParameterizeMarkApply
@@ -219,6 +225,9 @@ impl NormalizationRule for NormalizationRuleImpl {
     ) -> Result<bool, DatabaseError> {
         match self {
             NormalizationRuleImpl::ColumnPruning => ColumnPruning.apply(plan, arena, loader),
+            NormalizationRuleImpl::SkipResidualFilterColumns => {
+                SkipResidualFilterColumns.apply(plan, arena, loader)
+            }
             NormalizationRuleImpl::GreedyJoinReorder => {
                 GreedyJoinReorder.apply(plan, arena, loader)
             }
diff --git a/src/optimizer/rule/normalization/parameterized_index.rs b/src/optimizer/rule/normalization/parameterized_index.rs
index eb23e91e..937356ec 100644
--- a/src/optimizer/rule/normalization/parameterized_index.rs
+++ b/src/optimizer/rule/normalization/parameterized_index.rs
@@ -331,6 +331,10 @@ pub(crate) fn probe_index(
             probe
         }
         (Operator::TableScan(scan), _) if scan.limit == (None, None) => {
+            let key_count = keys.len();
+            if let Some(predicate) = scan.predicate {
+                constant_keys(predicate, keys, arena);
+            }
             let mut best: Option<ProbeIndex> = None;
             let mut probe = Vec::new();
             'indexes: for (position, info) in scan.index_infos.iter().enumerate() {
@@ -387,6 +391,7 @@ pub(crate) fn probe_index(
                     });
                 }
             }
+            keys.truncate(key_count);
             Ok(best)
         }
         _ => Ok(None),
@@ -429,11 +434,16 @@ fn use_probe_index(plan: &mut LogicalPlan, position: usize, arena: &PlanArena<'_
             let info = &mut scan.index_infos[position];
             info.lookup = Some(IndexLookup::Probe);
             info.residual_predicate = None;
-            (info.covered, info.cover_mapping) =
-                match index_covering(scan.deserializers.read_columns(), info.meta, arena) {
-                    Some(mapping) => (true, mapping),
-                    None => (false, None),
-                };
+            (info.covered, info.cover_mapping) = match index_covering(
+                scan.deserializers
+                    .read_columns()
+                    .map(|column| column.column),
+                info.meta,
+                arena,
+            ) {
+                Some(mapping) => (true, mapping),
+                None => (false, None),
+            };
             plan.physical_option = Some(PhysicalOption::new(
                 PlanImpl::IndexScan(Box::new(info.clone())),
                 info.sort_option.clone(),
@@ -815,6 +825,7 @@ mod inner_join_tests {
                 limit: (None, None),
                 index_infos,
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         )
diff --git a/src/optimizer/rule/normalization/pushdown_limit.rs b/src/optimizer/rule/normalization/pushdown_limit.rs
index 84f1a46c..a665cf3a 100644
--- a/src/optimizer/rule/normalization/pushdown_limit.rs
+++ b/src/optimizer/rule/normalization/pushdown_limit.rs
@@ -124,6 +124,9 @@ impl NormalizationRule for PushLimitIntoScan {
 
         if let Some(child) = only_child_mut(plan) {
             if let Operator::TableScan(scan_op) = &mut child.operator {
+                if scan_op.predicate.is_some() {
+                    return Ok(false);
+                }
                 scan_op.limit = (offset, limit);
                 let removed = replace_with_only_child(plan);
                 return Ok(removed);
diff --git a/src/optimizer/rule/normalization/pushdown_predicates.rs b/src/optimizer/rule/normalization/pushdown_predicates.rs
index 13356896..f8de79f5 100644
--- a/src/optimizer/rule/normalization/pushdown_predicates.rs
+++ b/src/optimizer/rule/normalization/pushdown_predicates.rs
@@ -326,6 +326,8 @@ impl NormalizationRule for PushPredicateIntoScan {
         let Some((scan_op, _)) = predicate_scan_mut(child, arena) else {
             return Ok(false);
         };
+        // Every residual must come from this filter for the scan to take it as its predicate.
+        let fresh = scan_op.index_infos.iter().all(|info| info.lookup.is_none());
 
         let mut changed = false;
         for IndexInfo {
@@ -363,13 +365,23 @@ impl NormalizationRule for PushPredicateIntoScan {
 
             *lookup = Some(IndexLookup::Static(detached.range));
             *residual_predicate = detached.residual;
-            let columns = scan_op.deserializers.read_columns();
+            let columns = scan_op
+                .deserializers
+                .read_columns()
+                .map(|column| column.column);
             (*covered, *cover_mapping) = match index_covering(columns, *meta, arena) {
                 Some(mapping) => (true, mapping),
                 None => (false, None),
             };
         }
 
+        let (predicate, having) = (filter_op.predicate, filter_op.having);
+        if let Operator::TableScan(scan_op) = &mut child.operator {
+            if fresh && !having {
+                scan_op.predicate = Some(predicate);
+                return Ok(replace_with_only_child(plan));
+            }
+        }
         Ok(changed)
     }
 }
@@ -753,8 +765,10 @@ mod tests {
                 &mut arena,
             )?;
 
-            let mut scan = best_plan.childrens.pop_only().childrens.pop_only();
+            // The filter becomes the scan's predicate unless initialization separates them.
+            let mut scan = best_plan.childrens.pop_only();
             if through_init {
+                scan = scan.childrens.pop_only();
                 assert!(matches!(scan.operator, Operator::ScalarQueryInit(_)));
                 scan = scan.childrens.pop_twins().0;
             }
@@ -834,6 +848,7 @@ mod tests {
                         stream_aggregate_hint: None,
                     }],
                     with_pk: false,
+                    predicate: None,
                 }),
                 Childrens::None,
             );
@@ -846,8 +861,7 @@ mod tests {
                 ),
                 &mut arena,
             )?;
-            let scan = plan.childrens.pop_only();
-            let Operator::TableScan(op) = scan.operator else {
+            let Operator::TableScan(op) = plan.operator else {
                 unreachable!()
             };
             assert_eq!(
@@ -1084,6 +1098,7 @@ mod tests {
                     },
                 ],
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         );
@@ -1111,8 +1126,7 @@ mod tests {
             &mut arena,
         )?;
 
-        let table_scan = best_plan.childrens.pop_only();
-        if let Operator::TableScan(op) = &table_scan.operator {
+        if let Operator::TableScan(op) = &best_plan.operator {
             let index_infos = &op.index_infos;
             assert_eq!(index_infos.len(), 2);
 
diff --git a/src/orm/mod.rs b/src/orm/mod.rs
index 8fdb6d58..93e1e693 100644
--- a/src/orm/mod.rs
+++ b/src/orm/mod.rs
@@ -3999,8 +3999,8 @@ mod tests {
             plan,
             concat!(
                 "Projection [orm_unit_users.name] [Project => (Sort Option: Follow)] ",
-                "Filter (orm_unit_users.id >= 4), Is Having: false [Filter => (Sort Option: Follow)] ",
-                "TableScan orm_unit_users -> [orm_unit_users.id, orm_unit_users.name] [SeqScan => (Sort Option: None)]"
+                "TableScan orm_unit_users -> [orm_unit_users.id, orm_unit_users.name], ",
+                "Filter: (orm_unit_users.id >= 4) [SeqScan => (Sort Option: None)]"
             ),
             "{plan}"
         );
@@ -4037,8 +4037,8 @@ mod tests {
             concat!(
                 "Projection [upper_name] [Project => (Sort Option: Follow)] ",
                 "Sort By orm_unit_users.age Desc Nulls Last [Sort => (Sort Option: OrderBy: (orm_unit_users.age Desc Nulls Last) ignore_prefix_len: 0)] ",
-                "Filter ((orm_unit_users.age is not null && ((orm_unit_users.age >= 18) && (orm_unit_users.age <= 25))) && ((orm_unit_users.name != Bob) && !(orm_unit_users.name = Missing))), Is Having: false ",
-                "[Filter => (Sort Option: Follow)] TableScan orm_unit_users -> [orm_unit_users.name, orm_unit_users.age] [SeqScan => (Sort Option: None)]"
+                "TableScan orm_unit_users -> [orm_unit_users.name, orm_unit_users.age], ",
+                "Filter: ((orm_unit_users.age is not null && ((orm_unit_users.age >= 18) && (orm_unit_users.age <= 25))) && ((orm_unit_users.name != Bob) && !(orm_unit_users.name = Missing))) [SeqScan => (Sort Option: None)]"
             ),
             "{expression_plan}"
         );
@@ -4060,8 +4060,7 @@ mod tests {
             concat!(
                 "Projection [orm_unit_users.id] [Project => (Sort Option: Follow)] ",
                 "Sort By 1 Asc Nulls Last [Sort => (Sort Option: OrderBy: (1 Asc Nulls Last) ignore_prefix_len: 0)] ",
-                "Filter (orm_unit_users.id != 3), Is Having: false ",
-                "[Filter => (Sort Option: Follow)] TableScan orm_unit_users -> [orm_unit_users.id] ",
+                "TableScan orm_unit_users -> [orm_unit_users.id], Filter: (orm_unit_users.id != 3) ",
                 "[IndexScan By pk_index => 1, 2, 3 Covered => (Sort Option: OrderBy: (orm_unit_users.id Asc Nulls Last) ignore_prefix_len: 0)]"
             ),
             "{list_plan}"
@@ -4082,8 +4081,8 @@ mod tests {
             nullable_plan,
             concat!(
                 "Projection [orm_unit_users.id] [Project => (Sort Option: Follow)] ",
-                "Filter (((orm_unit_users.age < 10) || (orm_unit_users.age > 30)) || orm_unit_users.age is null), Is Having: false ",
-                "[Filter => (Sort Option: Follow)] TableScan orm_unit_users -> [orm_unit_users.id, orm_unit_users.age] [SeqScan => (Sort Option: None)]"
+                "TableScan orm_unit_users -> [orm_unit_users.id, orm_unit_users.age], ",
+                "Filter: (((orm_unit_users.age < 10) || (orm_unit_users.age > 30)) || orm_unit_users.age is null) [SeqScan => (Sort Option: None)]"
             ),
             "{nullable_plan}"
         );
diff --git a/src/planner/mod.rs b/src/planner/mod.rs
index bf8df675..50e73dd2 100644
--- a/src/planner/mod.rs
+++ b/src/planner/mod.rs
@@ -601,6 +601,7 @@ mod tests {
                 limit: (None, None),
                 index_infos: Vec::new(),
                 with_pk: false,
+                predicate: None,
             }),
             Childrens::None,
         )
diff --git a/src/planner/operator/mod.rs b/src/planner/operator/mod.rs
index e35b7f14..28753450 100644
--- a/src/planner/operator/mod.rs
+++ b/src/planner/operator/mod.rs
@@ -272,6 +272,7 @@ pub struct PhysicalOption {
     pub(crate) sort_option: SortOption,
     pub(crate) row_count: RowCountOption,
     pub(crate) lookup_work: Option<usize>,
+    pub(crate) filtered_rows: Option<usize>,
     pub(crate) join_decided: bool,
 }
 
@@ -282,6 +283,7 @@ impl PhysicalOption {
             sort_option,
             row_count,
             lookup_work: None,
+            filtered_rows: None,
             join_decided: false,
         }
     }
@@ -326,6 +328,12 @@ impl PhysicalOption {
             | (PlanImpl::Project, Operator::Project(_), ChildCosts::Only(child)) => {
                 Some(input(child)?.work)
             }
+            (_, Operator::TableScan(_), _) if self.filtered_rows.is_some() => {
+                Some(cost.startup?.saturating_add(rows_work(
+                    rows?.saturating_add(self.filtered_rows?),
+                    cost.width,
+                )))
+            }
             (PlanImpl::MarkApply, Operator::MarkApply(op), ChildCosts::Twins(left, right))
                 if op.parameterized_probe().is_some() && !op.materialize =>
             {
@@ -1495,6 +1503,7 @@ mod tests {
             limit: (None, None),
             index_infos: Vec::new(),
             with_pk: false,
+            predicate: None,
         });
         assert_eq!(referenced_columns(&table_scan, &mut arena)?, vec![a, d]);
 
diff --git a/src/planner/operator/table_scan.rs b/src/planner/operator/table_scan.rs
index f17f97f9..7a3cbd41 100644
--- a/src/planner/operator/table_scan.rs
+++ b/src/planner/operator/table_scan.rs
@@ -18,7 +18,7 @@ use crate::errors::DatabaseError;
 use crate::expression::ScalarExpression;
 use crate::planner::operator::sort::SortField;
 use crate::planner::MetaArena;
-use crate::planner::{fmt_explain_list, fmt_table, Childrens, Explain, LogicalPlan};
+use crate::planner::{fmt_explain_list, fmt_table, Childrens, Explain, ExprRef, LogicalPlan};
 use crate::storage::{Bounds, TupleDeserializers};
 use crate::types::index::IndexInfo;
 use kite_sql_serde_macros::ReferenceSerialization;
@@ -34,11 +34,14 @@ pub struct TableScanOperator {
     // If pre_where is simple predicate, for example:  a > 1 then can calculate directly when read data.
     pub(crate) index_infos: Vec<IndexInfo>,
     pub(crate) with_pk: bool,
+    pub(crate) predicate: Option<ExprRef>,
 }
 
 impl TableScanOperator {
     pub(crate) fn columns(&self) -> impl Iterator<Item = ColumnRef> + '_ {
-        self.deserializers.read_columns()
+        self.deserializers
+            .read_columns()
+            .map(|column| column.column)
     }
 
     pub fn build(
@@ -87,6 +90,7 @@ impl TableScanOperator {
                 deserializers,
                 limit: (None, None),
                 with_pk,
+                predicate: None,
             }),
             Childrens::None,
         ))
@@ -104,6 +108,9 @@ impl Explain for TableScanOperator {
         f.write_str(" -> [")?;
         fmt_explain_list(self.columns(), ", ", arena, f)?;
         f.write_str("]")?;
+        if let Some(predicate) = self.predicate {
+            write!(f, ", Filter: {}", predicate.explain(arena))?;
+        }
         let (offset, limit) = self.limit;
         if let Some(limit) = limit {
             write!(f, ", Limit: {limit}")?;
diff --git a/src/planner/operator/visitor.rs b/src/planner/operator/visitor.rs
index de08a31c..190bdcb9 100644
--- a/src/planner/operator/visitor.rs
+++ b/src/planner/operator/visitor.rs
@@ -441,6 +441,9 @@ impl<'a, V: ExprVisitor<A>, A: MetaArena + ?Sized> OperatorVisitor<'a>
     }
 
     fn visit_table_scan(&mut self, op: &'a TableScanOperator) -> Result<(), DatabaseError> {
+        if let Some(expr) = op.predicate {
+            ExprVisitor::visit(self.visitor, expr, self.arena)?;
+        }
         for index_info in &op.index_infos {
             if let SortOption::OrderBy { fields, .. } = &index_info.sort_option {
                 for field in fields {
@@ -698,6 +701,7 @@ pub(crate) mod tests {
                 limit: (None, None),
                 index_infos: vec![index_info(expr(10), expr(11))],
                 with_pk: false,
+                predicate: None,
             }),
             Operator::FunctionScan(FunctionScanOperator { table_function }),
             Operator::Sort(SortOperator {
diff --git a/src/planner/operator/visitor_mut.rs b/src/planner/operator/visitor_mut.rs
index 4b9166be..3b5a99d5 100644
--- a/src/planner/operator/visitor_mut.rs
+++ b/src/planner/operator/visitor_mut.rs
@@ -458,6 +458,9 @@ impl<'a, V: ExprVisitorMut<A>, A: MetaArena + ?Sized> OperatorVisitorMut<'a>
     }
 
     fn visit_table_scan(&mut self, op: &'a mut TableScanOperator) -> Result<(), DatabaseError> {
+        if let Some(expr) = &mut op.predicate {
+            ExprVisitorMut::visit(self.visitor, expr, self.arena)?;
+        }
         for index_info in &mut op.index_infos {
             if let SortOption::OrderBy { fields, .. } = &mut index_info.sort_option {
                 for field in fields {
diff --git a/src/serdes/physical_option.rs b/src/serdes/physical_option.rs
index ebf0a955..85628222 100644
--- a/src/serdes/physical_option.rs
+++ b/src/serdes/physical_option.rs
@@ -78,6 +78,7 @@ impl ReferenceSerialization for PhysicalOption {
             sort_option: SortOption::decode(reader, context.as_deref_mut(), arena)?,
             row_count: RowCountOption::decode(reader, context, arena)?,
             lookup_work: None,
+            filtered_rows: None,
             join_decided: false,
         })
     }
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index 1395d868..bd43bb7c 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -1404,6 +1404,11 @@ impl TupleMapping {
         let mut target_len = 0;
 
         for index_idx in scan_to_index {
+            // No index field fills a `usize::MAX` slot, which stays NULL.
+            if index_idx == usize::MAX {
+                target_len += 1;
+                continue;
+            }
             if index_idx >= index_to_scan.len() {
                 index_to_scan.resize(index_idx + 1, usize::MAX);
             }
@@ -1931,6 +1936,7 @@ fn eq_to_res_scope<'a, T: StorageTransaction + 'a>(
 pub struct ScanColumn {
     pub(crate) column: ColumnRef,
     pub(crate) deserializer: TupleValueSerializableImpl,
+    pub(crate) pinned: bool,
 }
 
 impl ScanColumn {
@@ -1957,6 +1963,7 @@ impl TupleDeserializers {
                 .map(|column| ScanColumn {
                     column: *column,
                     deserializer: arena.column(*column).datatype().serializable(),
+                    pinned: true,
                 })
                 .collect(),
             total_len: table.columns_len(),
@@ -1964,11 +1971,8 @@ impl TupleDeserializers {
         }
     }
 
-    pub(crate) fn read_columns(&self) -> impl Iterator<Item = ColumnRef> + '_ {
-        self.columns
-            .iter()
-            .filter(|column| column.is_read())
-            .map(|column| column.column)
+    pub(crate) fn read_columns(&self) -> impl Iterator<Item = &ScanColumn> + '_ {
+        self.columns.iter().filter(|column| column.is_read())
     }
 
     pub(crate) fn retain(
@@ -1996,6 +2000,12 @@ impl TupleDeserializers {
         self.columns.truncate(read_len);
     }
 
+    pub(crate) fn unpin(&mut self) {
+        for column in &mut self.columns {
+            column.pinned = false;
+        }
+    }
+
     fn deserializers(&self) -> impl Iterator<Item = &TupleValueSerializableImpl> {
         self.columns.iter().map(|column| &column.deserializer)
     }
@@ -2530,6 +2540,7 @@ pub(crate) fn deserializers_for_test(
         .map(|column| ScanColumn {
             column,
             deserializer: TupleValueSerializableImpl::Boolean,
+            pinned: true,
         })
         .collect();
     TupleDeserializers {
diff --git a/tests/slt/crdb/join.slt b/tests/slt/crdb/join.slt
index b628f09e..b97c769c 100644
--- a/tests/slt/crdb/join.slt
+++ b/tests/slt/crdb/join.slt
@@ -1211,7 +1211,7 @@ insert into ij_c values (1,1),(2,2);
 query T
 explain select a.id,b.id from ij_a a,ij_b b where b.k=a.k and a.v>15 and b.v<250 and a.v<b.v;
 ----
-Projection [<slt:ignore>.id, <slt:ignore>.id] [Project => (Sort Option: Follow)] Filter (<slt:ignore>.v < <slt:ignore>.v), Is Having: false [Filter => (Sort Option: Follow)] Inner Join On cast (<slt:ignore>.k as Bigint) = <slt:ignore>.k [HashJoin => (Sort Option: None)] Projection [(ij_a.id) as (<slt:ignore>.id), (ij_a.k) as (<slt:ignore>.k), (ij_a.v) as (<slt:ignore>.v)] [Project => (Sort Option: Follow)] Filter ((ij_a.v) as (<slt:ignore>.v) > 15), Is Having: false [Filter => (Sort Option: Follow)] TableScan ij_a -> [ij_a.id, ij_a.k, ij_a.v] [SeqScan => (Sort Option: None)] Projection [(ij_b.id) as (<slt:ignore>.id), (ij_b.k) as (<slt:ignore>.k), (ij_b.v) as (<slt:ignore>.v)] [Project => (Sort Option: Follow)] Filter ((ij_b.v) as (<slt:ignore>.v) < 250), Is Having: false [Filter => (Sort Option: Follow)] TableScan ij_b -> [ij_b.id, ij_b.k, ij_b.v] [SeqScan => (Sort Option: None)]
+Projection [<slt:ignore>.id, <slt:ignore>.id] [Project => (Sort Option: Follow)] Filter (<slt:ignore>.v < <slt:ignore>.v), Is Having: false [Filter => (Sort Option: Follow)] Inner Join On cast (<slt:ignore>.k as Bigint) = <slt:ignore>.k [HashJoin => (Sort Option: None)] Projection [(ij_a.id) as (<slt:ignore>.id), (ij_a.k) as (<slt:ignore>.k), (ij_a.v) as (<slt:ignore>.v)] [Project => (Sort Option: Follow)] TableScan ij_a -> [ij_a.id, ij_a.k, ij_a.v], Filter: ((ij_a.v) as (<slt:ignore>.v) > 15) [SeqScan => (Sort Option: None)] Projection [(ij_b.id) as (<slt:ignore>.id), (ij_b.k) as (<slt:ignore>.k), (ij_b.v) as (<slt:ignore>.v)] [Project => (Sort Option: Follow)] TableScan ij_b -> [ij_b.id, ij_b.k, ij_b.v], Filter: ((ij_b.v) as (<slt:ignore>.v) < 250) [SeqScan => (Sort Option: None)]
 
 query II
 select a.id,b.id from ij_a a,ij_b b where b.k=a.k and a.v>15 and b.v<250 and a.v<b.v order by a.id,b.id;
diff --git a/tests/slt/stream_distinct_explain.slt b/tests/slt/stream_distinct_explain.slt
index daa965c7..c6abb431 100644
--- a/tests/slt/stream_distinct_explain.slt
+++ b/tests/slt/stream_distinct_explain.slt
@@ -35,19 +35,19 @@ drop index distinct_t.distinct_t_c1_index;
 query T
 explain select distinct c1 from distinct_t where c1 < 10 and c1 > 0;
 ----
-Projection [distinct_t.c1] [Project => (Sort Option: Follow)] Aggregate [] -> Group By [distinct_t.c1] [HashAggregate => (Sort Option: None)] Filter ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)), Is Having: false [Filter => (Sort Option: Follow)] TableScan distinct_t -> [distinct_t.c1] [SeqScan => (Sort Option: None)]
+Projection [distinct_t.c1] [Project => (Sort Option: Follow)] Aggregate [] -> Group By [distinct_t.c1] [HashAggregate => (Sort Option: None)] TableScan distinct_t -> [distinct_t.c1], Filter: ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)) [SeqScan => (Sort Option: None)]
 
 # forced spill aggregate
 query T
 explain select /*+ FORCE_AGG_SPILL */ c1, count(c2) from distinct_t where c1 < 10 and c1 > 0 group by c1;
 ----
-Projection [distinct_t.c1, Count(distinct_t.c2)] [Project => (Sort Option: Follow)] Aggregate [Count(distinct_t.c2)] -> Group By [distinct_t.c1] [StreamAggregate => (Sort Option: Follow)] Sort By distinct_t.c1 Asc Nulls Last [Sort => (Sort Option: OrderBy: (distinct_t.c1 Asc Nulls Last) ignore_prefix_len: 0)] Filter ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)), Is Having: false [Filter => (Sort Option: Follow)] TableScan distinct_t -> [distinct_t.c1, distinct_t.c2] [SeqScan => (Sort Option: None)]
+Projection [distinct_t.c1, Count(distinct_t.c2)] [Project => (Sort Option: Follow)] Aggregate [Count(distinct_t.c2)] -> Group By [distinct_t.c1] [StreamAggregate => (Sort Option: Follow)] Sort By distinct_t.c1 Asc Nulls Last [Sort => (Sort Option: OrderBy: (distinct_t.c1 Asc Nulls Last) ignore_prefix_len: 0)] TableScan distinct_t -> [distinct_t.c1, distinct_t.c2], Filter: ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)) [SeqScan => (Sort Option: None)]
 
 # forced spill sort also satisfies order by
 query T
 explain select /*+ FORCE_AGG_SPILL */ c1, count(c2) from distinct_t where c1 < 10 and c1 > 0 group by c1 order by c1;
 ----
-Projection [distinct_t.c1, Count(distinct_t.c2)] [Project => (Sort Option: Follow)] Aggregate [Count(distinct_t.c2)] -> Group By [distinct_t.c1] [StreamAggregate => (Sort Option: Follow)] Sort By distinct_t.c1 Asc Nulls Last [Sort => (Sort Option: OrderBy: (distinct_t.c1 Asc Nulls Last) ignore_prefix_len: 0)] Filter ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)), Is Having: false [Filter => (Sort Option: Follow)] TableScan distinct_t -> [distinct_t.c1, distinct_t.c2] [SeqScan => (Sort Option: None)]
+Projection [distinct_t.c1, Count(distinct_t.c2)] [Project => (Sort Option: Follow)] Aggregate [Count(distinct_t.c2)] -> Group By [distinct_t.c1] [StreamAggregate => (Sort Option: Follow)] Sort By distinct_t.c1 Asc Nulls Last [Sort => (Sort Option: OrderBy: (distinct_t.c1 Asc Nulls Last) ignore_prefix_len: 0)] TableScan distinct_t -> [distinct_t.c1, distinct_t.c2], Filter: ((distinct_t.c1 < 10) && (distinct_t.c1 > 0)) [SeqScan => (Sort Option: None)]
 
 statement ok
 drop table distinct_t;
diff --git a/tests/slt/where.slt b/tests/slt/where.slt
index bd333692..5d94f1a8 100644
--- a/tests/slt/where.slt
+++ b/tests/slt/where.slt
@@ -871,7 +871,7 @@ analyze table cp_c;
 query T
 explain select cp_a.id, cp_b.id from cp_a join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = 4;
 ----
-Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] Filter (cp_a.c1 = 4), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.id, cp_a.c1] [SeqScan => (Sort Option: None)]
+Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] TableScan cp_a -> [cp_a.id, cp_a.c1], Filter: (cp_a.c1 = 4) [SeqScan => (Sort Option: None)]
 
 query II
 select cp_a.id, cp_b.id from cp_a join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = 4;
@@ -881,7 +881,7 @@ select cp_a.id, cp_b.id from cp_a join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 =
 query T
 explain select cp_a.id, cp_b.id from cp_a, cp_b where cp_a.c1 = cp_b.c1 and cp_a.c1 = 4;
 ----
-Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] Filter (cp_a.c1 = 4), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.id, cp_a.c1] [SeqScan => (Sort Option: None)]
+Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] TableScan cp_a -> [cp_a.id, cp_a.c1], Filter: (cp_a.c1 = 4) [SeqScan => (Sort Option: None)]
 
 query II
 select cp_a.id, cp_b.id from cp_a, cp_b where cp_a.c1 = cp_b.c1 and cp_a.c1 = 4;
@@ -892,7 +892,7 @@ select cp_a.id, cp_b.id from cp_a, cp_b where cp_a.c1 = cp_b.c1 and cp_a.c1 = 4;
 query T
 explain select cp_a.id, cp_b.id, cp_c.id from cp_a, cp_b, cp_c where cp_a.c1 = cp_b.c1 and cp_b.c1 = cp_c.c1 and cp_a.c1 = 4;
 ----
-Projection [cp_a.id, cp_b.id, cp_c.id] [Project => (Sort Option: Follow)] Inner Join On cp_c.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] TableScan cp_c -> [cp_c.id, cp_c.c1] [IndexScan By cp_c_c1 => 4 Covered => (Sort Option: OrderBy: (cp_c.c1 Asc Nulls Last) ignore_prefix_len: 0)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] Filter (cp_a.c1 = 4), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.id, cp_a.c1] [SeqScan => (Sort Option: None)]
+Projection [cp_a.id, cp_b.id, cp_c.id] [Project => (Sort Option: Follow)] Inner Join On cp_c.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] TableScan cp_c -> [cp_c.id, cp_c.c1] [IndexScan By cp_c_c1 => 4 Covered => (Sort Option: OrderBy: (cp_c.c1 Asc Nulls Last) ignore_prefix_len: 0)] Inner Join On cp_b.c1 = cp_a.c1 [HashJoin => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)] TableScan cp_a -> [cp_a.id, cp_a.c1], Filter: (cp_a.c1 = 4) [SeqScan => (Sort Option: None)]
 
 query III
 select cp_a.id, cp_b.id, cp_c.id from cp_a, cp_b, cp_c where cp_a.c1 = cp_b.c1 and cp_b.c1 = cp_c.c1 and cp_a.c1 = 4;
@@ -903,7 +903,7 @@ select cp_a.id, cp_b.id, cp_c.id from cp_a, cp_b, cp_c where cp_a.c1 = cp_b.c1 a
 query T
 explain select id from cp_b where c1 = c2 and c2 = 4;
 ----
-Projection [cp_b.id] [Project => (Sort Option: Follow)] Filter ((cp_b.c1 = cp_b.c2) && (cp_b.c2 = 4)), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_b -> [cp_b.id, cp_b.c1, cp_b.c2] [IndexScan By cp_b_c1 => 4 => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [cp_b.id] [Project => (Sort Option: Follow)] TableScan cp_b -> [cp_b.id, cp_b.c1, cp_b.c2], Filter: ((cp_b.c1 = cp_b.c2) && (cp_b.c2 = 4)) [IndexScan By cp_b_c1 => 4 => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)]
 
 query I
 select id from cp_b where c1 = c2 and c2 = 4;
@@ -914,7 +914,7 @@ select id from cp_b where c1 = c2 and c2 = 4;
 query T
 explain select cp_a.id, cp_b.id from cp_a left join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = 4;
 ----
-Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] LeftOuter Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] Filter (cp_a.c1 = 4), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.id, cp_a.c1] [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [cp_a.id, cp_b.id] [Project => (Sort Option: Follow)] LeftOuter Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] TableScan cp_a -> [cp_a.id, cp_a.c1], Filter: (cp_a.c1 = 4) [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.id, cp_b.c1] [IndexScan By cp_b_c1 => 4 Covered => (Sort Option: OrderBy: (cp_b.c1 Asc Nulls Last) ignore_prefix_len: 0)]
 
 query II
 select cp_a.id, cp_b.id from cp_a left join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = 4;
@@ -936,7 +936,7 @@ select count(*), count(cp_b.id) from cp_a left join cp_b on cp_a.c1 = cp_b.c1 an
 query T
 explain select count(*), count(cp_a.id) from cp_a right join cp_b on cp_a.c1 = cp_b.c1 and cp_a.c1 = 4;
 ----
-Projection [Count(*), Count(cp_a.id)] [Project => (Sort Option: Follow)] Aggregate [Count(*), Count(cp_a.id)] [SimpleAggregate => (Sort Option: None)] RightOuter Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] Filter (cp_a.c1 = 4), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.id, cp_a.c1] [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.c1] [SeqScan => (Sort Option: None)]
+Projection [Count(*), Count(cp_a.id)] [Project => (Sort Option: Follow)] Aggregate [Count(*), Count(cp_a.id)] [SimpleAggregate => (Sort Option: None)] RightOuter Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] TableScan cp_a -> [cp_a.id, cp_a.c1], Filter: (cp_a.c1 = 4) [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.c1] [SeqScan => (Sort Option: None)]
 
 query II
 select count(*), count(cp_a.id) from cp_a right join cp_b on cp_a.c1 = cp_b.c1 and cp_a.c1 = 4;
@@ -947,7 +947,7 @@ select count(*), count(cp_a.id) from cp_a right join cp_b on cp_a.c1 = cp_b.c1 a
 query T
 explain select count(*) from cp_a join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = null;
 ----
-Projection [Count(*)] [Project => (Sort Option: Follow)] Aggregate [Count(*)] [SimpleAggregate => (Sort Option: None)] Inner Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] Filter (cp_a.c1 = cast (NULL as Integer)), Is Having: false [Filter => (Sort Option: Follow)] TableScan cp_a -> [cp_a.c1] [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.c1] [SeqScan => (Sort Option: None)]
+Projection [Count(*)] [Project => (Sort Option: Follow)] Aggregate [Count(*)] [SimpleAggregate => (Sort Option: None)] Inner Join On cp_a.c1 = cp_b.c1 [HashJoin => (Sort Option: None)] TableScan cp_a -> [cp_a.c1], Filter: (cp_a.c1 = cast (NULL as Integer)) [SeqScan => (Sort Option: None)] TableScan cp_b -> [cp_b.c1] [SeqScan => (Sort Option: None)]
 
 query I
 select count(*) from cp_a join cp_b on cp_a.c1 = cp_b.c1 where cp_a.c1 = null;
diff --git a/tests/slt/where_by_index_explain.slt b/tests/slt/where_by_index_explain.slt
index a2bffb1c..c64105ad 100644
--- a/tests/slt/where_by_index_explain.slt
+++ b/tests/slt/where_by_index_explain.slt
@@ -48,22 +48,22 @@ Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1
 query T
 explain select * from t1 where id = 0 and id != 0;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter (t1.id != 0), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [IndexScan By pk_index => 0 => (Sort Option: OrderBy: (t1.id Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: (t1.id != 0) [IndexScan By pk_index => 0 => (Sort Option: OrderBy: (t1.id Asc Nulls Last) ignore_prefix_len: 0)]
 
 query T
 explain select * from t1 where id = 0 or id != 0 limit 10;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] Filter ((t1.id = 0) || (t1.id != 0)), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [SeqScan => (Sort Option: None)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: ((t1.id = 0) || (t1.id != 0)) [SeqScan => (Sort Option: None)]
 
 query T
 explain select * from t1 where id = 0 and id != 0 and id = 3;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter (t1.id != 0), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [IndexScan By pk_index => Dummy => (Sort Option: OrderBy: (t1.id Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: (t1.id != 0) [IndexScan By pk_index => Dummy => (Sort Option: OrderBy: (t1.id Asc Nulls Last) ignore_prefix_len: 0)]
 
 query T
 explain select * from t1 where id = 0 and id != 0 or id = 3;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter (((t1.id = 0) && (t1.id != 0)) || (t1.id = 3)), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [SeqScan => (Sort Option: None)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: (((t1.id = 0) && (t1.id != 0)) || (t1.id = 3)) [SeqScan => (Sort Option: None)]
 
 query T
 explain select * from t1 where id > 0 and id = 3;
@@ -98,7 +98,7 @@ Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1
 query T
 explain select * from t1 where id >= 3 or id <= 9 limit 10;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] Filter ((t1.id >= 3) || (t1.id <= 9)), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [SeqScan => (Sort Option: None)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: ((t1.id >= 3) || (t1.id <= 9)) [SeqScan => (Sort Option: None)]
 
 query T
 explain select * from t1 where id <= 3 or id >= 9 limit 10;
@@ -113,7 +113,7 @@ Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1
 query T
 explain select * from t1 where (id >= 0 or id <= 3) and (id >= 9 or id <= 12) limit 10;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] Filter (((t1.id >= 0) || (t1.id <= 3)) && ((t1.id >= 9) || (t1.id <= 12))), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [SeqScan => (Sort Option: None)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Limit 10 [Limit => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: (((t1.id >= 0) || (t1.id <= 3)) && ((t1.id >= 9) || (t1.id <= 12))) [SeqScan => (Sort Option: None)]
 
 query T
 explain select * from t1 where id = 5 or (id > 5 and (id > 6 or id < 8) and id < 12);
@@ -123,7 +123,7 @@ Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1
 query T
 explain select * from t1 where c1 = 7 and c2 = 8;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter (t1.c2 = 8), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [IndexScan By u_c1_index => 7 => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: (t1.c2 = 8) [IndexScan By u_c1_index => 7 => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last) ignore_prefix_len: 0)]
 
 query T
 explain select * from t1 where c1 = 7 and c2 < 9;
@@ -138,7 +138,7 @@ Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1
 query T
 explain select * from t1 where c1 is null and c2 is null;
 ----
-Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter t1.c2 is null, Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2] [IndexScan By u_c1_index => NULL => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.id, t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.id, t1.c1, t1.c2], Filter: t1.c2 is null [IndexScan By u_c1_index => NULL => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last) ignore_prefix_len: 0)]
 
 query T
 explain select * from t1 where c1 > 0 and c1 < 8;
@@ -212,13 +212,13 @@ drop index t1.c2_index;
 query T
 explain select c1, c2 from t1 where c1 < 10 and c1 > 0 and c2 >0 and c2 < 10;
 ----
-Projection [t1.c1, t1.c2] [Project => (Sort Option: Follow)] Filter ((t1.c2 > 0) && (t1.c2 < 10)), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.c1, t1.c2] [IndexScan By p_index => ((0), (10)) Covered => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last, t1.c2 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.c1, t1.c2] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.c1, t1.c2], Filter: ((t1.c2 > 0) && (t1.c2 < 10)) [IndexScan By p_index => ((0), (10)) Covered => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last, t1.c2 Asc Nulls Last) ignore_prefix_len: 0)]
 
 # composite covered projection reorder
 query T
 explain select c2, c1 from t1 where c1 < 10 and c1 > 0 and c2 > 0 and c2 < 10;
 ----
-Projection [t1.c2, t1.c1] [Project => (Sort Option: Follow)] Filter ((t1.c2 > 0) && (t1.c2 < 10)), Is Having: false [Filter => (Sort Option: Follow)] TableScan t1 -> [t1.c1, t1.c2] [IndexScan By p_index => ((0), (10)) Covered => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last, t1.c2 Asc Nulls Last) ignore_prefix_len: 0)]
+Projection [t1.c2, t1.c1] [Project => (Sort Option: Follow)] TableScan t1 -> [t1.c1, t1.c2], Filter: ((t1.c2 > 0) && (t1.c2 < 10)) [IndexScan By p_index => ((0), (10)) Covered => (Sort Option: OrderBy: (t1.c1 Asc Nulls Last, t1.c2 Asc Nulls Last) ignore_prefix_len: 0)]
 
 
 statement ok
@@ -241,7 +241,7 @@ create index idx_cover on t_cover (c1, c2, c3);
 query T
 explain select c2, c3 from t_cover where c1 = 2;
 ----
-Projection [t_cover.c2, t_cover.c3] [Project => (Sort Option: Follow)] Filter (t_cover.c1 = 2), Is Having: false [Filter => (Sort Option: Follow)] TableScan t_cover -> [t_cover.c1, t_cover.c2, t_cover.c3] [SeqScan => (Sort Option: None)]
+Projection [t_cover.c2, t_cover.c3] [Project => (Sort Option: Follow)] TableScan t_cover -> [t_cover.c1, t_cover.c2, t_cover.c3], Filter: (t_cover.c1 = 2) [SeqScan => (Sort Option: None)]
 
 statement ok
 drop table t_cover;