icydb-core 0.257.17

//! Module: db::session::query::cache
//! Responsibility: session-owned shared query-plan cache and planner-visibility handoff.
//! Does not own: query planning semantics, execution, or cache-key fingerprint generation.
//! Boundary: resolves store visibility and memoizes prepared plans for typed and SQL callers.

mod identity;
mod template;

use crate::db::commit::CommitSchemaFingerprint;
use crate::{
    db::{
        DbSession, QueryError,
        executor::{
            EntityAuthority, SharedPreparedExecutionPlan,
            budget::{HardExecutionContext, direct_read_execution_context},
        },
        predicate::predicate_fingerprint_normalized,
        query::{
            intent::StructuralQuery,
            plan::{CardinalityTiebreakRoutePin, PreparedQueryParameterContract, VisibleIndexes},
            preparation::PreparationWork,
        },
        schema::{
            AcceptedSchemaSnapshot, PersistedIndexKeyItemSnapshot, PersistedIndexKeySnapshot,
            SchemaInfo,
        },
        session::{AcceptedSchemaCatalogContext, CacheEntryWeight, bounded_cache::BoundedCache},
    },
    error::InternalError,
    retained::RetainedBytes,
    traits::CanisterKind,
};
use icydb_diagnostic_code::{DiagnosticExecutionBudgetResource, DiagnosticExecutionLane};
use std::{cell::RefCell, collections::HashMap};

pub(in crate::db) use identity::QueryPlanVisibility;
use identity::{QueryPlanAcceptedSchema, QueryPlanCacheKey, SchemaCacheIdentity};
use template::PreparedQueryTemplate;

const SHARED_QUERY_PLAN_CACHE_MAX_ENTRIES: usize = 1024;
const SHARED_QUERY_PLAN_CACHE_MAX_RETAINED_BYTES: usize = 4 * 1024 * 1024;
const REQUEST_PLANNING_SHAPE_DOMAIN: u64 = 0x2210_0006_0000_0001;

/// Internal shared-plan cache outcome retained only for verbose `EXPLAIN`.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum QueryPlanCacheReuse {
    Hit,
    Miss,
}

impl QueryPlanCacheReuse {
    pub(in crate::db) const fn is_hit(self) -> bool {
        matches!(self, Self::Hit)
    }
}

type QueryPlanCache = BoundedCache<QueryPlanCacheKey, CachedQueryArtifact>;
type CachedPreparedPlanLookup = Option<(SharedPreparedExecutionPlan, QueryPlanCacheReuse)>;

#[derive(Clone, Debug)]
enum CachedQueryArtifact {
    PreparedPlan(SharedPreparedExecutionPlan),
    ParameterizedTemplate(PreparedQueryTemplate),
}

impl CachedQueryArtifact {
    const fn prepared_plan(&self) -> Option<&SharedPreparedExecutionPlan> {
        match self {
            Self::PreparedPlan(plan) => Some(plan),
            Self::ParameterizedTemplate(_) => None,
        }
    }

    const fn parameterized_template(&self) -> Option<&PreparedQueryTemplate> {
        match self {
            Self::PreparedPlan(_) => None,
            Self::ParameterizedTemplate(template) => Some(template),
        }
    }

    fn retained_plan(&self) -> Option<&SharedPreparedExecutionPlan> {
        match self {
            Self::PreparedPlan(plan) => Some(plan),
            Self::ParameterizedTemplate(template) => template.retained_plan(),
        }
    }
}

crate::retained::retained_fields!(CachedQueryArtifact {
    Self::PreparedPlan(plan) => [plan],
    Self::ParameterizedTemplate(template) => [template],
});

