perf: plan take operations through FilteredReadExec's range-read path

python/python/tests/test_mem_wal.py

@@ -111,7 +111,8 @@ def test_point_lookup_with_memtables(tmp_path):
     plan = planner.plan_lookup(pa.array([2], type=pa.int64()))
     assert plan.schema.names == ["id", "name"]
     assert plan.dataset_schema.names == ["id", "name"]
-    assert "Take" in plan.explain() or "Scan" in plan.explain()
+    explained = plan.explain()
+    assert "Take" in explained or "Scan" in explained or "LanceRead" in explained
     result = plan.to_table()
     assert len(result) == 1, "Expected exactly one row for id=2"
     assert result.column("name")[0].as_py() == "gen1_2", (

rust/lance-namespace-impls/src/dir.rs

@@ -12256,7 +12256,7 @@ mod tests {
             &plan_str,
             &[
                 "ProjectionExec: expr=[id@0 as id, name@2 as name",
-                "Take: columns=\"id, _rowid, (name)\"",
+                "projection=[name], source=stream(_rowid)",
                 "LanceRead: uri=",
                 "projection=[id]",
                 "row_id=true, row_addr=false",
@@ -12353,8 +12353,7 @@ mod tests {
                 "AnalyzeExec verbose=true",
                 "ProjectionExec: elapsed=",
                 "expr=[id@0 as id, name@2 as name",
-                "Take: elapsed=",
-                "columns=\"id, _rowid, (name)\"",
+                "projection=[name], source=stream(_rowid)",
                 "CoalesceBatchesExec: elapsed=",
                 "LanceRead: elapsed=",
                 "projection=[id]",

rust/lance/src/dataset/mem_wal/memtable/flush.rs

@@ -2172,7 +2172,7 @@ mod tests {
         crate::utils::test::assert_plan_node_equals(
             plan,
             "ProjectionExec: expr=[id@2 as id, text@3 as text, _score@1 as _score]
-  Take: ...
+  LanceRead: ..., source=stream(_rowid)
     CoalesceBatchesExec: ...
       MatchQuery: column=text, query=hello",
         )

rust/lance/src/dataset/write/merge_insert.rs

@@ -58,7 +58,9 @@ use crate::{
     },
     index::DatasetIndexInternalExt,
     io::exec::{
-        AddRowAddrExec, Planner, TakeExec, project,
+        AddRowAddrExec, Planner, TakeExec,
+        filtered_read::{FilteredReadExec, FilteredReadOptions},
+        project,
         scalar_index::{IndexLookup, MapIndexExec},
         utils::ReplayExec,
     },
@@ -734,14 +736,31 @@ impl MergeInsertJob {
             )?);
         }
 
-        // 4 - Take the mapped row ids
+        // 4 - Take the mapped row ids.  On the v2 storage format the take is
+        //     planned as a FilteredReadExec fed by the index-mapper stream:
+        //     the row ids become a row-id mask read through the planned
+        //     range-read path, which is considerably faster than TakeExec's
+        //     point-lookup path.  Both nodes have the same output contract
+        //     (input columns first, then the fetched columns).
+        //
+        //     INVARIANT: every site that replaces TakeExec with
+        //     FilteredReadExec must check is_legacy_storage() first.  The v1
+        //     reader's read_ranges_tasks is a stub that errors ("Attempt to
+        //     perform FilteredRead on v1 files"), so TakeExec remains the
+        //     only take that can read v1 files.
         let projection = self
             .dataset
             .empty_projection()
             .union_arrow_schema(schema.as_ref(), OnMissing::Error)?;
-        let mut target =
+        let mut target: Arc<dyn ExecutionPlan> = if self.dataset.is_legacy_storage() {
             Arc::new(TakeExec::try_new(self.dataset.clone(), index_mapper, projection)?.unwrap())
-                as Arc<dyn ExecutionPlan>;
+        } else {
+            Arc::new(FilteredReadExec::try_new(
+                self.dataset.clone(),
+                FilteredReadOptions::new(projection),
+                Some(index_mapper),
+            )?)
+        };
 
         // 5 - Take puts the row id and row addr at the beginning.  A full scan (used when there is
         //     no scalar index) puts the row id and addr at the end.  We need to match these up so

rust/lance/src/io/exec/count_pushdown.rs

@@ -147,6 +147,19 @@ fn try_rewrite(agg: &AggregateExec) -> DFResult<Option<Arc<dyn ExecutionPlan>>>
     };
 
     let options = filtered_read.options();
+    // A read with a row-stream source emits one row per input row; its count
+    // is driven by the input plan, not by fragment metadata.
+    //
+    // COUNT over such a read == the count of input rows whose id still
+    // exists.  A future CountLiveRowsExec could answer that without any
+    // column I/O: stream the input and, per batch, build the live-row set
+    // from fragment metadata + deletion vectors and count member rows
+    // (per row, so duplicates count).  Deferred because no plan shape today
+    // places a row-stream read under a COUNT — count plans request no
+    // columns, so the scanner never builds one.
+    if filtered_read.row_stream_input().is_some() {
+        return Ok(None);
+    }
     // A refine filter is a residual the index couldn't fully evaluate — it
     // needs column data to apply, which we can't.
     if options.refine_filter.is_some() {