perf: plan take operations through FilteredReadExec's range-read path by LuQQiu · Pull Request #7672 · lance-format/lance

LuQQiu · 2026-07-07T17:53:21Z

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

TakeExec reads rows through the v2 readers' point-lookup path with serial per-fragment opens; reading the same rows through FilteredReadExec's planned range reads is 2.3-2.7x faster (take-100 benchmark, warm NVMe). Generalize FilteredReadExec's input to "a plan that tells the node which rows to read", accepting two encodings detected from the input plan's schema: - the serialized IndexExprResult mask batch (existing behavior, unchanged) - any plan carrying a _rowid/_rowaddr column ("take mode", new): each batch is converted to an in-memory IndexExprResult and read through the same plan_scan -> read_fragment pipeline, then the columns are merged back into the input batch preserving row order, duplicates, and payload columns (e.g. _distance) - the same contract as TakeExec Scanner::take and merge_insert now plan takes as FilteredReadExec on the v2 storage format. TakeExec is unchanged and still used for legacy (v1) storage. The CoalesceTake optimizer rule also collapses stacked take-mode reads, and count-pushdown / distributed proto serialization explicitly skip take-mode nodes (distributed support is a follow-up). Adds a three-arm benchmark (TakeExec vs mask mode vs take mode) at rust/lance/benches/take_exec_compare.rs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

gen_range is deprecated and fails clippy -D warnings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Box the TakeRows variant (large size difference between variants) and allow print_stdout in the benchmark like the other benches. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

A random take touches many fragments (often one row per fragment) and try_join_all drives the opens concurrently but on a single task, so the CPU-bound part of ~100 fragment opens ran back to back (~4ms/query on a many-fragment dataset). Spawn one task per fragment open and per decode, matching the scan path (FilteredReadStream::try_new). Local take-100 (2M rows, 20 fragments): take mode 671 -> 858 QPS, now within 7% of mask mode (920 QPS) vs TakeExec at 445 QPS. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

codecov · 2026-07-07T18:33:41Z

Codecov Report

❌ Patch coverage is 88.35784% with 95 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/io/exec/filtered_read.rs	89.95%	50 Missing and 24 partials ⚠️
rust/lance/src/dataset/scanner.rs	76.59%	1 Missing and 10 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/io/exec/count_pushdown.rs	66.66%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

Review feedback: the node read as if it had "modes" (mode=take display, is_take() checks). Reframe it as one general I/O node, output = read(row_source, fields_to_read) + carry_columns, where only the row source varies: AllRows (no input), RowSet (one serialized IndexExprResult batch; sets are unordered and unique by nature), or RowStream (record batches whose order, duplicates, and columns are preserved). The source kind is derived from the input plan's schema, never configured. Renames only, no behavior change: FilteredReadInput -> RowSource, TakeRowsInput -> RowStreamSource, is_take() -> row_stream_input(), mode=take(_rowid) -> source=stream(_rowid) in plan text, and consumer checks (count pushdown, proto, CoalesceTake) now ask attributes instead of node identity. Also documents the legacy-storage invariant at both TakeExec swap sites (the v1 reader cannot serve FilteredRead) and the deferred count-through-streams rewrite in count_pushdown.rs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Review feedback: the three-arm comparison harness is a local tool, not a maintained benchmark. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

westonpace

Initial feedback. Some of these comments are pretty verbose and I'd prefer to trim them down if possible. Overall I think the strategy is fine.

Also, why can't we use row addresses as input unless we have stable row id? I would think we could use addresses or ids as input with or without stable row id.

westonpace · 2026-07-08T15:16:59Z