// Account both the map key and its FIFO copy. Shared subgraphs may be counted
// repeatedly, conservatively; HashMap/VecDeque spare buckets and allocator
// metadata are separate entry-count-bounded bookkeeping, not this payload cap.
fn artifact_retained_bytes(
    key: &QueryPlanCacheKey,
    artifact: &CachedQueryArtifact,
) -> Option<usize> {
    let mut bytes = RetainedBytes::new(SHARED_QUERY_PLAN_CACHE_MAX_RETAINED_BYTES);
    bytes.add(
        2 * size_of::<QueryPlanCacheKey>()
            + size_of::<CachedQueryArtifact>()
            + size_of::<CacheEntryWeight>()
            + 3 * size_of::<usize>(),
    )?;
    let before_key = bytes.total();
    bytes.visit(key)?;
    // The FIFO copy has the same retained graph: charge it without walking again.
    let key_heap_bytes = bytes.total() - before_key;
    bytes.add(key_heap_bytes)?;
    bytes.visit(artifact)?;
    Some(bytes.total())
}

thread_local! {
    // Keep one in-heap query-plan cache per store registry so fresh `DbSession`
    // facades can share prepared logical plans across update/query calls while
    // tests and multi-registry host processes remain isolated by registry
    // identity.
    static QUERY_PLAN_CACHES: RefCell<HashMap<usize, QueryPlanCache>> =
        RefCell::new(HashMap::default());
}

fn schema_info_for_plan_cache_authority(
    authority: &EntityAuthority,
    accepted_schema: &AcceptedSchemaSnapshot,
) -> Result<SchemaInfo, QueryError> {
    if let Some(schema_info) = authority.accepted_schema_info()
        && (!accepted_schema_has_expression_indexes(accepted_schema)
            || !schema_info.expression_indexes().is_empty())
    {
        return Ok(schema_info.clone());
    }

    let enum_catalog = authority
        .accepted_value_catalog_handle()
        .map_err(QueryError::execute)?
        .clone();
    Ok(SchemaInfo::from_accepted_snapshot_and_catalog(
        accepted_schema,
        enum_catalog,
        true,
    ))
}

fn accepted_schema_has_expression_indexes(accepted_schema: &AcceptedSchemaSnapshot) -> bool {
    accepted_schema
        .persisted_snapshot()
        .indexes()
        .iter()
        .any(|index| match index.key() {
            PersistedIndexKeySnapshot::FieldPath(_) => false,
            PersistedIndexKeySnapshot::Items(items) => items
                .iter()
                .any(|item| matches!(item, PersistedIndexKeyItemSnapshot::Expression(_))),
        })
}

// A compiled front-end artifact may retain an exact or policy-fallback plan,
// but an unavailable-evidence plan must keep flowing through this shared cache
// so the existing lifecycle-stamp check can observe a later Ready transition.
#[cfg(feature = "sql")]
pub(in crate::db::session) fn query_plan_requires_cardinality_lifecycle_recheck(
    prepared_plan: &SharedPreparedExecutionPlan,
) -> bool {
    prepared_plan
        .logical_plan()
        .cardinality_tiebreak()
        .unavailable_stamp()
        .is_some()
}

impl<C: CanisterKind> DbSession<C> {
    #[cfg(test)]
    pub(in crate::db::session) fn shared_query_cache_usage_for_tests(&self) -> (usize, usize) {
        self.with_query_plan_cache(|cache| {
            for (key, artifact, charged) in cache.retained_entries() {
                let key_bytes = RetainedBytes::measure(key, usize::MAX).expect("accountable key");
                let value_bytes =
                    RetainedBytes::measure(artifact, usize::MAX).expect("accountable artifact");
                // Independent composition includes the FIFO key copy and the
                // entry's reference-counted bookkeeping after lazy execution.
                let expected = 2 * key_bytes
                    + value_bytes
                    + size_of::<CacheEntryWeight>()
                    + 3 * size_of::<usize>();
                assert!(charged >= expected);
            }
            (cache.len(), cache.retained_weight())
        })
    }

