icydb-core 0.180.12

//! Module: db::session::sql::execute
//! Responsibility: session-owned SQL execution entrypoints that bind lowered SQL
//! commands onto structural planning, execution, and outward result shaping.
//! Does not own: SQL parsing or executor runtime internals.
//! Boundary: centralizes authority-aware SQL execution routing while keeping
//! only route and write wiring in child modules.

mod diagnostics;
mod explain;
mod global_aggregate;
mod metadata;
mod write;
mod write_returning;

#[cfg(feature = "diagnostics")]
use crate::db::executor::with_scalar_aggregate_terminal_attribution;
#[cfg(feature = "diagnostics")]
use crate::db::session::sql::{SqlExecutePhaseAttribution, measure_sql_stage};
#[cfg(feature = "diagnostics")]
use crate::error::InternalError;
use crate::{
    db::{
        DbSession, PersistedRow, QueryError,
        executor::{
            EntityAuthority, SharedPreparedExecutionPlan, StructuralGroupedProjectionResult,
        },
        query::intent::StructuralQuery,
        response::ResponseError,
        schema::AcceptedSchemaSnapshot,
        session::{
            AcceptedSchemaCatalogContext, finalize_structural_grouped_projection_result,
            sql::{
                CompiledSqlCommand, SqlCacheAttribution, SqlCompiledCommandExecutionContext,
                SqlStatementResult,
                projection::{SqlProjectionPayload, execute_sql_projection_rows_for_canister},
            },
            sql_grouped_cursor_from_bytes,
        },
        sql::lowering::LoweredSqlCommand,
        sql::parser::{SqlInsertSource, SqlInsertStatement},
    },
    error::ErrorClass,
    metrics::sink::{MetricsEvent, SqlWriteKind, record},
    traits::{CanisterKind, EntityValue},
};
use diagnostics::GroupedSqlDiagnosticsCollector;
#[cfg(feature = "diagnostics")]
use diagnostics::measure_execute_phase_with_physical_access;
#[cfg(test)]
use icydb_diagnostic_code::SqlLoweringCode;

// Collapse SQL execution failures into the stable error taxonomy used by the
// public metrics report instead of exposing internal query-error variants.
const fn sql_write_error_class(error: &QueryError) -> ErrorClass {
    match error {
        QueryError::Execute(err) => err.as_internal().class(),
        QueryError::Response(ResponseError::NotFound { .. }) => ErrorClass::NotFound,
        QueryError::Response(ResponseError::NotUnique { .. }) => ErrorClass::Conflict,
        QueryError::Validate(_)
        | QueryError::Plan(_)
        | QueryError::Intent(_)
        | QueryError::AccessRequirement(_) => ErrorClass::Unsupported,
    }
}

// Preserve the important INSERT shape distinction because `INSERT ... SELECT`
// has very different execution and debugging characteristics from VALUES.
const fn sql_insert_write_kind(statement: &SqlInsertStatement) -> SqlWriteKind {
    match &statement.source {
        SqlInsertSource::Values(_) => SqlWriteKind::Insert,
        SqlInsertSource::Select(_) => SqlWriteKind::InsertSelect,
    }
}

// Record only rejected SQL writes at the statement boundary. Successful writes
// are counted by the write executors after they know row cardinalities.
fn record_sql_write_error<E, C>(kind: SqlWriteKind, result: &Result<SqlStatementResult, QueryError>)
where
    E: PersistedRow<Canister = C> + EntityValue,
    C: CanisterKind,
{
    if let Err(error) = result {
        record(MetricsEvent::SqlWriteError {
            entity_path: E::PATH,
            kind,
            class: sql_write_error_class(error),
        });
    }
}

fn sql_statement_result_with_default_cache(
    result: Result<SqlStatementResult, QueryError>,
) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError> {
    result.map(|result| (result, SqlCacheAttribution::default()))
}

fn sql_write_statement_result_with_default_cache<E, C>(
    kind: SqlWriteKind,
    result: Result<SqlStatementResult, QueryError>,
) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
where
    E: PersistedRow<Canister = C> + EntityValue,
    C: CanisterKind,
{
    record_sql_write_error::<E, C>(kind, &result);
    sql_statement_result_with_default_cache(result)
}

