perf(filtered-read): reuse plan-time fragment metadata for the read stream#7792
perf(filtered-read): reuse plan-time fragment metadata for the read stream#7792LuQQiu wants to merge 7 commits into
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughFiltered reads now retain fragment handles in cached planning results and reuse them during stream construction. Take-shaped masked or indexed reads can conditionally coalesce small ordered batches while preserving stream ordering and completion behavior. ChangesTake-shaped filtered reads
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant FilteredReadExec
participant FilteredReadStream
participant FileFragment
participant RecordBatchStream
FilteredReadExec->>FilteredReadExec: Cache PlannedRead with plan and fragment handles
FilteredReadExec->>FilteredReadStream: Create stream with cached handles
FilteredReadStream->>FileFragment: Build scoped fragment reads
FileFragment->>RecordBatchStream: Produce ordered record batches
FilteredReadExec->>RecordBatchStream: Coalesce take-shaped batches
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance/src/io/exec/filtered_read.rs (1)
1978-2050: 🚀 Performance & Scalability | 🔵 TrivialConfirm the streaming-latency trade-off for sparse masked reads is acceptable.
When
is_sparse_planis true and total planned rows are belowtarget,coalesce_batchesonly emits once the inner stream is exhausted (ready_to_emitrequiresexhausted && !buffered.is_empty()), soconsolidated_streamwithholds all output until the entire masked scan across all touched fragments completes. Previously, per-fragment batches would stream out incrementally as each fragment finished. For a caller relying on early termination (e.g. aLimitExecsitting above a read whosescan_range_after_filtercouldn't be pushed down because of a recheck filter), this increases time-to-first-batch even though it doesn't change total I/O volume.This is likely intentional for the take-shaped scenario this PR targets, but worth confirming it doesn't regress latency for interactive/streaming consumers.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 1978 - 2050, Review the sparse masked-read path in the stream construction around is_sparse_plan and consolidated_stream, and confirm that withholding batches until all touched fragments finish is acceptable for interactive consumers and upstream early termination such as LimitExec. If the latency regression is not acceptable, constrain consolidation to the intended take-shaped scenario or preserve incremental per-fragment emission for other masked reads; otherwise document or validate the intentional trade-off without changing unrelated batch-size behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 2639-2709: The test_take_shaped_mask_consolidation coverage only
uses the default threading mode; extend it with a
FilteredReadThreadingMode::MultiplePartitions case. Configure the read options
with MultiplePartitions(n), execute the same take-shaped mask, and verify
consolidation preserves correct, disjoint row sets and expected batch results
across partitions without relying on single-partition ordering assumptions.
---
Outside diff comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 1978-2050: Review the sparse masked-read path in the stream
construction around is_sparse_plan and consolidated_stream, and confirm that
withholding batches until all touched fragments finish is acceptable for
interactive consumers and upstream early termination such as LimitExec. If the
latency regression is not acceptable, constrain consolidation to the intended
take-shaped scenario or preserve incremental per-fragment emission for other
masked reads; otherwise document or validate the intentional trade-off without
changing unrelated batch-size behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 7cd1a99c-22d0-4c8b-9d69-5cadc0d469f7
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
…anch Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…tream Masked reads loaded every fragment's metadata twice per query: once in get_or_create_plan_impl to compute the plan and again in FilteredReadStream::try_new to build the read. The plan cell now keeps the loaded fragments alongside the plan and stream construction reuses them. The distributed path (with_plan, plan computed on another node) carries no fragments and loads them at stream construction as before. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
e955296 to
8b5c856
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 2015-2033: Add a log::debug! inside the sparse-plan branch where
is_sparse_plan is computed, logging touched_fragments, planned_rows, and the
resolved batch_target_rows when the heuristic evaluates true. Keep consolidation
behavior unchanged and ensure the message identifies that masked-read
consolidation was triggered.
- Around line 337-446: Extract the consolidation utilities from filtered_read.rs
into a dedicated consolidate submodule, including consolidated_stream,
coalesce_batches, and the CONSOLIDATE_MIN_FRAGMENTS and
CONSOLIDATE_MAX_AVG_PLANNED_ROWS_PER_FRAGMENT constants. Update their callers
and visibility/imports so planning and execution behavior remains unchanged
while the main module retains only the wiring.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 4ccf83f8-cf9c-453e-bafd-c102cffd94be
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 2015-2033: Add a log::debug! inside the sparse-plan branch where
is_sparse_plan is computed, logging touched_fragments, planned_rows, and the
resolved batch_target_rows when the heuristic evaluates true. Keep consolidation
behavior unchanged and ensure the message identifies that masked-read
consolidation was triggered.
- Around line 337-446: Extract the consolidation utilities from filtered_read.rs
into a dedicated consolidate submodule, including consolidated_stream,
coalesce_batches, and the CONSOLIDATE_MIN_FRAGMENTS and
CONSOLIDATE_MAX_AVG_PLANNED_ROWS_PER_FRAGMENT constants. Update their callers
and visibility/imports so planning and execution behavior remains unchanged
while the main module retains only the wiring.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 4ccf83f8-cf9c-453e-bafd-c102cffd94be
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
🛑 Comments failed to post (2)
rust/lance/src/io/exec/filtered_read.rs (2)
337-446: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Consider extracting the consolidation utilities into a dedicated submodule.
consolidated_stream/coalesce_batches(plus the two threshold constants) are ~110 lines of self-contained, new algorithmic logic added directly into an already very large file (filtered_read.rsspans thousands of lines). Traced the state machine by hand (partial-buffer emission on oversized incoming batches, single-item passthrough, exhaustion handling) — the logic itself is correct and well-commented.As per coding guidelines, "Extract substantial new logic such as bin packing or scheduling into dedicated submodules instead of inlining it into large files." Moving this to e.g.
filtered_read/consolidate.rswould improve discoverability and keep the main file focused on planning/execution wiring.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 337 - 446, Extract the consolidation utilities from filtered_read.rs into a dedicated consolidate submodule, including consolidated_stream, coalesce_batches, and the CONSOLIDATE_MIN_FRAGMENTS and CONSOLIDATE_MAX_AVG_PLANNED_ROWS_PER_FRAGMENT constants. Update their callers and visibility/imports so planning and execution behavior remains unchanged while the main module retains only the wiring.Source: Coding guidelines
2015-2033: 📐 Maintainability & Code Quality | 🔵 Trivial
Consider a debug log when the sparse-plan consolidation heuristic fires.
The
is_sparse_plandecision (touched_fragments/planned_rows vs. thresholds) is computed once per query but never logged anywhere, making it hard to confirm in production whether a given masked read actually triggered consolidation. Alog::debug!withtouched_fragments,planned_rows, and the resolvedbatch_target_rowswould help diagnose unexpected batch shapes without needing to re-derive the heuristic from metrics.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2015 - 2033, Add a log::debug! inside the sparse-plan branch where is_sparse_plan is computed, logging touched_fragments, planned_rows, and the resolved batch_target_rows when the heuristic evaluates true. Keep consolidation behavior unchanged and ensure the message identifies that masked-read consolidation was triggered.
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance/src/io/exec/filtered_read.rs (1)
2690-2690: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winUse the required RecordBatch access style in the new test.
Access the named column with
batches[0]["value"]instead of the positional.column(0). As per coding guidelines: “Usecolumn_by_name()forRecordBatchcolumn access in production code; usebatch["column_name"]in tests.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` at line 2690, Update the test assertion near the RecordBatch access to use the named-column indexing style, replacing positional column access with batches[0]["value"] while preserving the existing expected-value comparison.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Line 2690: Update the test assertion near the RecordBatch access to use the
named-column indexing style, replacing positional column access with
batches[0]["value"] while preserving the existing expected-value comparison.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: d797de89-813d-484d-920e-f7610734ba1c
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
| /// fragments and the stream loads them itself. | ||
| struct PlannedRead { | ||
| plan: FilteredReadInternalPlan, | ||
| loaded_fragments: Option<Arc<Vec<LoadedFragment>>>, |
There was a problem hiding this comment.
Keeping the full Vec<LoadedFragment> in PlannedRead pins every candidate fragment's deletion vector, row-id sequence, row counts, and FileFragment for the lifetime of the exec, even though stream construction only consumes the ordered FileFragment handles. This changes the optimization from a transient O(N) metadata load into retained O(concurrent execs × N) state and can keep metadata alive past cache eviction on highly fragmented datasets.
Could PlannedRead instead retain only the plan-selected ordered (original_priority, Arc<FileFragment>) descriptors, with None for with_plan so the distributed fallback remains unchanged? That would eliminate the duplicate local load while preserving fragment order, retry behavior, and the wire contract without extending the lifetime of planning-only metadata.
There was a problem hiding this comment.
Thanks for the suggestion! Improved the PR according to the suggestion!
Planning-only metadata (deletion vectors, row-id sequences, row counts) now drops when planning finishes instead of living as long as the exec; stream construction only consumes the ordered FileFragment handles. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015RpPYGnQ9JYaP9B2BbujBn
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance/src/io/exec/filtered_read.rs (1)
424-443: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winFlush buffered batches before propagating input errors.
At Line 432,
?returns the input error whilethis.bufferedmay contain successfully read rows belowtarget; those rows are then dropped. Track a pending error, emit the buffered batch first, and return the error on the next poll. Add a regression test forOk(short_batch)followed byErr(...).As per coding guidelines, every bugfix and feature must have corresponding tests.
Proposed fix
struct Coalescer { input: SendableRecordBatchStream, schema: SchemaRef, target: usize, buffered: Vec<RecordBatch>, buffered_rows: usize, exhausted: bool, + pending_error: Option<DataFusionError>, } +if let Some(error) = this.pending_error.take() { + return Err(error); +} + -match this.input.try_next().await? { +match this.input.try_next().await { + Err(error) if !this.buffered.is_empty() => { + this.pending_error = Some(error); + let output = this.emit()?; + return Ok(Some((output, this))); + } + Err(error) => return Err(error), + Ok(Some(batch)) => { /* existing handling */ } + Ok(None) => this.exhausted = true, +}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 424 - 443, Update the try_unfold state machine around the input.try_next call to retain any input error as pending state when buffered rows exist, emit those rows first, and propagate the pending error on the following poll; preserve immediate error propagation when nothing is buffered. Add a regression test covering an Ok short batch followed by Err(...) and verify the short batch is emitted before the error.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 424-443: Update the try_unfold state machine around the
input.try_next call to retain any input error as pending state when buffered
rows exist, emit those rows first, and propagate the pending error on the
following poll; preserve immediate error propagation when nothing is buffered.
Add a regression test covering an Ok short batch followed by Err(...) and verify
the short batch is emitted before the error.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Pro Plus
Run ID: f84790f8-47f2-45c1-bb00-ed715d1ce088
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
…load # Conflicts: # rust/lance/src/io/exec/filtered_read.rs
There was a problem hiding this comment.
Note
Quiet mode is enabled, so only the most important comments were posted inline. Other review comments are grouped below.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rust/lance/src/io/exec/filtered_read.rs (1)
385-421: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
coalesce_batchespanics/errors immediately whentarget == 0.
ready_to_emit()returnsbuffered_rows >= target, which is trivially true (0 >= 0) before any input is ever read whentarget == 0. That routes straight intoemit()with an empty buffer, which returns the internal"coalesce_batches emitted with an empty buffer"error instead of reading any data. This is reachable fromRowStreamRead::apply(line ~2662), which derivesbatch_target_rowsfromread_options.batch_size— a user-settableOption<u32>viawith_batch_size— with no visible guard against0in this file (contrast withobtain_stream's masked-read path, which explicitly checksbatch_target_rows > 0before callingconsolidated_stream).🐛 Proposed fix: clamp target to at least 1
pub fn coalesce_batches( input: SendableRecordBatchStream, target: usize, ) -> impl Stream<Item = DataFusionResult<RecordBatch>> { + // A target of 0 would make ready_to_emit() fire before any input is + // read, emitting from an empty buffer; treat 0 as "no coalescing". + let target = target.max(1); struct Coalescer {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 385 - 421, Update coalesce_batches to normalize the target value to at least 1 before constructing or using Coalescer, so ready_to_emit cannot become true with an empty buffer when callers provide zero. Preserve existing batching behavior for positive targets and keep the change scoped to the target handling in coalesce_batches.
🟡 Other comments (1)
rust/lance/src/io/exec/filtered_read.rs-2739-2766 (1)
2739-2766: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winLatent panic via
HashMapindexing inattach_read_columns.
key_to_index[key]panics ifkeyis absent. It's currently unreachable becausekeyswas just filtered above to only contain keys present inkey_to_index, but that invariant lives several lines away and isn't enforced by the type system — a future refactor of the filter step could silently reintroduce a panic on production data. As per coding guidelines, "Avoid bare.unwrap(); useif let,match,let ... else,?, or combinators... If unavoidable, use.expect("reason")" and "Never use.unwrap(),.expect(),panic!(), orassert!()in library code for fallible operations."🛡️ Proposed fix: make the invariant explicit
- let indices = UInt32Array::from_iter_values(keys.values().iter().map(|key| key_to_index[key])); + let indices = UInt32Array::from_iter_values( + keys.values() + .iter() + .map(|key| *key_to_index.get(key).expect("key was filtered above to exist in key_to_index")), + );🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@rust/lance/src/io/exec/filtered_read.rs` around lines 2739 - 2766, In attach_read_columns, replace the direct key_to_index[key] lookup when building indices with a non-panicking lookup that explicitly handles missing keys, such as filtering or propagating an appropriate error. Preserve the existing input-order gathering behavior for present keys and ensure absent keys cannot trigger a HashMap indexing panic.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 385-421: Update coalesce_batches to normalize the target value to
at least 1 before constructing or using Coalescer, so ready_to_emit cannot
become true with an empty buffer when callers provide zero. Preserve existing
batching behavior for positive targets and keep the change scoped to the target
handling in coalesce_batches.
---
Other comments:
In `@rust/lance/src/io/exec/filtered_read.rs`:
- Around line 2739-2766: In attach_read_columns, replace the direct
key_to_index[key] lookup when building indices with a non-panicking lookup that
explicitly handles missing keys, such as filtering or propagating an appropriate
error. Preserve the existing input-order gathering behavior for present keys and
ensure absent keys cannot trigger a HashMap indexing panic.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: QUIET
Plan: Pro Plus
Run ID: 5574e2b7-0d5c-4149-aa40-d6cbe70c4320
📒 Files selected for processing (1)
rust/lance/src/io/exec/filtered_read.rs
…field PlannedRead was inserted mid-doc-comment, stealing FilteredReadExec's docs. StreamFragments no longer mirrors its fragments as a second handle vec; read_batch builds the handle list per batch instead. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015RpPYGnQ9JYaP9B2BbujBn
| // construction doesn't reload every fragment | ||
| let plan = | ||
| FilteredReadStream::plan_scan(&loaded_fragments, &evaluated_index, options); | ||
| let fragment_handles = loaded_fragments |
There was a problem hiding this comment.
Thanks—the updated version fixes the previous LoadedFragment lifetime issue: deletion vectors and row-id sequences now drop after planning. However, fragment_handles still contains every candidate fragment, not just those selected by plan.rows. Since FileFragment owns a cloned Fragment descriptor, sparse plans retain O(all fragments) descriptor state for the lifetime of each exec. This owned-handle API also makes RowStreamRead::read_batch allocate and Arc-clone an all-fragment vector for every input batch, adding an unmeasured O(batches × fragments) hot-path cost.
Stream construction only needs FileFragment, and FileFragment::new(dataset, metadata) is I/O-free. Could we instead enumerate the existing ordered options.fragments/dataset descriptors, filter by plan.rows, and construct handles only for selected fragments while preserving original priority? The row-stream path could then keep borrowing its existing loaded fragments. This removes the duplicate metadata load without a plan-lifetime sidecar, per-batch full-vector clone, or wire change. It would also be helpful to add a load-count regression test and a current-head base/PR benchmark, including a many-fragment row-stream case.
There was a problem hiding this comment.
thanks for the good advice! addressed the comments, please take a look
FileFragment::new is I/O-free, so nothing needs to be retained between planning and execution: plan_to_scoped_fragments enumerates the manifest descriptors and constructs handles only for the fragments the plan selects, with the candidate-list position as priority. The PlannedRead sidecar is gone, planning-only metadata drops when planning finishes, and the row-stream path no longer builds a per-batch handle vector. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015RpPYGnQ9JYaP9B2BbujBn
2 participants
Problem
Masked reads (
FilteredReadExecwith aRowSetinput) load every fragment's metadata twice per query:get_or_create_plan_implloads all fragments to compute the plan (plan_scanneeds deletion vectors / physical row counts), then throws the loaded set away;FilteredReadStream::try_newimmediately loads the same fragments again to build the read stream.Where it was found and how big it is
Found while stage-timing take-shaped masked reads (an exact row-id mask selecting ~100 scattered rows) on a 100M-row, 100-fragment dataset, 64 concurrent queries, page-cached NVMe. Per-query planning cost breaks down as:
The two loads together are 0.74 ms — ~64% of planning cost, all of it running in the first poll of the stream (i.e., on the caller's task). For a take-shaped query whose whole read is only a couple of milliseconds this is a significant fixed tax; it scales linearly with fragment count and query rate. Long scans amortize it to noise, which is why it went unnoticed.
Fix
The plan cell now keeps the loaded fragments alongside the resolved plan, and stream construction reuses them. The distributed path (
with_plan, plan computed on another node, e.g. a query node planning for executor nodes) carries no fragment metadata and loads at stream construction exactly as before — the plan phase stays independently callable.Tests
filtered_readsuite (28, single-threaded) andscannersuite (169) green; fmt + clippy clean. No behavior change — the same loaded set is used, just not loaded twice.🤖 Generated with Claude Code
https://claude.ai/code/session_015RpPYGnQ9JYaP9B2BbujBn