reddb-io-server 1.28.8

//! Canonical table query executor.
//!
//! Houses the path for plain `FROM table` queries: the
//! [`RuntimeTableExecutionContext`] struct and the dispatch
//! functions (`execute_runtime_canonical_table_query_indexed` and
//! `execute_runtime_canonical_table_node`) that walk the canonical
//! plan and produce `UnifiedRecord`s.
//!
//! Split out of `query_exec.rs` to isolate the ~400 lines of table
//! scan logic from the entry point and the expression router. Uses
//! `use super::*;` to inherit the parent executor's imports, plus
//! explicit imports from sibling helpers/indexed_scan submodules.

use super::helpers::{
    extract_all_eq_candidates, extract_bloom_key_for_pk, extract_entity_id_from_filter,
    extract_index_candidate, extract_select_column_names, extract_zone_predicates,
};
use super::indexed_scan::{
    extract_cross_index_predicates, find_range_predicate_with_sorted_index,
    try_sorted_index_filtered_by_set, try_sorted_index_lookup,
};
use super::*;
use crate::runtime::table_row_mvcc_resolver::TableRowMvccReadResolver;
use crate::storage::unified::manager::SegmentScanStats;
use reddb_rql::ast::geo_predicate::{geo_within_truth_test, GeoWithinPredicate};
use reddb_rql::sql_lowering::{
    effective_table_filter, effective_table_group_by_exprs, effective_table_having_filter,
    effective_table_projections,
};

thread_local! {
    static LAST_SEGMENT_SCAN_STATS: std::cell::Cell<SegmentScanStats> =
        const { std::cell::Cell::new(SegmentScanStats {
            segments_total: 0,
            segments_scanned: 0,
            segments_pruned: 0,
        }) };
}

pub(crate) fn take_last_segment_scan_stats() -> SegmentScanStats {
    LAST_SEGMENT_SCAN_STATS.with(|stats| stats.replace(SegmentScanStats::default()))
}

fn record_segment_scan_stats(stats: SegmentScanStats) {
    LAST_SEGMENT_SCAN_STATS.with(|slot| slot.set(stats));
}

fn filter_records_result(
    records: Vec<UnifiedRecord>,
    mut matches: impl FnMut(&UnifiedRecord) -> RedDBResult<bool>,
) -> RedDBResult<Vec<UnifiedRecord>> {
    let mut filtered = Vec::with_capacity(records.len());
    for record in records {
        if matches(&record)? {
            filtered.push(record);
        }
    }
    Ok(filtered)
}

fn runtime_filter_contains_compare_expr(filter: &Filter) -> bool {
    match filter {
        Filter::CompareExpr { .. } => true,
        Filter::And(left, right) | Filter::Or(left, right) => {
            runtime_filter_contains_compare_expr(left)
                || runtime_filter_contains_compare_expr(right)
        }
        Filter::Not(inner) => runtime_filter_contains_compare_expr(inner),
        _ => false,
    }
}

fn compiled_entity_filter_matches(
    db: &RedDB,
    compiled: &super::filter_compiled::CompiledEntityFilter,
    entity: &UnifiedEntity,
    filter: &Filter,
    table_name: &str,
    table_alias: &str,
) -> RedDBResult<bool> {
    compiled
        .try_evaluate(entity)?
        .try_resolve_with_fallback(|| {
            evaluate_entity_filter_result_with_db(Some(db), entity, filter, table_name, table_alias)
        })
}

fn optional_compiled_entity_filter_matches(
    db: &RedDB,
    compiled: Option<&super::filter_compiled::CompiledEntityFilter>,
    entity: &UnifiedEntity,
    filter: Option<&Filter>,
    table_name: &str,
    table_alias: &str,
) -> RedDBResult<bool> {
    match (compiled, filter) {
        (None, _) => Ok(true),
        (Some(compiled), Some(filter)) => {
            compiled_entity_filter_matches(db, compiled, entity, filter, table_name, table_alias)
        }
        (Some(_), None) => Ok(false),
    }
}

/// Filter borrowed index candidates before copying their projected fields.
/// Batch visitors may visit growing rows before sealed rows, so restore the
/// index's order explicitly. A finite window uses ordered single-ID visits to
/// avoid copying or evaluating candidates beyond the accepted prefix.
fn indexed_table_records(
    db: &RedDB,
    table: &str,
    entity_ids: &[EntityId],
    columns: &[String],
    limit: usize,
    mut matches: impl FnMut(&UnifiedEntity) -> RedDBResult<bool>,
) -> RedDBResult<Vec<UnifiedRecord>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let store = db.store();
    let Some(manager) = store.get_collection(table) else {
        return Ok(Vec::new());
    };
    let schema = manager.column_schema();
    let resolver = TableRowMvccReadResolver::current_statement();
    let mut records = Vec::with_capacity(entity_ids.len().min(limit));
    let mut failure = None;
    let mut consume = |position, entity: &UnifiedEntity| {
        if failure.is_some() {
            return false;
        }
        let result = (|| -> RedDBResult<bool> {
            crate::runtime::function_budget::charge(1)?;
            if resolver.resolve_read_candidate(entity).is_none()
                || !db.replica_allows_entity_at_read(table, entity)
                || !matches(entity)?
            {
                return Ok(true);
            }
            let record = match &entity.data {
                EntityData::Row(row)
                    if !columns.is_empty() && (row.named.is_some() || row.schema.is_some()) =>
                {
                    super::super::record_search::runtime_table_record_from_entity_ref_projected(
                        entity, columns,
                    )
                }
                _ => {
                    // Keep the owned materializer's model-specific behavior.
                    // Compact recovered rows also need their collection schema.
                    // At most one accepted entity is copied at a time.
                    let mut owned = entity.clone();
                    if let EntityData::Row(row) = &mut owned.data {
                        if row.schema.is_none() {
                            row.schema = schema.clone();
                        }
                    }
                    if columns.is_empty() {
                        super::super::record_search::runtime_table_record_lean_in_collection(
                            owned, table,
                        )
                    } else {
                        runtime_table_record_from_entity_projected(owned, columns)
                    }
                }
            };
            if let Some(record) = record {
                records.push((position, record));
            }
            Ok(records.len() < limit)
        })();
        match result {
            Ok(keep_scanning) => keep_scanning,
            Err(error) => {
                failure = Some(error);
                false
            }
        }
    };
    if limit < entity_ids.len() {
        for (position, id) in entity_ids.iter().enumerate() {
            let mut keep_scanning = true;
            manager.try_for_each_id(std::slice::from_ref(id), usize::MAX, |_, entity| {
                keep_scanning = consume(position, entity);
                keep_scanning
            })?;
            if !keep_scanning {
                break;
            }
        }
    } else {
        manager.try_for_each_id(entity_ids, usize::MAX, &mut consume)?;
    }
    if let Some(error) = failure {
        return Err(error);
    }
    records.sort_unstable_by_key(|(position, _)| *position);
    Ok(records.into_iter().map(|(_, record)| record).collect())
}

/// Build the JSON result from a set of entity IDs (from index lookup).
/// Scan entities sequentially but only process those in the candidate set (from hash index).
/// Faster than individual store.get() because HashMap iteration is sequential/cache-friendly.
pub(crate) struct RuntimeTableExecutionContext<'a> {
    pub(crate) query: &'a TableQuery,
    pub(crate) table_name: &'a str,
    pub(crate) table_alias: &'a str,
}

fn append_order_by_columns(cols: &mut Vec<String>, query: &TableQuery) -> bool {
    if cols.is_empty() {
        return false;
    }
    let mut appended = false;
    for clause in &query.order_by {
        if clause.expr.is_some() {
            continue;
        }
        let FieldRef::TableColumn { column, .. } = &clause.field else {
            continue;
        };
        if !cols.iter().any(|existing| existing == column) {
            cols.push(column.clone());
            appended = true;
        }
    }
    appended
}

fn project_added_order_by_columns(
    db: &RedDB,
    records: Vec<UnifiedRecord>,
    effective_projections: &[Projection],
    table_name: &str,
    table_alias: &str,
    has_added_order_by_columns: bool,
) -> RedDBResult<Vec<UnifiedRecord>> {
    if !has_added_order_by_columns {
        return Ok(records);
    }
    records
        .iter()
        .map(|record| {
            project_runtime_record_with_db(
                Some(db),
                record,
                effective_projections,
                Some(table_name),
                Some(table_alias),
                false,
                false,
            )
        })
        .collect()
}

fn finish_ordered_index_records(
    db: &RedDB,
    query: &TableQuery,
    effective_projections: &[Projection],
    mut records: Vec<UnifiedRecord>,
    table_name: &str,
    table_alias: &str,
    has_added_order_by_columns: bool,
) -> RedDBResult<Vec<UnifiedRecord>> {
    if !query.order_by.is_empty() {
        crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
        super::super::join_filter::try_sort_records_by_order_by_with_db(
            Some(db),
            &mut records,
            &query.order_by,
            Some(table_name),
            Some(table_alias),
        )?;
    }

    if let Some(offset) = query.offset {
        let offset = offset as usize;
        if offset < records.len() {
            records = records.into_iter().skip(offset).collect();
        } else {
            records.clear();
        }
    }
    if let Some(limit) = query.limit {
        records.truncate(limit as usize);
    }
    project_added_order_by_columns(
        db,
        records,
        effective_projections,
        table_name,
        table_alias,
        has_added_order_by_columns,
    )
}

fn resolve_table_row_by_logical_id(
    db: &RedDB,
    table: &str,
    logical_id: EntityId,
) -> Option<UnifiedEntity> {
    let store = db.store();
    TableRowMvccReadResolver::current_statement().resolve_logical_id(&store, table, logical_id)
}

fn projections_require_runtime_projection(projections: &[Projection]) -> bool {
    projections.iter().any(|projection| {
        matches!(
            projection,
            Projection::Function(_, _) | Projection::Expression(_, _) | Projection::Window { .. }
        ) || matches!(
            projection,
            Projection::Column(column) | Projection::Alias(column, _) if column.starts_with("LIT:")
        )
    })
}

#[derive(Clone, Copy)]
struct GeoDistancePredicate<'a> {
    column: &'a str,
    center_lat: f64,
    center_lon: f64,
    radius_km: f64,
}

fn execute_geo_candidate_scan(
    db: &RedDB,
    query: &TableQuery,
    filter: &Filter,
    effective_projections: &[Projection],
    candidate_ids: &std::collections::HashSet<u64>,
) -> RedDBResult<Vec<UnifiedRecord>> {
    let manager = db
        .store()
        .get_collection(query.table.as_str())
        .ok_or_else(|| RedDBError::NotFound(query.table.clone()))?;
    let table_name = query.table.as_str();
    let table_alias = query.alias.as_deref().unwrap_or(table_name);
    let snapshot = crate::runtime::impl_core::capture_current_snapshot();
    let hydrate_store = db.store();
    let mut records = Vec::new();
    crate::runtime::function_budget::try_scan(
        |visit| manager.try_scan_for_each(snapshot.as_ref(), usize::MAX, visit),
        |entity| {
            if !candidate_ids.contains(&entity.id.raw()) {
                return true;
            }
            if !db.replica_allows_entity_at_read(&query.table, entity) {
                return true;
            }
            let hydrated = super::super::impl_timeseries::hydrate_timeseries_entity_ref(
                hydrate_store.as_ref(),
                entity,
            );
            let Some(record) = runtime_table_record_from_entity(hydrated.into_owned()) else {
                return true;
            };
            if super::super::join_filter::evaluate_runtime_filter_with_db(
                Some(db),
                &record,
                filter,
                Some(table_name),
                Some(table_alias),
            ) {
                records.push(record);
            }
            true
        },
    )?;

    crate::runtime::window_phase::apply(
        Some(db),
        &mut records,
        effective_projections,
        Some(table_name),
        Some(table_alias),
    )?;

    let mut records = records
        .iter()
        .map(|record| {
            project_runtime_record_with_db(
                Some(db),
                record,
                effective_projections,
                Some(table_name),
                Some(table_alias),
                false,
                false,
            )
        })
        .collect::<RedDBResult<Vec<_>>>()?;

    if !query.order_by.is_empty() {
        crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
        super::super::join_filter::try_sort_records_by_order_by_with_db(
            Some(db),
            &mut records,
            &query.order_by,
            Some(table_name),
            Some(table_alias),
        )?;
    }
    if let Some(offset) = query.offset {
        let offset = offset as usize;
        if offset < records.len() {
            records = records.into_iter().skip(offset).collect();
        } else {
            records.clear();
        }
    }
    if let Some(limit) = query.limit {
        records.truncate(limit as usize);
    }

    Ok(records)
}

/// Candidate ids for any geo predicate in `filter` that an H3 index can
/// prune: `GEO_DISTANCE(col, lat, lon) < r` and
/// `GEO_WITHIN(col, POLYGON(...))`. The returned set is a *superset* of
/// the matching rows — the full WHERE expression still runs over every
/// candidate, so the index only ever removes work, never answers.
fn geo_h3_candidate_ids(
    filter: &Filter,
    table: &str,
    idx_store: &super::index_store::IndexStore,
) -> Option<std::collections::HashSet<u64>> {
    match filter {
        Filter::CompareExpr { lhs, op, rhs } => {
            if let Some(predicate) = geo_distance_predicate(lhs, *op, rhs)
                .or_else(|| geo_distance_predicate(rhs, flipped_compare_op_for_geo(*op), lhs))
            {
                return geo_distance_predicate_candidate_ids(predicate, table, idx_store);
            }
            let predicate = geo_within_truth_test(lhs, *op, rhs)
                .or_else(|| geo_within_truth_test(rhs, flipped_compare_op_for_geo(*op), lhs))?;
            geo_within_predicate_candidate_ids(&predicate, table, idx_store)
        }
        Filter::And(left, right) => {
            let left_ids = geo_h3_candidate_ids(left, table, idx_store);
            let right_ids = geo_h3_candidate_ids(right, table, idx_store);
            match (left_ids, right_ids) {
                (Some(mut left), Some(right)) => {
                    left.retain(|id| right.contains(id));
                    Some(left)
                }
                (Some(ids), None) | (None, Some(ids)) => Some(ids),
                (None, None) => None,
            }
        }
        Filter::Or(left, right) => {
            let mut left_ids = geo_h3_candidate_ids(left, table, idx_store)?;
            let right_ids = geo_h3_candidate_ids(right, table, idx_store)?;
            left_ids.extend(right_ids);
            Some(left_ids)
        }
        Filter::Not(_) => None,
        _ => None,
    }
}

fn geo_distance_predicate<'a>(
    lhs: &'a Expr,
    op: CompareOp,
    rhs: &'a Expr,
) -> Option<GeoDistancePredicate<'a>> {
    if !matches!(op, CompareOp::Lt | CompareOp::Le) {
        return None;
    }
    let radius_km = literal_f64(rhs)?;
    if radius_km.partial_cmp(&0.0) != Some(std::cmp::Ordering::Greater) {
        return None;
    }
    let Expr::FunctionCall { name, args, .. } = lhs else {
        return None;
    };
    if !(name.eq_ignore_ascii_case("GEO_DISTANCE") || name.eq_ignore_ascii_case("HAVERSINE")) {
        return None;
    }
    let [Expr::Column { field, .. }, lat, lon] = args.as_slice() else {
        return None;
    };
    let column = match field {
        FieldRef::TableColumn { column, .. } => column.as_str(),
        _ => return None,
    };
    Some(GeoDistancePredicate {
        column,
        center_lat: literal_f64(lat)?,
        center_lon: literal_f64(lon)?,
        radius_km,
    })
}