+/// Who names the rows that a [`FilteredReadExec`] reads.
+///
+/// The node itself is a single, general I/O operator:
+///
+/// ```text
+/// output = read(row_source, fields_to_read) ⊕ carry_columns
+///
+/// schema() = carry_columns + fields_to_read     (uniform for every source)
+/// ```
+///
+/// Only the row source varies, and the differences between the variants are
+/// inherent to what the sources *are* — not modes of the node:
+///
+/// - [`AllRows`](Self::AllRows): no input plan; every live row.  No carry
+///   columns.
+/// - [`RowSet`](Self::RowSet): a *set* of rows, delivered as exactly one
+///   batch in the serialized [`IndexExprResult`] wire layout (produced by
+///   e.g. [`ScalarIndexExec`](super::scalar_index::ScalarIndexExec) or a
+///   `_rowid IN (...)` filter).  Sets are unordered and unique by nature, so
+///   the output is in storage order and deduplicated.  No carry columns.
+/// - [`RowStream`](Self::RowStream): a *stream* of rows — ordinary record
+///   batches with a `_rowid`/`_rowaddr` column.  Streams have row order,
+///   duplicates, and their own columns, and all three are preserved: the
+///   stream's columns are carried through to the output next to the newly
+///   read fields (the contract [`TakeExec`](super::TakeExec) provides).
+///   Internally each batch is converted to an in-memory [`IndexExprResult`],
+///   so a stream is the streaming generalization of a set.
+///
+/// The source kind is *derived*, never configured: [`FilteredReadExec::try_new`]
+/// inspects the input plan's schema (the wire layouts are unmistakable —
+/// binary mask columns; anything else must carry a row id or row address
+/// column).


This comment seems excessively verbose and is hard to parse. I like the description of each of the options, but those descriptions should be on the enum choices themselves, and not in this header. Maybe just something like:

/// Describes which rows should be read /// /// This can be all rows, a specific set of rows, or the read can /// be a "take" which reads new columns into an existing set of rows /// using the `_rowid` or `_rowaddr`.

westonpace · 2026-07-08T15:21:50Z

+    ///
+    /// `index_input` generalizes to "a plan that tells this node which rows to
+    /// read" and accepts two encodings, detected from the plan's schema:
+    ///
+    /// - A serialized [`IndexExprResult`] batch (the wire layout emitted by
+    ///   [`ScalarIndexExec`](super::scalar_index::ScalarIndexExec)): a row
+    ///   *set* scoping a scan.  This is the classic behavior.
+    /// - Any other plan carrying a `_rowid` or `_rowaddr` column: a row
+    ///   *stream* of rows.  The node reads `options.projection` for exactly
+    ///   those rows and merges the columns into the stream's batches,
+    ///   preserving row order, duplicates, and the stream's own columns —
+    ///   the same contract as [`TakeExec`](super::TakeExec).  Fields already
+    ///   present in the stream's schema are not re-read.


This comment feels pretty repetitive. Also, we should rename index_input since it isn't necessarily index-specific anymore. Perhaps just:

/// Create a new filtered read /// /// `input` identifies which rows to read. This is parsed into /// a [`RowStreamSource`]

westonpace · 2026-07-08T15:24:12Z

+            if dataset.manifest.uses_stable_row_ids() {
+                return Err(Error::invalid_input_source(
+                    format!(
+                        "cannot read rows by '{}' on a dataset with stable row ids; the input plan must provide '{}'",
+                        ROW_ADDR, ROW_ID
+                    )
+                    .into(),
+                ));
+            }
+            ROW_ADDR


Why not? It almost seems like this would be the easier case since row addresses are what we ultimately need to locate the rows.

westonpace · 2026-07-08T15:49:56Z

+        // 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.


Do we really need this verbose comment?

I will clean up the verbose comments after all the code changes looks good (otherwise agent tend to add back a lot of comments when i delete them up front lollll)

westonpace · 2026-07-08T15:51:24Z

+    // 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.


Suggested change

// 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.

// We don't currently support count pushdown when the row selector

// is a row stream.

westonpace · 2026-07-08T16:02:48Z

+            // One read round per input batch: read each fragment's rows as a
+            // single batch, and prioritize earlier input batches so consuming
+            // the (ordered) output is not starved by later batches
+            scoped.priority = batch_index;


Hmm, plan_to_scoped_fragments is already going to assign a priority (based on fragment order). I'd rather not lose this. In other words, if we have 4 input batches and 2 fragments then instead of using priorities 0, 0, 1, 1, 2, 2, 3, 3 I'd rather see 0, 1, 2, 3, 4, 5, 6, 7.

Can batch_index be a counter instead so we can do something like...

for scoped in &mut scoped_fragments { scoped.priority += batch_index; } batch_index += scoped_fragments.len();

westonpace · 2026-07-08T16:17:21Z

+                )
+            })
+            .boxed()
+            .try_buffered(get_num_compute_intensive_cpus())


This probably should not be get_num_compute_intensive_cpus as we are not really trying to split up CPU work. It's not really I/O work either. It's more of a "prefetch" factor.