impl<C: CanisterKind> DbSession<C> {
    // Convert one grouped executor result plus SQL projection labels into the
    // statement result shape shared by normal and diagnostics SQL execution.
    fn grouped_sql_statement_result_from_result(
        columns: Vec<String>,
        fixed_scales: Vec<Option<u32>>,
        result: StructuralGroupedProjectionResult,
    ) -> Result<SqlStatementResult, QueryError> {
        let row_count = result.row_count();
        let grouped = finalize_structural_grouped_projection_result(result, None)?;
        let (rows, continuation_cursor, _) = grouped.into_rows_cursor_and_trace();
        let next_cursor = sql_grouped_cursor_from_bytes(continuation_cursor);

        Ok(SqlStatementResult::Grouped {
            columns,
            fixed_scales,
            rows,
            row_count,
            next_cursor,
        })
    }

    // Execute one SQL projection from one shared lower prepared plan plus
    // one thin SQL projection contract so cached and explicit-bypass paths
    // share the same final row-materialization shell.
    fn execute_sql_projection_from_structural_prepared_plan(
        &self,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        cache_attribution: SqlCacheAttribution,
    ) -> Result<(SqlProjectionPayload, SqlCacheAttribution), QueryError> {
        let (columns, fixed_scales) = projection.into_components();
        let (rows, row_count) =
            execute_sql_projection_rows_for_canister(&self.db, self.debug, prepared_plan)
                .map_err(QueryError::execute)?;

        Ok((
            SqlProjectionPayload::new(columns, fixed_scales, rows, row_count),
            cache_attribution,
        ))
    }

    // Execute one SQL projection and immediately shape it into the public
    // statement-result envelope. Diagnostics keeps using the payload-returning
    // sibling so it can measure response finalization separately.
    fn execute_sql_statement_from_structural_prepared_plan(
        &self,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        cache_attribution: SqlCacheAttribution,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError> {
        let (payload, cache_attribution) = self
            .execute_sql_projection_from_structural_prepared_plan(
                prepared_plan,
                projection,
                cache_attribution,
            )?;

        Ok((payload.into_statement_result(), cache_attribution))
    }

    // Execute one grouped SQL statement from one shared lowered prepared plan
    // plus one thin SQL projection contract. Normal and diagnostics surfaces
    // share this plan-to-statement shell; diagnostics only swaps response
    // finalization through the optional collector.
    fn execute_grouped_sql_core<T>(
        &self,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        diagnostics: Option<GroupedSqlDiagnosticsCollector<'_>>,
        execute_grouped: impl FnOnce(
            &Self,
            SharedPreparedExecutionPlan,
        )
            -> Result<(StructuralGroupedProjectionResult, T), QueryError>,
    ) -> Result<(SqlStatementResult, T), QueryError> {
        let (columns, fixed_scales) = projection.into_components();
        let (result, extra) = execute_grouped(self, prepared_plan)?;
        let statement_result = if let Some(diagnostics) = diagnostics {
            diagnostics.finalize_grouped_sql_statement::<C>(columns, fixed_scales, result)?
        } else {
            Self::grouped_sql_statement_result_from_result(columns, fixed_scales, result)?
        };

        Ok((statement_result, extra))
    }

    // Execute one grouped SQL statement through the shared grouped SQL core
    // without diagnostics response attribution.
    fn execute_grouped_sql_statement_from_prepared_plan<T>(
        &self,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        execute_grouped: impl FnOnce(
            &Self,
            SharedPreparedExecutionPlan,
        )
            -> Result<(StructuralGroupedProjectionResult, T), QueryError>,
    ) -> Result<(SqlStatementResult, T), QueryError> {
        self.execute_grouped_sql_core(prepared_plan, projection, None, execute_grouped)
    }

    // Diagnostics-only grouped SQL execution split that keeps runtime
    // invocation and session response-envelope finalization in separate
    // counters while sharing the same grouped SQL core as normal execution.
    #[cfg(feature = "diagnostics")]
    fn execute_grouped_sql_statement_with_response_attribution<T>(
        &self,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        execute_grouped: impl FnOnce(
            &Self,
            SharedPreparedExecutionPlan,
        )
            -> Result<(StructuralGroupedProjectionResult, T), QueryError>,
    ) -> Result<(SqlStatementResult, T, u64), QueryError> {
        let mut response_finalization_local_instructions = 0;
        let diagnostics =
            GroupedSqlDiagnosticsCollector::new(&mut response_finalization_local_instructions);
        let (statement_result, extra) = self.execute_grouped_sql_core(
            prepared_plan,
            projection,
            Some(diagnostics),
            execute_grouped,
        )?;

        Ok((
            statement_result,
            extra,
            response_finalization_local_instructions,
        ))
    }

    // Execute one SQL load query from a structural lowered query through the
    // shared lower query-plan cache while bypassing only the compiled SQL
    // command cache for lowered or aggregate-only paths.
    pub(in crate::db::session::sql) fn execute_sql_projection_from_structural_query_without_sql_compiled_cache(
        &self,
        query: StructuralQuery,
        authority: EntityAuthority,
        accepted_schema: &AcceptedSchemaSnapshot,
    ) -> Result<(SqlProjectionPayload, SqlCacheAttribution), QueryError> {
        let (prepared_plan, projection, cache_attribution) = self
            .sql_select_prepared_plan_for_accepted_authority(&query, authority, accepted_schema)?;

        self.execute_sql_projection_from_structural_prepared_plan(
            prepared_plan,
            projection,
            cache_attribution,
        )
    }

    /// Execute one compiled reduced SQL statement into one unified SQL payload.
    #[cfg(test)]
    pub(in crate::db) fn execute_compiled_sql<E>(
        &self,
        compiled: &CompiledSqlCommand,
    ) -> Result<SqlStatementResult, QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let (result, _) = self.execute_compiled_sql_with_cache_attribution::<E>(compiled)?;

        Ok(result)
    }