fn literal_f64(expr: &Expr) -> Option<f64> {
    match expr {
        Expr::Literal {
            value: Value::Float(value),
            ..
        } => Some(*value),
        Expr::Literal {
            value: Value::Integer(value),
            ..
        } => Some(*value as f64),
        Expr::Literal {
            value: Value::UnsignedInteger(value),
            ..
        } => Some(*value as f64),
        _ => None,
    }
}

fn flipped_compare_op_for_geo(op: CompareOp) -> CompareOp {
    match op {
        CompareOp::Eq => CompareOp::Eq,
        CompareOp::Ne => CompareOp::Ne,
        CompareOp::Lt => CompareOp::Gt,
        CompareOp::Le => CompareOp::Ge,
        CompareOp::Gt => CompareOp::Lt,
        CompareOp::Ge => CompareOp::Le,
    }
}

fn geo_distance_predicate_candidate_ids(
    predicate: GeoDistancePredicate<'_>,
    table: &str,
    idx_store: &super::index_store::IndexStore,
) -> Option<std::collections::HashSet<u64>> {
    let resolution = h3_index_resolution(table, predicate.column, idx_store)?;
    let cells = h3_cover_cells_for_geo_predicate(
        predicate.center_lat,
        predicate.center_lon,
        predicate.radius_km,
        resolution,
    );
    h3_cell_candidate_ids(&cells, table, predicate.column, idx_store)
}

/// Candidate ids for `GEO_WITHIN(col, POLYGON(...))`, taken from the same
/// H3 polyfill cover the `SEARCH SPATIAL WITHIN POLYGON` verb uses. A
/// polygon whose cover exceeds the shared cap yields `None`, sending the
/// query down the exact full scan.
fn geo_within_predicate_candidate_ids(
    predicate: &GeoWithinPredicate<'_>,
    table: &str,
    idx_store: &super::index_store::IndexStore,
) -> Option<std::collections::HashSet<u64>> {
    let resolution = h3_index_resolution(table, predicate.column, idx_store)?;
    let cells = crate::geo::h3::polygon_to_cover_cells(
        &predicate.vertices,
        resolution,
        crate::geo::h3::MAX_POLYGON_COVER_CELLS,
    )?;
    h3_cell_candidate_ids(&cells, table, predicate.column, idx_store)
}

/// The resolution of the H3 index registered on `(table, column)`, if any.
fn h3_index_resolution(
    table: &str,
    column: &str,
    idx_store: &super::index_store::IndexStore,
) -> Option<u8> {
    match idx_store.find_index_for_column(table, column)?.method {
        super::index_store::IndexMethodKind::H3 { resolution } => Some(resolution),
        _ => None,
    }
}

/// Entity ids stored under `cells` in the column's disk B-tree index.
fn h3_cell_candidate_ids(
    cells: &[u64],
    table: &str,
    column: &str,
    idx_store: &super::index_store::IndexStore,
) -> Option<std::collections::HashSet<u64>> {
    if cells.is_empty() {
        return None;
    }
    let keys: Vec<_> = cells
        .iter()
        .filter_map(|cell| reddb_types::value_to_canonical_key(&Value::UnsignedInteger(*cell)))
        .collect();
    if keys.is_empty() {
        return None;
    }
    let ids = idx_store
        .sorted
        .in_lookup_limited(table, column, &keys, usize::MAX)?;
    Some(ids.into_iter().map(|id| id.raw()).collect())
}

pub(crate) fn h3_cover_cells_for_geo_predicate(
    lat: f64,
    lon: f64,
    radius_km: f64,
    resolution: u8,
) -> Vec<u64> {
    let cell = crate::geo::h3::lat_lng_to_cell(lat, lon, resolution);
    if cell == 0 {
        return Vec::new();
    }
    let Some(k) = crate::geo::h3::radius_cover_ring(radius_km, resolution) else {
        return Vec::new();
    };
    crate::geo::h3::grid_disk(cell, k)
}