    #[cfg(test)]
    pub(in crate::db::session) fn clear_shared_query_cache_for_tests(&self, limit: usize) {
        self.with_query_plan_cache(|cache| {
            *cache = QueryPlanCache::new_weighted(SHARED_QUERY_PLAN_CACHE_MAX_ENTRIES, limit);
        });
    }

    #[cfg(all(test, feature = "sql"))]
    pub(in crate::db::session) fn retry_shared_query_cache_insertion_for_tests(
        &self,
        lane: DiagnosticExecutionLane,
        before_insert: impl FnOnce(),
    ) -> Result<(), QueryError> {
        let (key, artifact) = self.with_query_plan_cache(|cache| {
            cache
                .retained_entries()
                .next()
                .map(|(key, artifact, _)| (key.clone(), artifact.clone()))
                .expect("fixture has a prepared artifact")
        });
        // Isolate the final insertion boundary from earlier planner checks.
        before_insert();
        self.insert_shared_query_artifact(lane, key, artifact)
    }

    fn cached_cardinality_tiebreak_is_current(
        &self,
        authority: &EntityAuthority,
        prepared_plan: &SharedPreparedExecutionPlan,
    ) -> Result<bool, QueryError> {
        let Some(retained_stamp) = prepared_plan
            .logical_plan()
            .cardinality_tiebreak()
            .unavailable_stamp()
        else {
            return Ok(true);
        };
        let store = self
            .db
            .recovered_store(authority.store_path())
            .map_err(QueryError::execute)?;
        let current_stamp = store.exact_user_index_prefix_evidence_lifecycle_stamp();

        Ok(current_stamp == retained_stamp)
    }

    fn charge_request_planning_resource(
        &self,
        context: HardExecutionContext,
        resource: DiagnosticExecutionBudgetResource,
    ) -> Result<(), QueryError> {
        self.db
            .request_execution_scope()
            .charge(context, resource, 1)
            .map_err(InternalError::from)
            .map_err(QueryError::execute)
    }

    fn with_query_plan_cache<R>(&self, f: impl FnOnce(&mut QueryPlanCache) -> R) -> R {
        let scope_id = self.db.cache_scope_id();

        QUERY_PLAN_CACHES.with(|caches| {
            let mut caches = caches.borrow_mut();
            let cache = caches.entry(scope_id).or_insert_with(|| {
                QueryPlanCache::new_weighted(
                    SHARED_QUERY_PLAN_CACHE_MAX_ENTRIES,
                    SHARED_QUERY_PLAN_CACHE_MAX_RETAINED_BYTES,
                )
            });

            f(cache)
        })
    }

    fn lookup_shared_query_template_for_authority(
        &self,
        _authority: &EntityAuthority,
        cache_key: &QueryPlanCacheKey,
    ) -> Option<PreparedQueryTemplate> {
        self.with_query_plan_cache(|cache| {
            cache
                .get(cache_key)
                .and_then(CachedQueryArtifact::parameterized_template)
                .cloned()
        })
    }

    fn insert_shared_query_artifact(
        &self,
        lane: DiagnosticExecutionLane,
        cache_key: QueryPlanCacheKey,
        artifact: CachedQueryArtifact,
    ) -> Result<(), QueryError> {
        let retained_plan = artifact.retained_plan().cloned();
        // Finish sizing's instruction charge before changing cache contents or
        // attaching a lazy-growth handle. Retention ineligibility is not an error.
        let weight = PreparationWork::run(self.db.request_execution_scope(), lane, |_| {
            // An attached core cannot acquire another independent charge owner.
            if retained_plan
                .as_ref()
                .is_some_and(|plan| !plan.cache_retention_available())
            {
                return Ok(None);
            }
            Ok(artifact_retained_bytes(&cache_key, &artifact))
        })?;
        let Some(weight) = weight else {
            return Ok(());
        };
        self.with_query_plan_cache(|cache| {
            let admitted_entry = cache.insert_weighted(cache_key, artifact, weight);
            if let Some(entry) = admitted_entry
                && let Some(plan) = retained_plan
            {
                plan.attach_cache_retention(entry);
            }
        });
        Ok(())
    }

