[X-3269] Empty-plan discriminator: RewriteReadiness + per-branch CandidateMetadata

[zhuqi-lucas]

Summary

Empty candidate plans currently look indistinguishable to a cost function regardless of source. A fully-populated MV whose leading-sort-key stats prove absence of a rare literal via metadata alone (X-3269 case, e.g. WHERE composite_ticker = 'PQXX' on constituents_by_composite_ticker_v4) collapses to bare EmptyExec — same shape as a freshly-deployed MV whose ingest task hasn't run yet. Downstream discriminators can't tell them apart and end up either penalising both (X-3269: specialised MV loses to a 30s base-table scan) or trusting both (X-2174: unpopulated MVs silently return empty answers).

This PR fixes the semantics by splitting the two concerns across three layers:

1. LP-time filter in ViewMatcher::f_down via a new
   Materialized::rewrite_readiness() -> RewriteReadiness trait method
   (Ready | NotReady | Unknown, default Unknown). MVs reporting
   NotReady are filtered out of the candidate set at LP rewrite time.

2. Per-branch metadata that survives to the cost function via
   RewriteContext::candidates: Vec<CandidateMetadata>, aligned with
   OneOf::branches. CandidateMetadata::Materialized { table_ref, readiness }
   carries the provider's raw readiness verbatim, so cost functions can
   apply their own policy for the Unknown case (e.g. fall back to base
   scan cost) instead of being forced to trust every candidate equally.
   table_ref is stored as a TableReference (not stringified) so
   quoted / dotted identifiers round-trip losslessly through the
   physical-time catalog resolution.

3. Physical-time readiness refresh in ViewExploitationPlanner::plan_extension
   re-consults each Materialized provider after DataFusion has invoked
   TableProvider::scan() on every candidate branch. Providers that opt
   into ReadinessAnnotatedExec (see below) get their readiness read
   straight off the plan tree — atomically captured at scan time.
   Providers that haven't opted in fall back to catalog sampling, which
   is best-effort and can be stale under concurrent snapshot swaps.
   drop_not_ready_after_refresh then removes any candidate that
   transitioned to NotReady, preserving the invariant that
   NotReady never reaches the cost function.

Architecture

The three layers wire together like this. The provider is consulted twice
— once at LP-rewrite time (cold, last_modified may still be None) and
again at physical-planning time (after scan() has run, so any lazy index is
warm and readiness is definitive). NotReady is filtered by two gates so the
cost function only ever sees Base | Ready | Unknown.

flowchart TD
    A["SQL"] --> B["Logical planning"]
    B --> C["ViewMatcher::f_down<br/>(LP time)"]
    C -->|"provider.rewrite_readiness()<br/>#1 · cold, pre-scan"| D{"Readiness?"}
    D -->|NotReady| E["🚫 filtered — never enters OneOf"]
    D -->|Ready / Unknown| F["OneOf {<br/>branches: [Base, MV_a, MV_b, ...],<br/>candidates: [Base, Materialized{ref, readiness}, ...]<br/>}"]
    F --> G["Physical planning"]
    G -->|"TableProvider::scan()<br/>· lazy index warms here<br/>· provider wraps output in<br/>ReadinessAnnotatedExec<br/>with atomically-captured readiness"| H["ViewExploitationPlanner::plan_extension"]
    H -->|"prefer annotated readiness on the plan<br/>OR fall back to sampling<br/>provider.rewrite_readiness() #2"| I["refresh_candidate_readiness<br/>· update metadata with fresh value"]
    I --> J["drop_not_ready_after_refresh<br/>· post-scan NotReady 🚫"]
    J -->|only Base left| K["Short-circuit:<br/>return base physical plan"]
    J -->|multiple candidates| L["OneOfExec::try_new(cost_fn)"]
    L -->|"CostContext.candidates:<br/>[Base, Ready, Unknown]<br/>(NotReady impossible)"| M["Cost function"]
    M --> N["PruneCandidates picks min"]
    K --> Z["Execute"]
    N --> Z

Loading

Atomic-readiness opt-in: Sampling rewrite_readiness() again after
scan() has returned is racy — a concurrent snapshot swap can publish
new state between the two calls, letting an EmptyExec from the old
snapshot be labelled with readiness from the new one. Providers close
that race by capturing readiness atomically at scan time and returning
Arc::new(ReadinessAnnotatedExec::new(scan_output, captured_readiness)).
The refresh walks each candidate's plan tree for a wrapper and reads
its readiness verbatim; providers that haven't opted in keep the
(racy) sampling fallback for backward compatibility.