pub(crate) fn execute_runtime_canonical_table_query_indexed(
    db: &RedDB,
    query: &TableQuery,
    index_store: Option<&super::index_store::IndexStore>,
) -> RedDBResult<Vec<UnifiedRecord>> {
    let effective_projections = effective_table_projections(query);
    let effective_filter = effective_table_filter(query);
    let effective_group_by = effective_table_group_by_exprs(query);
    let effective_having = effective_table_having_filter(query);
    // Filtered/index-only shortcuts materialize source column names. An
    // alias needs the canonical projection node to rename the result.
    let requires_runtime_projection =
        projections_require_runtime_projection(&effective_projections)
            || (effective_filter.is_some()
                && effective_projections
                    .iter()
                    .any(|p| matches!(p, Projection::Alias(_, _) | Projection::Field(_, Some(_)))));
    let uses_document_projection =
        runtime_projections_use_document_path(&effective_projections, query);

    // Hypertable chunk pruning (PRD #850, Phase 0): when this table is a
    // hypertable and the WHERE clause constrains the time column, consult
    // the per-chunk time bounds. If no chunk overlaps the predicate
    // window, the scan is skipped entirely — by the pruner's soundness
    // contract there is provably no matching row, so returning an empty
    // result is exact, not a heuristic. Non-temporal predicates and
    // non-hypertable collections fall through untouched: the pruner keeps
    // every chunk conservatively, so `kept` is non-empty and the scan
    // proceeds as before.
    if let Some(spec) = db.hypertables().get(&query.table) {
        let chunks = db.hypertables().show_chunks(&query.table);
        if !chunks.is_empty() {
            let kept = crate::storage::query::planner::hypertable_pruning::prune_hypertable_chunks(
                &spec,
                &chunks,
                effective_filter.as_ref(),
            );
            if kept.is_empty() {
                return Ok(Vec::new());
            }
        }
    }

    // ── FROM SUBQUERY PATH (Fase 1.7 / W4 rebind): when the query's
    // source is a `(SELECT …) AS alias`, execute the inner query
    // recursively to get its records, then apply the outer query's
    // WHERE / ORDER BY / OFFSET / LIMIT on top of those records so
    // the user sees the canonical SQL semantics.
    //
    // Column scope: the outer sees the inner's projection aliases
    // verbatim because UnifiedRecord keys are string column names.
    // If the user writes `SELECT score FROM (SELECT a + b AS score
    // FROM t) AS s WHERE score > 10 ORDER BY score DESC LIMIT 5`,
    // the inner emits records keyed by `score` and the outer's
    // filter / sort resolve against that key directly.
    //
    // Only QueryExpr::Table nested shapes are supported here —
    // joins / unions / CTEs in FROM-subquery position error loudly.
    if let Some(reddb_rql::ast::TableSource::Subquery(inner)) = &query.source {
        match inner.as_ref() {
            reddb_rql::ast::QueryExpr::Table(inner_table) => {
                let mut records =
                    execute_runtime_canonical_table_query_indexed(db, inner_table, index_store)?;

                // Outer WHERE: re-evaluate the legacy filter walker
                // against each materialised record. The alias is the
                // outer query's alias (or the synthetic sentinel if
                // unaliased) so qualified column references resolve
                // back onto the inner projection keys.
                let outer_alias = query.alias.as_deref();
                if let Some(ref outer_filter) = effective_filter {
                    records = filter_records_result(records, |record| {
                        super::super::join_filter::evaluate_runtime_filter_result_with_db(
                            Some(db),
                            record,
                            outer_filter,
                            outer_alias,
                            outer_alias,
                        )
                    })?;
                }

                // Outer ORDER BY: sort the materialised records
                // using the same comparator as the normal table
                // path. Expression-shaped sort keys run through
                // expr_eval, bare columns through resolve_field.
                if !query.order_by.is_empty() {
                    // Issue #769 — the ORDER BY buffer is materialized in
                    // full before sorting; cap it.
                    crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
                    super::super::join_filter::try_sort_or_top_k_records_with_db(
                        Some(db),
                        &mut records,
                        &query.order_by,
                        query.offset,
                        query.limit,
                        outer_alias,
                        outer_alias,
                    )?;
                }

                // Outer OFFSET / LIMIT.
                if let Some(offset) = query.offset {
                    let offset = offset as usize;
                    if offset >= records.len() {
                        records.clear();
                    } else {
                        records.drain(..offset);
                    }
                }
                if let Some(limit) = query.limit {
                    records.truncate(limit as usize);
                }

                crate::runtime::window_phase::apply(
                    Some(db),
                    &mut records,
                    &effective_projections,
                    None,
                    outer_alias,
                )?;

                return records
                    .iter()
                    .map(|record| {
                        project_runtime_record_with_db(
                            Some(db),
                            record,
                            &effective_projections,
                            None,
                            outer_alias,
                            false,
                            false,
                        )
                    })
                    .collect();
            }
            other => {
                return Err(RedDBError::Query(format!(
                    "FROM subquery of kind {} is not yet supported — \
                     only nested SELECT lands in Fase 2 Week 4",
                    super::super::join_filter::query_expr_name(other)
                )));
            }
        }
    }

    // ── ULTRA-FAST PATH: entity_id lookup bypasses planner entirely ──
    if !requires_runtime_projection {
        if let Some(entity_id) = extract_entity_id_from_filter(&effective_filter) {
            let entity =
                resolve_table_row_by_logical_id(db, &query.table, EntityId::new(entity_id));
            if let Some(entity) = entity {
                return Ok(
                    super::super::record_search::runtime_table_record_lean_in_collection(
                        entity,
                        &query.table,
                    )
                    .into_iter()
                    .collect(),
                );
            }
            return Ok(Vec::new());
        }
    }

    let requires_mvcc_index_fallback =
        crate::runtime::impl_core::current_snapshot_requires_index_fallback();

    // ── GEO H3 CANDIDATE PATH ─────────────────────────────────────────────────
    // `WHERE GEO_DISTANCE(col, lat, lon) < r` over an H3-indexed column can
    // use the same covering-ring candidate set as SEARCH SPATIAL RADIUS, and
    // `WHERE GEO_WITHIN(col, POLYGON(...))` the same polyfill cover as SEARCH
    // SPATIAL WITHIN POLYGON. The full WHERE expression is still evaluated
    // after candidate pruning, so the index remains a pure optimization.
    if let (false, Some(idx_store), Some(ref filter)) =
        (requires_mvcc_index_fallback, index_store, &effective_filter)
    {
        if !is_universal_query_source(&query.table) {
            if let Some(candidate_ids) = geo_h3_candidate_ids(filter, &query.table, idx_store) {
                return execute_geo_candidate_scan(
                    db,
                    query,
                    filter,
                    &effective_projections,
                    &candidate_ids,
                );
            }
        }
    }

    // ── INDEX-ASSISTED PATH: sorted (BTREE) index for BETWEEN / >/>= ──
    //
    // Piggy-backs on `try_sorted_index_lookup`, which already knows how
    // to walk a `SortedIndexManager` for range predicates. Previously
    // the main execution path only looked at hash (equality) indices,
    // so `WHERE age BETWEEN 30 AND 40` always fell through to a full
    // scan even when a BTREE index on `age` existed.
    //
    // BUT: for `WHERE city='X' AND age > Y`, a bare sorted lookup on
    // `age` returns almost the entire table (e.g. `age > 18` = 99% of
    // rows). The real win is intersecting with the hash index on city
    // (5% of rows). Skip the sorted-only path when the filter has a
    // hash-indexed equality AND a sorted range predicate — the
    // cross-index bitmap block below does the right thing.
    let has_cross_index_candidate = index_store
        .zip(effective_filter.as_ref())
        .and_then(|(idx_store, filter)| {
            extract_cross_index_predicates(filter, &query.table, idx_store)
        })
        .is_some();
    if let (false, false, Some(idx_store), Some(ref filter), false, false) = (
        requires_mvcc_index_fallback,
        requires_runtime_projection,
        index_store,
        &effective_filter,
        has_cross_index_candidate,
        uses_document_projection,
    ) {
        let trace = std::env::var("REDDB_INDEX_TRACE").ok().as_deref() == Some("1");
        let sorted_lookup_limit = if query.order_by.is_empty() && query.offset.is_none() {
            query.limit.map(|l| l as usize)
        } else {
            None
        };
        let paged_result = idx_store.paged_btree_lookup(&query.table, filter)?;
        let uses_paged_range = paged_result.is_some();
        let sorted_res = match paged_result {
            Some(ids) => Some(ids),
            None => try_sorted_index_lookup(filter, &query.table, idx_store, sorted_lookup_limit)?,
        };
        if trace {
            eprintln!(
                "sorted_index_lookup table={} filter={:?} result={:?}",
                query.table,
                filter,
                sorted_res.as_ref().map(|v| v.len())
            );
        }
        if let Some(entity_ids) = sorted_res {
            // Even covered projections must fetch the candidate row so MVCC
            // can reject stale or tombstoned index IDs before materialization.
            let mut explicit_cols = extract_select_column_names(&effective_projections);
            let has_added_order_by_columns = append_order_by_columns(&mut explicit_cols, query);

            // Re-apply the full filter — when the filter is a compound AND, the
            // sorted lookup used only the range predicate to narrow candidates.
            // Residual predicates (equality, other ranges) must be checked here.
            //
            // But: when the filter is a *leaf* predicate (single Between /
            // Compare / In on a sorted-indexed column) the sorted lookup
            // already returned exactly the matching ids — every candidate
            // passes the residual check by construction. Skipping the
            // re-compile + re-evaluate shaves ~20% off range/filtered
            // scans on `SELECT * WHERE age BETWEEN X AND Y`.
            let table_name = query.table.as_str();
            let table_alias = query.alias.as_deref().unwrap_or(table_name);
            let filter_fully_covered = !uses_paged_range
                && matches!(
                    filter,
                    reddb_rql::ast::Filter::Between { .. }
                        | reddb_rql::ast::Filter::Compare { .. }
                        | reddb_rql::ast::Filter::In { .. }
                );
            let compiled_filter = if filter_fully_covered {
                None
            } else {
                let schema_arc = db
                    .store()
                    .get_collection(table_name)
                    .and_then(|m| m.column_schema());
                Some(match schema_arc.as_ref() {
                    Some(schema) => {
                        super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                            filter,
                            table_name,
                            table_alias,
                            schema.as_ref(),
                        )
                    }
                    None => super::filter_compiled::CompiledEntityFilter::compile(
                        filter,
                        table_name,
                        table_alias,
                    ),
                })
            };
            let store = db.store();
            // Use lean materialization (skip red_* system fields) when SELECT *
            // was requested.  Explicit projection columns still go through the full
            // path below so user-specified system fields (e.g. SELECT rid,
            // age FROM t) are not silently dropped.
            let lean = explicit_cols.is_empty(); // SELECT * → lean path
            let limit = if query.order_by.is_empty() && query.offset.is_none() {
                query.limit.map(|l| l as usize).unwrap_or(usize::MAX)
            } else {
                usize::MAX
            };

            // Lean/SELECT-* path uses the borrow-based
            // `SegmentManager::for_each_id` → `runtime_table_record_lean_ref`
            // combination to skip the `UnifiedEntity::clone` inside
            // `get_batch` (~20% of scan CPU on `select_range`/`filtered`).
            // Explicit projections borrow only their selected fields below.
            if lean {
                let manager = match store.get_collection(&query.table) {
                    Some(m) => m,
                    None => return Ok(Vec::new()),
                };
                let mut records: Vec<UnifiedRecord> =
                    Vec::with_capacity(entity_ids.len().min(limit));
                let table_row_resolver = TableRowMvccReadResolver::current_statement();
                crate::runtime::function_budget::charge(entity_ids.len() as u64)?;
                crate::runtime::function_budget::visit_result(
                    |visit| {
                        manager
                            .try_for_each_id(&entity_ids, usize::MAX, |_idx, entity| visit(entity))
                            .map_err(RedDBError::from)
                    },
                    |entity| {
                        if records.len() >= limit {
                            return Ok(false);
                        }
                        if table_row_resolver.resolve_read_candidate(entity).is_none() {
                            return Ok(true);
                        }
                        if !db.replica_allows_entity_at_read(&query.table, entity) {
                            return Ok(true);
                        }
                        if let Some(cf) = compiled_filter.as_ref() {
                            if !compiled_entity_filter_matches(
                                db,
                                cf,
                                entity,
                                filter,
                                table_name,
                                table_alias,
                            )? {
                                return Ok(true);
                            }
                        }
                        if let Some(record) =
                            super::super::record_search::runtime_table_record_lean_ref(entity)
                        {
                            records.push(record);
                        }
                        Ok(true)
                    },
                )?;
                return finish_ordered_index_records(
                    db,
                    query,
                    &effective_projections,
                    records,
                    table_name,
                    table_alias,
                    has_added_order_by_columns,
                );
            }

            let records = indexed_table_records(
                db,
                &query.table,
                &entity_ids,
                &explicit_cols,
                limit,
                |entity| {
                    optional_compiled_entity_filter_matches(
                        db,
                        compiled_filter.as_ref(),
                        entity,
                        Some(filter),
                        table_name,
                        table_alias,
                    )
                },
            )?;
            return finish_ordered_index_records(
                db,
                query,
                &effective_projections,
                records,
                table_name,
                table_alias,
                has_added_order_by_columns,
            );
        }
    }

    // ── CROSS-INDEX BITMAP AND: hash eq ∩ sorted range ──────────────────────────
    // Handles `WHERE city = 'X' AND age > 30` when city has a hash index and
    // age has a sorted index.
    //
    // Current single-index path: fetch ALL ~50K hash candidates → filter by age.
    // This path: iterate sorted range for age, check each ID against a HashSet
    // built from the hash candidates. Only fetch the ~1K intersection.
    //
    // Equivalent to PG's bitmap heap scan where two bitmap indexes are AND-ed
    // at word level before touching heap pages. Here we use HashSet instead of
    // actual bitmaps but the reduction in entity fetches is the same.
    if let (false, false, Some(idx_store), Some(ref filter), false) = (
        requires_mvcc_index_fallback,
        requires_runtime_projection,
        index_store,
        &effective_filter,
        uses_document_projection,
    ) {
        if let Some((eq_col, eq_bytes, range_filter)) =
            extract_cross_index_predicates(filter, &query.table, idx_store)
        {
            if let Some(idx) = idx_store.find_hash_index_for_column(&query.table, &eq_col) {
                // Keep candidate ownership through the resident intersection.
                if let Some(hash_ids) = idx_store
                    .hash
                    .lookup_union_owned(
                        &query.table,
                        &idx.hash_lookup_name(),
                        std::iter::once(eq_bytes.as_slice()),
                    )
                    .map_err(|error| RedDBError::Internal(error.to_string()))?
                {
                    if !hash_ids.is_empty() {
                        // The range index orders candidates by its own key, not
                        // by ORDER BY. Keep the full set until the SQL window is
                        // applied, including when OFFSET appears without a sort.
                        let limit = if query.order_by.is_empty() && query.offset.is_none() {
                            query.limit.map(|l| l as usize).unwrap_or(usize::MAX)
                        } else {
                            usize::MAX
                        };
                        // Build HashSet from the smaller (hash) candidate set
                        let hash_set =
                            idx_store.candidate_set(hash_ids.iter().map(|id| id.raw()))?;
                        // Stream sorted range, collect only IDs in hash_set
                        if let Some(intersection_ids) = try_sorted_index_filtered_by_set(
                            range_filter,
                            &query.table,
                            idx_store,
                            &hash_set,
                            limit,
                        )? {
                            let table_name = query.table.as_str();
                            let table_alias = query.alias.as_deref().unwrap_or(table_name);
                            let schema_arc = db
                                .store()
                                .get_collection(table_name)
                                .and_then(|m| m.column_schema());
                            let compiled_filter =
                                effective_filter.as_ref().map(|f| match schema_arc.as_ref() {
                                    Some(schema) => {
                                        super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                                            f,
                                            table_name,
                                            table_alias,
                                            schema.as_ref(),
                                        )
                                    }
                                    None => super::filter_compiled::CompiledEntityFilter::compile(
                                        f,
                                        table_name,
                                        table_alias,
                                    ),
                                });
                            let mut explicit_cols =
                                extract_select_column_names(&effective_projections);
                            let has_added_order_by_columns =
                                append_order_by_columns(&mut explicit_cols, query);

                            let records = indexed_table_records(
                                db,
                                &query.table,
                                &intersection_ids,
                                &explicit_cols,
                                limit,
                                |entity| {
                                    optional_compiled_entity_filter_matches(
                                        db,
                                        compiled_filter.as_ref(),
                                        entity,
                                        Some(filter),
                                        table_name,
                                        table_alias,
                                    )
                                },
                            )?;
                            return finish_ordered_index_records(
                                db,
                                query,
                                &effective_projections,
                                records,
                                table_name,
                                table_alias,
                                has_added_order_by_columns,
                            );
                        }
                    }
                }
            }
        }
    }

    // ── TID BITMAP PATH: AND multiple hash indexes for multi-predicate queries ──
    // `WHERE a = 1 AND b = 2 [AND range_col op val]` with hash indexes on a, b
    // and optional sorted index on range_col:
    // - Look up each equality index → admitted 64-bit candidate bitmap
    // - AND the Roaring bitmaps within each high-32 partition (smallest-first)
    // - Optionally narrow further via sorted range scan filtered by the intersection set
    // - Fetch only the surviving rows; re-apply full compiled filter for residual predicates
    // Only fires when ≥2 indexed equality columns exist in the filter.
    if let (false, false, Some(idx_store), Some(ref filter), false) = (
        requires_mvcc_index_fallback,
        requires_runtime_projection,
        index_store,
        &effective_filter,
        uses_document_projection,
    ) {
        let mut eq_candidates: Vec<(String, Vec<u8>, reddb_types::Value)> = Vec::new();
        extract_all_eq_candidates(filter, &mut eq_candidates);
        // Avoid allocating bitmap metadata for predicates without resident
        // postings. Each lookup below rechecks the backing after this hint.
        eq_candidates.retain(|(column, _, _)| {
            idx_store
                .find_hash_index_for_column(&query.table, column)
                .is_some_and(|index| {
                    !idx_store
                        .hash
                        .is_paged(&query.table, &index.hash_lookup_name())
                })
        });

        if eq_candidates.len() >= 2 {
            let metadata_bytes = eq_candidates
                .len()
                .checked_mul(std::mem::size_of::<
                    super::super::index_store::candidates::CandidateBitmap,
                >())
                .ok_or_else(|| {
                    RedDBError::Internal("index bitmap metadata size overflow".into())
                })?;
            let _bitmap_metadata = idx_store.candidate_workspace(metadata_bytes)?;
            let mut bitmaps = Vec::new();
            bitmaps
                .try_reserve_exact(eq_candidates.len())
                .map_err(|error| RedDBError::Internal(error.to_string()))?;
            for (col, val_bytes, _val) in &eq_candidates {
                if let Some(idx) = idx_store.find_hash_index_for_column(&query.table, col) {
                    if let Some(ids) = idx_store
                        .hash
                        .lookup_union_owned(
                            &query.table,
                            &idx.hash_lookup_name(),
                            std::iter::once(val_bytes.as_slice()),
                        )
                        .map_err(|error| RedDBError::Internal(error.to_string()))?
                    {
                        bitmaps.push(idx_store.candidate_bitmap(ids)?);
                    }
                }
            }

            // Only use bitmap AND when we have ≥2 bitmaps (otherwise single-index path below)
            if bitmaps.len() >= 2 {
                // Intersect all bitmaps — start from the smallest for best performance
                bitmaps.sort_unstable_by_key(|b| b.len());
                let mut intersection = bitmaps.remove(0);
                for other in &bitmaps {
                    intersection.intersect_with(other); // word-level AND — O(n/32)
                }
                if intersection.is_empty() {
                    return Ok(Vec::new());
                }
                // Convert to HashSet<u64> for compatibility with try_sorted_index_filtered_by_set.
                // The intersection is already small (post-AND), so this conversion is cheap.
                drop(bitmaps);
                drop(_bitmap_metadata);
                let intersection_set = idx_store.candidate_set(intersection.iter())?;
                drop(intersection);
                let intersection = intersection_set;

                let table_name = query.table.as_str();
                let table_alias = query.alias.as_deref().unwrap_or(table_name);
                let schema_arc = db
                    .store()
                    .get_collection(table_name)
                    .and_then(|m| m.column_schema());
                // Always re-apply the full filter — residual predicates (range, OR, etc.)
                // not captured by the eq-bitmap path must still be checked.
                let compiled_filter = match schema_arc.as_ref() {
                    Some(schema) => {
                        super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                            filter,
                            table_name,
                            table_alias,
                            schema.as_ref(),
                        )
                    }
                    None => super::filter_compiled::CompiledEntityFilter::compile(
                        filter,
                        table_name,
                        table_alias,
                    ),
                };

                let limit = if query.order_by.is_empty() && query.offset.is_none() {
                    query.limit.map(|l| l as usize).unwrap_or(usize::MAX)
                } else {
                    usize::MAX
                };
                let mut explicit_cols = extract_select_column_names(&effective_projections);
                let has_added_order_by_columns = append_order_by_columns(&mut explicit_cols, query);
                // Optional: narrow via sorted range index filtered by the eq-intersection set.
                // Handles `WHERE city='NYC' AND status='active' AND age > 30` fully in-index.
                // Uses find_range_predicate_with_sorted_index (not extract_cross_index_predicates)
                // because the eq predicates are already handled by the bitmap intersection above.
                let entity_ids = if let Some(range_filter) =
                    find_range_predicate_with_sorted_index(filter, table_name, idx_store)
                {
                    // Sorted-range scan filtered by the eq-bitmap intersection set
                    if let Some(range_ids) = try_sorted_index_filtered_by_set(
                        range_filter,
                        table_name,
                        idx_store,
                        &intersection,
                        limit,
                    )? {
                        range_ids
                    } else {
                        // Sorted range not applicable; use eq-intersection directly
                        {
                            let mut candidates = idx_store.collect_candidates(
                                intersection.iter().copied().map(EntityId::new),
                            )?;
                            candidates
                                .as_mut_slice()
                                .sort_unstable_by_key(|id| id.raw());
                            candidates
                        }
                    }
                } else {
                    // No range predicate; just sort eq-intersection for sequential access
                    {
                        let mut candidates = idx_store
                            .collect_candidates(intersection.iter().copied().map(EntityId::new))?;
                        candidates
                            .as_mut_slice()
                            .sort_unstable_by_key(|id| id.raw());
                        candidates
                    }
                };

                let records = indexed_table_records(
                    db,
                    &query.table,
                    &entity_ids,
                    &explicit_cols,
                    limit,
                    |entity| {
                        compiled_entity_filter_matches(
                            db,
                            &compiled_filter,
                            entity,
                            filter,
                            table_name,
                            table_alias,
                        )
                    },
                )?;
                return finish_ordered_index_records(
                    db,
                    query,
                    &effective_projections,
                    records,
                    table_name,
                    table_alias,
                    has_added_order_by_columns,
                );
            }
        }
    }

    // ── INDEX-ASSISTED PATH: use hash index for O(1) equality lookups ──
    if let (false, false, Some(idx_store), Some(ref filter), false) = (
        requires_mvcc_index_fallback,
        requires_runtime_projection,
        index_store,
        &effective_filter,
        uses_document_projection,
    ) {
        if let Some((column, value_bytes)) = extract_index_candidate(filter) {
            if let Some(idx) = idx_store.find_hash_index_for_column(&query.table, &column) {
                let mut entity_ids = idx_store
                    .hash_lookup(&query.table, idx.hash_lookup_name().as_ref(), &value_bytes)
                    .map_err(|err| {
                        RedDBError::Internal(format!("hash index lookup failed: {err}"))
                    })?;
                if !entity_ids.is_empty() {
                    let limit = query.limit.map(|l| l as usize).unwrap_or(usize::MAX);

                    // When the hash result is much larger than LIMIT and the filter has
                    // secondary predicates (e.g. AND(city='X', age>30)), fetching and
                    // cloning all N candidates is wasteful — a sequential scan with
                    // early termination visits ~limit/combined_selectivity entities and
                    // stops, vs N entity clones.  Fall through to the fast-scan path below.
                    //
                    // Threshold: N > limit * 20. For LIMIT=100, N must be >2000 before
                    // we skip; for LIMIT=usize::MAX (no LIMIT), we never skip.
                    //
                    // Simple equality filter (no secondary predicates): truncate candidate
                    // list to limit instead — any limit rows from the hash set are valid
                    // without fetching the rest.
                    let is_simple_eq = matches!(
                        effective_filter.as_ref(),
                        Some(Filter::Compare {
                            op: CompareOp::Eq,
                            ..
                        })
                    );
                    if is_simple_eq && limit < entity_ids.len() {
                        // No secondary predicate — any N rows are valid.
                        entity_ids.truncate(limit);
                    } else if !is_simple_eq
                        && limit < usize::MAX
                        && entity_ids.len() > limit.saturating_mul(20)
                    {
                        // Compound filter + large candidate set: let fast scan handle it.
                        // Drop entity_ids and fall through to for_each_entity_zoned below.
                    } else {
                        // Compile and re-apply the FULL filter.
                        // The hash lookup extracted only one equality predicate from the AND tree
                        // (e.g. city = 'X'); secondary predicates (e.g. age > 30) would be silently
                        // dropped without this step, producing wrong row counts.
                        //
                        // Use compile_with_schema when the collection schema is available so that
                        // user column accesses use O(1) index lookup (RowFieldFast) instead of
                        // O(n) linear search (RowField) — matters for 50K+ candidate sets.
                        let table_name = query.table.as_str();
                        let table_alias = query.alias.as_deref().unwrap_or(table_name);
                        let schema_arc = db
                            .store()
                            .get_collection(table_name)
                            .and_then(|m| m.column_schema());
                        let compiled_filter = effective_filter.as_ref().map(|f| {
                            match schema_arc
                        .as_ref()
                    {
                        Some(schema) => {
                            super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                                f,
                                table_name,
                                table_alias,
                                schema.as_ref(),
                            )
                        }
                        None => super::filter_compiled::CompiledEntityFilter::compile(
                            f,
                            table_name,
                            table_alias,
                        ),
                    }
                        });
                        let explicit_cols = extract_select_column_names(&effective_projections);
                        let records = indexed_table_records(
                            db,
                            &query.table,
                            &entity_ids,
                            &explicit_cols,
                            limit,
                            |entity| {
                                optional_compiled_entity_filter_matches(
                                    db,
                                    compiled_filter.as_ref(),
                                    entity,
                                    effective_filter.as_ref(),
                                    table_name,
                                    table_alias,
                                )
                            },
                        )?;
                        return Ok(records);
                    } // end else (hash batch path)
                }
            }
        }
    }

    // ── GLOBAL ATTRIBUTE PATH: `SELECT * WHERE passport = '...'` ──
    //
    // SELECT without FROM parses as the universal source (`table = "any"`).
    // Before falling back to the canonical universal scan, narrow the
    // collection set using declared collection contracts / column schemas when
    // the WHERE clause references plain attributes. Collections with unknown
    // shape stay in the candidate set so dynamic graph/document collections
    // can still match.
    if effective_filter.is_some()
        && is_universal_query_source(&query.table)
        && effective_group_by.is_empty()
        && effective_having.is_none()
        && query.expand.is_none()
        && !effective_projections.iter().any(|p| {
            matches!(
                p,
                Projection::Function(_, _) | Projection::Expression(_, _) | Projection::Window { .. }
            ) || matches!(p, Projection::Column(column) | Projection::Alias(column, _) if column.starts_with("LIT:"))
        })
    {
        let filter = effective_filter.as_ref().ok_or_else(|| {
            RedDBError::Internal(
                "global attribute scan selected without a WHERE clause".into(),
            )
        })?;
        if let Some(candidate_collections) =
            universal_candidate_collections_for_filter(db, query, filter)
        {
            let table_name = query.table.as_str();
            let table_alias = query.alias.as_deref().unwrap_or(table_name);
            let mut records = scan_runtime_universal_source_records_limited(
                db,
                Some(candidate_collections.as_slice()),
                None,
            )?;
            records.retain(|record| {
                super::super::join_filter::evaluate_runtime_filter_with_db(
                    Some(db),
                    record,
                    filter,
                    Some(table_name),
                    Some(table_alias),
                )
            });

            if !query.order_by.is_empty() {
                crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
                super::super::join_filter::try_sort_records_by_order_by_with_db(
                    Some(db),
                    &mut records,
                    &query.order_by,
                    Some(table_name),
                    Some(table_alias),
                )?;
            }

            if let Some(offset) = query.offset {
                let offset = offset as usize;
                if offset < records.len() {
                    records = records.into_iter().skip(offset).collect();
                } else {
                    records.clear();
                }
            }
            if let Some(limit) = query.limit {
                records.truncate(limit as usize);
            }

            if !matches!(effective_projections.as_slice(), [Projection::All]) {
                records = records
                    .iter()
                    .map(|record| {
                        project_runtime_record_with_db(
                            Some(db),
                            record,
                            &effective_projections,
                            Some(table_name),
                            Some(table_alias),
                            false,
                            false,
                        )
                    })
                    .collect::<crate::RedDBResult<Vec<_>>>()?;
            }

            return Ok(records);
        }
    }

    // ── FAST PATH: Simple filtered scan — bypass planner for basic WHERE queries ──
    // Evaluates the filter directly on raw entity data to avoid materializing
    // UnifiedRecord for every entity in the collection.
    // Excludes universal entity sources (e.g. "any") which span all collections.
    let tag_series_candidates_for_filter = db
        .collection_contract(&query.table)
        .filter(|contract| contract.declared_model == crate::catalog::CollectionModel::TimeSeries)
        .and_then(|_| {
            effective_filter.as_ref().and_then(|filter| {
                let store = db.store();
                super::super::impl_timeseries::timeseries_tag_index_series_candidates(
                    store.as_ref(),
                    &query.table,
                    filter,
                )
            })
        });
    let tag_index_can_filter = tag_series_candidates_for_filter.is_some();

    if effective_filter.is_some()
        && !effective_filter
            .as_ref()
            .is_some_and(runtime_filter_contains_compare_expr)
        && (!effective_filter
            .as_ref()
            .is_some_and(|filter| runtime_filter_uses_document_path(filter, query))
            || tag_index_can_filter)
        && effective_group_by.is_empty()
        && effective_having.is_none()
        && query.expand.is_none()
        && !requires_runtime_projection
        && !uses_document_projection
        && !is_universal_query_source(&query.table)
    {
        let manager = db
            .store()
            .get_collection(query.table.as_str())
            .ok_or_else(|| RedDBError::NotFound(query.table.clone()))?;
        let store = db.store();

        let filter = effective_filter.as_ref().ok_or_else(|| {
            RedDBError::Internal("filtered runtime scan selected without a WHERE clause".into())
        })?;
        let tag_series_candidates = tag_series_candidates_for_filter;
        if tag_series_candidates
            .as_ref()
            .is_some_and(std::collections::HashSet::is_empty)
        {
            return Ok(Vec::new());
        }
        let table_name = query.table.as_str();
        let table_alias = query.alias.as_deref().unwrap_or(table_name);
        let explicit_limit = query.limit;
        // Issue #550 — the sequential scan terminates once `records.len()
        // >= limit` so the cap must cover the rows OFFSET will skip later;
        // otherwise `WHERE … LIMIT N OFFSET M` returns `N - M` rows instead
        // of `N`. ORDER BY is applied after the scan, so when a sort key is
        // present the cap is invalidated (capping at `off + lim` would
        // surface whichever rows the storage layer happens to yield first,
        // not the sort-ordered slice). Matches the unfiltered fast path's
        // `scan_cap` shape further down in this file.
        let limit = match (query.offset, explicit_limit) {
            _ if !query.order_by.is_empty() => usize::MAX,
            (Some(off), Some(lim)) => (off as usize).saturating_add(lim as usize),
            (None, Some(lim)) => lim as usize,
            _ => usize::MAX,
        };

        // Bloom filter: extract PK key for segment pruning
        let bloom_key = extract_bloom_key_for_pk(filter);
        if let Some(ref key) = bloom_key {
            if !manager.bloom_may_contain_key(key) {
                return Ok(Vec::new());
            }
        }

        let residual_filter = if tag_series_candidates.is_some() {
            super::super::impl_timeseries::strip_indexed_timeseries_tag_filter(filter)
        } else {
            Some(filter.clone())
        };

        // Zone map: extract range/equality predicates for segment-level pruning.
        // Sealed segments whose column min/max proves no row can match are skipped.
        let mut zone_raw: Vec<(
            String,
            reddb_types::Value,
            crate::storage::unified::segment::ZoneColPredKind,
        )> = Vec::new();
        if let Some(filter) = residual_filter.as_ref() {
            extract_zone_predicates(filter, &mut zone_raw);
        }
        // Reconstruct lifetime-bound ZoneColPred refs from the owned Vec.
        let zone_preds: Vec<(&str, crate::storage::unified::segment::ZoneColPred<'_>)> = zone_raw
            .iter()
            .map(|(col, val, kind)| {
                use crate::storage::unified::segment::{ZoneColPred, ZoneColPredKind};
                let pred = match kind {
                    ZoneColPredKind::Eq => ZoneColPred::Eq(val),
                    ZoneColPredKind::Gt => ZoneColPred::Gt(val),
                    ZoneColPredKind::Gte => ZoneColPred::Gte(val),
                    ZoneColPredKind::Lt => ZoneColPred::Lt(val),
                    ZoneColPredKind::Lte => ZoneColPred::Lte(val),
                };
                (col.as_str(), pred)
            })
            .collect();

        // Extract explicit column names for projection pushdown
        let mut select_cols = extract_select_column_names(&effective_projections);
        let has_added_order_by_columns = append_order_by_columns(&mut select_cols, query);

        // Compile the filter ONCE before iterating. When the collection
        // schema is available, use compile_with_schema to pre-resolve
        // user column names to positional indices — O(1) access per field
        // per row instead of O(n schema search) for bulk-inserted entities.
        let schema_arc = manager.column_schema();
        let compiled = residual_filter
            .as_ref()
            .map(|filter| match schema_arc.as_ref() {
                Some(schema) => super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                    filter,
                    table_name,
                    table_alias,
                    schema.as_ref(),
                ),
                None => super::filter_compiled::CompiledEntityFilter::compile(
                    filter,
                    table_name,
                    table_alias,
                ),
            });
        // Pre-filter at entity level, only materialize records that pass.
        // Uses zone-map-aware iteration: sealed segments whose column zones
        // prove no row can match the predicate are skipped entirely.
        //
        // B3 optimisation: when select_cols is non-empty, use the ref-based
        // projected materialiser — avoids cloning the whole entity, only
        // clones the K selected field values instead of all N.
        //
        // Schema-index precomputation: for bulk-inserted (columnar) entities,
        // resolve projected column names → schema positions once before the
        // scan loop. Each row then does O(1) indexed access instead of
        // O(schema_len) linear search per (row, column) pair.
        let schema_col_indices: Option<Vec<(usize, usize)>> = if !select_cols.is_empty() {
            schema_arc.as_ref().map(|schema| {
                select_cols
                    .iter()
                    .enumerate()
                    .filter_map(|(ci, col)| schema.iter().position(|s| s == col).map(|si| (ci, si)))
                    .collect()
            })
        } else {
            None
        };

        // A5 — parallel scan: when there's no explicit LIMIT and the collection
        // is large enough, project borrowed entities while parallelising filter eval
        // across sealed segments using std::thread::scope. Sequential path kept
        // for LIMIT queries so the early-exit optimisation still works.
        let entity_count = manager.count();
        // Secret evaluation uses statement-local key and authorization context,
        // which storage scan workers do not inherit. Undeclared schemas may
        // contain SECRET values even when no column type advertises them.
        let filter_requires_statement_context = residual_filter.as_ref().is_some_and(|filter| {
            let Some(contract) = db.collection_contract(table_name) else {
                return true;
            };
            if contract.declared_columns.is_empty() {
                return true;
            }
            let mut sensitivity_source = UnifiedRecord::new();
            for column in &contract.declared_columns {
                if column.data_type.eq_ignore_ascii_case("secret") {
                    sensitivity_source.set(&column.name, Value::Secret(Vec::new()));
                }
            }
            crate::runtime::join_filter::filter_uses_secret(
                filter,
                &sensitivity_source,
                Some(table_name),
                Some(table_alias),
            )
        });
        let use_parallel = !crate::runtime::function_budget::active()
            && !filter_requires_statement_context
            && explicit_limit.is_none()
            && entity_count >= crate::storage::query::executors::parallel_scan::MIN_PARALLEL_ROWS;

        // SELECT * with lean materialization: skip the 6 heavy red_* system fields
        // (collection, kind, type, capabilities, sequence_id, row_id) while keeping
        // the two timestamp fields that external adapters commonly parse.
        let lean_select_star = select_cols.is_empty();

        let mut records: Vec<UnifiedRecord> = Vec::new();
        let hydrate_store = store;
        let snapshot = crate::runtime::impl_core::capture_current_snapshot();
        if use_parallel {
            let predicate_error = parking_lot::Mutex::new(None);
            let (matching, scan_stats) =
                manager.try_scan_project_zoned_with_stats(snapshot.as_ref(), &zone_preds, usize::MAX, |entity| {
                    (|| -> RedDBResult<Option<UnifiedRecord>> {
                    if let Some(candidates) = tag_series_candidates.as_ref() {
                        let EntityData::TimeSeries(point) = &entity.data else {
                            return Ok(None);
                        };
                        if !point.series_id.is_some_and(|id| candidates.contains(&id)) {
                            return Ok(None);
                        }
                        crate::runtime::observe_timeseries_tag_index_point();
                    }
                    if !db.replica_allows_entity_at_read(&query.table, entity) {
                        return Ok(None);
                    }
                    let hydrated = super::super::impl_timeseries::hydrate_timeseries_entity_ref(
                        hydrate_store.as_ref(),
                        entity,
                    );
                    if !optional_compiled_entity_filter_matches(
                        db, compiled.as_ref(), &hydrated, residual_filter.as_ref(), table_name, table_alias,
                    )? {
                        return Ok(None);
                    }
                    let record = if !select_cols.is_empty() {
                        if let Some(ref idx_map) = schema_col_indices {
                            super::super::record_search::runtime_table_record_with_col_indices(
                                &hydrated,
                                &select_cols,
                                idx_map,
                            )
                            .or_else(|| {
                                super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                    &hydrated, &select_cols,
                                )
                            })
                            .or_else(|| {
                                runtime_table_record_from_entity_projected(hydrated.as_ref().clone(), &select_cols)
                            })
                        } else {
                            super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                &hydrated,
                                &select_cols,
                            )
                            .or_else(|| {
                                runtime_table_record_from_entity_projected(
                                    hydrated.as_ref().clone(),
                                    &select_cols,
                                )
                            })
                        }
                    } else if lean_select_star {
                        super::super::record_search::runtime_table_record_lean_in_collection(
                            hydrated.as_ref().clone(),
                            &query.table,
                        )
                    } else {
                        runtime_table_record_from_entity(hydrated.as_ref().clone())
                    };
                    Ok(record)
                    })().unwrap_or_else(|error| {
                        let mut failure = predicate_error.lock();
                        if failure.is_none() {
                            *failure = Some(error);
                        }
                        None
                    })
                })?;
            if let Some(error) = predicate_error.into_inner() {
                return Err(error);
            }
            record_segment_scan_stats(scan_stats);
            records = matching;
        } else {
            let scan_stats = crate::runtime::function_budget::try_scan_result(
                |visit| {
                    manager.try_scan_for_each_zoned_with_stats(
                        snapshot.as_ref(),
                        &zone_preds,
                        usize::MAX,
                        visit,
                    )
                },
                |entity| {
                    if records.len() >= limit {
                        return Ok(false); // stop iteration
                    }
                    if !db.replica_allows_entity_at_read(&query.table, entity) {
                        return Ok(true);
                    }
                    if let Some(candidates) = tag_series_candidates.as_ref() {
                        let EntityData::TimeSeries(point) = &entity.data else {
                            return Ok(true);
                        };
                        if !point.series_id.is_some_and(|id| candidates.contains(&id)) {
                            return Ok(true);
                        }
                        crate::runtime::observe_timeseries_tag_index_point();
                    }
                    let hydrated = super::super::impl_timeseries::hydrate_timeseries_entity_ref(
                        hydrate_store.as_ref(),
                        entity,
                    );
                    if optional_compiled_entity_filter_matches(
                        db,
                        compiled.as_ref(),
                        &hydrated,
                        residual_filter.as_ref(),
                        table_name,
                        table_alias,
                    )? {
                        let record = if !select_cols.is_empty() {
                            // Fast columnar path: use pre-computed schema indices when available.
                            if let Some(ref idx_map) = schema_col_indices {
                                super::super::record_search::runtime_table_record_with_col_indices(
                                &hydrated,
                                &select_cols,
                                idx_map,
                            )
                            .or_else(|| {
                                super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                    &hydrated, &select_cols,
                                )
                            })
                            .or_else(|| {
                                runtime_table_record_from_entity_projected(hydrated.as_ref().clone(), &select_cols)
                            })
                            } else {
                                super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                &hydrated,
                                &select_cols,
                            )
                            .or_else(|| {
                                runtime_table_record_from_entity_projected(hydrated.as_ref().clone(), &select_cols)
                            })
                            }
                        } else if lean_select_star {
                            super::super::record_search::runtime_table_record_lean_in_collection(
                                hydrated.as_ref().clone(),
                                &query.table,
                            )
                        } else {
                            runtime_table_record_from_entity(hydrated.as_ref().clone())
                        };
                        if let Some(record) = record {
                            records.push(record);
                        }
                    }
                    Ok(true) // continue
                },
            )?;
            record_segment_scan_stats(scan_stats);
        }

        // Apply ORDER BY — Schwartzian transform extracts keys once (O(n))
        // instead of per-comparison (O(n log n) HashMap lookups).
        if !query.order_by.is_empty() {
            // Issue #769 — cap the materialized sort buffer.
            crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
            super::super::join_filter::try_sort_records_by_order_by_with_db(
                Some(db),
                &mut records,
                &query.order_by,
                Some(table_name),
                Some(table_alias),
            )?;
        }

        // Apply OFFSET, then LIMIT. Order matters: SQL semantics drop the
        // first `OFFSET` filtered rows, then keep the next `LIMIT`. The
        // scan above terminates at `offset + limit` records (when there is
        // no ORDER BY) so we still get the right slice; with ORDER BY the
        // scan is uncapped above and the sort runs before OFFSET/LIMIT.
        if let Some(offset) = query.offset {
            let offset = offset as usize;
            if offset < records.len() {
                records = records.into_iter().skip(offset).collect();
            } else {
                records.clear();
            }
        }
        if let Some(limit) = explicit_limit {
            records.truncate(limit as usize);
        }

        return project_added_order_by_columns(
            db,
            records,
            &effective_projections,
            table_name,
            table_alias,
            has_added_order_by_columns,
        );
    }

    // ── FAST PATH: Unfiltered scan — bypass planner for simple SELECT * ──
    // Skipped when the projection list contains scalar function calls
    // (e.g. VERIFY_PASSWORD(...), UPPER(...)), since the fast path
    // returns raw records without running project_runtime_record.
    if effective_filter.is_none()
        && effective_group_by.is_empty()
        && effective_having.is_none()
        && query.expand.is_none()
        && !requires_runtime_projection
        && !uses_document_projection
    {
        // An unordered finite window can discard its prefix during the scan.
        // Ordered queries retain their full input for the downstream sort.
        let pushed_window = query.limit.filter(|_| query.order_by.is_empty());
        let mut records = match pushed_window {
            Some(limit) => scan_runtime_table_source_records_window(
                db,
                query.table.as_str(),
                usize::try_from(query.offset.unwrap_or(0)).unwrap_or(usize::MAX),
                usize::try_from(limit).unwrap_or(usize::MAX),
            )?,
            None => scan_runtime_table_source_records_limited(db, query.table.as_str(), None)?,
        };
        let table_name = query.table.as_str();
        let table_alias = query.alias.as_deref().unwrap_or(table_name);

        if !query.order_by.is_empty() {
            // Issue #769 — cap the materialized sort buffer.
            crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
            super::super::join_filter::try_sort_or_top_k_records_with_db(
                Some(db),
                &mut records,
                &query.order_by,
                query.offset,
                query.limit,
                Some(table_name),
                Some(table_alias),
            )?;
        }

        if pushed_window.is_none() {
            if let Some(offset) = query.offset {
                let offset = offset as usize;
                if offset < records.len() {
                    records = records.into_iter().skip(offset).collect();
                } else {
                    records.clear();
                }
            }

            if let Some(limit) = query.limit {
                records.truncate(limit as usize);
            }
        }

        // Explicit column names and aliases must be projected on the fast
        // scan too. Returning raw records silently turns SELECT col AS alias
        // into SELECT *, including when the name is quoted.
        // SESSIONIZE consumes its key/time inputs in the outer executor.
        if query.sessionize.is_none()
            && !matches!(effective_projections.as_slice(), [Projection::All])
        {
            crate::runtime::retention_filter::apply(
                &mut records,
                db.collection_contract(query.table.as_str()).as_ref(),
            );
            return records
                .iter()
                .map(|record| {
                    let mut projected = project_runtime_record_with_db(
                        Some(db),
                        record,
                        &effective_projections,
                        Some(table_name),
                        Some(table_alias),
                        false,
                        false,
                    )?;
                    // Keep the system envelope for wire metadata and stream
                    // resume; result.columns still describes only the SELECT.
                    for key in [
                        "rid",
                        "collection",
                        "kind",
                        "tenant",
                        "created_at",
                        "updated_at",
                    ] {
                        if projected.get(key).is_none() {
                            if let Some(value) = record.get(key) {
                                projected.set(key, value.clone());
                            }
                        }
                    }
                    Ok(projected)
                })
                .collect();
        }
        return Ok(records);
    }

    let plan = CanonicalPlanner::new(db).build(&QueryExpr::Table(query.clone()));
    let table_name = query.table.as_str();
    let table_alias = query.alias.as_deref().unwrap_or(table_name);
    let context = RuntimeTableExecutionContext {
        query,
        table_name,
        table_alias,
    };
    execute_runtime_canonical_table_node(db, &plan.root, &context)
}

