feat(scanner): plan every read through the logical planner - #8561
Closed
wjones127 wants to merge 28 commits into
Closed
feat(scanner): plan every read through the logical planner#8561wjones127 wants to merge 28 commits into
wjones127 wants to merge 28 commits into
Conversation
Fragment readers build batches from the projected field list, which carries no schema-level metadata, so batches arrived without the dataset's metadata even though `output_schema` declares it. Callers ending in a projection re-stamp it on the way out and never notice; callers reading this node's output directly saw empty metadata. Restamp in both `get_stream` branches. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
In take mode the node reads one output row per input row, in input order, so its input's ordering holds on its output. It inherited the input's `PlanProperties` — partitioning and emission — but then replaced the equivalence properties wholesale, dropping the ordering. Re-express the input's ordering against the output schema, whose column positions differ, stopping the prefix at the first expression the output schema cannot name. Scan mode is unchanged: it is a leaf with no input ordering to carry. No plan changes: the physical rule set reads equivalence properties only in EnforceDistribution, which does not repartition below a take. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Prototype of the "DataFusion Logical Plans for the Lance Scanner" design. `Scanner::create_plan` builds physical plans imperatively today. This adds an alternative path that builds a `LogicalPlan`, prefetches index and fragment metadata into a `ScanPlanningContext`, derives optimizer rules from that context, and lowers via an `ExtensionPlanner`. Off by default; enabled with `LANCE_LOGICAL_SCAN_PLANNER=1`. The only change outside the new module is a 5-line dispatch in `Scanner::create_plan`. Unsupported query shapes return `NotSupported` rather than falling back silently, so the equivalence tests stay meaningful. Covers filtered scan + projection + limit, flat KNN, ANN, prefilter, postfilter, and combined KNN/ANN over partially-indexed data. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Ports every FTS query shape reachable from `Scanner::create_plan` to the logical-plan prototype: Match, Phrase, Boost, Boolean, MultiMatch, row and list-element granularity, index and flat access paths, partially-indexed splits, fast_search, fragment restriction, and both directions of `query_filter` in prefilter and postfilter form. All FTS support lives in one file (`logical/fts.rs`) exposing five entry points — nodes, prefetch, rules, lowering, requirement collection — as an experiment in what an index plugin's planning surface would look like. Three structural changes to the existing stages: - A new stage 0 (`logical/prepare.rs`) normalizes the scanner's queries before the plan is built. An `FtsQuery` may omit its column and document granularity, and resolving those needs I/O that stage 1 cannot do and a rule cannot defer, since node output schemas depend on granularity. - Metadata-dependent input validation moved into stage 2. An error raised from `OptimizerRule::rewrite` returns wrapped as `DataFusionError::External` and loses the Lance `Error` variant callers match on. - `TakeSettings` carries the scanner's fragment list and batch size to every take, including takes a rule creates. `with_fragments` reaches a search only through the take's read options. Also adds `VectorRerankNode` for a postfilter vector `query_filter`, and prepends DataFusion's `JoinSelection` to the physical rule set: the stock planner emits `HashJoinExec` with `PartitionMode::Auto`, which panics at execute() without it. 45 tests, 35 of which execute both paths and compare rows.
Closes the shapes where the logical path returned wrong rows rather than an error, which is what makes it usable as an oracle: - `ordering`: sort below limit/offset, ordering columns unioned into the take. This was rejecting ~59 tests as unsupported, hiding real failures behind noise. - `refine_factor`: the over-fetch was already there, only the exact re-rank was missing. `ExpandVectorRefine` builds the same subtree `SplitPartiallyIndexedSearch` does — "re-rank approximate candidates exactly" is one logical operation with two reasons to reach for it. - `fast_search`: one feature, three manifestations — no index at all becomes `EmptyRelation`, a scalar index filter sets `only_indexed_fragments`, partial coverage skips the flat branch. - Flat KNN lost global distance order at parallelism > 1: `FilteredReadExec` advertises no output ordering, so the take above the top-k made `execute_plan` pick a plain coalesce. Fixed by stating the ordering in the logical plan. - `_rowcreatedatversion` / `_rowlastupdatedatversion` projections. - ANN delta segments covering none of the scan's fragments are pruned. Also prepends `JoinSelection` to the physical rules: `get_physical_optimizer` is tuned for hand-built plans, and `DefaultPhysicalPlanner` emits a `HashJoinExec` with `PartitionMode::Auto`, which panics at `execute()` without it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`SplitPartiallyIndexedSearch` and `SplitPartiallyIndexedFts` were the same rewrite twice: split a search on how much of the scan its index covers, answer the covered part from the index and the rest by brute force, merge. Data overlays are the same statement at row granularity — an overlay committed after an index was built leaves the index describing values that no longer exist, so those rows are a coverage gap too. Both rules are replaced by one `SplitOnIndexCoverage` plus a `SplittableSearch` trait. What counts as coverage and how the branches merge stay per-index-kind; the split does not. Overlay stale rows need no machinery of their own — they are one more entry in the gap list, narrowing that branch's scan by row set instead of by fragment, and the same union absorbs them. Two things fell out of the unification: * `fast_search` becomes one statement instead of three. `IndexCoverage:: indexed_only()` drops the brute-force branch but keeps the block mask: "answer from indices only" means a gap goes unanswered, and a stale entry is not an answer either. * The one legitimate difference between index kinds is now a parameter. A vector or scalar index can be blocked by row address whatever segment produced it, so an opaque segment is `Covering`; a BM25 score depends on the whole indexed document set, so for FTS it is `Opaque` and a relevant overlay makes the index `Unusable`. The rule reaches two of the three cases. A scalar index query is not a node in the logical plan — it is derived in the scan leaf from pushed-down predicates — so its split lives in `LanceScanSource::scan_impl`. Unifying that one means giving the index query a logical node, which is beyond this spike; the finding is recorded on `stale_rows_branch`. `ensure_supported` no longer rejects overlaid datasets. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`rows_as_index_input` mirrored `Scanner::row_ids_as_take_input` and wrapped the serialized mask in a `OneShotExec`. The mask is a materialized batch, so a memory source expresses the same thing without the one-shot restriction. This does not make the scan re-executable, and an earlier note claiming that a row-restricted scan was single-execution while a fragment-restricted one was not is wrong. Measured on a plain filtered scan with no index and no overlay, both planning paths return the full result once and zero rows on a second `execute()`: `FilteredReadStream` drains one shared `task_stream` per instance. Single-execution is a property of the leaf, not of how the leaf was narrowed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
DataFusion documents an `OptimizerRule` as computing "the same results, but in a potentially more efficient way", and directs semantics-changing rewrites to `AnalyzerRule`. Measured against that, most of the spike's rules were misfiled: skip access-path resolution or the coverage split and a search silently returns the wrong rows. `ResolveVectorAccessPath`, `ResolveFtsAccessPath`, `SplitOnIndexCoverage`, and `ExpandVectorRefine` become analyzer rules. `UseFtsCompoundScorer` stays an optimizer rule — it is the only one that fits the trait's contract, since dropping it still returns the same rows. Two things follow. The analyzer runs each rule exactly once, where the optimizer re-runs the list until the plan stops changing, so the marker fields that existed only so a rule could recognize its own output are gone: `input_fully_indexed`, `refine_expanded`, and their `PartialEq`/`Hash` plumbing. And the analyzer checks `InvariantLevel::Executable` after its last rule, so `VectorSearchNode` and `FtsLeafNode` now reject an unresolved access path there — a rule that fails to fire is a planning error instead of a silent brute-force fallback. `ResolvePrefilterSource` stays in the optimizer. It is equally mandatory but answers "is there a predicate below the search?", which `PushDownFilter` is what settles, so it cannot run in a stage that precedes the optimizer. The `resolution.is_some()` checks in the two resolvers are kept and documented: they read like idempotence guards but are not. The builder resolves a search to `Flat` when its input already is the candidate set, and that is a decision. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The scalar index query was derived inside `TableProvider::scan`, so no rule could see it and its overlay coverage split had to be written by hand as an imperative `UnionExec` in `scan_impl`. Add a `filter_plan` slot to `LanceScanSource` with a public accessor, fill it from a new `ResolveScalarIndexQuery` rule, and recognize a resolved scan as one more `SplittableSearch`. All three coverage splits now go through `SplitOnIndexCoverage`. The scan half must observe `PushDownFilter`'s output, so the rule is registered in both stages, parameterized by `SplitScope`: searches in the analyzer, scans in the optimizer, before `PushDownLimit`. Being an optimizer rule, the scan half re-fires, so it is guarded by the block its own output carries. `fast_search` moves to `SplittableSearch::honors_fast_search` — it may drop a search's brute-force branch but not a scan's overlay re-read, which repairs the index result rather than extending it. Fixes `test_btree_overlay_row_level_precision` and `test_btree_overlay_masked_under_fast_search` under the flag. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Closes out the spike's remaining risk items. Guard list is now compiler-enforced: `ensure_supported` destructures `Scanner` exhaustively, so adding a field fails to compile until someone decides what this path does with it. That surfaced three shapes the path had been silently ignoring rather than rejecting — explicit `materialization_style`, a projection of only dynamic expressions, and a `_rowoffset` predicate. Oracle now compares rows in order by default. 22 of 24 unordered call sites were defensive; the 4 that remain are an all-ties FTS score sort and three coverage-split unions, each with its reason at the call site. Adds `benches/logical_scan_planner.rs`, comparing planning latency and execution for scan-heavy and search-heavy shapes. Execution is within 1% on filtered scans; planning is 2-4x more expensive. `SessionState` is built from the session `execute_plan` will run the plan on, rather than a bare `SessionConfig`, so lowering sees the runtime it will execute against. This removes the intermittent `ResourcesExhausted` failures under parallel test load. Also fixes a pre-existing bug in `FilteredReadExec`: it declared an `output_schema` carrying the dataset's schema metadata but its fragment readers emitted batches without it. Callers that end in a projection re-stamp the metadata and never notice; callers that emit the node's output directly do. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Planning built a `SessionState` per query, which re-populated DataFusion's entire function catalog via `with_default_features()` to plan a scan that consults none of it. That was ~37us per query — more than the imperative path spends building a whole plan for a trivial scan. It looked structural: each Lance rule holds an `Arc<ScanPlanningContext>`, so the rule list varies per query, and registering rules is what `SessionState` is for. But the rules never needed to be on the state. `Analyzer::with_rules` and `Optimizer::with_rules` are standalone, and `SessionState` already implements `OptimizerConfig` — `SessionState::optimize` itself is just those two calls. So the state now carries only what is query-independent (config, runtime env, catalog, Lance's physical rules) and is cached on `(session_id, target_partitions)`, while the rule lists are supplied per call. Planning, before -> after, against the imperative path: full_scan 64us -> 29us (imperative 28us) filtered_scan 115us -> 72us (imperative 55us) filtered_scan_with_limit 138us -> 99us (imperative 84us) ann 234us -> 198us (imperative 64us) ann_prefiltered 332us -> 279us (imperative 98us) The saving is 35-53us and near-constant across shapes, which is the signature of a fixed cost removed. Plain scans now plan at parity. The residual gap on searches is shape-dependent rule cost and is unaddressed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
fts.rs had grown to 1790 lines holding every FTS concern at once. Split it into a directory: one module per node type (leaf, compound, scorer, match_filter) plus prefetch, builder, rules, and planner. Largest module is now 397 lines. The five entry points the rest of the planner uses are unchanged; only where they live moved. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
nodes.rs, rules.rs, and planner.rs each held every index type's contribution at once, so adding an index type meant editing three shared files and reading past two other index types to do it. Reorganize so the framework is split by stage and each index type keeps its own node, rules, and lowering together: vector/ and fts/ now reach the framework through the same five entry points, and take/ holds late materialization. Cross-index concerns that no single index can decide stay central: rule ordering in mod.rs, and the coverage split — which serves every search node type through the SplittableSearch trait — in coverage.rs. Largest module drops from 1790 to 501 lines. No behavior change. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
tests.rs was 1505 lines. Split it so the equivalence oracle and dataset fixtures live in one harness module and the tests themselves sit next to the area they cover: scan, vector, fts, and whole-planner properties. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The logical path rejected v1 datasets outright. It only needed to: v1 and v2 differ at exactly one node, the scan leaf, and only when it reads ranges — `Fragment::take_all_tasks` is implemented for v1 while `read_ranges_tasks` is not, so takes, index lookups and every search node already work unchanged. `source/v1.rs` routes that one node through the frozen legacy builder via the existing `versions::filtered_read` seam, rather than reimplementing it. The legacy builder does not honor the `TableProvider` contract, so the leaf re-establishes it on the way out: a filter where the predicate was not applied (the plain fragment scan branch, which unlike the statistics pushdown and scalar-index branches leaves it to the caller), and a projection where the schema does not match. With both in place v1 reports exact filter pushdown like v2, and keeps the statistics pushdown that is its only page pruning. The legacy builder reads its options straight off `&Scanner`, so the leaf captures an `Arc<Scanner>` — set only on v1 — instead of changing that frozen signature. Equivalence tests run the plain-scan shapes against both storage versions, plus one case per legacy branch.
`include_deleted_rows` is a leaf option on both storage versions: the v2 leaf forwards it to `FilteredReadOptions::with_deleted_rows`, the v1 leaf to the legacy builder's `make_deletions_null`. It survives a coverage split, and should — the branches read disjoint fragments, so between them they still emit every deleted row once. Combined with a search it is rejected with the same error the imperative path gives, since a deleted row has no row id for a search to return. `strict_batch_size` is applied to the finished plan rather than modelled in it: the node delegates its plan properties to its input and only reshapes batches, so no rule has anything to say about it.
`ScanPlanningContext` now narrows a searched column's index to the segments `with_index_segments` named, validating the request the way the imperative path does: the segments must exist, cover the queried column, and belong to one logical index. That surfaced a divergence in metric resolution. A search that names no metric was being resolved to the element type's default and then compared against the index, so an index built with a non-default metric was silently downgraded to brute force. The imperative path adopts the index's metric in that case; the search node now records whether the metric was asked for, and only falls back when it was and disagrees. With explicit segments a disagreement is an error rather than a fallback, since the caller named segments that cannot answer their question. Also fixes an indexed search resolving to an empty segment list — when the scan's fragments reach none of the index's segments, there is no index to search and the query is brute force.
The builder restates a search's distance ordering above the take, because a logical `Sort` is the only way to say "this is the result's order". Whether the take actually preserved that ordering is a physical fact, so only a physical rule can drop the restatement. Adding `EnforceSorting` to the physical rule set does, now that take-mode `FilteredReadExec` advertises the ordering it carries through. A flat KNN plan is one sort — the top-k — instead of two.
Stages 2-4 read the `Scanner` for four things: the dataset, the fragment restriction, `fast_search`, and the requested index segments. All four are already on the scan leaf (or, for the requested metric, on the search node), so `ScanPlanningContext::collect` now takes only the plan. Any plan whose leaf is a `LanceScanSource` can be lowered, not just one the scanner's builder produced. That makes a DataFrame API possible: `SessionContext::read_lance_dataset` reads a dataset through the logical path's own leaf, `DataFrame::nearest` and `DataFrame::full_text_search` stack searches on it, and `DataFrame::lance_plan` lowers the result. A vector search whose input is already a search result scores exactly rather than consulting an index, matching what the scanner does for a vector search over a full-text filter. Not gated by `LANCE_LOGICAL_SCAN_PLANNER`: there is no imperative equivalent.
The builder emits a logical `Aggregate` and lets projection pushdown find the columns it reads, replacing the imperative path's hand-built `AggregateExec` and its by-hand `agg_projection` take. Two things this needed. Aggregate and group expressions are aliased to their unqualified names, so `sum(i)` does not become `sum(lance.i)` in the output schema. And the scan leaf now honors an empty projection — `COUNT(*)` asks for no columns, and the row address the reader needs as a stand-in is projected away rather than leaking into the aggregate's input.
A multivector column lowers to `Scanner::multivec_ann` in the imperative path. Nothing in the vector lowering here inspects the column type, so a multivector search would have built a single-vector fanout over it and returned wrong rows. Report it as unsupported until the node exists.
An indexed batch expands to a union of one search per query vector; a brute-force batch stays one node, since KNNVectorDistanceExec already scores every query in a single pass over the rows.
Multivector lowers to the per-vector fanout plus XTR scoring. A `_rowid` or `_rowaddr` predicate becomes a row restriction on the scan instead of a filter. `_rowoffset` gets a node whose fragment offsets are loaded in stage 2, since building it is the one physical constructor that does IO. Also fixes AddRowOffsetExec's statistics, which reported the input's column count while its schema had one more, panicking anything that read them by index.
The logical path read every projected column in one pass, so an explicit `materialization_style` was rejected and the default was only approximated. The leaf now splits the read the way `Scanner::filtered_read_source` does: with a refine filter it reads the cheap columns alongside the filter and takes the rest afterwards, then restates the requested projection so the leaf still returns exactly what it was asked for. `blob_handling` comes with it — it types the blob columns the leaf reads, and the width heuristic asks it whether a blob comes back as a description. Legacy (v1) storage opts out: its statistics-pushdown branch takes its projection off the `Scanner` and ignores the one it is handed, so narrowing here would add a take without narrowing the read.
A wide column that the refine filter also reads was counted as deferrable before the filter's own columns were added back, so the eager read ended up covering everything the take was supposed to fetch and the take failed with "the input plan already contains every projected field". Decide after the union instead, which is also the order `Scanner::calc_eager_projection` uses.
`SimplifyProjection` removed a projection that restated its input exactly, but left adjacent projections alone, so a plan that reorders columns above a node that already projects them kept both. A column-only projection over another projection is just a projection over the same input, so fold it — and re-examine the result, which is often the no-op the old rule handled. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The result vector was sized by subtracting the block list's length from the row count of the fragments being read. The block list is not restricted to those fragments — a plan may materialize an index over a subset of them — so the subtraction could underflow. Saturate instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The logical read path was behind `LANCE_LOGICAL_SCAN_PLANNER=1`, next to the imperative `Scanner::create_plan`. Make it the only path: the flag is gone, and with it the hand-written physical builder — the `(nearest, full_text_query)` dispatch, the vector and FTS source builders, the take and prefilter plumbing, and the limit pushdown the logical `Limit` expresses. The legacy (v1) exec nodes stay; the logical path calls them for legacy storage. Full-text results are now ordered by `_score` descending then `_rowid` ascending, stated in the plan rather than inherited from whichever operator sorted last. Without the tie-break a query with a limit and the same query without one disagreed on which of two equally relevant rows came first. The planner's tests compared the two paths against each other, which says nothing once there is one path. They now state what a query's answer is: each fixture derives its data from the row's `i`, so the expected rows are computed in Rust and checked against the scan, column values included. Approximate search asserts a recall floor against a brute-force oracle. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Contributor
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
#8557 added a logical-plan read path behind
LANCE_LOGICAL_SCAN_PLANNER=1, off by default, next to the imperativeScanner::create_plan. This makes it the only read path: the flag is gone, and with it the hand-written physical builder — about 3,200 lines ofscanner.rs, including the(nearest, full_text_query)dispatch, the vector and FTS source builders, the take and prefilter plumbing, and the limit pushdown that the logicalLimitnow expresses. The legacy (v1) exec nodes stay, since the logical path calls them for legacy storage.The bulk of the diff is tests, not deletion. Most of the planner's tests planned a query both ways and compared the results, which says nothing once there is only one way. They now state what the query's answer is: each fixture derives its data from the row's
i, so the expected rows can be computed in Rust and compared against what the scan returned, column values included. Approximate search asserts a recall floor against a brute-force oracle instead, and the tests that pin a plan's shape keep doing that.Flipping the default also brought out the last piece of parity the flagged path was missing: when a prefilter's predicate is answered exactly by a scalar index, the search now consumes the index lookup itself instead of a read of the rows it selects, as
prefilter_sourcedid.Two behavior changes are worth a look:
_scoredescending, then_rowidascending, stated as a sort above the take rather than inherited from whichever operator happened to sort last. The tie-break is new: relevance ties previously came out in whatever order the FTS operators produced, which was not the same for a query with a limit and the same query without one.ProjectionExecsits above each search to put_rowidand_distancein the order the take expects. Same rows, same reads.Two changes outside the scanner come along because the new path reaches them:
SimplifyProjectionnow folds a column-only projection into the projection below it, and re-examines the result, so the stacked projections the lowering produces collapse instead of surviving into the executed plan.row_ids_for_maskcomputed aVeccapacity by subtracting the block list's length from the row count, which underflows when a plan materializes an index over a subset of fragments the block list ranges beyond.Not included
LanceTableProviderstill plans throughScanner::create_planrather than being a thin wrapper overLanceScanSource. Consolidating the two needs somewhere for the legacy-storage escape hatch to live — the v1 leaf hands the wholeScannerto the frozen legacy builder, so the provider cannot stop building one — and it changes a public provider's pushdown behavior, which is worth its own PR.Stacked on #8557.