The case for a large value, is when we have lots of input rows from a very fast source and a very small number of fragments. For example, maybe we have 1M input rows, available almost immediately, in batches of 8Ki, and each batch hits one fragment. Without enough buffering here we might not get enough I/O parallelism.

The only case I see for a small value, is when we have carry-through columns that are large. In that case if we buffer too much, we will accumulate a lot of carry-through data, which creates a lot of RAM pressure.

Maybe for now we hard-code it to something like 64. I think, in all current cases, we are unlikely to get more than 1 or 2 input batches anyways, so it is a moot point.

If we do start to have large input streams in the future then we should tie it into the memory pool reservation system. We should try and grab a reservation for the carry-through data as each batch of input arrives. If we succeed, we spawn a map_batch call on it. If we fail, then we pause reading the input until we can get the reservation.

westonpace · 2026-07-08T16:20:41Z

+            // target stays at the scanner batch size because take output
+            // batches mirror the input batches, and `batch_size` is a
+            // documented maximum for output batch sizes.
+            let coalesced = Arc::new(CoalesceBatchesExec::new(input, self.get_batch_size()));


Should the coalescing be an implicit part of the filtered read? I guess it could work either way but I feel like coalescing is very important here.

westonpace · 2026-07-08T16:21:23Z

+            if let Some(fragments) = &self.fragments {
+                read_options = read_options.with_fragments(Arc::new(fragments.clone()));
+            }


What does a "take with fragments" mean? Do you have any tests that cover this case?

westonpace · 2026-07-08T16:25:21Z

+
+        // Align the read rows (dataset order, deduplicated) back to the
+        // input's row order via the key column
+        let read_keys = read_data


I believe the code in take had a fast path to skip alignment if it wasn't needed (the order was already correct). Is that happening here?

A row-stream read now gathers its input into rounds of batch_size rows (explicit option, then LANCE_DEFAULT_BATCH_SIZE, then a flat 8192) before planning each read: tiny batches merge so planning overhead amortizes, and oversized batches split so a round's decoded output stays bounded. The external CoalesceBatchesExec in Scanner::take is gone - the node sizes its own rounds, and every row-stream consumer (e.g. merge_insert) now gets coalescing for free. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

A physical address already encodes (fragment, offset), so an address-keyed round now builds its read ranges directly instead of translating through the row-id sequence (which would misread addresses as stable row ids - the reason this case was previously rejected). Row ids keep the mask path; addresses at deleted rows drop like stale keys. Scanner::take can now hand any keyed input to FilteredReadExec, shrinking the TakeExec fallback to legacy storage only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Round reads now keep both levels of I/O-scheduler ordering (rounds strictly ordered, fragments in dataset order within a round) via a stride instead of flattening each round to one priority. The round pipeline depth is a named prefetch constant (64) rather than the CPU count - the work is I/O bound and the window is what keeps the scheduler saturated on long input streams. When a round's read returns exactly one row per input row with an identical key sequence (storage-ordered input), the hash-map alignment and permutation are skipped. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Renames per review: the enum is RowSelector, try_new's parameter is simply `input`, and the enum/constructor docs shrink to short forms with per-variant descriptions on the variants. The legacy-storage and count-pushdown comments trim to one-liners. Also adds a unit test for a fragment-scoped take (in-scope key read, out-of-scope key dropped). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

LuQQiu and others added 4 commits July 3, 2026 10:33

bench: use rand 0.9 random_range in take_exec_compare

4114b80

gen_range is deprecated and fails clippy -D warnings. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

style: fix clippy findings in take mode

d94ef90

Box the TakeRows variant (large size difference between variants) and allow print_stdout in the benchmark like the other benches. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

github-actions Bot added A-python Python bindings A-namespace Namespace impls performance labels Jul 7, 2026

LuQQiu and others added 2 commits July 7, 2026 19:31

chore: keep the take benchmark harness out of the tree

173d816

Review feedback: the three-arm comparison harness is a local tool, not a maintained benchmark. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

westonpace self-requested a review July 8, 2026 15:06

westonpace requested changes Jul 8, 2026

View reviewed changes

LuQQiu and others added 4 commits July 8, 2026 12:13

Uh oh!

Conversation

LuQQiu commented Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Design: one I/O node over a row source

Design notes (from review)

Behavior change

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

Testing

Uh oh!

codecov Bot commented Jul 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

westonpace left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

LuQQiu commented Jul 7, 2026 •

edited

Loading

codecov Bot commented Jul 7, 2026 •

edited

Loading