The library ships layers 1–3 plus the atomic-readiness vehicle. The provider
(e.g. atlas's MaterializedView<Index>) implements rewrite_readiness however
fits its lifecycle model and optionally wraps its scan output in
ReadinessAnnotatedExec; the cost function branches on the readiness
value however fits its cost model. The atlas companion PR wires up both:

Layer	Responsibility	Owned by
Materialized::rewrite_readiness	Report lifecycle state (cheap check)	Provider
f_down NotReady filter	Drop pre-scan NotReady from candidate set	This PR
Per-branch CandidateMetadata	Carry readiness value through to cost fn	This PR
ReadinessAnnotatedExec wrapper	Atomically propagate readiness with the scan output	Provider opt-in; wrapper defined here
plan_extension refresh + filter	Re-consult (annotated or sampled) after scan, drop post-scan NotReady	This PR
deprioritize_empty_candidates	Trust Ready+empty, penalise Unknown+empty	Cost fn (atlas side)

File organisation

All readiness-related types and helpers live in src/rewrite/readiness.rs:
RewriteReadiness, CandidateMetadata, ReadinessAnnotatedExec,
readiness_from_plan, refresh_candidate_readiness,
drop_not_ready_after_refresh, resolve_current_readiness. The pre-move
locations (crate::materialized::RewriteReadiness,
crate::rewrite::exploitation::{CandidateMetadata, ReadinessAnnotatedExec, readiness_from_plan}) still resolve via pub use re-exports, so downstream
callers don't have to change their imports.

Changes

Materialized trait (in src/materialized.rs)

• New RewriteReadiness enum (defined in src/rewrite/readiness.rs,
  re-exported here for backward compat): Ready / NotReady / Unknown.
• New fn rewrite_readiness(&self) -> RewriteReadiness with Unknown
  default — backward-compatible for providers that don't distinguish
  lifecycle states.

CandidateMetadata + ReadinessAnnotatedExec (in src/rewrite/readiness.rs)

• CandidateMetadata::Materialized { table_ref: TableReference, readiness }
  carries the provider's raw readiness verbatim. TableReference (not
  String) so quoted / dotted identifiers round-trip losslessly.
  Sorted deterministically by TableReference in f_down so alignment
  survives downstream LP rewrites via with_exprs_and_inputs.
• ReadinessAnnotatedExec transparent wrapper carrying a
  RewriteReadiness value alongside an ExecutionPlan, so providers can
  publish the readiness that corresponds to the snapshot their scan()
  read from.
• readiness_from_plan depth-first walker used by the refresh to
  extract annotated readiness from a candidate's physical input.

ViewMatcher::f_down

• Filters NotReady MVs before they enter the candidate set.
• Encodes Ready / Unknown values into CandidateMetadata::Materialized.
• Pre-sorts MV candidates so OneOf::inputs() becomes a no-op.
• Keeps the base branch pinned at branches[0].

ViewExploitationPlanner::plan_extension

• Calls refresh_candidate_readiness — prefers ReadinessAnnotatedExec
  when a candidate's physical_input carries one; falls back to catalog
  sampling otherwise.
• Filters NotReady post-scan transitions via drop_not_ready_after_refresh.
• Short-circuits to the base physical plan directly when only Base remains.

Test coverage (61 lib tests, all passing)

New in this PR (organised by concern):

• Metadata field — f_down_encodes_ready_and_unknown_readiness_verbatim,
  candidate_metadata_partial_eq_discriminates_readiness,
  one_of_alignment_survives_rebuild_with_unknown_readiness.
• LP-time filter — f_down_excludes_not_ready_and_admits_ready_and_unknown,
  f_down_returns_original_lp_when_every_mv_is_not_ready,
  f_down_reflects_provider_readiness_on_each_rewrite.
• Physical-time refresh — 5 tests covering post-scan Unknown → Ready,
  Unknown → NotReady when scan uncovers unpopulated, missing-provider
  fallback, Base variant preservation, and Base-at-index-0 invariant.
• Atomic-readiness wrapper — refresh_prefers_annotated_readiness_over_catalog_sampling
  (intentional disagreement between annotated Ready and sampled Unknown,
  asserts annotated wins), refresh_falls_back_to_sampling_when_input_not_annotated,
  readiness_from_plan_walks_nested_children,
  readiness_from_plan_returns_none_when_no_annotation,
  readiness_annotated_exec_delegates_properties_and_children.
• NotReady filter — 3 tests covering the happy path, backward-compat
  passthrough when metadata is missing, and the collapse-to-Base short-circuit.
