refactor(layout): keep Filter out of lower, and let a segment scan keep its own rows

A flat layout lowers to a bare segment scan again, as on develop. Every
plan now produces only the rows its split selects, a segment scan
included, so the scan's source applies the mask; a Filter over a scan
compiles to the same source.

Signed-off-by: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NWYmfHTFtaEdhdyd3u5hv3

Author: claude

vortex-layout/src/plan/lower.rs

@@ -25,7 +25,6 @@ use crate::layouts::list::VALIDITY_CHILD_INDEX;
 use crate::layouts::struct_::Struct;
 use crate::layouts::struct_::StructLayout;
 use crate::plan::ConcatPlan;
-use crate::plan::FilterPlan;
 use crate::plan::ListPackPlan;
 use crate::plan::PackPlan;
 use crate::plan::PlanChildren;
@@ -37,12 +36,9 @@ use crate::plan::TakePlan;
 ///
 /// The root operator is built immediately. Its child container owns a hidden clone of the source
 /// layout and lowers each child independently on first access.
-///
-/// A flat layout lowers to a [`Filter`](crate::plan::Filter) over its segment scan, so the scan
-/// returns only the rows it is executed with.
 pub fn lower(layout: &LayoutRef) -> VortexResult<PlanRef> {
     if let Some(layout) = layout.as_opt::<Flat>() {
-        return Ok(FilterPlan::new(lower_flat(layout).into_plan()).into_plan());
+        return Ok(lower_flat(layout).into_plan());
     }
     if let Some(layout) = layout.as_opt::<Chunked>() {
         return Ok(lower_chunked(layout)?.into_plan());

vortex-layout/src/plan/pipeline/tests.rs

@@ -33,7 +33,6 @@ use crate::OwnedLayoutChildren;
 use crate::layouts::chunked::ChunkedLayout;
 use crate::layouts::flat::FlatLayout;
 use crate::layouts::struct_::StructLayout;
-use crate::plan::Filter;
 use crate::plan::PlanRef;
 use crate::plan::SegmentScan;
 use crate::plan::lower;
@@ -521,42 +520,34 @@ fn aligned_fields_stream_one_struct_per_chunk() -> VortexResult<()> {
     assert_view(&expected, &(0..ROWS), &mask, run.arrays)
 }
 
-/// A bare segment scan returns every row of its range whatever it is told to care about, and a
-/// filter over the same scan returns only the selected rows.
+/// A segment scan keeps the rows its split selects, and so does a filter over the same scan,
+/// which runs as the same source.
 #[rstest]
 #[case::every_other(Sel::EveryOther)]
 #[case::sparse(Sel::Rows(&[1, 4]))]
 #[case::nothing(Sel::None)]
-fn bare_scan_is_dense_and_filter_keeps_the_selection(#[case] sel: Sel) -> VortexResult<()> {
+fn scan_keeps_the_selection_with_or_without_a_filter(#[case] sel: Sel) -> VortexResult<()> {
     let mut store = Store::default();
     let values = PrimitiveArray::from_iter(0..ROWS as i32).into_array();
-    let filtered = lower(&store.flat(&values)?)?;
-    assert!(filtered.is::<Filter>());
-    let scan = filtered.child_required(0)?;
+    let scan = lower(&store.flat(&values)?)?;
     assert!(scan.is::<SegmentScan>());
+    let filtered = crate::plan::FilterPlan::new(scan.clone()).into_plan();
 
     let rows = 3..9;
     let mask = sel.mask(6);
-    let mut ctx = SESSION.create_execution_ctx();
-
-    let dense = run(
-        &store,
-        &scan,
-        rows.clone(),
-        mask.clone(),
-        delivery(Delivery::Fifo),
-    )?;
-    assert_eq!(dense.arrays.len(), 1);
-    assert_arrays_eq!(dense.arrays[0], values.slice(3..9)?, &mut ctx);
-
-    let kept = run(
-        &store,
-        &filtered,
-        rows.clone(),
-        mask.clone(),
-        delivery(Delivery::Fifo),
-    )?;
-    assert_view(&values, &rows, &mask, kept.arrays)?;
+    for plan in [&scan, &filtered] {
+        let kept = run(
+            &store,
+            plan,
+            rows.clone(),
+            mask.clone(),
+            delivery(Delivery::Fifo),
+        )?;
+        if mask.all_false() {
+            assert_eq!(reads(&kept.events), 0, "nothing selected reads nothing");
+        }
+        assert_view(&values, &rows, &mask, kept.arrays)?;
+    }
     Ok(())
 }
 

vortex-layout/src/plan/plans/segment_scan.rs

@@ -105,9 +105,8 @@ impl PlanVTable for SegmentScan {
         mask: &Mask,
         compiler: &mut Compiler<'_>,
     ) -> VortexResult<Option<Chain>> {
-        let _ = mask;
-        // A bare scan produces every row of its range; a filter over it keeps the selected ones.
-        compiler.scan(plan, rows, None)
+        // A scan keeps the selected rows itself, as every plan produces only those.
+        compiler.scan(plan, rows, Some(mask.clone()))
     }
 
     fn reach(