/// Materialize computed SELECT-alias values referenced by ORDER BY onto the
/// records before sorting (#1982). The planner places the sort node *below*
/// the projection node, so a computed alias such as `GEO_DISTANCE(...) AS dist`
/// is not yet a column when the sort runs — resolving the bare name yields
/// `None` for every row, the comparator treats all rows as equal, and the
/// stable sort leaves them in nondeterministic scan order (ORDER BY silently
/// no-ops). We compute the alias with the same `project_runtime_record_with_db`
/// the projection node uses — which resolves document-body fields correctly on
/// the lean records — and inject it as a virtual column; the sort then finds
/// it, and the projection node re-projects it idempotently. Re-evaluating the
/// expression directly at sort time does NOT work: the lean records do not
/// expose document-body fields to the bare expression evaluator.
fn materialize_order_by_computed_keys(
    db: &RedDB,
    records: &mut [UnifiedRecord],
    order_by: &[OrderByClause],
    projections: &[Projection],
    table_name: &str,
    table_alias: &str,
    document_projection: bool,
    entity_projection: bool,
) {
    let referenced: Vec<String> = order_by
        .iter()
        .filter(|clause| clause.expr.is_none())
        .filter_map(|clause| match &clause.field {
            FieldRef::TableColumn { table, column } if table.is_empty() => Some(column.clone()),
            _ => None,
        })
        .collect();
    if referenced.is_empty() {
        return;
    }
    let needed: Vec<&Projection> = projections
        .iter()
        .filter(|projection| {
            matches!(
                projection,
                Projection::Function(..) | Projection::Expression(..)
            ) && referenced
                .iter()
                .any(|name| *name == projection_name(projection))
        })
        .collect();
    if needed.is_empty() {
        return;
    }
    for record in records.iter_mut() {
        for projection in &needed {
            let name = projection_name(projection);
            if record.get(name.as_str()).is_some() {
                continue;
            }
            if let Ok(projected) = project_runtime_record_with_db(
                Some(db),
                record,
                std::slice::from_ref(*projection),
                Some(table_name),
                Some(table_alias),
                document_projection,
                entity_projection,
            ) {
                if let Some(value) = projected.get(name.as_str()) {
                    let value = value.clone();
                    record.set(name.as_str(), value);
                }
            }
        }
    }
}