    /// Execute one owned compiled reduced SQL statement into one unified SQL payload.
    pub(in crate::db) fn execute_compiled_sql_owned<E>(
        &self,
        compiled: CompiledSqlCommand,
    ) -> Result<SqlStatementResult, QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let (result, _) = self.execute_compiled_sql_owned_with_cache_attribution::<E>(compiled)?;

        Ok(result)
    }

    // Keep one perf-only execution entrypoint that returns cache attribution
    // together with planner/runtime instruction splits for shell-facing tools.
    #[cfg(feature = "diagnostics")]
    fn execute_non_select_compiled_sql_with_phase_attribution<E>(
        &self,
        compiled: &CompiledSqlCommand,
    ) -> Result<
        (
            SqlStatementResult,
            SqlCacheAttribution,
            SqlExecutePhaseAttribution,
        ),
        QueryError,
    >
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if matches!(compiled, CompiledSqlCommand::Select { .. }) {
            return Err(QueryError::execute(
                InternalError::query_executor_invariant(
                    "non-select SQL phase attribution helper received SELECT",
                ),
            ));
        }

        let (
            scalar_aggregate_terminal,
            ((execute_local_instructions, store_local_instructions), result),
        ) = with_scalar_aggregate_terminal_attribution(|| {
            measure_execute_phase_with_physical_access(|| {
                self.execute_compiled_sql_with_cache_attribution::<E>(compiled)
            })
        });
        let (result, cache_attribution) = result?;
        let mut phase_attribution = SqlExecutePhaseAttribution::from_execute_total_and_store_total(
            execute_local_instructions,
            store_local_instructions,
        );
        phase_attribution.scalar_aggregate_terminal = scalar_aggregate_terminal;

        Ok((result, cache_attribution, phase_attribution))
    }

    #[cfg(feature = "diagnostics")]
    fn execute_select_compiled_sql_with_phase_attribution_from_resolver<E>(
        &self,
        query: &StructuralQuery,
        resolve_plan: impl FnOnce() -> Result<
            (
                SharedPreparedExecutionPlan,
                crate::db::session::sql::SqlProjectionContract,
                SqlCacheAttribution,
            ),
            QueryError,
        >,
    ) -> Result<
        (
            SqlStatementResult,
            SqlCacheAttribution,
            SqlExecutePhaseAttribution,
        ),
        QueryError,
    >
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if query.has_grouping() {
            let (planner_local_instructions, resolved_query_plan) = measure_sql_stage(resolve_plan);
            let (prepared_plan, projection, cache_attribution) = resolved_query_plan?;

            let ((execute_local_instructions, store_local_instructions), statement_result) =
                measure_execute_phase_with_physical_access(move || {
                    self.execute_grouped_sql_statement_with_response_attribution(
                        prepared_plan,
                        projection,
                        |session, prepared_plan| {
                            let plan = prepared_plan.typed_clone::<E>();
                            session.execute_grouped_with_cursor(
                                plan,
                                None,
                                |executor, plan, cursor| {
                                    executor
                                        .execute_grouped_paged_with_cursor_traced_with_phase_attribution(
                                            plan, cursor,
                                        )
                                },
                            )
                            .map(|(result, _trace, phase_attribution)| {
                                (result, phase_attribution)
                            })
                        },
                    )
                });
            let (
                statement_result,
                grouped_phase_attribution,
                response_finalization_local_instructions,
            ) = statement_result?;

            return Ok((
                statement_result,
                cache_attribution,
                SqlExecutePhaseAttribution {
                    planner_local_instructions,
                    store_local_instructions,
                    executor_invocation_local_instructions: execute_local_instructions
                        .saturating_sub(response_finalization_local_instructions),
                    executor_local_instructions: execute_local_instructions
                        .saturating_sub(store_local_instructions)
                        .saturating_sub(response_finalization_local_instructions),
                    response_finalization_local_instructions,
                    grouped_stream_local_instructions: grouped_phase_attribution
                        .stream_local_instructions,
                    grouped_fold_local_instructions: grouped_phase_attribution
                        .fold_local_instructions,
                    grouped_finalize_local_instructions: grouped_phase_attribution
                        .finalize_local_instructions,
                    grouped_count: grouped_phase_attribution.grouped_count,
                    scalar_aggregate_terminal:
                        crate::db::executor::ScalarAggregateTerminalAttribution::none(),
                },
            ));
        }

        let (planner_local_instructions, resolved_query_plan) = measure_sql_stage(resolve_plan);
        let (prepared_plan, projection, cache_attribution) = resolved_query_plan?;

        let ((execute_local_instructions, store_local_instructions), payload) =
            measure_execute_phase_with_physical_access(move || {
                self.execute_sql_projection_from_structural_prepared_plan(
                    prepared_plan,
                    projection,
                    SqlCacheAttribution::default(),
                )
                .map(|(payload, _)| payload)
            });
        let payload = payload?;
        let (response_finalization_local_instructions, statement_result) =
            measure_sql_stage(|| Ok::<_, QueryError>(payload.into_statement_result()));
        let statement_result = statement_result?;

        Ok((
            statement_result,
            cache_attribution,
            SqlExecutePhaseAttribution {
                planner_local_instructions,
                store_local_instructions,
                executor_invocation_local_instructions: execute_local_instructions,
                executor_local_instructions: execute_local_instructions
                    .saturating_sub(store_local_instructions),
                response_finalization_local_instructions,
                grouped_stream_local_instructions: 0,
                grouped_fold_local_instructions: 0,
                grouped_finalize_local_instructions: 0,
                grouped_count: crate::db::executor::GroupedCountAttribution::none(),
                scalar_aggregate_terminal:
                    crate::db::executor::ScalarAggregateTerminalAttribution::none(),
            },
        ))
    }

    // Execute one compiled SQL command while preserving diagnostics-only
    // cache, planning, executor, and response-finalization phase attribution
    // at the session/executor handoff.
    #[cfg(feature = "diagnostics")]
    #[expect(
        dead_code,
        reason = "explicit compiled SQL diagnostics can still enter without a compile context; query endpoint diagnostics use the context-aware sibling"
    )]
    pub(in crate::db) fn execute_compiled_sql_with_phase_attribution<E>(
        &self,
        compiled: &CompiledSqlCommand,
    ) -> Result<
        (
            SqlStatementResult,
            SqlCacheAttribution,
            SqlExecutePhaseAttribution,
        ),
        QueryError,
    >
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        match compiled {
            CompiledSqlCommand::Select { query, .. } => self
                .execute_select_compiled_sql_with_phase_attribution_from_resolver::<E>(
                    query,
                    || self.sql_select_prepared_plan_for_entity::<E>(query),
                ),
            CompiledSqlCommand::Delete { .. }
            | CompiledSqlCommand::GlobalAggregate { .. }
            | CompiledSqlCommand::Explain(..)
            | CompiledSqlCommand::Insert(..)
            | CompiledSqlCommand::Update(..)
            | CompiledSqlCommand::DescribeEntity
            | CompiledSqlCommand::ShowIndexesEntity
            | CompiledSqlCommand::ShowColumnsEntity
            | CompiledSqlCommand::ShowEntities { .. }
            | CompiledSqlCommand::ShowStores { .. }
            | CompiledSqlCommand::ShowMemory => {
                self.execute_non_select_compiled_sql_with_phase_attribution::<E>(compiled)
            }
        }
    }

    #[cfg(feature = "diagnostics")]
    pub(in crate::db) fn execute_compiled_sql_context_with_phase_attribution<E>(
        &self,
        context: &SqlCompiledCommandExecutionContext,
    ) -> Result<
        (
            SqlStatementResult,
            SqlCacheAttribution,
            SqlExecutePhaseAttribution,
        ),
        QueryError,
    >
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        match context.command() {
            CompiledSqlCommand::Select { query, .. } => {
                self.execute_select_compiled_sql_with_phase_attribution_from_resolver::<E>(
                    query,
                    || {
                        if let Some((prepared_plan, projection)) =
                            context.command().cached_select_plan(
                                context.schema_fingerprint_method_version(),
                                context.schema_fingerprint(),
                            )
                        {
                            return Ok((
                                prepared_plan,
                                projection,
                                SqlCacheAttribution::shared_query_plan_cache_hit(),
                            ));
                        }

                        let authority = match context.accepted_authority() {
                            Some(authority) => authority.clone(),
                            None => context
                                .accepted_catalog()
                                .accepted_entity_authority_for::<E>()
                                .map_err(QueryError::execute)?,
                        };

                        let resolved = self
                            .sql_select_prepared_plan_for_accepted_authority_with_schema_fingerprint(
                            query,
                            authority,
                            context.accepted_schema(),
                            context.schema_fingerprint(),
                        );
                        if let Ok((prepared_plan, projection, _)) = &resolved {
                            context.command().set_cached_select_plan(
                                context.schema_fingerprint_method_version(),
                                context.schema_fingerprint(),
                                prepared_plan.clone(),
                                projection.clone(),
                            );
                        }

                        resolved
                    },
                )
            }
            CompiledSqlCommand::Explain(lowered) => {
                let (
                    scalar_aggregate_terminal,
                    ((execute_local_instructions, store_local_instructions), result),
                ) = with_scalar_aggregate_terminal_attribution(|| {
                    measure_execute_phase_with_physical_access(|| {
                        self.execute_explain_sql_with_catalog_cache_attribution::<E>(
                            lowered,
                            context.accepted_catalog(),
                            context.accepted_authority(),
                        )
                    })
                });
                let (result, cache_attribution) = result?;
                let mut phase_attribution =
                    SqlExecutePhaseAttribution::from_execute_total_and_store_total(
                        execute_local_instructions,
                        store_local_instructions,
                    );
                phase_attribution.scalar_aggregate_terminal = scalar_aggregate_terminal;

                Ok((result, cache_attribution, phase_attribution))
            }
            compiled => self.execute_non_select_compiled_sql_with_phase_attribution::<E>(compiled),
        }
    }

    fn execute_select_compiled_sql_with_cache_attribution<E>(
        &self,
        query: &StructuralQuery,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let catalog = self
            .accepted_schema_catalog_context_for_query::<E>()
            .map_err(QueryError::execute)?;
        let authority = catalog
            .accepted_entity_authority_for::<E>()
            .map_err(QueryError::execute)?;

        let (prepared_plan, projection, cache_attribution) = self
            .sql_select_prepared_plan_for_accepted_authority_with_schema_fingerprint(
                query,
                authority,
                catalog.snapshot(),
                catalog.fingerprint(),
            )?;

        self.execute_select_compiled_sql_from_prepared_plan::<E>(
            query,
            prepared_plan,
            projection,
            cache_attribution,
        )
    }

    fn execute_select_compiled_sql_with_context<E>(
        &self,
        query: &StructuralQuery,
        context: &SqlCompiledCommandExecutionContext,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if let Some((prepared_plan, projection)) = context.command().cached_select_plan(
            context.schema_fingerprint_method_version(),
            context.schema_fingerprint(),
        ) {
            return self.execute_select_compiled_sql_from_prepared_plan::<E>(
                query,
                prepared_plan,
                projection,
                SqlCacheAttribution::shared_query_plan_cache_hit(),
            );
        }

        let authority = match context.accepted_authority() {
            Some(authority) => authority.clone(),
            None => context
                .accepted_catalog()
                .accepted_entity_authority_for::<E>()
                .map_err(QueryError::execute)?,
        };

        let resolved = self
            .sql_select_prepared_plan_for_accepted_authority_with_schema_fingerprint(
                query,
                authority,
                context.accepted_schema(),
                context.schema_fingerprint(),
            );
        if let Ok((prepared_plan, projection, _)) = &resolved {
            context.command().set_cached_select_plan(
                context.schema_fingerprint_method_version(),
                context.schema_fingerprint(),
                prepared_plan.clone(),
                projection.clone(),
            );
        }
        let (prepared_plan, projection, cache_attribution) = resolved?;

        self.execute_select_compiled_sql_from_prepared_plan::<E>(
            query,
            prepared_plan,
            projection,
            cache_attribution,
        )
    }

    fn execute_select_compiled_sql_from_prepared_plan<E>(
        &self,
        query: &StructuralQuery,
        prepared_plan: SharedPreparedExecutionPlan,
        projection: crate::db::session::sql::SqlProjectionContract,
        cache_attribution: SqlCacheAttribution,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if query.has_grouping() {
            let (statement_result, ()) = self.execute_grouped_sql_statement_from_prepared_plan(
                prepared_plan,
                projection,
                |session, prepared_plan| {
                    let plan = prepared_plan.typed_clone::<E>();
                    session
                        .execute_grouped_with_trace(plan, None)
                        .map(|(result, _trace)| (result, ()))
                },
            )?;

            return Ok((statement_result, cache_attribution));
        }

        self.execute_sql_statement_from_structural_prepared_plan(
            prepared_plan,
            projection,
            cache_attribution,
        )
    }

    fn execute_explain_sql_with_cache_attribution<E>(
        &self,
        lowered: &LoweredSqlCommand,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let catalog = self
            .accepted_schema_catalog_context_for_query::<E>()
            .map_err(QueryError::execute)?;
        self.execute_explain_sql_with_catalog_cache_attribution::<E>(lowered, &catalog, None)
    }

    fn execute_explain_sql_with_catalog_cache_attribution<E>(
        &self,
        lowered: &LoweredSqlCommand,
        catalog: &AcceptedSchemaCatalogContext,
        accepted_authority: Option<&EntityAuthority>,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let authority = match accepted_authority {
            Some(authority) => authority.clone(),
            None => catalog
                .accepted_entity_authority_for::<E>()
                .map_err(QueryError::execute)?,
        };
        let schema_info = catalog.accepted_schema_info_for::<E>();

        if let Some(explain) = self.explain_lowered_sql_execution_for_authority(
            lowered,
            authority.clone(),
            catalog.snapshot(),
            &schema_info,
        )? {
            return Ok((
                SqlStatementResult::Explain(explain),
                SqlCacheAttribution::default(),
            ));
        }

        self.explain_lowered_sql_for_authority(lowered, authority, catalog.snapshot(), &schema_info)
            .map(SqlStatementResult::Explain)
            .map(|result| (result, SqlCacheAttribution::default()))
    }

    fn execute_metadata_compiled_sql_with_default_cache<E>(
        &self,
        compiled: &CompiledSqlCommand,
    ) -> Option<Result<(SqlStatementResult, SqlCacheAttribution), QueryError>>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let result = match compiled {
            CompiledSqlCommand::DescribeEntity => self.describe_entity_sql_statement_result::<E>(),
            CompiledSqlCommand::ShowIndexesEntity => self.show_indexes_sql_statement_result::<E>(),
            CompiledSqlCommand::ShowColumnsEntity => self.show_columns_sql_statement_result::<E>(),
            CompiledSqlCommand::ShowEntities { verbose } => {
                self.show_entities_sql_statement_result(*verbose)
            }
            CompiledSqlCommand::ShowStores { verbose } => {
                Ok(self.show_stores_sql_statement_result(*verbose))
            }
            CompiledSqlCommand::ShowMemory => Ok(self.show_memory_sql_statement_result()),
            CompiledSqlCommand::Select { .. }
            | CompiledSqlCommand::Delete { .. }
            | CompiledSqlCommand::GlobalAggregate { .. }
            | CompiledSqlCommand::Explain(_)
            | CompiledSqlCommand::Insert(_)
            | CompiledSqlCommand::Update(_) => return None,
        };

        Some(sql_statement_result_with_default_cache(result))
    }

    #[cfg(any(test, feature = "diagnostics"))]
    pub(in crate::db) fn execute_compiled_sql_with_cache_attribution<E>(
        &self,
        compiled: &CompiledSqlCommand,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if let Some(result) = self.execute_metadata_compiled_sql_with_default_cache::<E>(compiled) {
            return result;
        }

        match compiled {
            CompiledSqlCommand::Select { query, .. } => {
                self.execute_select_compiled_sql_with_cache_attribution::<E>(query)
            }
            CompiledSqlCommand::Delete { query, returning } => {
                let result =
                    self.execute_sql_delete_statement::<E>(query.as_ref(), returning.as_ref());
                sql_write_statement_result_with_default_cache::<E, C>(SqlWriteKind::Delete, result)
            }
            CompiledSqlCommand::GlobalAggregate { command } => {
                self.execute_global_aggregate_statement::<E>(*command.clone())
            }
            CompiledSqlCommand::Explain(lowered) => {
                self.execute_explain_sql_with_cache_attribution::<E>(lowered)
            }
            CompiledSqlCommand::Insert(statement) => {
                let result = self.execute_sql_insert_statement::<E>(statement);
                sql_write_statement_result_with_default_cache::<E, C>(
                    sql_insert_write_kind(statement),
                    result,
                )
            }
            CompiledSqlCommand::Update(statement) => {
                let result = self.execute_sql_update_statement::<E>(statement);
                sql_write_statement_result_with_default_cache::<E, C>(SqlWriteKind::Update, result)
            }
            CompiledSqlCommand::DescribeEntity
            | CompiledSqlCommand::ShowIndexesEntity
            | CompiledSqlCommand::ShowColumnsEntity
            | CompiledSqlCommand::ShowEntities { .. }
            | CompiledSqlCommand::ShowStores { .. }
            | CompiledSqlCommand::ShowMemory => unreachable!("metadata SQL handled above"),
        }
    }

    #[cfg(any(test, feature = "diagnostics"))]
    #[expect(
        dead_code,
        reason = "available for cache-attribution tests over compile contexts; normal query execution uses owned or diagnostics context entrypoints"
    )]
    pub(in crate::db) fn execute_compiled_sql_context_with_cache_attribution<E>(
        &self,
        context: &SqlCompiledCommandExecutionContext,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        match context.command() {
            CompiledSqlCommand::Select { query, .. } => {
                self.execute_select_compiled_sql_with_context::<E>(query, context)
            }
            CompiledSqlCommand::Explain(lowered) => self
                .execute_explain_sql_with_catalog_cache_attribution::<E>(
                    lowered,
                    context.accepted_catalog(),
                    context.accepted_authority(),
                ),
            compiled => self.execute_compiled_sql_with_cache_attribution::<E>(compiled),
        }
    }

    pub(in crate::db) fn execute_compiled_sql_owned_with_cache_attribution<E>(
        &self,
        compiled: CompiledSqlCommand,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        if let Some(result) = self.execute_metadata_compiled_sql_with_default_cache::<E>(&compiled)
        {
            return result;
        }

        match compiled {
            CompiledSqlCommand::Select { query, .. } => {
                self.execute_select_compiled_sql_with_cache_attribution::<E>(query.as_ref())
            }
            CompiledSqlCommand::Delete { query, returning } => {
                let result =
                    self.execute_sql_delete_statement::<E>(query.as_ref(), returning.as_ref());
                sql_write_statement_result_with_default_cache::<E, C>(SqlWriteKind::Delete, result)
            }
            CompiledSqlCommand::GlobalAggregate { command } => {
                self.execute_global_aggregate_statement::<E>(*command)
            }
            CompiledSqlCommand::Explain(lowered) => {
                self.execute_explain_sql_with_cache_attribution::<E>(&lowered)
            }
            CompiledSqlCommand::Insert(statement) => {
                let kind = sql_insert_write_kind(&statement);
                let result = self.execute_sql_insert_statement::<E>(&statement);
                sql_write_statement_result_with_default_cache::<E, C>(kind, result)
            }
            CompiledSqlCommand::Update(statement) => {
                let result = self.execute_sql_update_statement::<E>(&statement);
                sql_write_statement_result_with_default_cache::<E, C>(SqlWriteKind::Update, result)
            }
            CompiledSqlCommand::DescribeEntity
            | CompiledSqlCommand::ShowIndexesEntity
            | CompiledSqlCommand::ShowColumnsEntity
            | CompiledSqlCommand::ShowEntities { .. }
            | CompiledSqlCommand::ShowStores { .. }
            | CompiledSqlCommand::ShowMemory => unreachable!("metadata SQL handled above"),
        }
    }

    pub(in crate::db) fn execute_compiled_sql_context_owned_with_cache_attribution<E>(
        &self,
        context: SqlCompiledCommandExecutionContext,
    ) -> Result<(SqlStatementResult, SqlCacheAttribution), QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        match context.command() {
            CompiledSqlCommand::Select { query, .. } => {
                self.execute_select_compiled_sql_with_context::<E>(query, &context)
            }
            CompiledSqlCommand::Explain(lowered) => self
                .execute_explain_sql_with_catalog_cache_attribution::<E>(
                    lowered,
                    context.accepted_catalog(),
                    context.accepted_authority(),
                ),
            _ => {
                self.execute_compiled_sql_owned_with_cache_attribution::<E>(context.into_command())
            }
        }
    }

    pub(in crate::db) fn execute_compiled_sql_context_owned<E>(
        &self,
        context: SqlCompiledCommandExecutionContext,
    ) -> Result<SqlStatementResult, QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let (result, _) =
            self.execute_compiled_sql_context_owned_with_cache_attribution::<E>(context)?;

        Ok(result)
    }

    /// Compile and then execute one parsed reduced SQL statement into one
    /// unified SQL payload for session-owned tests.
    #[cfg(test)]
    pub(in crate::db) fn execute_sql_statement_inner<E>(
        &self,
        sql: &str,
    ) -> Result<SqlStatementResult, QueryError>
    where
        E: PersistedRow<Canister = C> + EntityValue,
    {
        let statement = crate::db::session::sql::parse_sql_statement(sql)?;
        let (compiled, _, _) = match statement {
            crate::db::sql::parser::SqlStatement::Insert(_)
            | crate::db::sql::parser::SqlStatement::Update(_)
            | crate::db::sql::parser::SqlStatement::Delete(_) => {
                self.compile_sql_update_with_cache_attribution::<E>(sql)?
            }
            crate::db::sql::parser::SqlStatement::Select(_)
            | crate::db::sql::parser::SqlStatement::Explain(_)
            | crate::db::sql::parser::SqlStatement::Describe(_)
            | crate::db::sql::parser::SqlStatement::ShowIndexes(_)
            | crate::db::sql::parser::SqlStatement::ShowColumns(_)
            | crate::db::sql::parser::SqlStatement::ShowEntities(_)
            | crate::db::sql::parser::SqlStatement::ShowStores(_)
            | crate::db::sql::parser::SqlStatement::ShowMemory(_) => {
                self.compile_sql_query_with_cache_attribution::<E>(sql)?
            }
            crate::db::sql::parser::SqlStatement::Ddl(_) => {
                return Err(QueryError::sql_lowering(
                    SqlLoweringCode::SqlDdlExecutionUnsupported,
                ));
            }
        };

        self.execute_compiled_sql_owned::<E>(compiled)
    }
}