• End-to-end via plan_extension — 2 tests driving the actual planner
  method through a SessionContext, covering the cold-lazy Ready promotion
  and the short-circuit-to-base scenarios.

Affects

Downstream cost functions consuming RewriteContext::candidates() and MV
providers overriding rewrite_readiness(). The atlas repo has a companion
PR that:

• Overrides MaterializedView<Index>::rewrite_readiness() to report Ready
  when Index::get_last_modified().is_some() and Unknown otherwise.
  Never returns NotReady — atlas cannot cheaply distinguish cold-lazy
  from unpopulated, so Unknown + cost-fn fallback handles both.
• Modifies deprioritize_empty_candidates to skip the +INFINITY penalty
  when a candidate is Materialized { readiness: Ready } (trust the empty
  as predicate-pruned); the pre-readiness "penalize every empty" behaviour
  is retained as a fallback for Unknown and for older callsites that
  don't wire CandidateMetadata through.
• Verified end-to-end against ny2 staging: PQXX-shape query on
  external.etfglobal.constituents_v1 picks the specialised MV, returns
  EmptyExec in microseconds instead of the previous 30 s base-table
  scan.

[Copilot]

Pull request overview

This PR extends the view-matching rewrite metadata so downstream cost functions can distinguish “populated MV pruned to empty by predicates” from “unpopulated MV (zero files)” when an empty physical plan collapses to EmptyExec.

Changes:

• Add Materialized::file_count() -> Option<usize> (default None) to expose total MV file count (ignoring predicates).
• Extend RewriteContext with per-branch candidate_table_refs and candidate_file_counts, plus builder/getter APIs.
• Populate these per-branch signals when constructing OneOf candidates in ViewMatchingRewriter::f_down.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
src/rewrite/exploitation.rs	Threads per-branch MV identity + file-count into RewriteContext during OneOf candidate construction.
src/materialized.rs	Adds optional file_count() hook to the Materialized trait for populated-ness discrimination.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[xudong963]

I thought more, considering this is a lib, it might be better to reduce coupling with the storage layout.

The cost model only needs to know whether an MV is safe to use, while file_count is an unreliable proxy for that lifecycle state.

How about considering exposing an explicit Ready | NotReady | Unknown status from the same published snapshot used for scanning, and excluding NotReady candidates in ViewMatchingRewriter. This keeps lifecycle correctness in the provider instead of coupling the cost model to storage layout.

pub enum RewriteReadiness {
    Ready,
    NotReady,
    Unknown,
}

pub trait Materialized: ListingTableLike {
    fn rewrite_readiness(&self) -> RewriteReadiness {
        RewriteReadiness::Unknown
    }
}

On our side:

match self.index() {
    Some(index) if index.get_last_modified().is_some() => RewriteReadiness::Ready,
    Some(_) => RewriteReadiness::NotReady,
    None => RewriteReadiness::Unknown,
}

[zhuqi-lucas]

I thought more, considering this is a lib, it might be better to reduce coupling with the storage layout.

The cost model only needs to know whether an MV is safe to use, while file_count is an unreliable proxy for that lifecycle state.

How about considering exposing an explicit Ready | NotReady | Unknown status from the same published snapshot used for scanning, and excluding NotReady candidates in ViewMatchingRewriter. This keeps lifecycle correctness in the provider instead of coupling the cost model to storage layout.

pub enum RewriteReadiness {
    Ready,
    NotReady,
    Unknown,
}

pub trait Materialized: ListingTableLike {
    fn rewrite_readiness(&self) -> RewriteReadiness {
        RewriteReadiness::Unknown
    }
}

On our side:

match self.index() {
    Some(index) if index.get_last_modified().is_some() => RewriteReadiness::Ready,
    Some(_) => RewriteReadiness::NotReady,
    None => RewriteReadiness::Unknown,
}

Thanks @xudong963 , great idea, so the emptyexec after this must be the prune cases, in progress.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.

[zhuqi-lucas]

@xudong963 Updated the design now, and added rich tests.

[xudong963]

This readiness check happens during logical rewriting, before the candidate scan is planned. In our internal case, lazy_index is enabled and a cold index has get_last_modified() == None until Index::query() is called from FileScanTable::scan(). With the proposed Atlas mapping, a valid but not-yet-loaded MV is therefore classified as NotReady and filtered here, so the scan that would initialize it is never planned.

Returning Unknown does not fully solve this either: Unknown is admitted below, but CandidateMetadata::Materialized retains only table_ref, so the cost function cannot distinguish it from Ready despite being documented as responsible for the unknown-state policy.