pub(crate) fn execute_runtime_canonical_table_node(
    db: &RedDB,
    node: &crate::storage::query::planner::CanonicalLogicalNode,
    context: &RuntimeTableExecutionContext<'_>,
) -> RedDBResult<Vec<UnifiedRecord>> {
    reddb_engine::query::runtime::table_plan::execute_table_plan_with_host(
        &RuntimeTablePlanHost { db, context },
        node,
        context.query,
    )
}

struct RuntimeTablePlanHost<'a, 'b> {
    db: &'a RedDB,
    context: &'a RuntimeTableExecutionContext<'b>,
}
impl reddb_engine::query::runtime::table_plan::TablePlanHost for RuntimeTablePlanHost<'_, '_> {
    type Error = RedDBError;
    fn scan(
        &self,
        _node: &crate::storage::query::planner::CanonicalLogicalNode,
    ) -> RedDBResult<Vec<UnifiedRecord>> {
        let db = self.db;
        let context = self.context;
        let effective_filter = effective_table_filter(context.query);
        let effective_projections = effective_table_projections(context.query);
        let uses_document_projection =
            runtime_projections_use_document_path(&effective_projections, context.query);
        // ── FAST PATH 1: Direct entity_id lookup (O(1) instead of full scan) ──
        if !projections_require_runtime_projection(&effective_projections)
            && !uses_document_projection
        {
            if let Some(entity_id) = extract_entity_id_from_filter(&effective_filter) {
                let entity = resolve_table_row_by_logical_id(
                    db,
                    &context.query.table,
                    EntityId::new(entity_id),
                );
                if let Some(entity) = entity {
                    if !db.replica_allows_entity_at_read(&context.query.table, &entity) {
                        return Ok(Vec::new());
                    }
                    return Ok(
                        super::super::record_search::runtime_table_record_lean_in_collection(
                            entity,
                            &context.query.table,
                        )
                        .into_iter()
                        .collect(),
                    );
                }
                return Ok(Vec::new());
            }
        }

        // ── FAST PATH 2: Filtered scan with entity-level pre-filter ──
        // Evaluates the WHERE clause directly on raw entity data, only
        // creating UnifiedRecord for entities that match the filter.
        // Skip for universal sources ("any") which need cross-collection scanning.
        if effective_filter.is_some()
                && !effective_filter
                    .as_ref()
                    .is_some_and(runtime_filter_contains_compare_expr)
                && !effective_filter.as_ref().is_some_and(|filter| {
                    runtime_filter_uses_document_path(filter, context.query)
                })
                && !effective_projections.iter().any(|p| {
                        matches!(p, Projection::Function(_, _) | Projection::Expression(_, _))
                            || matches!(p, Projection::Column(column) | Projection::Alias(column, _) if column.starts_with("LIT:"))
                    })
                && !uses_document_projection
                && !is_universal_query_source(context.query.table.as_str())
            {
                let manager = db
                    .store()
                    .get_collection(context.query.table.as_str())
                    .ok_or_else(|| RedDBError::NotFound(context.query.table.clone()))?;

                let filter = effective_filter.as_ref().ok_or_else(|| {
                    RedDBError::Internal(
                        "canonical filtered scan selected without a WHERE clause".into(),
                    )
                })?;
                let table_name = context.table_name;
                let table_alias = context.table_alias;
                let limit = usize::try_from(context.query.limit.unwrap_or(10000)).unwrap_or(usize::MAX);

                // The canonical parent still evaluates WHERE and ORDER BY
                // before applying aliases. Keep its source fields available.
                let select_cols = if effective_projections.iter().any(|p| {
                    matches!(p, Projection::Alias(_, _) | Projection::Field(_, Some(_)))
                }) {
                    Vec::new()
                } else {
                    extract_select_column_names(&effective_projections)
                };
                let schema_arc = manager.column_schema();
                let compiled = match schema_arc.as_ref() {
                    Some(schema) => {
                        super::filter_compiled::CompiledEntityFilter::compile_with_schema(
                            filter,
                            table_name,
                            table_alias,
                            schema.as_ref(),
                        )
                    }
                    None => super::filter_compiled::CompiledEntityFilter::compile(
                        filter,
                        table_name,
                        table_alias,
                    ),
                };
                // Schema-index precomputation: same optimisation as the indexed scan path.
                // Resolve projected column names → schema positions once before the loop.
                let schema_col_indices: Option<Vec<(usize, usize)>> = if !select_cols.is_empty() {
                    schema_arc.as_ref().map(|schema| {
                        select_cols
                            .iter()
                            .enumerate()
                            .filter_map(|(ci, col)| {
                                schema.iter().position(|s| s == col).map(|si| (ci, si))
                            })
                            .collect()
                    })
                } else {
                    None
                };

                let mut records: Vec<UnifiedRecord> = Vec::new();
                let snapshot = crate::runtime::impl_core::capture_current_snapshot();
                crate::runtime::function_budget::try_scan_result(
            |visit| manager.try_scan_for_each(snapshot.as_ref(), usize::MAX, visit),
            |entity| {
                    if records.len() >= limit {
                        return Ok(false);
                    }
                    if !db.replica_allows_entity_at_read(&context.query.table, entity) {
                        return Ok(true);
                    }
                    if compiled_entity_filter_matches(
                        db,
                        &compiled,
                        entity,
                        filter,
                        table_name,
                        table_alias,
                    )? {
                        let record = if !select_cols.is_empty() {
                            if let Some(ref idx_map) = schema_col_indices {
                                super::super::record_search::runtime_table_record_with_col_indices(
                                    entity, &select_cols, idx_map,
                                )
                                .or_else(|| {
                                    super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                        entity, &select_cols,
                                    )
                                })
                                .or_else(|| {
                                    runtime_table_record_from_entity_projected(
                                        entity.clone(),
                                        &select_cols,
                                    )
                                })
                            } else {
                                super::super::record_search::runtime_table_record_from_entity_ref_projected(
                                    entity,
                                    &select_cols,
                                )
                                .or_else(|| {
                                    runtime_table_record_from_entity_projected(
                                        entity.clone(),
                                        &select_cols,
                                    )
                                })
                            }
                        } else {
                            runtime_table_record_from_entity(entity.clone())
                        };
                        if let Some(record) = record {
                            records.push(record);
                        }
                    }
                    Ok(true)
                })?;
                return Ok(records);
            }

        // ── DEFAULT: Full scan with LIMIT pushdown ──
        let scan_cap = reddb_engine::query::runtime::table_plan::table_scan_cap(context.query);
        scan_runtime_table_source_records_limited(db, context.query.table.as_str(), scan_cap)
    }
    fn try_filter_lookup(&self) -> RedDBResult<Option<Vec<UnifiedRecord>>> {
        let db = self.db;
        let context = self.context;
        let effective_filter = effective_table_filter(context.query);
        let effective_projections = effective_table_projections(context.query);
        let uses_document_projection =
            runtime_projections_use_document_path(&effective_projections, context.query);
        // ── FAST PATH: Direct entity_id lookup (O(1)) ──
        if !projections_require_runtime_projection(&effective_projections)
            && !uses_document_projection
        {
            if let Some(entity_id) = extract_entity_id_from_filter(&effective_filter) {
                let entity = resolve_table_row_by_logical_id(
                    db,
                    &context.query.table,
                    EntityId::new(entity_id),
                );
                if let Some(entity) = entity {
                    return Ok(Some(
                        super::super::record_search::runtime_table_record_lean_in_collection(
                            entity,
                            &context.query.table,
                        )
                        .into_iter()
                        .collect(),
                    ));
                }
                return Ok(Some(Vec::new()));
            }
        }

        Ok(None)
    }
    fn filter_records(
        &self,
        records: Vec<UnifiedRecord>,
        document_gate: bool,
    ) -> RedDBResult<Vec<UnifiedRecord>> {
        let db = self.db;
        let context = self.context;
        let Some(filter) = effective_table_filter(context.query) else {
            return Ok(records);
        };
        filter_records_result(records, |record| {
            // Preserve the document capability/root gate for JSON-column tenant paths.
            if document_gate
                && !(runtime_record_has_document_capability(record)
                    || runtime_record_carries_filter_document_roots(record, &filter, context.query))
            {
                return Ok(false);
            }
            evaluate_runtime_filter_result_with_db(
                Some(db),
                record,
                &filter,
                Some(context.table_name),
                Some(context.table_alias),
            )
        })
    }
    fn sort_records(
        &self,
        node: &crate::storage::query::planner::CanonicalLogicalNode,
        records: &mut Vec<UnifiedRecord>,
    ) -> RedDBResult<()> {
        let db = self.db;
        let context = self.context;
        if !context.query.order_by.is_empty() {
            // Issue #769 — cap the materialized sort buffer.
            crate::runtime::materialization_limit::guard(db, "sort", records.len())?;
            // #1982 — materialize computed SELECT aliases referenced by
            // ORDER BY before sorting, so the sort finds the column instead
            // of resolving it to None and leaving rows unordered.
            let effective_projections = effective_table_projections(context.query);
            let document_projection = node.operator == "document_sort"
                || runtime_projections_use_document_path(&effective_projections, context.query);
            let entity_projection = node.operator == "entity_sort";
            materialize_order_by_computed_keys(
                db,
                records,
                &context.query.order_by,
                &effective_projections,
                context.table_name,
                context.table_alias,
                document_projection,
                entity_projection,
            );
            super::super::join_filter::try_sort_records_by_order_by_with_db(
                Some(db),
                records,
                &context.query.order_by,
                Some(context.table_name),
                Some(context.table_alias),
            )?;
        } else if node.operator == "entity_sort" {
            records.sort_by(compare_runtime_ranked_records);
        }
        Ok(())
    }
    fn rank_records(&self, records: &mut Vec<UnifiedRecord>) {
        records.sort_by(compare_runtime_ranked_records);
    }
    fn project_records(
        &self,
        node: &crate::storage::query::planner::CanonicalLogicalNode,
        mut records: Vec<UnifiedRecord>,
    ) -> RedDBResult<Vec<UnifiedRecord>> {
        let db = self.db;
        let context = self.context;
        let document_projection = node.operator == "document_projection";
        let entity_projection = node.operator == "entity_projection";
        let effective_projections = effective_table_projections(context.query);
        // Issue #590 slice 7b — apply the window phase (ROW_NUMBER,
        // RANK, DENSE_RANK, LAG, LEAD) between filter/sort and the
        // final per-row projection. The window phase materialises a
        // virtual column on each record under the projection's
        // alias; the `Projection::Window` arm in
        // `project_runtime_record_with_db` then reads that column
        // back like any other field.
        crate::runtime::window_phase::apply(
            Some(db),
            &mut records,
            &effective_projections,
            Some(context.table_name),
            Some(context.table_alias),
        )?;
        records
            .iter()
            .map(|record| {
                project_runtime_record_with_db(
                    Some(db),
                    record,
                    &effective_projections,
                    Some(context.table_name),
                    Some(context.table_alias),
                    document_projection,
                    entity_projection,
                )
            })
            .collect()
    }
    fn query_error(&self, message: String) -> RedDBError {
        RedDBError::Query(message)
    }
}

