diff --git a/crates/omnigraph-compiler/src/catalog/schema_ir.rs b/crates/omnigraph-compiler/src/catalog/schema_ir.rs index 3546302e0..efd3f9a4c 100644 --- a/crates/omnigraph-compiler/src/catalog/schema_ir.rs +++ b/crates/omnigraph-compiler/src/catalog/schema_ir.rs @@ -47,7 +47,13 @@ pub struct SchemaIdentityDomain(String); impl SchemaIdentityDomain { pub fn new() -> Self { - Self(Ulid::new().to_string()) + Self::from_ulid(Ulid::new()) + } + + /// The domain of an already-minted ULID; the engine mints through its + /// identity seam. + pub fn from_ulid(ulid: Ulid) -> Self { + Self(ulid.to_string()) } pub fn parse(value: impl AsRef) -> Result { diff --git a/crates/omnigraph/src/branch_names.rs b/crates/omnigraph/src/branch_names.rs index a93ffad69..7f573ad07 100644 --- a/crates/omnigraph/src/branch_names.rs +++ b/crates/omnigraph/src/branch_names.rs @@ -20,7 +20,7 @@ pub(crate) const INCARNATION_LEN: usize = 26; /// Mint a fresh branch incarnation. pub(crate) fn mint_incarnation() -> String { - ulid::Ulid::new().to_string() + crate::dst_ids::new_ulid().to_string() } /// The native Lance ref name for one incarnation of a logical branch. diff --git a/crates/omnigraph/src/db/mod.rs b/crates/omnigraph/src/db/mod.rs index 61820d572..51e2240e2 100644 --- a/crates/omnigraph/src/db/mod.rs +++ b/crates/omnigraph/src/db/mod.rs @@ -18,6 +18,7 @@ pub use omnigraph::{ }; pub(crate) use omnigraph::{DeferredTableFork, WriteAuthorityToken, WriteTxn}; pub(crate) use omnigraph::{export_blob_values, logical_row_image}; +pub(crate) use schema_state::SchemaContractText; use crate::error::{OmniError, Result}; diff --git a/crates/omnigraph/src/db/omnigraph.rs b/crates/omnigraph/src/db/omnigraph.rs index ad619607e..6d26c23f1 100644 --- a/crates/omnigraph/src/db/omnigraph.rs +++ b/crates/omnigraph/src/db/omnigraph.rs @@ -51,11 +51,12 @@ use super::manifest::{ GenesisManifestAttempt, ManifestChange, Snapshot, TableRegistration, TableTombstone, }; use super::schema_state::{ - SCHEMA_SOURCE_FILENAME, load_validated_schema_contract, - load_validated_schema_contract_for_source, read_accepted_schema_ir, read_schema_state_identity, - recover_schema_state_files, schema_ir_uri, schema_source_staging_uri, schema_source_uri, - schema_state_uri, validate_schema_contract, validate_schema_ir_against_snapshot, - write_schema_contract, write_schema_contract_staging, + SCHEMA_SOURCE_FILENAME, SchemaContractText, load_validated_schema_contract, + load_validated_schema_contract_for_source, read_accepted_schema_ir, read_schema_contract_text, + read_schema_contract_text_for_source, read_schema_state_identity, recover_schema_state_files, + render_schema_contract, schema_ir_uri, schema_source_staging_uri, schema_source_uri, + schema_state_uri, validate_schema_contract, validate_schema_contract_text, + validate_schema_ir_against_snapshot, write_schema_contract, write_schema_contract_staging, }; use super::{ ReadTarget, ResolvedTarget, SCHEMA_APPLY_LOCK_BRANCH, SnapshotId, is_internal_system_branch, @@ -420,7 +421,7 @@ impl Omnigraph { let schema_shape = read_schema_shape_from_source(schema_source)?; let resolution = omnigraph_compiler::initialize_schema_ir_with_actor_provenance( - SchemaIdentityDomain::new(), + SchemaIdentityDomain::from_ulid(crate::dst_ids::new_ulid()), &schema_shape, options.actor_provenance, ) @@ -440,6 +441,12 @@ impl Omnigraph { let schema_identity_domain = schema_ir.schema_identity_domain.as_str().to_string(); let mut catalog = build_catalog_from_ir(&schema_ir)?; fixup_physical_schemas(&mut catalog)?; + let (_, ir_json, state_json) = render_schema_contract(&schema_ir)?; + let contract = SchemaContractText { + source: schema_source.to_string(), + ir_json, + state_json, + }; // Every init write needs atomic create-if-absent (the root init claim, // the strict-mode `_schema.pg` defence, and each Lance commit), so @@ -521,8 +528,7 @@ impl Omnigraph { // remove the graph directory manually before retrying `init`. let coordinator = match init_commit_phase( &root, - schema_source, - &schema_ir, + &contract, &catalog, &storage, !schema_pg_claimed, @@ -586,6 +592,16 @@ impl Omnigraph { }; let session = lance_access.data_session(); + let catalog = Arc::new(catalog); + let read_caches = Arc::new(crate::runtime_cache::ReadCaches { + session: Arc::clone(&session), + handles: Arc::new(crate::runtime_cache::TableHandleCache::default()), + accepted_catalog: crate::runtime_cache::AcceptedCatalogMemo::default(), + compiled_queries: crate::runtime_cache::CompiledQueryCache::default(), + }); + read_caches + .accepted_catalog + .memoize(contract, Arc::clone(&catalog)); Ok(Self { root_uri: root.clone(), storage, @@ -595,15 +611,12 @@ impl Omnigraph { // warm across reads, writes, and maintenance. Mutable control // metadata uses the context's separate zero-cache session; both // sessions reuse the process-wide object-store registry. - table_store: TableStore::new(&root, session.clone()), + table_store: TableStore::new(&root, session), runtime_cache: RuntimeCache::default(), feed_cut_cache: tokio::sync::RwLock::new(None), - read_caches: Arc::new(crate::runtime_cache::ReadCaches { - session, - handles: Arc::new(crate::runtime_cache::TableHandleCache::default()), - }), + read_caches, schema_view: Arc::new(ArcSwap::from_pointee(HandleSchemaView { - catalog: Arc::new(catalog), + catalog, source: Arc::new(schema_source.to_string()), schema_ir_hash: accepted_schema_ir_hash, schema_identity_domain, @@ -758,15 +771,26 @@ impl Omnigraph { "graph at '{root}' is missing its schema files: '_schema.pg' was not found although '__manifest' is readable; restore the matching '_schema.pg', '_schema.ir.json', and '__schema_state.json' contract from a backup, or rebuild a fresh graph from an existing export or backup" ))); }; - let (accepted_ir, accepted_state) = - load_validated_schema_contract_for_source(&root, Arc::clone(&storage), &schema_source) - .await?; + let contract = + read_schema_contract_text_for_source(&root, storage.as_ref(), schema_source).await?; + let (accepted_ir, accepted_state) = validate_schema_contract_text(&contract)?; validate_schema_ir_against_snapshot(&accepted_ir, &coordinator.snapshot())?; let schema_identity_domain = accepted_ir.schema_identity_domain.as_str().to_string(); let mut catalog = build_catalog_from_ir(&accepted_ir)?; fixup_physical_schemas(&mut catalog)?; let session = lance_access.data_session(); + let catalog = Arc::new(catalog); + let schema_source = Arc::new(contract.source.clone()); + let read_caches = Arc::new(crate::runtime_cache::ReadCaches { + session: Arc::clone(&session), + handles: Arc::new(crate::runtime_cache::TableHandleCache::default()), + accepted_catalog: crate::runtime_cache::AcceptedCatalogMemo::default(), + compiled_queries: crate::runtime_cache::CompiledQueryCache::default(), + }); + read_caches + .accepted_catalog + .memoize(contract, Arc::clone(&catalog)); let db = Self { root_uri: root.clone(), storage, @@ -776,16 +800,13 @@ impl Omnigraph { // warm across reads, writes, and maintenance. Mutable control // metadata uses the context's separate zero-cache session; both // sessions reuse the process-wide object-store registry. - table_store: TableStore::new(&root, session.clone()), + table_store: TableStore::new(&root, session), runtime_cache: RuntimeCache::default(), feed_cut_cache: tokio::sync::RwLock::new(None), - read_caches: Arc::new(crate::runtime_cache::ReadCaches { - session, - handles: Arc::new(crate::runtime_cache::TableHandleCache::default()), - }), + read_caches, schema_view: Arc::new(ArcSwap::from_pointee(HandleSchemaView { - catalog: Arc::new(catalog), - source: Arc::new(schema_source), + catalog, + source: schema_source, schema_ir_hash: accepted_state.schema_ir_hash, schema_identity_domain, })), @@ -977,11 +998,25 @@ impl Omnigraph { /// [`Self::load_accepted_catalog_with_schema_gate_held`] unless they perform /// that post-resolution identity join themselves. async fn build_accepted_catalog_with_schema_gate_held(&self) -> Result> { - let (schema_ir, _) = - load_validated_schema_contract(self.uri(), Arc::clone(&self.storage)).await?; + let text = read_schema_contract_text(self.uri(), self.storage.as_ref()).await?; + if let Some(catalog) = self.read_caches.accepted_catalog.get(&text) { + return Ok(catalog); + } + let (schema_ir, _) = validate_schema_contract_text(&text)?; let mut catalog = build_catalog_from_ir(&schema_ir)?; fixup_physical_schemas(&mut catalog)?; - Ok(Arc::new(catalog)) + crate::instrumentation::record_catalog_build(); + let catalog = Arc::new(catalog); + self.read_caches + .accepted_catalog + .memoize(text, Arc::clone(&catalog)); + Ok(catalog) + } + + /// The per-graph read caches (`ReadCaches`): table handles, the accepted + /// catalog memo, and the compiled-query cache. + pub(crate) fn read_caches(&self) -> &Arc { + &self.read_caches } /// Read customer source and its full accepted schema, including protected @@ -1871,6 +1906,7 @@ impl Omnigraph { async fn invalidate_read_caches(&self) { self.runtime_cache.invalidate_all().await; self.read_caches.handles.invalidate_all().await; + // `accepted_catalog` and `compiled_queries` are content-keyed: no invalidation. // Hygiene, like the caches above: the incarnation key already makes a // stale feed cut miss, but a same-branch refresh clears it so an // e_tag-less substrate cannot serve a recreated branch's projection @@ -4190,8 +4226,7 @@ enum InitCommitError { async fn init_commit_phase( root: &str, - schema_source: &str, - schema_ir: &SchemaIR, + contract: &SchemaContractText, catalog: &Catalog, storage: &Arc, write_schema_pg: bool, @@ -4201,14 +4236,14 @@ async fn init_commit_phase( if write_schema_pg { let schema_path = join_uri(root, SCHEMA_SOURCE_FILENAME); storage - .write_text(&schema_path, schema_source) + .write_text(&schema_path, &contract.source) .await .map_err(InitCommitError::BeforePhysicalInit)?; crate::failpoints::maybe_fail(crate::failpoints::names::INIT_AFTER_SCHEMA_PG_WRITTEN) .map_err(InitCommitError::BeforePhysicalInit)?; } - write_schema_contract(root, storage.as_ref(), schema_ir) + write_schema_contract(root, storage.as_ref(), contract) .await .map_err(InitCommitError::BeforePhysicalInit)?; crate::failpoints::maybe_fail(crate::failpoints::names::INIT_AFTER_SCHEMA_CONTRACT_WRITTEN) diff --git a/crates/omnigraph/src/db/schema_state.rs b/crates/omnigraph/src/db/schema_state.rs index edc8277ba..2ba6ee41f 100644 --- a/crates/omnigraph/src/db/schema_state.rs +++ b/crates/omnigraph/src/db/schema_state.rs @@ -27,6 +27,11 @@ pub(crate) const SCHEMA_STATE_STAGING_FILENAME: &str = "__schema_state.json.stag const SCHEMA_STATE_FORMAT_VERSION: u32 = 2; const SCHEMA_IDENTITY_VERSION: u32 = 2; +const MISSING_SCHEMA_CONTRACT_MESSAGE: &str = "graph is missing the mandatory identity-bearing schema contract (_schema.ir.json and __schema_state.json); automatic bootstrap is not supported"; +const INCOMPLETE_SCHEMA_CONTRACT_MESSAGE: &str = "graph schema contract is incomplete: _schema.ir.json and __schema_state.json must both be present"; +const INVALID_SCHEMA_IR_SUBJECT: &str = "accepted compiled schema contract"; +const INVALID_SCHEMA_STATE_SUBJECT: &str = "graph schema state"; + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct SchemaState { pub(crate) format_version: u32, @@ -69,8 +74,8 @@ pub(crate) async fn load_validated_schema_contract( root_uri: &str, storage: Arc, ) -> Result<(SchemaIR, SchemaState)> { - let source = storage.read_text(&schema_source_uri(root_uri)).await?; - load_validated_schema_contract_for_source(root_uri, storage, &source).await + let text = read_schema_contract_text(root_uri, storage.as_ref()).await?; + validate_schema_contract_text(&text) } /// Validate the complete durable schema contract against source bytes already @@ -82,24 +87,83 @@ pub(crate) async fn load_validated_schema_contract_for_source( storage: Arc, source: &str, ) -> Result<(SchemaIR, SchemaState)> { - let current_source_shape = compile_schema_source(source)?; - let (persisted_ir, state) = match read_schema_contract(root_uri, storage.as_ref()).await? { - SchemaContractRead::Present { ir, state } => (*ir, state), + let text = read_schema_contract_text_for_source(root_uri, storage.as_ref(), source.to_string()) + .await?; + validate_schema_contract_text(&text) +} + +/// The three durable schema-contract files as read, byte-exact. Memo key of +/// `ReadCaches::accepted_catalog` (see `AcceptedCatalogMemo`). +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SchemaContractText { + pub(crate) source: String, + pub(crate) ir_json: String, + pub(crate) state_json: String, +} + +/// Read the schema contract's bytes: three `read_text` and two `exists` calls +/// (the per-query drift detection the lifecycle tests pin). A missing or +/// incomplete contract reports an unparseable source first; a storage error +/// from the existence probes is reported before a parse error, and so is a +/// failing IR or state read on the present-contract path. +pub(crate) async fn read_schema_contract_text( + root_uri: &str, + storage: &dyn StorageAdapter, +) -> Result { + let source = storage.read_text(&schema_source_uri(root_uri)).await?; + read_schema_contract_text_for_source(root_uri, storage, source).await +} + +/// [`read_schema_contract_text`] over a source already read under the root +/// schema gate. +pub(crate) async fn read_schema_contract_text_for_source( + root_uri: &str, + storage: &dyn StorageAdapter, + source: String, +) -> Result { + match read_schema_contract(root_uri, storage).await? { + SchemaContractRead::Present { + ir_json, + state_json, + } => Ok(SchemaContractText { + source, + ir_json, + state_json, + }), SchemaContractRead::MissingAll => { - return Err(schema_lock_conflict( - "graph is missing the mandatory identity-bearing schema contract (_schema.ir.json and __schema_state.json); automatic bootstrap is not supported", - )); + compile_schema_source(&source)?; + Err(schema_lock_conflict(MISSING_SCHEMA_CONTRACT_MESSAGE)) } SchemaContractRead::PartialMissing => { - return Err(schema_lock_conflict( - "graph schema contract is incomplete: _schema.ir.json and __schema_state.json must both be present", - )); + compile_schema_source(&source)?; + Err(schema_lock_conflict(INCOMPLETE_SCHEMA_CONTRACT_MESSAGE)) } - }; + } +} - validate_persisted_schema_contract(&persisted_ir, &state)?; +/// Validate contract bytes: source compiled, IR and state parsed, IR checked +/// against the state, source shape checked against the state. +pub(crate) fn validate_schema_contract_text( + text: &SchemaContractText, +) -> Result<(SchemaIR, SchemaState)> { + let current_source_shape = compile_schema_source(&text.source)?; + let (ir, state) = parse_schema_contract(&text.ir_json, &text.state_json)?; + validate_persisted_schema_contract(&ir, &state)?; validate_current_source_matches(&state, ¤t_source_shape)?; - Ok((persisted_ir, state)) + Ok((ir, state)) +} + +fn parse_schema_contract(ir_json: &str, state_json: &str) -> Result<(SchemaIR, SchemaState)> { + let ir = serde_json::from_str::(ir_json) + .map_err(|err| invalid_contract_file(INVALID_SCHEMA_IR_SUBJECT, SCHEMA_IR_FILENAME, err))?; + let state = serde_json::from_str::(state_json).map_err(|err| { + invalid_contract_file(INVALID_SCHEMA_STATE_SUBJECT, SCHEMA_STATE_FILENAME, err) + })?; + Ok((ir, state)) +} + +fn invalid_contract_file(subject: &str, file: &str, err: impl std::fmt::Display) -> OmniError { + schema_lock_conflict(format!("{subject} in {file} is invalid: {err}")) } /// Read only the durable schema-identity marker. Schema apply promotes this @@ -113,25 +177,39 @@ pub(crate) async fn read_schema_state_identity( ) -> Result { let text = storage.read_text(&schema_state_uri(root_uri)).await?; let state = serde_json::from_str::(&text).map_err(|err| { - schema_lock_conflict(format!( - "graph schema state in {} is invalid: {}", - SCHEMA_STATE_FILENAME, err - )) + invalid_contract_file(INVALID_SCHEMA_STATE_SUBJECT, SCHEMA_STATE_FILENAME, err) })?; validate_schema_state_envelope(&state)?; Ok(state) } +/// The IR and state JSON of `schema_ir` as the contract writers put them on +/// the object store, with the state they encode. +pub(crate) fn render_schema_contract( + schema_ir: &SchemaIR, +) -> Result<(SchemaState, String, String)> { + let ir_json = schema_ir_pretty_json(schema_ir) + .map_err(|err| OmniError::manifest_internal(err.to_string()))?; + let state = SchemaState::from_ir(schema_ir)?; + let state_json = serde_json::to_string_pretty(&state).map_err(|err| { + OmniError::manifest_internal(format!("serialize schema state error: {}", err)) + })?; + Ok((state, ir_json, state_json)) +} + +/// Write the IR and state of an already-rendered contract to their final +/// filenames; init keeps the same text to seed its accepted-catalog memo. pub(crate) async fn write_schema_contract( root_uri: &str, storage: &dyn StorageAdapter, - schema_ir: &SchemaIR, -) -> Result { + contract: &SchemaContractText, +) -> Result<()> { write_schema_contract_to( storage, &schema_ir_uri(root_uri), &schema_state_uri(root_uri), - schema_ir, + &contract.ir_json, + &contract.state_json, ) .await } @@ -144,31 +222,27 @@ pub(crate) async fn write_schema_contract_staging( storage: &dyn StorageAdapter, schema_ir: &SchemaIR, ) -> Result { + let (state, ir_json, state_json) = render_schema_contract(schema_ir)?; write_schema_contract_to( storage, &schema_ir_staging_uri(root_uri), &schema_state_staging_uri(root_uri), - schema_ir, + &ir_json, + &state_json, ) - .await + .await?; + Ok(state) } async fn write_schema_contract_to( storage: &dyn StorageAdapter, ir_uri: &str, state_uri: &str, - schema_ir: &SchemaIR, -) -> Result { - let ir_json = schema_ir_pretty_json(schema_ir) - .map_err(|err| OmniError::manifest_internal(err.to_string()))?; - let state = SchemaState::from_ir(schema_ir)?; - let state_json = serde_json::to_string_pretty(&state).map_err(|err| { - OmniError::manifest_internal(format!("serialize schema state error: {}", err)) - })?; - - storage.write_text(ir_uri, &ir_json).await?; - storage.write_text(state_uri, &state_json).await?; - Ok(state) + ir_json: &str, + state_json: &str, +) -> Result<()> { + storage.write_text(ir_uri, ir_json).await?; + storage.write_text(state_uri, state_json).await } pub(crate) async fn read_accepted_schema_ir( @@ -176,9 +250,13 @@ pub(crate) async fn read_accepted_schema_ir( storage: Arc, ) -> Result { match read_schema_contract(root_uri, storage.as_ref()).await? { - SchemaContractRead::Present { ir, state } => { + SchemaContractRead::Present { + ir_json, + state_json, + } => { + let (ir, state) = parse_schema_contract(&ir_json, &state_json)?; validate_persisted_schema_contract(&ir, &state)?; - Ok(*ir) + Ok(ir) } SchemaContractRead::MissingAll => Err(schema_lock_conflict( "graph is missing the mandatory identity-bearing schema contract; automatic bootstrap is not supported", @@ -322,10 +400,7 @@ pub(crate) fn schema_state_staging_uri(root_uri: &str) -> String { } enum SchemaContractRead { - Present { - ir: Box, - state: SchemaState, - }, + Present { ir_json: String, state_json: String }, MissingAll, PartialMissing, } @@ -344,21 +419,9 @@ async fn read_schema_contract( (true, true) => { let ir_json = storage.read_text(&ir_uri).await?; let state_json = storage.read_text(&state_uri).await?; - let ir = serde_json::from_str::(&ir_json).map_err(|err| { - schema_lock_conflict(format!( - "accepted compiled schema contract in {} is invalid: {}", - SCHEMA_IR_FILENAME, err - )) - })?; - let state = serde_json::from_str::(&state_json).map_err(|err| { - schema_lock_conflict(format!( - "graph schema state in {} is invalid: {}", - SCHEMA_STATE_FILENAME, err - )) - })?; Ok(SchemaContractRead::Present { - ir: Box::new(ir), - state, + ir_json, + state_json, }) } _ => Ok(SchemaContractRead::PartialMissing), diff --git a/crates/omnigraph/src/exec/query.rs b/crates/omnigraph/src/exec/query.rs index e282e20ca..182e7d736 100644 --- a/crates/omnigraph/src/exec/query.rs +++ b/crates/omnigraph/src/exec/query.rs @@ -68,10 +68,7 @@ impl Omnigraph { // the applying handle could pair that new snapshot with the old catalog. let (resolved, catalog) = self.capture_read_view(target).await?; - let query_decl = omnigraph_compiler::find_named_query(query_source, query_name) - .map_err(|e| OmniError::manifest(e.to_string()))?; - let type_ctx = typecheck_query(&catalog, &query_decl)?; - let ir = lower_query(&catalog, &query_decl, &type_ctx)?; + let ir = self.compile_named_query(&catalog, query_source, query_name)?; let needs_graph = ir .pipeline @@ -120,10 +117,7 @@ impl Omnigraph { // live-target query. let (snapshot, catalog) = self.capture_historical_read_view(version).await?; - let query_decl = omnigraph_compiler::find_named_query(query_source, query_name) - .map_err(|e| OmniError::manifest(e.to_string()))?; - let type_ctx = typecheck_query(&catalog, &query_decl)?; - let ir = lower_query(&catalog, &query_decl, &type_ctx)?; + let ir = self.compile_named_query(&catalog, query_source, query_name)?; let needs_graph = ir .pipeline @@ -151,6 +145,29 @@ impl Omnigraph { ) .await } + + /// Compile `query_name` from `query_source` against `catalog`, cached in + /// `ReadCaches::compiled_queries`; errors are never cached. INPUT CONTRACT: + /// a hit needs the memoized `Arc` from `build_accepted_catalog_with_schema_gate_held`. + fn compile_named_query( + &self, + catalog: &Arc, + query_source: &str, + query_name: &str, + ) -> Result> { + let cache = &self.read_caches().compiled_queries; + let key = crate::runtime_cache::CompiledQueryCache::key_for(query_source, query_name); + if let Some(ir) = cache.get(catalog, &key) { + return Ok(ir); + } + let query_decl = omnigraph_compiler::find_named_query(query_source, query_name) + .map_err(|e| OmniError::manifest(e.to_string()))?; + let type_ctx = typecheck_query(catalog, &query_decl)?; + let ir = Arc::new(lower_query(catalog, &query_decl, &type_ctx)?); + crate::instrumentation::record_query_compile(); + cache.insert(catalog, key, Arc::clone(&ir)); + Ok(ir) + } } // ─── Search mode ───────────────────────────────────────────────────────────── diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs index c3cfa98b0..3870196e4 100644 --- a/crates/omnigraph/src/instrumentation.rs +++ b/crates/omnigraph/src/instrumentation.rs @@ -261,6 +261,16 @@ pub struct QueryIoProbes { /// scan. Lets a test pin that a pruned search scan reads only its needed /// columns and names Lance's scoring column explicitly. pub node_scan_projections: Arc>>>>, + /// Misses of `ReadCaches::accepted_catalog`, counted in + /// `Omnigraph::build_accepted_catalog_with_schema_gate_held`. + pub catalog_builds: Arc, + /// Misses of `ReadCaches::compiled_queries`, counted in + /// `Omnigraph::compile_named_query`. + pub query_compiles: Arc, + /// Full-text validations entered (`TableStore::validate_full_text_demand`): + /// scans with a full-text query, a SQL-string filter, or a `contains_tokens` + /// demand in a typed filter; other scans record nothing. + pub fts_validations: Arc, } /// The two candidate plans of the rrf prefilter gate. Over FTS-index-covered @@ -759,6 +769,21 @@ pub(crate) fn record_ann_prefilter_verdict(verdict: RrfGateVerdict) { }); } +/// Probe-only: one accepted-catalog build (a memo miss). +pub(crate) fn record_catalog_build() { + let _ = current(|p| p.catalog_builds.fetch_add(1, Ordering::Relaxed)); +} + +/// Probe-only: one named-query compilation (a compiled-query cache miss). +pub(crate) fn record_query_compile() { + let _ = current(|p| p.query_compiles.fetch_add(1, Ordering::Relaxed)); +} + +/// Probe-only: one full-text validation entered by a scan. +pub(crate) fn record_fts_validation() { + let _ = current(|p| p.fts_validations.fetch_add(1, Ordering::Relaxed)); +} + /// Record `commits` walked into a change-feed poll's first-parent chain. No-op /// when no probes are installed (production). pub(crate) fn record_feed_commits_visited(commits: usize) { diff --git a/crates/omnigraph/src/runtime_cache.rs b/crates/omnigraph/src/runtime_cache.rs index 4951c9dde..21e7d9fe2 100644 --- a/crates/omnigraph/src/runtime_cache.rs +++ b/crates/omnigraph/src/runtime_cache.rs @@ -1,9 +1,12 @@ use std::collections::{HashMap, VecDeque}; use std::hash::Hash; -use std::sync::Arc; +use std::sync::{Arc, Weak}; use lance::Dataset; use lance::session::Session; +use omnigraph_compiler::catalog::Catalog; +use omnigraph_compiler::ir::QueryIR; +use sha2::{Digest, Sha256}; use tokio::sync::Mutex; use crate::db::{ResolvedTarget, Snapshot}; @@ -425,6 +428,124 @@ impl Default for TableHandleCacheInner { pub struct ReadCaches { pub session: Arc, pub handles: Arc, + /// The accepted catalog built by + /// `Omnigraph::build_accepted_catalog_with_schema_gate_held`, memoized on + /// the contract's exact bytes. + pub accepted_catalog: AcceptedCatalogMemo, + /// Named queries compiled against an accepted catalog. + pub compiled_queries: CompiledQueryCache, +} + +/// Memo of the last accepted catalog, keyed on the contract's byte-exact text +/// (`SchemaContractText`), which the caller reads on every call, so a +/// rewritten contract misses. One entry: a graph has one accepted contract at +/// a time. +#[derive(Default)] +pub struct AcceptedCatalogMemo { + inner: std::sync::Mutex>>, +} + +struct AcceptedCatalogEntry { + text: crate::db::SchemaContractText, + catalog: Arc, +} + +impl AcceptedCatalogMemo { + pub(crate) fn get(&self, text: &crate::db::SchemaContractText) -> Option> { + let entry = self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone()?; + (entry.text == *text).then(|| Arc::clone(&entry.catalog)) + } + + pub(crate) fn memoize(&self, text: crate::db::SchemaContractText, catalog: Arc) { + *self + .inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = + Some(Arc::new(AcceptedCatalogEntry { text, catalog })); + } +} + +/// Bound on distinct `(query source, query name)` pairs a handle keeps +/// compiled. Small: a deployment's named queries are a fixed set. +const COMPILED_QUERY_CACHE_CAP: usize = 256; + +/// Named queries compiled against an accepted catalog, keyed on a SHA-256 +/// digest of the source bytes plus the query name (the digest, not the source, +/// is kept). A hit also needs the entry's `Weak` to upgrade to the +/// caller's `Arc` (`Arc::ptr_eq`): a rebuilt or dropped catalog never hits. +pub struct CompiledQueryCache { + inner: std::sync::Mutex>, +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub(crate) struct CompiledQueryKey { + source_digest: [u8; 32], + name: String, +} + +struct CompiledQueryEntry { + catalog: Weak, + ir: Arc, +} + +impl Default for CompiledQueryCache { + fn default() -> Self { + Self { + inner: std::sync::Mutex::new(LruMap::new(COMPILED_QUERY_CACHE_CAP)), + } + } +} + +impl CompiledQueryCache { + pub(crate) fn key_for(source: &str, name: &str) -> CompiledQueryKey { + CompiledQueryKey { + source_digest: Sha256::digest(source.as_bytes()).into(), + name: name.to_string(), + } + } + + pub(crate) fn get( + &self, + catalog: &Arc, + key: &CompiledQueryKey, + ) -> Option> { + self.inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(key) + .filter(|entry| { + entry + .catalog + .upgrade() + .is_some_and(|held| Arc::ptr_eq(&held, catalog)) + }) + .map(|entry| Arc::clone(&entry.ir)) + } + + pub(crate) fn insert(&self, catalog: &Arc, key: CompiledQueryKey, ir: Arc) { + self.inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert( + key, + CompiledQueryEntry { + catalog: Arc::downgrade(catalog), + ir, + }, + ); + } + + #[cfg(test)] + fn len(&self) -> usize { + self.inner + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .len() + } } impl std::fmt::Debug for ReadCaches { @@ -609,4 +730,53 @@ edge Likes: Person -> Person {} // Sanity that the shared index really is the artifact's full catalog. assert!(knows.csr("Knows").is_some() && knows.csr("Likes").is_some()); } + + /// A hit needs the live `Arc` the entry was compiled under: an + /// equal catalog under another `Arc` misses; once that `Arc` is dropped the + /// entry pins nothing, never hits again, and stays counted until evicted. + #[tokio::test] + async fn compiled_query_cache_hit_requires_the_live_compiled_under_catalog() { + const SCHEMA: &str = "node Person { name: String @key }"; + let dir = tempfile::tempdir().unwrap(); + let db = crate::db::Omnigraph::init(dir.path().to_str().unwrap(), SCHEMA) + .await + .unwrap(); + let (_, built) = db.capture_current_read_view().await.unwrap(); + let compile = |catalog: &Catalog, source: &str, name: &str| { + let decl = omnigraph_compiler::find_named_query(source, name).unwrap(); + let ctx = + omnigraph_compiler::query::typecheck::typecheck_query(catalog, &decl).unwrap(); + Arc::new(omnigraph_compiler::lower_query(catalog, &decl, &ctx).unwrap()) + }; + let cache = CompiledQueryCache::default(); + let source = "query q() { match { $p: Person } return { $p.name } }"; + let key = CompiledQueryCache::key_for(source, "q"); + + let first = Arc::new((*built).clone()); + let ir = compile(&first, source, "q"); + cache.insert(&first, key.clone(), Arc::clone(&ir)); + assert!(Arc::ptr_eq(&cache.get(&first, &key).unwrap(), &ir)); + + let equal = Arc::new((*built).clone()); + assert!( + cache.get(&equal, &key).is_none(), + "an equal catalog under another Arc must miss" + ); + + let dropped = Arc::downgrade(&first); + drop(first); + assert!( + dropped.upgrade().is_none(), + "the entry must hold no Arc to its catalog" + ); + let source_r = "query r() { match { $p: Person } return { $p.name } }"; + let key_r = CompiledQueryCache::key_for(source_r, "r"); + cache.insert(&equal, key_r.clone(), compile(&equal, source_r, "r")); + assert!( + cache.get(&equal, &key).is_none(), + "an entry whose catalog was dropped never hits under a later catalog" + ); + assert!(cache.get(&equal, &key_r).is_some()); + assert_eq!(cache.len(), 2, "the dead entry stays until evicted"); + } } diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs index c5603c52a..0a43f147e 100644 --- a/crates/omnigraph/src/table_store.rs +++ b/crates/omnigraph/src/table_store.rs @@ -113,10 +113,57 @@ const ORDERED_SCAN_MAX_INPUT_BATCH_BYTES: u64 = ORDERED_SCAN_MEMORY_BYTES / 4; pub(crate) struct ScanTuning<'a> { scanner: &'a mut Scanner, full_text_columns: Option>, + /// FTS-index demand of the typed filters set through this surface, unioned + /// over every `filter_expr` call although Lance keeps only the last filter: + /// a fail-closed over-approximation. + filter_demand: FtsFilterDemand, +} + +/// Which FTS-index columns a filter expression reads through +/// `contains_tokens(column, ...)`. `all_columns` is the fail-closed verdict +/// for a call whose first argument is not a plain column. +#[derive(Debug, Default)] +pub(crate) struct FtsFilterDemand { + all_columns: bool, + columns: HashSet, +} + +impl FtsFilterDemand { + fn from_filter(filter: &Expr) -> Self { + let mut demand = Self::default(); + filter + .apply(|expr| { + if let Expr::ScalarFunction(function) = expr + && function.name() == "contains_tokens" + { + match function.args.first() { + Some(Expr::Column(column)) => { + demand.columns.insert(column.name.clone()); + } + // An unfamiliar expression must not bypass the gate. + _ => demand.all_columns = true, + } + } + Ok(TreeNodeRecursion::Continue) + }) + .expect("the visitor returns Ok on every node"); + demand + } + + fn merge(&mut self, other: Self) { + self.all_columns |= other.all_columns; + self.columns.extend(other.columns); + } + + fn is_empty(&self) -> bool { + !self.all_columns && self.columns.is_empty() + } } impl ScanTuning<'_> { pub(crate) fn filter_expr(&mut self, filter: Expr) -> &mut Self { + self.filter_demand + .merge(FtsFilterDemand::from_filter(&filter)); self.scanner.filter_expr(filter); self } @@ -1863,16 +1910,23 @@ impl TableStore { let mut tuning = ScanTuning { scanner: &mut scanner, full_text_columns: None, + filter_demand: FtsFilterDemand::default(), }; configure(&mut tuning)?; let columns = tuning.full_text_columns; - Ok((scanner, has_ordering, columns)) + let filter_demand = tuning.filter_demand; + Ok((scanner, has_ordering, columns, filter_demand)) })(); + let has_sql_filter = filter.is_some(); let dataset = ds.clone(); Box::pin(async move { - let (scanner, has_ordering, columns) = prepared?; - Self::validate_full_text_scan(&dataset, &scanner, columns).await?; + let (scanner, has_ordering, columns, filter_demand) = prepared?; + if has_sql_filter { + Self::validate_full_text_scan(&dataset, &scanner, columns).await?; + } else if columns.is_some() || !filter_demand.is_empty() { + Self::validate_full_text_demand(&dataset, columns, filter_demand).await?; + } if has_ordering { Self::execute_bounded_ordered_scan( scanner, @@ -1898,26 +1952,26 @@ impl TableStore { scanner: &Scanner, columns: Option>, ) -> Result<()> { - let mut all_columns = columns.as_ref().is_some_and(HashSet::is_empty); + let filter_demand = match scanner.get_expr_filter().map_err(OmniError::storage)? { + Some(filter) => FtsFilterDemand::from_filter(&filter), + None => FtsFilterDemand::default(), + }; + Self::validate_full_text_demand(ds, columns, filter_demand).await + } + + /// The validation proper, over an already-derived demand: `columns` is + /// the full-text query's column set (`Some(empty)` = every FTS column), + /// `filter_demand` what the filters read through `contains_tokens`. + async fn validate_full_text_demand( + ds: &Dataset, + columns: Option>, + filter_demand: FtsFilterDemand, + ) -> Result<()> { + crate::instrumentation::record_fts_validation(); + let all_columns = + columns.as_ref().is_some_and(HashSet::is_empty) || filter_demand.all_columns; let mut requested = columns.unwrap_or_default(); - if let Some(filter) = scanner.get_expr_filter().map_err(OmniError::storage)? { - filter - .apply(|expr| { - if let Expr::ScalarFunction(function) = expr - && function.name() == "contains_tokens" - { - match function.args.first() { - Some(Expr::Column(column)) => { - requested.insert(column.name.clone()); - } - // An unfamiliar expression must not bypass the gate. - _ => all_columns = true, - } - } - Ok(TreeNodeRecursion::Continue) - }) - .map_err(OmniError::datafusion)?; - } + requested.extend(filter_demand.columns); if !all_columns && requested.is_empty() { return Ok(()); } @@ -4017,7 +4071,9 @@ impl TableStore { scanner.filter(f).map_err(OmniError::storage)?; } scanner.with_fragments(combine_committed_with_staged(ds, staged)); - Self::validate_full_text_scan(ds, &scanner, None).await?; + if filter.is_some() { + Self::validate_full_text_scan(ds, &scanner, None).await?; + } let stream = scanner .try_into_stream() .await @@ -4330,11 +4386,13 @@ impl TableStore { return self.count_rows(ds, filter).await; } let mut scanner = ds.scan(); - if let Some(f) = filter { - scanner.filter(&f).map_err(OmniError::storage)?; + if let Some(f) = &filter { + scanner.filter(f).map_err(OmniError::storage)?; } scanner.with_fragments(combine_committed_with_staged(ds, staged)); - Self::validate_full_text_scan(ds, &scanner, None).await?; + if filter.is_some() { + Self::validate_full_text_scan(ds, &scanner, None).await?; + } let count = scanner.count_rows().await.map_err(OmniError::storage)?; Ok(count as usize) } @@ -6003,6 +6061,54 @@ mod tests { RecordBatch::try_new(schema, vec![col]).unwrap() } + /// `FtsFilterDemand::from_filter` names the columns a typed filter reads + /// through `contains_tokens`, at any nesting; a call whose first argument + /// is not a plain column fails closed to every column. + #[test] + fn fts_filter_demand_names_contains_tokens_columns_and_fails_closed() { + use datafusion::logical_expr::{Cast, expr::ScalarFunction}; + use datafusion::prelude::{col, lit}; + use lance_datafusion::udf::CONTAINS_TOKENS_UDF; + + let contains_tokens = |args: Vec| { + Expr::ScalarFunction(ScalarFunction::new_udf( + Arc::new(CONTAINS_TOKENS_UDF.clone()), + args, + )) + }; + let body = || HashSet::from(["body".to_string()]); + + let plain = FtsFilterDemand::from_filter(&contains_tokens(vec![col("body"), lit("x")])); + assert!(!plain.all_columns); + assert_eq!(plain.columns, body()); + + let cast = FtsFilterDemand::from_filter(&contains_tokens(vec![ + Expr::Cast(Cast::new(Box::new(col("body")), DataType::Utf8)), + lit("x"), + ])); + assert!(cast.all_columns, "a wrapped column fails closed"); + + let swapped = FtsFilterDemand::from_filter(&contains_tokens(vec![lit("x"), col("body")])); + assert!( + swapped.all_columns, + "a literal in column position fails closed" + ); + + let bare = FtsFilterDemand::from_filter(&contains_tokens(vec![])); + assert!(bare.all_columns, "a call with no arguments fails closed"); + + let nested = FtsFilterDemand::from_filter( + &(!contains_tokens(vec![col("body"), lit("x")])).and(col("a").eq(lit(1))), + ); + assert!(!nested.all_columns); + assert_eq!(nested.columns, body(), "a call under NOT and AND is found"); + + assert!( + FtsFilterDemand::from_filter(&col("a").eq(lit(1))).is_empty(), + "a filter without the call demands nothing" + ); + } + #[tokio::test] async fn ordered_scan_spills_under_explicit_memory_and_scratch_bounds() { const ROWS: usize = 20_000; diff --git a/crates/omnigraph/tests/search.rs b/crates/omnigraph/tests/search.rs index 4b2488bf7..088c2fe45 100644 --- a/crates/omnigraph/tests/search.rs +++ b/crates/omnigraph/tests/search.rs @@ -3323,6 +3323,56 @@ async fn uncertified_full_text_refuses_all_search_routes_but_not_ordinary_reads( ); } +/// A plain `nearest` over a type with full-text-indexed String columns runs no +/// full-text validation; a full-text query over the same fixture runs one, so +/// the zero is a skip and not a dead probe. +#[tokio::test] +#[serial] +async fn plain_nearest_skips_the_full_text_validation() { + use omnigraph::instrumentation::{QueryIoProbes, with_query_io_probes}; + use std::sync::atomic::Ordering; + + let dir = tempfile::tempdir().unwrap(); + let mut db = init_search_db(&dir).await; + + let probes = QueryIoProbes::default(); + let nearest = with_query_io_probes(probes.clone(), async { + query_main( + &mut db, + SEARCH_QUERIES, + "vector_search", + &vector_param("$q", &[0.1, 0.2, 0.3, 0.4]), + ) + .await + }) + .await + .unwrap(); + assert!(nearest.num_rows() > 0); + assert_eq!( + probes.fts_validations.load(Ordering::Relaxed), + 0, + "a scan with no full-text query and no contains_tokens demand skips the validation" + ); + + let probes = QueryIoProbes::default(); + let text = with_query_io_probes(probes.clone(), async { + query_main( + &mut db, + SEARCH_QUERIES, + "text_search", + ¶ms(&[("$q", "Learning")]), + ) + .await + }) + .await + .unwrap(); + assert!(text.num_rows() > 0); + assert!( + probes.fts_validations.load(Ordering::Relaxed) > 0, + "a full-text query validates its index coverage" + ); +} + // ─── RRF hybrid search ───────────────────────────────────────────────────── #[tokio::test] diff --git a/crates/omnigraph/tests/warm_read_cost.rs b/crates/omnigraph/tests/warm_read_cost.rs index 23ab75305..dbea63dd9 100644 --- a/crates/omnigraph/tests/warm_read_cost.rs +++ b/crates/omnigraph/tests/warm_read_cost.rs @@ -15,8 +15,8 @@ use omnigraph::instrumentation::{QueryIoProbes, with_query_io_probes, with_trave use helpers::cost::{cost_harness, last_manifest_reads, measure}; use helpers::{ - MUTATION_QUERIES, TEST_QUERIES, commit_many, count_rows, first_column_sorted, init_and_load, - mixed_params, mutate_branch, mutate_main, params, + MUTATION_QUERIES, TEST_QUERIES, TEST_SCHEMA, commit_many, count_rows, first_column_sorted, + init_and_load, mixed_params, mutate_branch, mutate_main, params, }; /// A warm same-branch read must do ZERO `__manifest` object-store reads and must @@ -1018,3 +1018,154 @@ async fn single_edge_query_builds_only_referenced_edge() { "a query referencing only `knows` must build only that edge, not all catalog edges" ); } + +/// Warm queries over unchanged contract bytes build no catalog and compile each +/// named query once (the key digests the source); a SchemaApply rewrites the +/// bytes, so both handles see the added type and the next query rebuilds once. +#[tokio::test] +async fn warm_query_memoizes_catalog_and_compiled_query_until_schema_apply() { + let dir = tempfile::tempdir().unwrap(); + let writer = init_and_load(&dir).await; + let uri = dir.path().to_str().unwrap(); + let reader = Omnigraph::open(uri).await.unwrap(); + let no_params = params(&[]); + let total_people = |probes: QueryIoProbes| { + with_query_io_probes( + probes, + reader.query( + ReadTarget::branch("main"), + TEST_QUERIES, + "total_people", + &no_params, + ), + ) + }; + + let probes = QueryIoProbes::default(); + total_people(probes.clone()).await.unwrap(); + total_people(probes.clone()).await.unwrap(); + assert_eq!( + probes.catalog_builds.load(Ordering::Relaxed), + 0, + "open memoizes the catalog it validated, so unchanged contract bytes build nothing" + ); + assert_eq!( + probes.query_compiles.load(Ordering::Relaxed), + 1, + "the same source and name against the same catalog compile once" + ); + + with_query_io_probes( + probes.clone(), + reader.query( + ReadTarget::branch("main"), + TEST_QUERIES, + "adults", + &no_params, + ), + ) + .await + .unwrap(); + assert_eq!(probes.catalog_builds.load(Ordering::Relaxed), 0); + assert_eq!( + probes.query_compiles.load(Ordering::Relaxed), + 2, + "a different named query from the same source is its own compile" + ); + assert_eq!( + probes.fts_validations.load(Ordering::Relaxed), + 0, + "a read with a typed filter, no full-text query and no SQL-string filter runs no full-text validation" + ); + + let limited = "query total_people() { match { $p: Person } return { $p.name } limit 1 }"; + let names = with_query_io_probes( + probes.clone(), + reader.query( + ReadTarget::branch("main"), + limited, + "total_people", + &no_params, + ), + ) + .await + .unwrap(); + assert_eq!( + probes.query_compiles.load(Ordering::Relaxed), + 3, + "the same query name in a second source is a second compile: the key digests the source" + ); + assert_eq!(names.num_rows(), 1); + assert_eq!( + names.concat_batches().unwrap().schema().field(0).name(), + "p.name", + "the second source answers with its own rows" + ); + total_people(probes.clone()).await.unwrap(); + assert_eq!( + probes.query_compiles.load(Ordering::Relaxed), + 3, + "the first source's entry survives the second's" + ); + + let projects_query = "query projects() { match { $p: Project } return { $p.name } }"; + let before = writer + .query( + ReadTarget::branch("main"), + projects_query, + "projects", + &no_params, + ) + .await + .expect_err("before apply_schema the catalog has no `Project`"); + assert!( + before.to_string().contains("unknown node type `Project`"), + "the refusal is the typecheck's unknown type, got: {before}" + ); + + let desired = format!("{TEST_SCHEMA}\nnode Project {{\n name: String @key\n}}\n"); + writer.apply_schema(&desired).await.unwrap(); + + let through_writer = writer + .query( + ReadTarget::branch("main"), + projects_query, + "projects", + &no_params, + ) + .await + .expect("the applying handle answers a query naming the added type"); + assert_eq!(through_writer.num_rows(), 0, "the new type has no rows yet"); + + let probes = QueryIoProbes::default(); + let projects = with_query_io_probes( + probes.clone(), + reader.query( + ReadTarget::branch("main"), + projects_query, + "projects", + &no_params, + ), + ) + .await + .expect("the next query after another handle's SchemaApply must see the added type"); + assert_eq!(projects.num_rows(), 0); + assert_eq!( + probes.catalog_builds.load(Ordering::Relaxed), + 1, + "rewritten contract bytes miss the memo and rebuild the catalog" + ); + assert_eq!(probes.query_compiles.load(Ordering::Relaxed), 1); + + total_people(probes.clone()).await.unwrap(); + assert_eq!( + probes.catalog_builds.load(Ordering::Relaxed), + 1, + "the rebuilt catalog is memoized in turn" + ); + assert_eq!( + probes.query_compiles.load(Ordering::Relaxed), + 2, + "a query cached under the old catalog recompiles under the rebuilt one" + ); +} diff --git a/docs/dev/execution.md b/docs/dev/execution.md index 4eed548ef..8b6694160 100644 --- a/docs/dev/execution.md +++ b/docs/dev/execution.md @@ -86,7 +86,13 @@ The executor hoists a filter only when its bindings and operation make the move semantically safe. Pushable scalar expressions use structured DataFusion/Lance expressions with case-preserved column identities. Search prefilters remain on the same scanner as the search operation. Multi-binding or unsupported -expressions stay in the engine at their lowered position. +expressions stay in the engine at their lowered position. A scan with a +SQL-string filter inspects its planned filter for full-text demand (a +`contains_tokens` call); full-text index coverage +(`FullTextIndexRebuildRequired`) is checked only when a full-text query or a +`contains_tokens` demand is present. A scan with no full-text query, no +SQL-string filter and no `contains_tokens` call in its typed filters skips +both. String-built SQL is retained only at explicitly documented compatibility seams. The camel-case regression and its two-parser boundary are recorded in diff --git a/docs/releases/v0.11.0.md b/docs/releases/v0.11.0.md index 12f08e378..449c2f10a 100644 --- a/docs/releases/v0.11.0.md +++ b/docs/releases/v0.11.0.md @@ -253,6 +253,16 @@ Notes accumulate here until the release is cut. needs plus Lance's ranking column; the vector column is read only when returned. +- **Warm queries skip repeated catalog builds, compiles, and full-text + checks.** The accepted catalog is memoized on the schema contract's exact + bytes: the bytes are still read on every query, so a schema apply by + another handle is observed on the next query and any drift still fails + validation, but an unchanged contract no longer rebuilds the catalog. + Named queries are compiled once per source digest, query name, and catalog + build (256 entries per handle; errors are never cached). A scan with no + full-text query, no SQL-string filter and no `contains_tokens` demand in its + typed filters skips the full-text index validation. + ## Cluster configuration