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

[LuQQiu]

Motivation

TakeExec materializes rows through the v2 readers' point-lookup path (ReadBatchParams::Indices)
with per-fragment reader opens driven on a single task. Reading the same rows through
FilteredReadExec's planned range reads is much faster. Take-100 benchmark (100 random row
addresses per query, warm NVMe, heavy ~1KiB string column; harness kept out of the tree):

arm	QPS	P50	vs TakeExec
TakeExec	~104	9.1 ms	1x
FilteredReadExec, row-set input (like _rowid IN (...))	255	3.7 ms	2.45x
FilteredReadExec, row-stream input (this PR)	227	4.4 ms	2.18x

Carrying a payload column (e.g. _distance/_score) through the row-stream read costs ~2% —
the capability a row-set input cannot express.

Design: one I/O node over a row source

FilteredReadExec stays a single, general I/O node:

output   = read(row_source, fields_to_read) ⊕ carry_columns
schema() = carry_columns + fields_to_read          (uniform for every source)

Only the row source — who names the rows — varies, and the differences between sources are
inherent to what they are, not modes of the node:

row source	delivered as	output rows	carry columns
all rows	(no input plan)	storage order	none
row set	one serialized IndexExprResult batch (scalar index result, _rowid IN) — existing behavior, unchanged	storage order, deduplicated (sets are unordered + unique by nature)	none
row stream (new)	ordinary record batches with a _rowid/_rowaddr column	the stream's order and duplicates, preserved	the stream's own columns (e.g. _distance), preserved

The source kind is derived from the input plan's schema at construction — never configured
(the wire layouts are unmistakable; anything else must carry a row id/address column). Each
stream batch is converted to an in-memory IndexExprResult and read through the same
plan_scan -> plan_to_scoped_fragments -> read_fragment pipeline as a set; the read
columns are then aligned (hash on the key column, O(N)) and merged back into the batch —
including nested-struct merges, via the same calculate_output_schema + merge_with_schema
contract as TakeExec. Streaming batch-by-batch: no input draining, same memory profile as
TakeExec.

Call sites: Scanner::take() and merge_insert's indexed take now plan takes as
FilteredReadExec on the v2 storage format. Plan text shows
LanceRead: ..., projection=[...], source=stream(_rowid) where Take: columns=... appeared
before.

Design notes (from review)

• Why does TakeExec remain for legacy (v1) storage? The v1 file reader's
  read_ranges_tasks is a stub that errors ("Attempt to perform FilteredRead on v1 files") —
  FilteredReadExec physically cannot read v1 files. v1 datasets keep bit-for-bit yesterday's
  behavior; TakeExec becomes deletable when v1 read support is dropped. The invariant is
  commented at both swap sites.
• Why does count pushdown skip row-stream reads? Their output count is driven by the
  input plan, not fragment metadata. It IS implementable without column I/O (stream the
  input; per batch, count rows whose id is in the live-row set built from fragment metadata +
  deletion vectors — per row, so duplicates count); deferred because no plan shape places a
  row-stream read under a COUNT today (count plans request no columns, so no read node is
  built). Recipe recorded in count_pushdown.rs.
• Why align/merge instead of reading in request order? Readers require ascending offsets
  (TakeExec itself sorts and inverse-permutes); duplicates and stable-row-id address order
  make request-order reads impossible. The alignment is O(N) per batch and ~2% of query time.

Behavior change

Input rows whose row id/address no longer exists (stale index results pointing at deleted
rows) are now silently dropped on non-stable-row-id datasets, where TakeExec failed with a
row-count mismatch error. This matches the existing stable-row-id behavior and FTS
deleted-row expectations.

Follow-ups (out of scope, marked in code)

• Distributed serialization of row-stream sources (filtered_read_exec_to_proto returns
  not_supported).
• mem_wal vector-search take site (dormant today; needs a keep-unmatched/null-fill option for
  NULL _rowid memtable rows).
• Count pushdown through row-stream reads (recipe above), and an optional per-fragment reader
  cache for many-batch reads.

Testing

• Row-stream unit tests: order/duplicate/payload preservation, nested struct merge, stable
  row ids, stale-id drops, construction errors, with_new_children, execution without a
  tokio runtime.
• Full lance lib suite, --features slow_tests query integration suite (incl. FTS
  stale-rowid), lance-namespace-impls; plan-text assertions updated.

🤖 Generated with Claude Code

[codecov]

Codecov Report

❌ Patch coverage is 87.46398% with 87 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/io/exec/filtered_read.rs	89.18%	48 Missing and 18 partials ⚠️
rust/lance/src/dataset/scanner.rs	78.72%	3 Missing and 7 partials ⚠️
rust/lance/src/io/exec/filtered_read_proto.rs	28.57%	5 Missing ⚠️
rust/lance/src/io/exec/optimizer.rs	73.33%	2 Missing and 2 partials ⚠️
rust/lance/src/dataset/write/merge_insert.rs	87.50%	0 Missing and 1 partial ⚠️
rust/lance/src/io/exec/count_pushdown.rs	66.66%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!