/// Does `record` carry the root column of every document path the
/// filter traverses? Used to relax the `document_path_filter`
/// capability gate for dotted tenant/JSON-column predicates on plain
/// tables (#638): `meta.tenant` resolves against the row's real `meta`
/// column, so the row should be evaluated even without a "document"
/// capability. Returns `false` when the filter references no resolvable
/// dotted root (e.g. a foreign-table qualifier with no dotted path), so
/// genuine non-document rows in a document collection stay gated.
fn runtime_record_carries_filter_document_roots(
    record: &UnifiedRecord,
    filter: &Filter,
    query: &TableQuery,
) -> bool {
    let mut roots: Vec<Option<String>> = Vec::new();
    collect_filter_document_path_roots(filter, query, &mut roots);
    !roots.is_empty()
        && roots.iter().all(|root| {
            root.as_ref()
                .is_some_and(|root| record.get(root.as_str()).is_some())
        })
}

/// Collect, for each document-path field the filter references, the
/// root column its path traverses (`meta` for `meta.tenant`). A
/// document-path field with no dotted root (a bare foreign-table
/// qualifier) contributes `None` so the caller treats it as
/// unresolvable. Mirrors the shape of `runtime_filter_uses_document_path`.
fn collect_filter_document_path_roots(
    filter: &Filter,
    query: &TableQuery,
    out: &mut Vec<Option<String>>,
) {
    match filter {
        Filter::Compare { field, .. }
        | Filter::IsNull(field)
        | Filter::IsNotNull(field)
        | Filter::In { field, .. }
        | Filter::Between { field, .. }
        | Filter::Like { field, .. }
        | Filter::StartsWith { field, .. }
        | Filter::EndsWith { field, .. }
        | Filter::Contains { field, .. } => {
            push_field_document_path_root(field, query, out);
        }
        Filter::CompareFields { left, right, .. } => {
            push_field_document_path_root(left, query, out);
            push_field_document_path_root(right, query, out);
        }
        Filter::CompareExpr { lhs, rhs, .. } => {
            collect_expr_document_path_roots(lhs, query, out);
            collect_expr_document_path_roots(rhs, query, out);
        }
        Filter::And(left, right) | Filter::Or(left, right) => {
            collect_filter_document_path_roots(left, query, out);
            collect_filter_document_path_roots(right, query, out);
        }
        Filter::Not(inner) => collect_filter_document_path_roots(inner, query, out),
    }
}

fn collect_expr_document_path_roots(
    expr: &reddb_rql::ast::Expr,
    query: &TableQuery,
    out: &mut Vec<Option<String>>,
) {
    use reddb_rql::ast::Expr;
    match expr {
        Expr::Literal { .. } | Expr::Parameter { .. } => {}
        Expr::Column { field, .. } => push_field_document_path_root(field, query, out),
        Expr::BinaryOp { lhs, rhs, .. } => {
            collect_expr_document_path_roots(lhs, query, out);
            collect_expr_document_path_roots(rhs, query, out);
        }
        Expr::UnaryOp { operand, .. } => collect_expr_document_path_roots(operand, query, out),
        Expr::Cast { inner, .. } => collect_expr_document_path_roots(inner, query, out),
        Expr::FunctionCall { args, .. } => {
            for arg in args {
                collect_expr_document_path_roots(arg, query, out);
            }
        }
        Expr::Case {
            branches, else_, ..
        } => {
            for (cond, val) in branches {
                collect_expr_document_path_roots(cond, query, out);
                collect_expr_document_path_roots(val, query, out);
            }
            if let Some(e) = else_ {
                collect_expr_document_path_roots(e, query, out);
            }
        }
        Expr::IsNull { operand, .. } => collect_expr_document_path_roots(operand, query, out),
        Expr::InList { target, values, .. } => {
            collect_expr_document_path_roots(target, query, out);
            for v in values {
                collect_expr_document_path_roots(v, query, out);
            }
        }
        Expr::Between {
            target, low, high, ..
        } => {
            collect_expr_document_path_roots(target, query, out);
            collect_expr_document_path_roots(low, query, out);
            collect_expr_document_path_roots(high, query, out);
        }
        // Opaque shapes can't be proven to traverse a known root —
        // contribute an unresolvable marker so the gate stays strict.
        Expr::Subquery { .. } | Expr::WindowFunctionCall { .. } => out.push(None),
    }
}

fn runtime_projections_use_document_path(projections: &[Projection], query: &TableQuery) -> bool {
    projections
        .iter()
        .any(|projection| runtime_projection_uses_document_path(projection, query))
}

fn runtime_projection_uses_document_path(projection: &Projection, query: &TableQuery) -> bool {
    match projection {
        Projection::Column(name) | Projection::Alias(name, _) => {
            name.split_once('.').is_some_and(|(head, tail)| {
                tail.contains('.')
                    || (head != query.table.as_str() && query.alias.as_deref() != Some(head))
            })
        }
        Projection::Field(field, _) => runtime_field_ref_uses_document_path(field, query),
        Projection::Expression(filter, _) => runtime_filter_uses_document_path(filter, query),
        Projection::Function(_, _) | Projection::All | Projection::Window { .. } => false,
    }
}

/// Push the resolvable document-path root for `field`, if it is one.
/// A dotted column (`meta.tenant`) yields `Some("meta")`. A bare
/// foreign-table qualifier (document-path by virtue of the qualifier,
/// not a dotted name) yields `None` — unresolvable for presence checks.
/// Flat same-table columns are not document paths and contribute nothing.
fn push_field_document_path_root(
    field: &FieldRef,
    query: &TableQuery,
    out: &mut Vec<Option<String>>,
) {
    if !runtime_field_ref_uses_document_path(field, query) {
        return;
    }
    match field {
        FieldRef::TableColumn { column, .. } => match column.split_once('.') {
            Some((root, _)) => out.push(Some(root.to_string())),
            None => out.push(None),
        },
        _ => out.push(None),
    }
}

fn runtime_filter_uses_document_path(filter: &Filter, query: &TableQuery) -> bool {
    match filter {
        Filter::Compare { field, .. }
        | Filter::IsNull(field)
        | Filter::IsNotNull(field)
        | Filter::In { field, .. }
        | Filter::Between { field, .. }
        | Filter::Like { field, .. }
        | Filter::StartsWith { field, .. }
        | Filter::EndsWith { field, .. }
        | Filter::Contains { field, .. } => runtime_field_ref_uses_document_path(field, query),
        Filter::CompareFields { left, right, .. } => {
            runtime_field_ref_uses_document_path(left, query)
                || runtime_field_ref_uses_document_path(right, query)
        }
        // Mirror the planner's `filter_uses_document_path`: only route
        // a CompareExpr to the document-path path when an operand
        // actually traverses a document path. Flat predicates such as
        // the tenant-iso policy `col = CURRENT_TENANT()` stay on the
        // regular filter path.
        Filter::CompareExpr { lhs, rhs, .. } => {
            runtime_expr_uses_document_path(lhs, query)
                || runtime_expr_uses_document_path(rhs, query)
        }
        Filter::And(left, right) | Filter::Or(left, right) => {
            runtime_filter_uses_document_path(left, query)
                || runtime_filter_uses_document_path(right, query)
        }
        Filter::Not(inner) => runtime_filter_uses_document_path(inner, query),
    }
}

/// Runtime twin of the planner's `expr_uses_document_path` — must stay
/// in sync so FAST PATH 2 gating agrees with the operator the planner
/// chose. See `planner::logical_helpers::expr_uses_document_path`.
fn runtime_expr_uses_document_path(expr: &reddb_rql::ast::Expr, query: &TableQuery) -> bool {
    use reddb_rql::ast::Expr;
    match expr {
        Expr::Literal { .. } | Expr::Parameter { .. } => false,
        Expr::Column { field, .. } => runtime_field_ref_uses_document_path(field, query),
        Expr::BinaryOp { lhs, rhs, .. } => {
            runtime_expr_uses_document_path(lhs, query)
                || runtime_expr_uses_document_path(rhs, query)
        }
        Expr::UnaryOp { operand, .. } => runtime_expr_uses_document_path(operand, query),
        Expr::Cast { inner, .. } => runtime_expr_uses_document_path(inner, query),
        Expr::FunctionCall { args, .. } => args
            .iter()
            .any(|a| runtime_expr_uses_document_path(a, query)),
        Expr::Case {
            branches, else_, ..
        } => {
            branches.iter().any(|(cond, val)| {
                runtime_expr_uses_document_path(cond, query)
                    || runtime_expr_uses_document_path(val, query)
            }) || else_
                .as_ref()
                .is_some_and(|e| runtime_expr_uses_document_path(e, query))
        }
        Expr::IsNull { operand, .. } => runtime_expr_uses_document_path(operand, query),
        Expr::InList { target, values, .. } => {
            runtime_expr_uses_document_path(target, query)
                || values
                    .iter()
                    .any(|v| runtime_expr_uses_document_path(v, query))
        }
        Expr::Between {
            target, low, high, ..
        } => {
            runtime_expr_uses_document_path(target, query)
                || runtime_expr_uses_document_path(low, query)
                || runtime_expr_uses_document_path(high, query)
        }
        Expr::Subquery { .. } | Expr::WindowFunctionCall { .. } => true,
    }
}

fn runtime_field_ref_uses_document_path(field: &FieldRef, query: &TableQuery) -> bool {
    match field {
        FieldRef::TableColumn { table, column } => {
            column.contains('.')
                || (!table.is_empty()
                    && table != &query.table
                    && query.alias.as_deref() != Some(table.as_str()))
        }
        _ => false,
    }
}

fn universal_candidate_collections_for_filter(
    db: &RedDB,
    query: &TableQuery,
    filter: &Filter,
) -> Option<Vec<String>> {
    let mut fields = Vec::new();
    if !collect_universal_filter_field_roots(filter, query, &mut fields) || fields.is_empty() {
        return None;
    }
    fields.sort();
    fields.dedup();

    Some(
        db.store()
            .list_collections()
            .into_iter()
            .filter(|collection| universal_collection_may_have_fields(db, collection, &fields))
            .collect(),
    )
}

fn collect_universal_filter_field_roots(
    filter: &Filter,
    query: &TableQuery,
    out: &mut Vec<String>,
) -> bool {
    match filter {
        Filter::Compare { field, .. }
        | Filter::IsNull(field)
        | Filter::IsNotNull(field)
        | Filter::In { field, .. }
        | Filter::Between { field, .. }
        | Filter::Like { field, .. }
        | Filter::StartsWith { field, .. }
        | Filter::EndsWith { field, .. }
        | Filter::Contains { field, .. } => push_universal_filter_field_root(field, query, out),
        Filter::CompareFields { left, right, .. } => {
            push_universal_filter_field_root(left, query, out)
                && push_universal_filter_field_root(right, query, out)
        }
        Filter::And(left, right) => {
            collect_universal_filter_field_roots(left, query, out)
                && collect_universal_filter_field_roots(right, query, out)
        }
        // OR/NOT and expression filters can be soundly executed by the
        // existing universal plan. Keep them there until we have a candidate
        // algebra that can prove union/complement collection sets.
        Filter::Or(_, _) | Filter::Not(_) | Filter::CompareExpr { .. } => false,
    }
}

fn push_universal_filter_field_root(
    field: &FieldRef,
    query: &TableQuery,
    out: &mut Vec<String>,
) -> bool {
    let FieldRef::TableColumn { table, column } = field else {
        return false;
    };
    if !table.is_empty()
        && !is_universal_query_source(table)
        && query.alias.as_deref() != Some(table.as_str())
    {
        return false;
    }
    let root = column
        .split_once('.')
        .map_or(column.as_str(), |(root, _)| root);
    if is_universal_runtime_field(root) {
        return false;
    }
    out.push(root.to_string());
    true
}

fn is_universal_runtime_field(field: &str) -> bool {
    matches!(
        field,
        "rid"
            | "row_id"
            | "entity_id"
            | "collection"
            | "red_collection"
            | "kind"
            | "red_kind"
            | "tenant"
            | "created_at"
            | "updated_at"
            | "red_entity_type"
            | "red_capabilities"
            | "red_sequence_id"
    )
}

fn universal_collection_may_have_fields(db: &RedDB, collection: &str, fields: &[String]) -> bool {
    let contract = db.collection_contract(collection);
    if let Some(contract) = contract.as_ref() {
        let declared_match = fields.iter().all(|field| {
            contract
                .declared_columns
                .iter()
                .any(|column| column.name.eq_ignore_ascii_case(field))
        });
        if declared_match && !contract.declared_columns.is_empty() {
            return true;
        }
        if matches!(contract.schema_mode, crate::catalog::SchemaMode::Strict)
            && !contract.declared_columns.is_empty()
        {
            return false;
        }
    }

    let Some(manager) = db.store().get_collection(collection) else {
        return false;
    };
    if let Some(schema) = manager.column_schema() {
        if !schema.is_empty() {
            return fields.iter().all(|field| {
                schema
                    .iter()
                    .any(|column| column.eq_ignore_ascii_case(field))
            });
        }
    }

    if let Some(contract) = contract.as_ref() {
        if !universal_contract_may_have_dynamic_fields(contract) {
            return universal_contract_has_static_fields(contract, fields);
        }
        if universal_contract_has_static_fields(contract, fields) {
            return true;
        }
        return universal_collection_has_observed_fields(&manager, fields);
    }

    universal_collection_has_observed_fields(&manager, fields)
}

fn universal_contract_has_static_fields(
    contract: &crate::physical::CollectionContract,
    fields: &[String],
) -> bool {
    match contract.declared_model {
        crate::catalog::CollectionModel::Graph => fields_match_any(
            fields,
            &["label", "node_type", "from_rid", "to_rid", "weight"],
        ),
        crate::catalog::CollectionModel::Vector => {
            fields_match_any(fields, &["dimension", "content"])
        }
        crate::catalog::CollectionModel::TimeSeries | crate::catalog::CollectionModel::Metrics => {
            fields_match_any(
                fields,
                &[
                    "metric",
                    "timestamp_ns",
                    "timestamp",
                    "time",
                    "value",
                    "tags",
                ],
            )
        }
        crate::catalog::CollectionModel::Queue => fields_match_any(
            fields,
            &["position", "payload", "attempts", "acked", "priority"],
        ),
        crate::catalog::CollectionModel::Kv
        | crate::catalog::CollectionModel::Config
        | crate::catalog::CollectionModel::Vault => fields_match_any(fields, &["key", "value"]),
        _ => false,
    }
}