Could readiness be resolved after the physical inputs have been planned, when lazy index initialization has completed, or otherwise be propagated so Unknown must be handled explicitly? A cold-start test with lazy_index = true would catch this case later in our internal.

[zhuqi-lucas]

Good catch!

[xudong963]

Unknown is documented as leaving the policy to the cost function, but this value is discarded before constructing RewriteContext. f_down admits both Ready and Unknown, and both are encoded as CandidateMetadata::Materialized { table_ref }, so a cost function cannot distinguish them through this API.

This also makes it unsafe to assume that every materialized EmptyExec is predicate-pruned, especially because Unknown is the default for existing providers. Could readiness be retained in CandidateMetadata, or could Unknown be handled explicitly rather than being erased?

[zhuqi-lucas]

Readiness be retained in CandidateMetadata is a good suggestion.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

src/rewrite/exploitation.rs:609

• drop_not_ready_after_refresh returns the (potentially misaligned) candidates slice unchanged when candidates.len() != physical_inputs.len(). If a caller accidentally provides a non-empty but mismatched metadata vector, this will preserve the misalignment into OneOfExec, where downstream cost functions may index by branch and panic or misattribute readiness. Safer behavior is to preserve the inputs but drop metadata (treat as backward-compat “no metadata”).

    if candidates.len() != physical_inputs.len() {
        return (physical_inputs.to_vec(), candidates.to_vec());
    }

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.

[xudong963]

Could we put the Readiness-related struct/APIs into a seperate file?

[xudong963]

This fast path makes the post-scan safety check depend on an undocumented monotonicity assumption. rewrite_readiness() currently describes the provider’s current state; its public contract does not promise that Ready can never transition to NotReady.

If f_down observes Ready and scan() subsequently discovers that the snapshot is unusable, this path retains the stale Ready value. drop_not_ready_after_refresh cannot remove the candidate, and the cost function may trust its EmptyExec, recreating the silent-empty-result failure this PR is intended to prevent.

Could we refresh every Materialized candidate after scan, or make monotonic readiness an explicit API capability?

[xudong963]

Re-reading readiness after scan() only observes the provider’s current state; it does not guarantee that this is the same snapshot that was used to construct the corresponding physical_input.

For example, scan() could produce an EmptyExec from an unknown or unpublished snapshot, and a concurrent refresh could publish a Ready snapshot before this lookup runs. The old EmptyExec would then be labeled Ready and trusted as predicate-pruned.

Could readiness, or a snapshot generation/token, be captured atomically during scan() and propagated with the physical input instead of being sampled independently afterward?

[zhuqi-lucas]

Be captured atomically during scan() , this is a great idea, addressing this.

[xudong963]

TableReference::to_string() followed by parse_str() is not lossless for quoted or case-sensitive identifiers, or identifiers whose components contain dots. Display does not preserve quoting, while parse_str() normalizes and splits the resulting string.

This can make the physical-time refresh resolve the wrong provider or fail and silently retain stale LP-time readiness. Since TableReference already implements Clone, Eq, Hash, and ordering traits, could CandidateMetadata retain it directly, or at least store to_quoted_string()?

[zhuqi-lucas]

Good point!

[zhuqi-lucas]

Could we put the Readiness-related struct/APIs into a seperate file?

Good suggestion!

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

src/rewrite/exploitation.rs:396

• f_down sorts MV candidates using table_ref.to_string(). Elsewhere (e.g. readiness resolution) the code explicitly avoids going through Display because it’s not lossless for quoted/dotted identifiers; using to_string() here can therefore collapse distinct identifiers and undermine the determinism/alignment invariant. Prefer ordering directly on TableReference instead of its display string (and avoid per-candidate allocations).

            let mut candidates = candidates;
            candidates.sort_by_key(|a| a.1.to_string());

[xudong963]

ReadinessAnnotatedExec is not actually transparent to DataFusion. It delegates only properties, execution, and partition statistics, while ExecutionPlan also exposes optimizer-facing behavior such as input distribution/order requirements, maintains_input_order, repartitioning, limit/filter/sort pushdown, metrics, with_preserve_order, and downcast_delegate. The defaults are not necessarily equivalent to the wrapped scan.

Because drop_not_ready_after_refresh retains the original physical input, this wrapper survives into OneOfExec and eventually into the selected execution plan, where it can block optimizations or appear as an unknown node to physical-plan codecs.

Since the annotation is only needed during plan_extension, could we strip it from the physical input immediately after extracting readiness? Alternatively, the wrapper would need to proxy all relevant behavior/type identity and be covered by optimizer and serialization integration tests.

[zhuqi-lucas]

Good point, thanks @xudong963 , addressed in latest PR.