    fn lookup_shared_query_plan_for_authority(
        &self,
        authority: &EntityAuthority,
        cache_key: &QueryPlanCacheKey,
    ) -> Result<CachedPreparedPlanLookup, QueryError> {
        let cached = self.with_query_plan_cache(|cache| {
            cache
                .get(cache_key)
                .and_then(CachedQueryArtifact::prepared_plan)
                .cloned()
        });
        if let Some(prepared_plan) = cached
            && self.cached_cardinality_tiebreak_is_current(authority, &prepared_plan)?
        {
            return Ok(Some((prepared_plan, QueryPlanCacheReuse::Hit)));
        }

        Ok(None)
    }

    fn resolve_shared_query_plan_for_authority(
        &self,
        authority: &EntityAuthority,
        build_cache_key: impl FnOnce(&PreparationWork<'_>) -> Result<QueryPlanCacheKey, QueryError>,
        planning_context: HardExecutionContext,
        build_prepared_plan: impl FnOnce(
            &PreparationWork<'_>,
        ) -> Result<SharedPreparedExecutionPlan, QueryError>,
    ) -> Result<(SharedPreparedExecutionPlan, QueryPlanCacheReuse), QueryError> {
        // Key construction, hashing/comparison and current lifecycle validation
        // complete before reuse or compilation, without overlapping its interval.
        let (cache_key, cached_plan) = PreparationWork::run(
            self.db.request_execution_scope(),
            planning_context.lane(),
            |work| {
                let cache_key = build_cache_key(work)?;
                let cached = self.lookup_shared_query_plan_for_authority(authority, &cache_key)?;
                Ok((cache_key, cached))
            },
        )?;
        if let Some(cached_plan) = cached_plan {
            return Ok(cached_plan);
        }

        self.charge_request_planning_resource(
            planning_context,
            DiagnosticExecutionBudgetResource::PlanCompilations,
        )?;

        // Finish the request's instruction watermark before publishing an
        // artifact. A failed final charge must leave the cache without it.
        let prepared_plan = PreparationWork::run(
            self.db.request_execution_scope(),
            planning_context.lane(),
            build_prepared_plan,
        )?;
        self.insert_shared_query_artifact(
            planning_context.lane(),
            cache_key,
            CachedQueryArtifact::PreparedPlan(prepared_plan.clone()),
        )?;

        Ok((prepared_plan, QueryPlanCacheReuse::Miss))
    }

    pub(in crate::db::session) fn visible_indexes_for_accepted_schema(
        schema_info: &SchemaInfo,
        visibility: QueryPlanVisibility,
    ) -> Result<VisibleIndexes, QueryError> {
        Ok(match visibility {
            QueryPlanVisibility::StoreReady => {
                let visible_indexes = VisibleIndexes::accepted_schema_visible(schema_info)
                    .map_err(QueryError::execute)?;
                debug_assert!(visible_indexes.accepted_field_path_contracts_are_consistent());
                debug_assert!(visible_indexes.accepted_expression_contracts_are_consistent());
                debug_assert!(visible_indexes.accepted_semantic_contracts_are_consistent());
                debug_assert_eq!(
                    visible_indexes.accepted_expression_index_count(),
                    Some(visible_indexes.accepted_expression_indexes().len()),
                );

                visible_indexes
            }
            #[cfg(feature = "sql")]
            QueryPlanVisibility::PrimaryOnly => VisibleIndexes::accepted_schema_primary_only(),
            QueryPlanVisibility::StoreNotReady => VisibleIndexes::none(),
        })
    }

    pub(in crate::db) fn query_plan_visibility_for_store_path(
        &self,
        store_path: &'static str,
    ) -> Result<QueryPlanVisibility, QueryError> {
        let store = self
            .db
            .recovered_store(store_path)
            .map_err(QueryError::execute)?;
        let visibility = if store.index_state() == crate::db::IndexState::Ready {
            QueryPlanVisibility::StoreReady
        } else {
            QueryPlanVisibility::StoreNotReady
        };

        Ok(visibility)
    }

    pub(in crate::db) fn cached_shared_query_plan_for_accepted_authority_with_schema_fingerprint(
        &self,
        authority: EntityAuthority,
        accepted_schema: &AcceptedSchemaSnapshot,
        schema_fingerprint: CommitSchemaFingerprint,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<SharedPreparedExecutionPlan, QueryError> {
        let visibility = self.query_plan_visibility_for_store_path(authority.store_path())?;
        let schema = QueryPlanAcceptedSchema::from_accepted_schema_with_fingerprint(
            accepted_schema,
            schema_fingerprint,
            authority.accepted_runtime_root_identity(),
            authority
                .accepted_schema_authority()
                .map_err(QueryError::execute)?
                .revision(),
        );
        self.cached_shared_query_plan_for_accepted_authority_with_schema_and_visibility(
            authority, schema, visibility, query, lane,
        )
        .map(|(plan, _reuse)| plan)
    }

    /// Compile one authenticated cardinality-selected continuation route without
    /// consulting current cardinality or replacing the ordinary initial-query cache entry.
    pub(in crate::db) fn shared_query_plan_for_accepted_authority_with_route_pin(
        &self,
        authority: EntityAuthority,
        accepted_schema: &AcceptedSchemaSnapshot,
        schema_fingerprint: CommitSchemaFingerprint,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
        route_pin: CardinalityTiebreakRoutePin,
    ) -> Result<Option<SharedPreparedExecutionPlan>, QueryError> {
        let visibility = self.query_plan_visibility_for_store_path(authority.store_path())?;
        let planning_context =
            direct_read_execution_context(&authority, lane, REQUEST_PLANNING_SHAPE_DOMAIN);
        self.charge_request_planning_resource(
            planning_context,
            DiagnosticExecutionBudgetResource::PlanningSteps,
        )?;
        self.charge_request_planning_resource(
            planning_context,
            DiagnosticExecutionBudgetResource::PlanCompilations,
        )?;
        PreparationWork::run(self.db.request_execution_scope(), lane, |work| {
            let schema_info = schema_info_for_plan_cache_authority(&authority, accepted_schema)?;
            let planning_state =
                query.prepare_scalar_planning_state_with_schema_info(schema_info, work)?;
            let visible_indexes = Self::visible_indexes_for_accepted_schema(
                planning_state.schema_info(),
                visibility,
            )?;
            let plan = query.build_plan_with_visible_indexes_from_scalar_planning_state(
                &visible_indexes,
                planning_state,
                work,
            )?;
            let Some(plan) = Self::apply_pinned_cardinality_tiebreak(
                &authority,
                visible_indexes.accepted_semantic_index_contracts(),
                plan,
                route_pin,
                work,
            )?
            else {
                return Ok(None);
            };
            let prepared_plan =
                SharedPreparedExecutionPlan::from_plan(authority, plan, schema_fingerprint, work)
                    .map_err(QueryError::execute)?;

            Ok(Some(prepared_plan))
        })
    }

    #[cfg(feature = "sql")]
    pub(in crate::db) fn cached_primary_only_query_plan_for_accepted_authority_with_schema_fingerprint(
        &self,
        authority: EntityAuthority,
        accepted_schema: &AcceptedSchemaSnapshot,
        schema_fingerprint: CommitSchemaFingerprint,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<SharedPreparedExecutionPlan, QueryError> {
        let visibility = match self.query_plan_visibility_for_store_path(authority.store_path())? {
            QueryPlanVisibility::StoreReady | QueryPlanVisibility::PrimaryOnly => {
                QueryPlanVisibility::PrimaryOnly
            }
            QueryPlanVisibility::StoreNotReady => QueryPlanVisibility::StoreNotReady,
        };
        let schema = QueryPlanAcceptedSchema::from_accepted_schema_with_fingerprint(
            accepted_schema,
            schema_fingerprint,
            authority.accepted_runtime_root_identity(),
            authority
                .accepted_schema_authority()
                .map_err(QueryError::execute)?
                .revision(),
        );

        self.cached_shared_query_plan_for_accepted_authority_with_schema_and_visibility(
            authority, schema, visibility, query, lane,
        )
        .map(|(plan, _reuse)| plan)
    }

    pub(in crate::db) fn cached_shared_query_plan_for_accepted_authority_with_catalog(
        &self,
        authority: EntityAuthority,
        catalog: &AcceptedSchemaCatalogContext,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<SharedPreparedExecutionPlan, QueryError> {
        self.cached_shared_query_plan_for_accepted_authority_with_catalog_and_reuse(
            authority, catalog, query, lane,
        )
        .map(|(plan, _reuse)| plan)
    }

    pub(in crate::db) fn cached_shared_query_plan_for_accepted_authority_with_catalog_and_reuse(
        &self,
        authority: EntityAuthority,
        catalog: &AcceptedSchemaCatalogContext,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<(SharedPreparedExecutionPlan, QueryPlanCacheReuse), QueryError> {
        let visibility = self.query_plan_visibility_for_store_path(authority.store_path())?;
        let schema = QueryPlanAcceptedSchema::from_catalog(catalog);

        self.cached_shared_query_plan_for_accepted_authority_with_schema_and_visibility(
            authority, schema, visibility, query, lane,
        )
    }

    fn cached_shared_query_plan_for_accepted_authority_with_schema_and_visibility(
        &self,
        authority: EntityAuthority,
        schema: QueryPlanAcceptedSchema<'_>,
        visibility: QueryPlanVisibility,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<(SharedPreparedExecutionPlan, QueryPlanCacheReuse), QueryError> {
        let planning_context =
            direct_read_execution_context(&authority, lane, REQUEST_PLANNING_SHAPE_DOMAIN);
        self.charge_request_planning_resource(
            planning_context,
            DiagnosticExecutionBudgetResource::PlanningSteps,
        )?;
        let schema_identity = schema.identity();
        if let Some(cached) = self.try_cached_filterless_query_plan_for_authority(
            &authority,
            schema_identity,
            visibility,
            query,
            lane,
        )? {
            return Ok(cached);
        }
        let schema_info =
            schema_info_for_plan_cache_authority(&authority, schema.accepted_schema())?;
        if query.trivial_scalar_load_fast_path_eligible_with_schema(&schema_info) {
            return self.cached_trivial_scalar_load_plan_for_authority(
                authority,
                schema_identity,
                schema_info,
                visibility,
                query,
                planning_context,
            );
        }

        // Filtered hits still normalize current operands and derive identity.
        // Finish their instruction interval before cache lookup; it must not
        // overlap the separate miss/rebind construction interval below.
        let (planning_state, parameter_contract, normalized_predicate_fingerprint) =
            PreparationWork::run(self.db.request_execution_scope(), lane, |work| {
                let planning_state =
                    query.prepare_scalar_planning_state_with_schema_info(schema_info, work)?;
                let parameter_contract = query
                    .filter_predicate_fully_covers_expression()
                    .then(|| planning_state.normalized_predicate())
                    .flatten()
                    .map(|predicate| {
                        PreparedQueryParameterContract::from_normalized_predicate(predicate, work)
                    })
                    .transpose()
                    .map_err(QueryError::execute)?
                    .flatten();
                let fingerprint = planning_state
                    .normalized_predicate()
                    .map(|predicate| predicate_fingerprint_normalized(predicate, work))
                    .transpose()
                    .map_err(QueryError::execute)?;

                Ok((planning_state, parameter_contract, fingerprint))
            })?;
        if let Some(parameter_contract) = parameter_contract {
            let bound_predicate_fingerprint =
                normalized_predicate_fingerprint.ok_or_else(QueryError::invariant)?;

            return self.resolve_parameterized_query_plan_for_authority(
                &authority,
                schema,
                schema_identity,
                visibility,
                query,
                planning_state,
                parameter_contract,
                bound_predicate_fingerprint,
                planning_context,
            );
        }
        let visible_indexes =
            Self::visible_indexes_for_accepted_schema(planning_state.schema_info(), visibility)?;
        self.resolve_shared_query_plan_for_authority(
            &authority,
            |work| {
                QueryPlanCacheKey::for_authority_with_normalized_predicate_fingerprint(
                    authority.clone(),
                    schema_identity,
                    visibility,
                    query,
                    normalized_predicate_fingerprint,
                    work,
                )
            },
            planning_context,
            |work| {
                let plan = query.build_plan_with_visible_indexes_from_scalar_planning_state(
                    &visible_indexes,
                    planning_state,
                    work,
                )?;
                let plan = self.apply_exact_cardinality_tiebreak(
                    &authority,
                    visible_indexes.accepted_semantic_index_contracts(),
                    plan,
                    work,
                )?;

                SharedPreparedExecutionPlan::from_plan(
                    authority.clone(),
                    plan,
                    schema.fingerprint(),
                    work,
                )
                .map_err(QueryError::execute)
            },
        )
    }

    #[expect(
        clippy::too_many_arguments,
        reason = "parameterized cache binding keeps schema, visibility, and current values explicit"
    )]
    fn resolve_parameterized_query_plan_for_authority(
        &self,
        authority: &EntityAuthority,
        schema: QueryPlanAcceptedSchema<'_>,
        schema_identity: SchemaCacheIdentity,
        visibility: QueryPlanVisibility,
        query: &StructuralQuery,
        planning_state: crate::db::query::plan::PreparedScalarPlanningState<'_>,
        parameter_contract: PreparedQueryParameterContract,
        bound_predicate_fingerprint: [u8; 32],
        planning_context: HardExecutionContext,
    ) -> Result<(SharedPreparedExecutionPlan, QueryPlanCacheReuse), QueryError> {
        let (cache_key, cached_template, reused_plan) = PreparationWork::run(
            self.db.request_execution_scope(),
            planning_context.lane(),
            |work| {
                let cache_key = QueryPlanCacheKey::for_authority_with_parameter_contract(
                    authority.clone(),
                    schema_identity,
                    visibility,
                    query,
                    parameter_contract,
                    work,
                )?;
                let template =
                    self.lookup_shared_query_template_for_authority(authority, &cache_key);
                let mut reused = template
                    .as_ref()
                    .and_then(|template| template.reused_bound_plan(bound_predicate_fingerprint));
                if let Some(plan) = &reused
                    && !self.cached_cardinality_tiebreak_is_current(authority, plan)?
                {
                    reused = None;
                }
                Ok((cache_key, template, reused))
            },
        )?;
        if let Some(prepared_plan) = reused_plan {
            return Ok((prepared_plan, QueryPlanCacheReuse::Hit));
        }
        if let Some(mut template) = cached_template {
            let prepared_plan = PreparationWork::run(
                self.db.request_execution_scope(),
                planning_context.lane(),
                |work| {
                    let bound = template.bind(query, planning_state, work)?;
                    let bound = self.apply_exact_cardinality_tiebreak(
                        authority,
                        template.candidate_indexes(),
                        bound,
                        work,
                    )?;
                    SharedPreparedExecutionPlan::from_plan(
                        authority.clone(),
                        bound,
                        schema.fingerprint(),
                        work,
                    )
                    .map_err(QueryError::execute)
                },
            )?;
            // Binding is synchronous and does not publish into the cache. Keep
            // the checked-out memo unchanged until the complete plan succeeds.
            template.remember_bound_plan(bound_predicate_fingerprint, prepared_plan.clone());
            let artifact = CachedQueryArtifact::ParameterizedTemplate(template);
            self.insert_shared_query_artifact(planning_context.lane(), cache_key, artifact)?;

            return Ok((prepared_plan, QueryPlanCacheReuse::Hit));
        }

        self.charge_request_planning_resource(
            planning_context,
            DiagnosticExecutionBudgetResource::PlanCompilations,
        )?;

        let (prepared_plan, mut template) = PreparationWork::run(
            self.db.request_execution_scope(),
            planning_context.lane(),
            |work| {
                let visible_indexes = Self::visible_indexes_for_accepted_schema(
                    planning_state.schema_info(),
                    visibility,
                )?;
                let plan = query.build_plan_with_visible_indexes_from_scalar_planning_state(
                    &visible_indexes,
                    planning_state,
                    work,
                )?;
                let plan = self.apply_exact_cardinality_tiebreak(
                    authority,
                    visible_indexes.accepted_semantic_index_contracts(),
                    plan,
                    work,
                )?;
                let template = PreparedQueryTemplate::new(
                    visible_indexes.accepted_semantic_index_contracts(),
                    work,
                )?;
                let prepared_plan = SharedPreparedExecutionPlan::from_plan(
                    authority.clone(),
                    plan,
                    schema.fingerprint(),
                    work,
                )
                .map_err(QueryError::execute)?;
                Ok((prepared_plan, template))
            },
        )?;
        template.remember_bound_plan(bound_predicate_fingerprint, prepared_plan.clone());
        let artifact = CachedQueryArtifact::ParameterizedTemplate(template);
        self.insert_shared_query_artifact(planning_context.lane(), cache_key, artifact)?;

        Ok((prepared_plan, QueryPlanCacheReuse::Miss))
    }

    fn try_cached_filterless_query_plan_for_authority(
        &self,
        authority: &EntityAuthority,
        schema_identity: SchemaCacheIdentity,
        visibility: QueryPlanVisibility,
        query: &StructuralQuery,
        lane: DiagnosticExecutionLane,
    ) -> Result<CachedPreparedPlanLookup, QueryError> {
        if query.has_scalar_filter() {
            return Ok(None);
        }

        PreparationWork::run(self.db.request_execution_scope(), lane, |work| {
            let cache_key = QueryPlanCacheKey::for_authority_with_normalized_predicate_fingerprint(
                authority.clone(),
                schema_identity,
                visibility,
                query,
                None,
                work,
            )?;
            self.lookup_shared_query_plan_for_authority(authority, &cache_key)
        })
    }

    fn cached_trivial_scalar_load_plan_for_authority(
        &self,
        authority: EntityAuthority,
        schema_identity: SchemaCacheIdentity,
        schema_info: SchemaInfo,
        visibility: QueryPlanVisibility,
        query: &StructuralQuery,
        planning_context: HardExecutionContext,
    ) -> Result<(SharedPreparedExecutionPlan, QueryPlanCacheReuse), QueryError> {
        self.resolve_shared_query_plan_for_authority(
            &authority,
            |work| {
                QueryPlanCacheKey::for_authority_with_normalized_predicate_fingerprint(
                    authority.clone(),
                    schema_identity,
                    visibility,
                    query,
                    None,
                    work,
                )
            },
            planning_context,
            |work| {
                let Some(plan) =
                    query.try_build_trivial_scalar_load_plan_with_schema_info(schema_info, work)?
                else {
                    return Err(QueryError::invariant());
                };

                SharedPreparedExecutionPlan::from_plan(
                    authority.clone(),
                    plan,
                    schema_identity.fingerprint(),
                    work,
                )
                .map_err(QueryError::execute)
            },
        )
    }
}