fn universal_contract_may_have_dynamic_fields(
    contract: &crate::physical::CollectionContract,
) -> bool {
    matches!(
        contract.declared_model,
        crate::catalog::CollectionModel::Document
            | crate::catalog::CollectionModel::Graph
            | crate::catalog::CollectionModel::Hll
            | crate::catalog::CollectionModel::Sketch
            | crate::catalog::CollectionModel::Filter
            | crate::catalog::CollectionModel::Mixed
    ) || matches!(
        contract.declared_model,
        crate::catalog::CollectionModel::Table
    ) && !matches!(contract.schema_mode, crate::catalog::SchemaMode::Strict)
}

fn universal_collection_has_observed_fields(
    manager: &crate::storage::unified::manager::SegmentManager,
    fields: &[String],
) -> bool {
    if fields.is_empty() {
        return false;
    }
    let mut seen = vec![false; fields.len()];
    manager.for_each_entity(|entity| {
        for (idx, field) in fields.iter().enumerate() {
            if !seen[idx] && universal_entity_has_observed_field(entity, field) {
                seen[idx] = true;
            }
        }
        !seen.iter().all(|value| *value)
    });
    seen.into_iter().all(|value| value)
}

fn universal_entity_has_observed_field(
    entity: &crate::storage::unified::UnifiedEntity,
    field: &str,
) -> bool {
    match &entity.data {
        crate::storage::unified::EntityData::Row(row) => row
            .iter_fields()
            .any(|(name, _)| name.eq_ignore_ascii_case(field)),
        crate::storage::unified::EntityData::Node(node) => node
            .properties
            .keys()
            .any(|name| name.eq_ignore_ascii_case(field)),
        crate::storage::unified::EntityData::Edge(edge) => edge
            .properties
            .keys()
            .any(|name| name.eq_ignore_ascii_case(field)),
        _ => false,
    }
}

fn fields_match_any(fields: &[String], candidates: &[&str]) -> bool {
    fields.iter().all(|field| {
        candidates
            .iter()
            .any(|candidate| candidate.eq_ignore_ascii_case(field))
    })
}

#[cfg(test)]
mod tests {
    use crate::runtime::mvcc::{clear_current_connection_id, set_current_connection_id};
    use crate::storage::unified::EntityId;
    use crate::{RedDBOptions, RedDBRuntime};
    use reddb_rql::ast::{CompareOp, FieldRef, Filter, QueryExpr};
    use reddb_types::Value;

    fn rt() -> RedDBRuntime {
        RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("in-memory runtime")
    }

    #[test]
    fn cold_published_table_reads_match_resident_sql_results() {
        use crate::storage::engine::BTree;
        use reddb_engine::storage::collection_read::PagedCollection;
        use std::collections::BTreeMap;
        use std::sync::Arc;
        let directory = tempfile::tempdir().expect("SQL fixture");
        let options = RedDBOptions::persistent(directory.path().join("cold-sql.rdb"))
            .with_memory_budget(128 * 1024 * 1024)
            .with_storage_profile(
                crate::storage::profile::StorageDeployPreset::PrimaryReplicaDev.selection(),
            )
            .expect("profile");
        let runtime = RedDBRuntime::with_options(options).expect("paged runtime");
        exec(
            &runtime,
            "CREATE TABLE cold_rows (id INTEGER, cohort INTEGER, payload TEXT)",
        );
        let fields = vec![
            "id".to_string(),
            "cohort".to_string(),
            "payload".to_string(),
        ];
        runtime
            .db()
            .store()
            .bulk_insert(
                "cold_rows",
                (1..=4096)
                    .map(|id| {
                        let values = vec![
                            Value::Integer(id),
                            Value::Integer(1),
                            Value::text(if id == 4096 {
                                "x".repeat(32 * 1024)
                            } else {
                                "preserved".to_string()
                            }),
                        ];
                        let mut entity = crate::storage::UnifiedEntity::table_row(
                            EntityId::new(0),
                            "cold_rows",
                            0,
                            vec![],
                        );
                        entity.data = crate::storage::EntityData::Row(
                            crate::storage::unified::entity::RowData::with_names(
                                values,
                                fields.clone(),
                            ),
                        );
                        entity
                    })
                    .collect(),
            )
            .expect("bulk SQL fixture");
        exec(
            &runtime,
            "CREATE INDEX cold_rows_id ON cold_rows (id) USING HASH",
        );
        let queries = [
            "SELECT id FROM cold_rows ORDER BY id",
            "SELECT id FROM cold_rows WHERE cohort >= 0 ORDER BY id",
            "SELECT id FROM cold_rows WHERE id = 5",
            "SELECT COUNT(*) AS n FROM cold_rows",
        ];
        let observe = |sql: &str| {
            runtime.invalidate_result_cache();
            runtime
                .execute_query(sql)
                .expect("SQL read")
                .result
                .records
                .iter()
                .map(|row| {
                    row.iter_fields()
                        .map(|(key, value)| (key.to_string(), value.clone()))
                        .collect::<BTreeMap<_, _>>()
                })
                .collect::<Vec<_>>()
        };
        let expected: Vec<_> = queries.iter().map(|sql| observe(sql)).collect();
        assert_eq!(
            expected.iter().map(Vec::len).collect::<Vec<_>>(),
            vec![4096, 4096, 1, 1]
        );
        let store = runtime.db().store();
        store.persist().expect("durable base");
        let manager = store.get_collection("cold_rows").expect("rows");
        let root = store
            .collection_root_page("cold_rows")
            .expect("published root");
        let tree = Arc::new(BTree::with_root(
            Arc::clone(store.pager().expect("pager")),
            root,
        ));
        // Install the real storage read seam without enabling the unfinished
        // runtime profile or changing any managed deployment configuration.
        manager.install_backing(
            Arc::new(PagedCollection::new(tree)),
            root,
            store.format_version(),
            128 * 1024,
            manager.count() as u64,
        );
        let mut mismatches = Vec::new();
        for (sql, expected) in queries.iter().zip(expected) {
            let actual = observe(sql);
            if actual != expected {
                mismatches.push(format!(
                    "{sql}: expected {} rows, got {}; expected prefix {:?}, actual prefix {:?}",
                    expected.len(),
                    actual.len(),
                    expected.iter().take(3).collect::<Vec<_>>(),
                    actual.iter().take(3).collect::<Vec<_>>()
                ));
            }
        }
        assert!(
            mismatches.is_empty(),
            "cold SQL paths differ: {mismatches:#?}"
        );
        let (base, _) = manager.current_backing_overlay().expect("cold generation");
        manager.install_backing(
            base,
            root,
            store.format_version(),
            4096,
            manager.count() as u64,
        );
        runtime.invalidate_result_cache();
        assert!(
            runtime
                .execute_query("SELECT id FROM cold_rows WHERE cohort >= 0 ORDER BY id")
                .is_err(),
            "a late disk decode error must not publish the successful prefix"
        );
    }

    #[test]
    fn projected_filtered_scans_do_not_clone_unreturned_payloads() {
        use crate::storage::no_malloc_hot_paths::measure_allocations;
        use crate::storage::unified::entity::RowData;
        use crate::storage::{EntityData, UnifiedEntity};

        let observe = |payload_bytes, columnar| {
            let runtime = rt();
            exec(
                &runtime,
                "CREATE TABLE projected_scan (id INTEGER, cohort INTEGER, payload BLOB)",
            );
            let manager = runtime
                .db()
                .store()
                .get_collection("projected_scan")
                .expect("manager");
            let schema = std::sync::Arc::new(vec!["id".into(), "cohort".into(), "payload".into()]);
            for id in 1..=4096 {
                let mut entity = UnifiedEntity::table_row(
                    EntityId::new(1_000_000 + id),
                    "projected_scan",
                    id,
                    vec![],
                );
                let values = vec![
                    Value::Integer(id as i64),
                    Value::Integer(0),
                    Value::Blob(vec![7; payload_bytes]),
                ];
                let row = if columnar {
                    RowData {
                        columns: values,
                        named: None,
                        schema: Some(schema.clone()),
                    }
                } else {
                    RowData::with_names(values, schema.as_ref().clone())
                };
                entity.data = EntityData::Row(row);
                manager.insert(entity).expect("row");
            }
            // One growing segment keeps the allocation observation on this
            // thread while exercising the parallel-eligible query branch.
            assert_eq!(manager.stats().sealed_count, 0);
            let sql = "SELECT id FROM projected_scan WHERE cohort >= 0";
            runtime.execute_query(sql).expect("warm planner");
            runtime.invalidate_result_cache();
            let (result, allocations) = measure_allocations(|| runtime.execute_query(sql));
            let result = result.expect("projected scan");
            assert_eq!(result.result.records.len(), 4096);
            let ids: std::collections::BTreeSet<_> = result
                .result
                .records
                .iter()
                .map(|row| match row.get("id") {
                    Some(Value::Integer(id)) => *id,
                    other => panic!("integer id: {other:?}"),
                })
                .collect();
            assert_eq!(ids, (1..=4096).collect());
            let ordered = runtime.execute_query(
                "SELECT id FROM projected_scan WHERE cohort >= 0 ORDER BY id DESC LIMIT 2 OFFSET 1"
            ).expect("ordered projected scan");
            assert_eq!(ordered.result.records.len(), 2);
            assert_eq!(
                ordered.result.records[0].get("id"),
                Some(&Value::Integer(4095))
            );
            assert_eq!(
                ordered.result.records[1].get("id"),
                Some(&Value::Integer(4094))
            );
            let selected = runtime
                .execute_query("SELECT payload FROM projected_scan WHERE id = 4096")
                .expect("selected payload");
            assert_eq!(selected.result.records.len(), 1);
            assert_eq!(
                selected.result.records[0].get("payload"),
                Some(&Value::Blob(vec![7; payload_bytes]))
            );
            allocations
        };
        for columnar in [false, true] {
            let baseline = observe(0, columnar);
            let measured = observe(8192, columnar);
            eprintln!(
                "projected scan allocation bytes (columnar={columnar}): {} -> {}",
                baseline.allocated_bytes, measured.allocated_bytes
            );
            assert!(measured.allocated_bytes <= baseline.allocated_bytes + 1024 * 1024,
                "unreturned payloads must not be cloned before projection: {baseline:?} -> {measured:?}");
        }
    }

    #[test]
    fn indexed_projections_do_not_clone_unreturned_payloads() {
        use crate::storage::no_malloc_hot_paths::measure_allocations;
        use crate::storage::unified::entity::RowData;
        use crate::storage::{EntityData, UnifiedEntity};

        let cases = [
            (
                "sorted",
                vec!["cohort USING BTREE"],
                "WHERE cohort >= 0 ORDER BY id DESC LIMIT 2 OFFSET 1",
            ),
            ("hash", vec!["cohort USING HASH"], "WHERE cohort = 0"),
            (
                "cross",
                vec!["cohort USING HASH", "id USING BTREE"],
                "WHERE cohort = 0 AND id >= 1",
            ),
            (
                "bitmap",
                vec!["cohort USING HASH", "bucket USING HASH"],
                "WHERE cohort = 0 AND bucket = 0",
            ),
        ];
        let mut regressions = Vec::new();
        for (name, indexes, predicate) in cases {
            for columnar in [false, true] {
                let observe = |payload_bytes| {
                    let runtime = rt();
                    exec(&runtime, "CREATE TABLE indexed_projection (id INTEGER, cohort INTEGER, bucket INTEGER, payload BLOB)");
                    let manager = runtime
                        .db()
                        .store()
                        .get_collection("indexed_projection")
                        .expect("manager");
                    let schema = std::sync::Arc::new(vec![
                        "id".into(),
                        "cohort".into(),
                        "bucket".into(),
                        "payload".into(),
                    ]);
                    for id in 1..=256 {
                        let mut entity = UnifiedEntity::table_row(
                            EntityId::new(1_000_000 + id),
                            "indexed_projection",
                            id,
                            vec![],
                        );
                        let values = vec![
                            Value::Integer(id as i64),
                            Value::Integer(0),
                            Value::Integer(0),
                            Value::Blob(vec![7; payload_bytes]),
                        ];
                        entity.data = EntityData::Row(if columnar {
                            RowData {
                                columns: values,
                                named: None,
                                schema: Some(schema.clone()),
                            }
                        } else {
                            RowData::with_names(values, schema.as_ref().clone())
                        });
                        manager.insert(entity).expect("row");
                    }
                    for (number, definition) in indexes.iter().enumerate() {
                        let (column, method) =
                            definition.split_once(" USING ").expect("index definition");
                        exec(&runtime, &format!("CREATE INDEX projection_{number} ON indexed_projection ({column}) USING {method}"));
                        assert!(runtime
                            .index_store_ref()
                            .find_index_for_column("indexed_projection", column)
                            .is_some());
                    }
                    let sql = format!("SELECT id FROM indexed_projection {predicate}");
                    runtime.execute_query(&sql).expect("warm planner");
                    runtime.invalidate_result_cache();
                    let (result, allocations) = measure_allocations(|| runtime.execute_query(&sql));
                    let result = result.expect("indexed projection");
                    let actual: Vec<_> = result
                        .result
                        .records
                        .iter()
                        .map(|row| {
                            assert!(row.get("payload").is_none());
                            match row.get("id") {
                                Some(Value::Integer(id)) => *id,
                                other => panic!("integer id: {other:?}"),
                            }
                        })
                        .collect();
                    if name == "sorted" {
                        assert_eq!(actual, vec![255, 254]);
                    } else {
                        assert_eq!(
                            actual
                                .into_iter()
                                .collect::<std::collections::BTreeSet<_>>(),
                            (1..=256).collect()
                        );
                    }
                    allocations
                };
                let baseline = observe(0);
                let measured = observe(16 * 1024);
                eprintln!(
                    "indexed {name} allocation bytes (columnar={columnar}): {} -> {}",
                    baseline.allocated_bytes, measured.allocated_bytes
                );
                if measured.allocated_bytes > baseline.allocated_bytes + 1024 * 1024 {
                    regressions.push(format!(
                        "{name}, columnar={columnar}: {baseline:?} -> {measured:?}"
                    ));
                }
            }
        }
        assert!(
            regressions.is_empty(),
            "unreturned indexed payloads were cloned: {regressions:#?}"
        );
    }

    #[test]
    fn indexed_candidates_preserve_order_windows_and_errors_across_segments() {
        use crate::storage::unified::entity::RowData;
        use crate::storage::{EntityData, UnifiedEntity};
        let runtime = rt();
        exec(
            &runtime,
            "CREATE TABLE index_window (id INTEGER, payload BLOB)",
        );
        let manager = runtime
            .db()
            .store()
            .get_collection("index_window")
            .expect("manager");
        for id in 1..=4 {
            let mut entity =
                UnifiedEntity::table_row(EntityId::new(1_000_000 + id), "index_window", id, vec![]);
            entity.data = EntityData::Row(RowData::with_names(
                vec![Value::Integer(id as i64), Value::Blob(vec![7; 8192])],
                vec!["id".into(), "payload".into()],
            ));
            manager.insert(entity).expect("row");
            if id == 2 {
                manager.force_seal().expect("sealed prefix");
            }
        }
        assert_eq!(manager.stats().sealed_count, 1);
        let candidates = [2, 4, 1, 99, 3].map(|id| EntityId::new(1_000_000 + id));
        let columns = vec!["id".to_string()];
        let ids = |records: Vec<crate::storage::query::unified::UnifiedRecord>| {
            records
                .into_iter()
                .map(|record| record.get("id").cloned().expect("id"))
                .collect::<Vec<_>>()
        };
        let records = super::indexed_table_records(
            &runtime.db(),
            "index_window",
            &candidates,
            &columns,
            usize::MAX,
            |_| Ok(true),
        )
        .expect("all candidates");
        assert_eq!(ids(records), [2, 4, 1, 3].map(Value::Integer));
        let mut visited = Vec::new();
        let records = super::indexed_table_records(
            &runtime.db(),
            "index_window",
            &candidates,
            &columns,
            1,
            |entity| {
                visited.push(entity.id.raw());
                Ok(entity.id.raw() % 2 == 1)
            },
        )
        .expect("filtered prefix");
        assert_eq!(ids(records), vec![Value::Integer(1)]);
        assert_eq!(visited, vec![1_000_002, 1_000_004, 1_000_001]);
        let result = super::indexed_table_records(
            &runtime.db(),
            "index_window",
            &candidates,
            &columns,
            usize::MAX,
            |entity| {
                if entity.id.raw() == 1_000_003 {
                    Err(crate::RedDBError::Internal(
                        "injected indexed predicate failure".into(),
                    ))
                } else {
                    Ok(true)
                }
            },
        );
        assert!(
            result.is_err(),
            "a predicate failure must reject the accepted prefix"
        );
        let zero = super::indexed_table_records(
            &runtime.db(),
            "index_window",
            &candidates,
            &columns,
            0,
            |_| panic!("LIMIT 0 must not evaluate candidates"),
        )
        .expect("zero window");
        assert!(zero.is_empty());
    }

    #[test]
    fn indexed_compact_rows_restore_schema_and_selected_payloads() {
        use crate::storage::unified::entity::RowData;
        use crate::storage::{EntityData, UnifiedEntity};
        let runtime = rt();
        exec(
            &runtime,
            "CREATE TABLE compact_index (id INTEGER, tenant_id TEXT, payload BLOB)",
        );
        let manager = runtime
            .db()
            .store()
            .get_collection("compact_index")
            .expect("manager");
        manager.set_column_schema_if_empty(vec!["id".into(), "tenant_id".into(), "payload".into()]);
        let mut entity =
            UnifiedEntity::table_row(EntityId::new(1_000_001), "compact_index", 1, vec![]);
        entity.data = EntityData::Row(RowData::new(vec![
            Value::Integer(1),
            Value::text("tenant-a"),
            Value::Blob(vec![7; 8192]),
        ]));
        manager.insert(entity).expect("compact row");
        manager
            .get_with(EntityId::new(1_000_001), |entity| {
                let EntityData::Row(row) = &entity.data else {
                    panic!("row")
                };
                assert!(row.named.is_none() && row.schema.is_none());
            })
            .expect("resident raw row");
        let candidates = [EntityId::new(1_000_001)];
        for columns in [vec![], vec!["id".into(), "tenant".into(), "payload".into()]] {
            let records = super::indexed_table_records(
                &runtime.db(),
                "compact_index",
                &candidates,
                &columns,
                usize::MAX,
                |_| Ok(true),
            )
            .expect("schema-restored projection");
            assert_eq!(records.len(), 1);
            assert_eq!(records[0].get("id"), Some(&Value::Integer(1)));
            assert_eq!(records[0].get("tenant"), Some(&Value::text("tenant-a")));
            assert_eq!(records[0].get("payload"), Some(&Value::Blob(vec![7; 8192])));
        }
    }

    #[test]
    fn canonical_table_plan_keeps_ordering_projection_and_wide_bounds() {
        let runtime = rt();
        exec(&runtime, "CREATE TABLE plan_values (id INT)");
        exec(
            &runtime,
            "INSERT INTO plan_values (id) VALUES (1), (2), (3), (4)",
        );
        let result = runtime.execute_query(
            "SELECT id * 2 AS doubled FROM plan_values WHERE id >= 2 ORDER BY id DESC LIMIT 1 OFFSET 1",
        ).expect("canonical read plan");
        assert_eq!(result.result.records.len(), 1);
        assert_eq!(
            result.result.records[0].get("doubled"),
            Some(&Value::Integer(6))
        );
        let result = runtime.execute_query(
            "SELECT id * 2 AS doubled FROM plan_values LIMIT 9223372036854775807 OFFSET 9223372036854775807",
        ).expect("wide SQL scan bounds must not wrap or panic");
        assert!(result.result.records.is_empty());
    }

    #[test]
    fn sql_hash_equality_uses_registered_candidates_before_and_after_reopen() {
        let directory = tempfile::tempdir().expect("directory");
        let path = directory.path().join("indexed.rdb");
        let sql = "SELECT seq, item FROM indexed WHERE seq = 42";
        let parsed = reddb_rql::parser::parse(sql).expect("parse").query;
        let QueryExpr::Table(query) = parsed else {
            panic!("table query")
        };
        let filter = reddb_rql::sql_lowering::effective_table_filter(&query).expect("filter");
        let (column, bytes) = super::extract_index_candidate(&filter).expect("index candidate");
        assert_eq!(column, "seq");
        for reopen in [false, true] {
            let runtime =
                RedDBRuntime::with_options(RedDBOptions::persistent(&path)).expect("runtime");
            if !reopen {
                runtime
                    .execute_query("CREATE TABLE indexed (seq INTEGER, item TEXT, unused TEXT)")
                    .expect("table");
                runtime.execute_query("INSERT INTO indexed (seq, item, unused) VALUES (42, 'answer', 'unreturned'), (43, 'other', 'unreturned')").expect("rows");
                runtime
                    .execute_query("CREATE INDEX indexed_seq ON indexed (seq) USING HASH")
                    .expect("index");
            }
            let index = runtime
                .index_store_ref()
                .find_index_for_column("indexed", &column)
                .expect("registered index");
            let candidates = runtime
                .index_store_ref()
                .hash_lookup("indexed", index.hash_lookup_name().as_ref(), &bytes)
                .expect("lookup");
            assert_eq!(candidates.len(), 1, "reopen={reopen}, filter={filter:?}");
            let result = runtime.execute_query(sql).expect("select");
            assert_eq!(result.result.records.len(), 1);
            assert_eq!(
                result.result.records[0].get("item"),
                Some(&Value::text("answer"))
            );
            assert_eq!(
                result.result.stats.segments_scanned, 0,
                "point lookup should not scan; reopen={reopen}"
            );
            let missing = runtime
                .execute_query("SELECT item FROM indexed WHERE seq = 99")
                .expect("indexed miss");
            assert!(missing.result.records.is_empty());
        }
    }

    #[test]
    fn hash_miss_keeps_sql_numeric_equality_fallback() {
        let runtime = rt();
        exec(&runtime, "CREATE TABLE numeric_hash (n FLOAT, item TEXT)");
        exec(
            &runtime,
            "INSERT INTO numeric_hash (n, item) VALUES (42.0, 'answer')",
        );
        let sql = "SELECT item FROM numeric_hash WHERE n = 42";
        let before = runtime.execute_query(sql).expect("scan");
        assert_eq!(before.result.records.len(), 1);
        exec(
            &runtime,
            "CREATE INDEX numeric_hash_n ON numeric_hash (n) USING HASH",
        );
        let after = runtime
            .execute_query(sql)
            .expect("indexed numeric equality");
        assert_eq!(after.result.records.len(), 1);
        assert_eq!(
            after.result.records[0].get("item"),
            Some(&Value::text("answer"))
        );
    }

    fn exec(rt: &RedDBRuntime, sql: &str) {
        rt.execute_query(sql)
            .unwrap_or_else(|err| panic!("{sql}: {err:?}"));
    }

    #[test]
    fn where_arithmetic_errors_do_not_switch_evaluators() {
        let rt = rt();
        exec(&rt, "CREATE TABLE where_errors (id INT)");
        exec(&rt, "INSERT INTO where_errors (id) VALUES (1)");

        let overflow = rt
            .execute_query("SELECT id FROM where_errors WHERE 9223372036854775807 + 1 > 0")
            .expect_err("WHERE integer overflow must propagate");
        assert!(
            overflow.to_string().contains("arithmetic overflow"),
            "unexpected overflow error: {overflow}"
        );

        let division = rt
            .execute_query("SELECT id FROM where_errors WHERE 1 / 0 > 0")
            .expect_err("WHERE division by zero must propagate");
        assert!(
            division.to_string().contains("division by zero"),
            "unexpected division error: {division}"
        );
    }

    #[test]
    fn where_sparse_document_fields_are_null_without_hiding_errors() {
        let rt = rt();
        exec(&rt, "CREATE DOCUMENT sparse_values");
        exec(
            &rt,
            r#"INSERT INTO sparse_values DOCUMENT VALUES ({"name":"absent"}), ({"name":"present","n":1})"#,
        );
        for predicate in ["n + 1 > 1", "ABS(n) > 0"] {
            assert_eq!(
                text_values(
                    &rt,
                    &format!("SELECT name FROM sparse_values WHERE {predicate}"),
                    "name",
                ),
                ["present"],
            );
        }
        let error = rt
            .execute_query("SELECT name FROM sparse_values WHERE n / 0 > 0")
            .expect_err("present document values must still report arithmetic errors");
        assert!(error.to_string().contains("division by zero"), "{error}");

        exec(&rt, "CREATE TABLE strict_values (n INT)");
        exec(&rt, "INSERT INTO strict_values (n) VALUES (1)");
        let error = rt
            .execute_query("SELECT n FROM strict_values WHERE missing + 1 > 1")
            .expect_err("unknown table columns must remain errors");
        assert!(error.to_string().contains("unknown column"), "{error}");
    }

    #[test]
    fn where_omitted_nullable_columns_are_null_before_enrichment() {
        let rt = rt();
        exec(&rt, "CREATE TABLE nullable_values (id INT, n INT)");
        exec(&rt, "INSERT INTO nullable_values (id) VALUES (1)");
        exec(&rt, "INSERT INTO nullable_values (id, n) VALUES (2, 1)");
        for predicate in ["n + 1 > 1", "ABS(v.n) > 0"] {
            let result = rt
                .execute_query(&format!(
                    "SELECT id FROM nullable_values AS v WHERE {predicate}"
                ))
                .expect("omitted nullable columns do not reject the query");
            assert_eq!(result.result.records.len(), 1);
            assert_eq!(result.result.records[0].get("id"), Some(&Value::Integer(2)));
        }
        let error = rt
            .execute_query("SELECT id FROM nullable_values WHERE n / 0 > 0")
            .expect_err("present values still report arithmetic failures");
        assert!(error.to_string().contains("division by zero"), "{error}");
    }

    fn rid(rt: &RedDBRuntime, table: &str, id: i64) -> u64 {
        let result = rt
            .execute_query(&format!("SELECT rid FROM {table} WHERE id = {id}"))
            .unwrap_or_else(|err| panic!("select rid from {table}: {err:?}"));
        match result.result.records[0].get("rid") {
            Some(Value::UnsignedInteger(id)) => *id,
            Some(Value::Integer(id)) => *id as u64,
            other => panic!("expected rid, got {other:?}"),
        }
    }

    fn text_values(rt: &RedDBRuntime, sql: &str, column: &str) -> Vec<String> {
        let result = rt
            .execute_query(sql)
            .unwrap_or_else(|err| panic!("{sql}: {err:?}"));
        result
            .result
            .records
            .iter()
            .map(|record| match record.get(column) {
                Some(Value::Text(value)) => value.to_string(),
                other => panic!("expected text column {column}, got {other:?}"),
            })
            .collect()
    }

    #[test]
    fn covered_sorted_index_rejects_tombstoned_stale_candidate() {
        let rt = rt();
        set_current_connection_id(51001);
        exec(
            &rt,
            "CREATE TABLE mvcc_idx_stale (id INT, status TEXT, marker TEXT)",
        );
        exec(
            &rt,
            "INSERT INTO mvcc_idx_stale (id, status, marker) VALUES (1, 'gone', 'stable')",
        );
        exec(
            &rt,
            "CREATE INDEX idx_mvcc_stale_status ON mvcc_idx_stale (status) USING BTREE",
        );
        let stale_id = rid(&rt, "mvcc_idx_stale", 1);

        exec(&rt, "DELETE FROM mvcc_idx_stale WHERE id = 1");
        rt.index_store_ref()
            .index_entity_insert(
                "mvcc_idx_stale",
                EntityId::new(stale_id),
                &[("status".to_string(), Value::text("gone".to_string()))],
            )
            .expect("force stale index entry");

        let indexed = text_values(
            &rt,
            "SELECT status FROM mvcc_idx_stale WHERE status IN ('gone')",
            "status",
        );
        let scanned = text_values(
            &rt,
            "SELECT status FROM mvcc_idx_stale WHERE marker = 'stable'",
            "status",
        );
        assert_eq!(indexed, scanned);
        assert!(indexed.is_empty());
        clear_current_connection_id();
    }

    #[test]
    fn global_attribute_candidates_use_declared_columns_but_keep_dynamic_collections() {
        let rt = rt();
        exec(&rt, "CREATE TABLE travelers (passport TEXT, name TEXT)");
        exec(&rt, "CREATE TABLE pets (tag TEXT, name TEXT)");
        exec(&rt, "CREATE VECTOR embeddings DIM 2 METRIC cosine");
        exec(&rt, "CREATE TIMESERIES metrics RETENTION 7 d");
        exec(&rt, "CREATE QUEUE jobs");
        exec(
            &rt,
            "INSERT INTO social NODE (label, node_type, passport) \
             VALUES ('person', 'Person', 'ABC123123')",
        );
        exec(
            &rt,
            "INSERT INTO places NODE (label, node_type, name) \
             VALUES ('city', 'Place', 'Paris')",
        );

        let query = match reddb_rql::parser::parse("SELECT * WHERE passport = 'ABC123123'")
            .expect("parse")
            .query
        {
            QueryExpr::Table(query) => query,
            other => panic!("expected table query, got {other:?}"),
        };
        let filter = Filter::Compare {
            field: FieldRef::column("", "passport"),
            op: CompareOp::Eq,
            value: Value::text("ABC123123"),
        };
        let candidates =
            super::universal_candidate_collections_for_filter(&rt.db(), &query, &filter)
                .expect("plain attribute filter should produce candidates");

        assert!(candidates.iter().any(|name| name == "travelers"));
        assert!(candidates.iter().any(|name| name == "social"));
        assert!(!candidates.iter().any(|name| name == "pets"));
        assert!(!candidates.iter().any(|name| name == "places"));

        let content_fields = vec!["content".to_string()];
        assert!(super::universal_collection_may_have_fields(
            &rt.db(),
            "embeddings",
            &content_fields
        ));
        assert!(!super::universal_collection_may_have_fields(
            &rt.db(),
            "metrics",
            &content_fields
        ));

        let tags_fields = vec!["tags".to_string()];
        assert!(super::universal_collection_may_have_fields(
            &rt.db(),
            "metrics",
            &tags_fields
        ));
        assert!(!super::universal_collection_may_have_fields(
            &rt.db(),
            "embeddings",
            &tags_fields
        ));

        let payload_fields = vec!["payload".to_string()];
        assert!(super::universal_collection_may_have_fields(
            &rt.db(),
            "jobs",
            &payload_fields
        ));
        assert!(!super::universal_collection_may_have_fields(
            &rt.db(),
            "travelers",
            &payload_fields
        ));
    }
}