icydb-core 0.257.7

//! Module: query::intent::cache_key
//! Responsibility: canonical shared-cache identity normalization for structural queries.
//! Does not own: planner validation, executor runtime behavior, or SQL surface routing.
//! Boundary: turns semantic query intent into one explicit derived-hash cache key.

use crate::{
    db::{
        predicate::MissingRowPolicy,
        query::{
            builder::aggregate::AggregateExpr,
            intent::{model::QueryModel, state::GroupedIntent},
            plan::{
                AggregateSemanticKeyRef, OrderDirection, OrderSpec, PreparedQueryParameterContract,
                QueryMode,
                expr::{Expr, Function, ProjectionField, ProjectionSelection},
            },
        },
    },
    error::InternalError,
    value::{Value, hash_value},
};
use std::rc::Rc;

///
/// StructuralQueryCacheKey
///
/// Canonical semantic identity for the shared structural query-plan cache.
/// This key is intentionally explicit: normalization owns semantic equivalence,
/// while `Hash` ownership stays mechanical at the map boundary.
/// Clones share immutable content; query edits replace the memoized key rather
/// than mutating it. Accepted runtime authority stays in the session key shell.
///

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
pub(in crate::db) struct StructuralQueryCacheKey(Rc<StructuralQueryCacheKeyData>);

// One completed payload per construction, shared by memo and cache key copies.
// Keep it private and non-Clone so the handle owns all key sharing.
#[derive(Debug, Eq, Hash, PartialEq)]
struct StructuralQueryCacheKeyData {
    mode: QueryModeCacheKey,
    predicate: Option<[u8; 32]>,
    parameter_contract: Option<PreparedQueryParameterContract>,
    filter_expr: Option<ProjectionExprCacheKey>,
    order: Option<Vec<OrderTermCacheKey>>,
    distinct: bool,
    projection: ProjectionCacheKey,
    grouping: Option<GroupingCacheKey>,
    consistency: ConsistencyCacheKey,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum QueryModeCacheKey {
    Load { limit: Option<u32>, offset: u32 },
    Delete { limit: Option<u32>, offset: u32 },
}

// Value identity uses the existing canonical value hash while preserving one
// stable fallback when some nested structured value cannot hash cleanly.
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum ValueCacheKey {
    Canonical([u8; 16]),
    HashError(DiagnosticCacheKey),
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
struct DiagnosticCacheKey {
    code: u16,
    origin: u16,
}

impl DiagnosticCacheKey {
    fn from_internal_error(err: &InternalError) -> Self {
        let diagnostic = err.diagnostic();

        Self {
            code: diagnostic.error_code().raw(),
            origin: diagnostic.origin() as u16,
        }
    }
}

///
/// OrderTermCacheKey
///
/// Canonical representation of one `ORDER BY` term in the structural query
/// cache key.
/// Typed expression identity, rather than its display label, keeps cache hits
/// from crossing different operand domains or sort layouts.
///

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct OrderTermCacheKey {
    expr: ProjectionExprCacheKey,
    direction: OrderDirectionCacheKey,
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum OrderDirectionCacheKey {
    Asc,
    Desc,
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum ProjectionCacheKey {
    All,
    Fields(Vec<String>),
    // Cached projections retain output aliases as well as evaluation semantics.
    Exprs(Vec<(ProjectionExprCacheKey, Option<String>)>),
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
enum ProjectionExprCacheKey {
    Field(String),
    FieldPath {
        root: String,
        segments: Vec<String>,
    },
    Literal(ValueCacheKey),
    FunctionCall {
        function: Function,
        args: Vec<Self>,
    },
    Unary {
        op: UnaryOpCacheKey,
        expr: Box<Self>,
    },
    Case {
        when_then_arms: Vec<CaseWhenArmCacheKey>,
        else_expr: Box<Self>,
    },
    Binary {
        op: BinaryOpCacheKey,
        left: Box<Self>,
        right: Box<Self>,
    },
    Aggregate(AggregateCacheKey),
}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct CaseWhenArmCacheKey {
    condition: ProjectionExprCacheKey,
    result: ProjectionExprCacheKey,
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum BinaryOpCacheKey {
    Or,
    And,
    Eq,
    Ne,
    Lt,
    Lte,
    Gt,
    Gte,
    Add,
    Sub,
    Mul,
    Div,
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum UnaryOpCacheKey {
    Not,
}

///
/// AggregateCacheKey
///
/// Canonical aggregate identity shared by projected and grouped aggregate
/// cache entries. It records only the semantic pieces that affect planner
/// reuse.
///

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct AggregateCacheKey {
    kind_tag: u8,
    input_expr: Option<Box<ProjectionExprCacheKey>>,
    filter_expr: Option<Box<ProjectionExprCacheKey>>,
    distinct: bool,
}

///
/// GroupingCacheKey
///
/// Canonical identity for the grouped-query portion of a structural cache key.
/// This captures grouping fields, aggregate slots, grouped `HAVING`
/// expressions, and the configured grouping limits so grouped plans only reuse
/// compatible shapes.
/// This is a canonicalized grouped structural/cache identity surface;
/// prepared/template identity remains outside this key and stays syntax-bound.
///

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct GroupingCacheKey {
    // The enclosing query-plan key already carries accepted schema identity,
    // so canonical field/path identity is sufficient and cannot cross a slot remap.
    group_fields: Vec<ProjectionExprCacheKey>,
    aggregates: Vec<AggregateCacheKey>,
    having_expr: Option<ProjectionExprCacheKey>,
    max_groups: u64,
    max_group_bytes: u64,
}

#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum ConsistencyCacheKey {
    Ignore,
    Error,
}

impl StructuralQueryCacheKey {
    pub(in crate::db::query) fn from_query_model_with_normalized_predicate_fingerprint(
        model: &QueryModel,
        predicate_fingerprint: Option<[u8; 32]>,
    ) -> Self {
        Self::from_query_model_with_optional_predicate_key(model, predicate_fingerprint, None)
    }

    pub(in crate::db::query) fn from_query_model_with_parameter_contract(
        model: &QueryModel,
        parameter_contract: PreparedQueryParameterContract,
    ) -> Self {
        Self::from_query_model_with_optional_predicate_key(model, None, Some(parameter_contract))
    }

    // Build the shared structural cache key from one optional predicate-key
    // fragment so callers that already computed canonical predicate identity
    // do not walk the same normalized tree twice.
    fn from_query_model_with_optional_predicate_key(
        model: &QueryModel,
        predicate: Option<[u8; 32]>,
        parameter_contract: Option<PreparedQueryParameterContract>,
    ) -> Self {
        let scalar = model.scalar_intent_for_cache_key();
        // The fully-covered parameter contract is the semantic filter
        // authority. Equivalent IN filters may lower to different-width OR
        // expression trees, so authored expression arity must not enter the
        // reusable template identity.
        let filter_expr = if parameter_contract.is_some() {
            None
        } else {
            scalar.filter.as_ref().and_then(|filter| {
                filter
                    .logical_filter_expr()
                    .map(ProjectionExprCacheKey::from_expr)
            })
        };
        Self(Rc::new(StructuralQueryCacheKeyData {
            mode: QueryModeCacheKey::from_query_mode(model.mode()),
            // Canonical scalar `filter_expr` owns semantic filter identity when
            // present. The derived predicate key remains only for plans that
            // still have no planner-owned semantic filter expression.
            predicate: if filter_expr.is_some() {
                None
            } else {
                predicate
            },
            parameter_contract,
            filter_expr,
            order: scalar
                .order
                .as_ref()
                .map(OrderTermCacheKey::from_order_spec),
            distinct: scalar.distinct,
            projection: ProjectionCacheKey::from_projection_selection(&scalar.projection_selection),
            grouping: model
                .grouped_intent_for_cache_key()
                .map(GroupingCacheKey::from_grouped_intent),
            consistency: ConsistencyCacheKey::from_missing_row_policy(
                model.consistency_for_cache_key(),
            ),
        }))
    }
}

impl QueryModeCacheKey {
    const fn from_query_mode(mode: QueryMode) -> Self {
        match mode {
            QueryMode::Load(spec) => Self::Load {
                limit: spec.limit(),
                offset: spec.offset(),
            },
            QueryMode::Delete(spec) => Self::Delete {
                limit: spec.limit(),
                offset: spec.offset(),
            },
        }
    }
}

impl ValueCacheKey {
    fn from_value(value: &Value) -> Self {
        match hash_value(value) {
            Ok(digest) => Self::Canonical(digest),
            Err(err) => Self::HashError(DiagnosticCacheKey::from_internal_error(&err)),
        }
    }
}

impl OrderTermCacheKey {
    fn from_order_spec(order: &OrderSpec) -> Vec<Self> {
        order
            .fields
            .iter()
            .map(|term| Self {
                expr: ProjectionExprCacheKey::from_expr(term.expr()),
                direction: OrderDirectionCacheKey::from_order_direction(term.direction()),
            })
            .collect()
    }
}

impl OrderDirectionCacheKey {
    const fn from_order_direction(direction: OrderDirection) -> Self {
        match direction {
            OrderDirection::Asc => Self::Asc,
            OrderDirection::Desc => Self::Desc,
        }
    }
}

impl ProjectionCacheKey {
    fn from_projection_selection(projection: &ProjectionSelection) -> Self {
        match projection {
            ProjectionSelection::All => Self::All,
            ProjectionSelection::Fields(fields) => Self::Fields(
                fields
                    .iter()
                    .map(|field| field.as_str().to_string())
                    .collect(),
            ),
            ProjectionSelection::Exprs(fields) => Self::Exprs(
                fields
                    .iter()
                    .map(|ProjectionField::Scalar { expr, alias }| {
                        (
                            ProjectionExprCacheKey::from_expr(expr),
                            alias.as_ref().map(|alias| alias.as_str().to_string()),
                        )
                    })
                    .collect(),
            ),
        }
    }
}

impl ProjectionExprCacheKey {
    fn from_expr(expr: &Expr) -> Self {
        match expr {
            Expr::Field(field) => Self::Field(field.as_str().to_string()),
            Expr::FieldPath(path) => Self::FieldPath {
                root: path.root().as_str().to_string(),
                segments: path.segments().to_vec(),
            },
            Expr::Literal(value) => Self::Literal(ValueCacheKey::from_value(value)),
            Expr::FunctionCall { function, args } => Self::FunctionCall {
                function: *function,
                args: args.iter().map(Self::from_expr).collect(),
            },
            Expr::Unary { op, expr } => Self::Unary {
                op: UnaryOpCacheKey::from_unary_op(*op),
                expr: Box::new(Self::from_expr(expr.as_ref())),
            },
            Expr::Case {
                when_then_arms,
                else_expr,
            } => Self::Case {
                when_then_arms: when_then_arms
                    .iter()
                    .map(CaseWhenArmCacheKey::from_arm)
                    .collect(),
                else_expr: Box::new(Self::from_expr(else_expr.as_ref())),
            },
            Expr::Binary { op, left, right } => Self::Binary {
                op: BinaryOpCacheKey::from_binary_op(*op),
                left: Box::new(Self::from_expr(left.as_ref())),
                right: Box::new(Self::from_expr(right.as_ref())),
            },
            Expr::Aggregate(aggregate) => {
                Self::Aggregate(AggregateCacheKey::from_aggregate_expr(aggregate))
            }
            #[cfg(test)]
            Expr::Alias { expr, name: _ } => Self::from_expr(expr.as_ref()),
        }
    }

    fn from_group_field(field: crate::db::query::plan::GroupFieldRef<'_>) -> Self {
        match field.as_scalar_path() {
            Some(path) => Self::FieldPath {
                root: path.path().root().as_str().to_string(),
                segments: path.path().segments().to_vec(),
            },
            None => Self::Field(field.field().to_string()),
        }
    }
}

impl BinaryOpCacheKey {
    const fn from_binary_op(op: crate::db::query::plan::expr::BinaryOp) -> Self {
        match op {
            crate::db::query::plan::expr::BinaryOp::Or => Self::Or,
            crate::db::query::plan::expr::BinaryOp::And => Self::And,
            crate::db::query::plan::expr::BinaryOp::Eq => Self::Eq,
            crate::db::query::plan::expr::BinaryOp::Ne => Self::Ne,
            crate::db::query::plan::expr::BinaryOp::Lt => Self::Lt,
            crate::db::query::plan::expr::BinaryOp::Lte => Self::Lte,
            crate::db::query::plan::expr::BinaryOp::Gt => Self::Gt,
            crate::db::query::plan::expr::BinaryOp::Gte => Self::Gte,
            crate::db::query::plan::expr::BinaryOp::Add => Self::Add,
            crate::db::query::plan::expr::BinaryOp::Sub => Self::Sub,
            crate::db::query::plan::expr::BinaryOp::Mul => Self::Mul,
            crate::db::query::plan::expr::BinaryOp::Div => Self::Div,
        }
    }
}

impl UnaryOpCacheKey {
    const fn from_unary_op(op: crate::db::query::plan::expr::UnaryOp) -> Self {
        match op {
            crate::db::query::plan::expr::UnaryOp::Not => Self::Not,
        }
    }
}

impl CaseWhenArmCacheKey {
    fn from_arm(arm: &crate::db::query::plan::expr::CaseWhenArm) -> Self {
        Self {
            condition: ProjectionExprCacheKey::from_expr(arm.condition()),
            result: ProjectionExprCacheKey::from_expr(arm.result()),
        }
    }
}

impl AggregateCacheKey {
    fn from_aggregate_expr(aggregate: &AggregateExpr) -> Self {
        Self::from_semantic_key(AggregateSemanticKeyRef::from_aggregate_expr(aggregate))
    }

    fn from_group_aggregate_spec(aggregate: &crate::db::query::plan::GroupAggregateSpec) -> Self {
        Self::from_semantic_key(aggregate.semantic_key())
    }

    // Labels are presentation, not identity: Decimal(1) and U256(1) both render
    // as "1" but have different query semantics. Reuse structural expression
    // keys after the shared aggregate owner decides COUNT/DISTINCT equivalence.
    fn from_semantic_key(identity: AggregateSemanticKeyRef<'_>) -> Self {
        Self {
            kind_tag: identity.kind().fingerprint_tag(),
            input_expr: identity
                .input_expr()
                .map(|expr| Box::new(ProjectionExprCacheKey::from_expr(expr))),
            filter_expr: identity
                .filter_expr()
                .map(|expr| Box::new(ProjectionExprCacheKey::from_expr(expr))),
            distinct: identity.distinct(),
        }
    }
}

impl GroupingCacheKey {
    fn from_grouped_intent(grouped: &GroupedIntent) -> Self {
        Self {
            group_fields: grouped
                .group
                .group_fields
                .iter()
                .map(ProjectionExprCacheKey::from_group_field)
                .collect(),
            aggregates: grouped
                .group
                .aggregates
                .iter()
                .map(AggregateCacheKey::from_group_aggregate_spec)
                .collect(),
            having_expr: grouped
                .having_expr
                .as_ref()
                .map(ProjectionExprCacheKey::from_expr),
            max_groups: grouped.group.execution.max_groups,
            max_group_bytes: grouped.group.execution.max_group_bytes,
        }
    }
}

impl ConsistencyCacheKey {
    const fn from_missing_row_policy(policy: MissingRowPolicy) -> Self {
        match policy {
            MissingRowPolicy::Ignore => Self::Ignore,
            MissingRowPolicy::Error => Self::Error,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::{
        db::{
            predicate::MissingRowPolicy,
            query::{
                builder::aggregate,
                intent::StructuralQuery,
                plan::{
                    AggregateKind, GroupAggregateSpec, OrderDirection, OrderSpec, OrderTerm,
                    expr::{Alias, Expr, ProjectionField, ProjectionSelection},
                },
            },
        },
        retained::RetainedBytes,
        types::{Decimal, U256},
        value::Value,
    };

    use super::{AggregateCacheKey, OrderTermCacheKey, ProjectionCacheKey};
    use std::{
        hash::{BuildHasher, BuildHasherDefault, DefaultHasher},
        rc::Rc,
    };

    #[test]
    fn shared_structural_key_preserves_content_identity_and_retention() {
        let make_key = || {
            StructuralQuery::new(MissingRowPolicy::Ignore)
                .projection_selection(ProjectionSelection::Exprs(vec![ProjectionField::Scalar {
                    expr: Expr::Aggregate(aggregate::sum("amount".repeat(100))),
                    alias: Some(Alias::new("output".repeat(100))),
                }]))
                .structural_cache_key_with_normalized_predicate_fingerprint(None)
        };
        let key = make_key();
        let cloned = key.clone();
        // Sharing is the resource contract of key cloning, not pointer-based identity.
        assert!(Rc::ptr_eq(&key.0, &cloned.0));
        let independent = make_key();
        assert!(!Rc::ptr_eq(&key.0, &independent.0));
        assert_eq!(key, independent);
        let hash = BuildHasherDefault::<DefaultHasher>::default();
        assert_eq!(hash.hash_one(&key), hash.hash_one(&independent));
        // The handle must preserve the payload's existing content hash.
        assert_eq!(hash.hash_one(&key), hash.hash_one(key.0.as_ref()));

        let bytes = RetainedBytes::measure(&key, usize::MAX).unwrap();
        let payload_bytes = RetainedBytes::measure(key.0.as_ref(), usize::MAX).unwrap();
        assert!(bytes >= size_of_val(&key) + 2 * size_of::<usize>() + payload_bytes);
        assert_eq!(RetainedBytes::measure(&key, bytes), Some(bytes));
        assert!(RetainedBytes::measure(&key, bytes - 1).is_none());
        drop(key);
        assert_eq!(cloned, independent);
        assert_eq!(RetainedBytes::measure(&cloned, usize::MAX), Some(bytes));
    }

    #[test]
    fn projection_cache_keys_preserve_optional_alias_text() {
        let key = |alias: Option<&str>| {
            ProjectionCacheKey::from_projection_selection(&ProjectionSelection::Exprs(vec![
                ProjectionField::Scalar {
                    expr: Expr::Field("label".into()),
                    alias: alias.map(Alias::new),
                },
            ]))
        };
        let aliases = [None, Some(""), Some("first"), Some("second"), Some("FIRST")];
        for left in aliases {
            for right in aliases {
                assert_eq!(key(left) == key(right), left == right);
            }
        }
    }

    #[test]
    fn projection_cache_retention_includes_alias_backing() {
        let alias = "output".repeat(100);
        let key = ProjectionCacheKey::from_projection_selection(&ProjectionSelection::Exprs(vec![
            ProjectionField::Scalar {
                expr: Expr::Field("label".into()),
                alias: Some(Alias::new(alias.as_str())),
            },
        ]));
        let bytes = RetainedBytes::measure(&key, usize::MAX).unwrap();
        assert!(
            bytes
                >= size_of::<ProjectionCacheKey>()
                    + size_of::<(super::ProjectionExprCacheKey, Option<String>)>()
                    + "label".len()
                    + alias.len()
        );
        assert_eq!(RetainedBytes::measure(&key, bytes), Some(bytes));
        assert!(RetainedBytes::measure(&key, bytes - 1).is_none());
        let cloned = key.clone();
        assert_eq!(RetainedBytes::measure(&cloned, usize::MAX), Some(bytes));
        assert_eq!(key, cloned);
    }

    #[test]
    fn order_cache_keys_preserve_typed_operands_and_direction() {
        let key = |expr, direction| {
            OrderTermCacheKey::from_order_spec(&OrderSpec {
                fields: vec![OrderTerm::new(expr, direction)],
            })
        };
        let decimal = Expr::Literal(Value::Decimal(Decimal::from(1_u64)));
        let wide = Expr::Literal(Value::U256(U256::from(1_u64)));
        for (left, right) in [
            (decimal.clone(), wide.clone()),
            (
                Expr::Aggregate(aggregate::AggregateExpr::from_expression_input(
                    AggregateKind::Sum,
                    decimal,
                )),
                Expr::Aggregate(aggregate::AggregateExpr::from_expression_input(
                    AggregateKind::Sum,
                    wide,
                )),
            ),
        ] {
            assert_ne!(
                key(left, OrderDirection::Asc),
                key(right, OrderDirection::Asc)
            );
        }
        let field = Expr::Field("label".into());
        assert_ne!(
            key(field.clone(), OrderDirection::Asc),
            key(field, OrderDirection::Desc),
        );
        assert_eq!(
            key(Expr::Aggregate(aggregate::count()), OrderDirection::Asc),
            key(
                Expr::Aggregate(aggregate::AggregateExpr::from_expression_input(
                    AggregateKind::Count,
                    Expr::Literal(Value::Nat64(1)),
                )),
                OrderDirection::Asc,
            ),
        );
    }

    #[test]
    fn order_cache_retention_includes_nested_operands() {
        let field = "amount".repeat(100);
        let filter = "filter".repeat(100);
        let order = OrderSpec {
            fields: vec![OrderTerm::new(
                Expr::Aggregate(
                    aggregate::sum(field.clone())
                        .with_filter_expr(Expr::Field(filter.clone().into())),
                ),
                OrderDirection::Desc,
            )],
        };
        let key = OrderTermCacheKey::from_order_spec(&order);
        let bytes = RetainedBytes::measure(&key, usize::MAX).unwrap();
        assert!(
            bytes
                >= size_of_val(&key)
                    + size_of::<OrderTermCacheKey>()
                    + 2 * size_of::<super::ProjectionExprCacheKey>()
                    + field.len()
                    + filter.len()
        );
        assert_eq!(RetainedBytes::measure(&key, bytes), Some(bytes));
        assert!(RetainedBytes::measure(&key, bytes - 1).is_none());
    }

    #[test]
    fn aggregate_cache_keys_preserve_typed_inputs_and_canonical_equivalence() {
        let from_input = |kind, value| {
            aggregate::AggregateExpr::from_expression_input(kind, Expr::Literal(value))
        };
        for kind in [AggregateKind::Sum, AggregateKind::Min, AggregateKind::Max] {
            let decimal = from_input(kind, Value::Decimal(Decimal::from(1_u64)));
            let wide = from_input(kind, Value::U256(U256::from(1_u64)));
            assert_ne!(
                AggregateCacheKey::from_aggregate_expr(&decimal),
                AggregateCacheKey::from_aggregate_expr(&wide),
            );
            for aggregate in [decimal, wide] {
                assert_eq!(
                    AggregateCacheKey::from_aggregate_expr(&aggregate),
                    AggregateCacheKey::from_group_aggregate_spec(
                        &GroupAggregateSpec::from_aggregate_expr(aggregate.clone()),
                    ),
                );
            }
        }
        for value in [Value::Nat64(1), Value::U256(U256::from(1_u64))] {
            assert_eq!(
                AggregateCacheKey::from_aggregate_expr(&aggregate::count()),
                AggregateCacheKey::from_aggregate_expr(&from_input(AggregateKind::Count, value)),
            );
        }
        assert_ne!(
            AggregateCacheKey::from_aggregate_expr(
                &from_input(AggregateKind::Count, Value::Nat64(1)).distinct(),
            ),
            AggregateCacheKey::from_aggregate_expr(
                &from_input(AggregateKind::Count, Value::U256(U256::from(1_u64))).distinct(),
            ),
        );
        for aggregate in [aggregate::min_by("amount"), aggregate::max_by("amount")] {
            assert_eq!(
                AggregateCacheKey::from_aggregate_expr(&aggregate),
                AggregateCacheKey::from_aggregate_expr(&aggregate.clone().distinct()),
            );
        }
        assert_eq!(
            AggregateCacheKey::from_aggregate_expr(&from_input(
                AggregateKind::Sum,
                Value::Nat64(1)
            )),
            AggregateCacheKey::from_aggregate_expr(&from_input(
                AggregateKind::Sum,
                Value::Decimal(Decimal::from(1_u64)),
            )),
        );
    }

    #[test]
    fn aggregate_cache_retention_includes_both_structural_operands() {
        let field = "input".repeat(100);
        let filter = "filter".repeat(100);
        let aggregate =
            aggregate::sum(field.clone()).with_filter_expr(Expr::Field(filter.clone().into()));
        let key = AggregateCacheKey::from_aggregate_expr(&aggregate);
        let bytes = RetainedBytes::measure(&key, usize::MAX).unwrap();
        let minimum = size_of::<AggregateCacheKey>()
            + 2 * size_of::<super::ProjectionExprCacheKey>()
            + field.len()
            + filter.len();
        assert!(bytes >= minimum);
        assert_eq!(RetainedBytes::measure(&key, bytes), Some(bytes));
        assert!(RetainedBytes::measure(&key, bytes - 1).is_none());
        let cloned = key.clone();
        assert_eq!(RetainedBytes::measure(&cloned, usize::MAX), Some(bytes));
        assert_eq!(key, cloned);
    }

    #[test]
    fn scalar_and_grouped_aggregate_cache_identity_stays_shared() {
        let aggregate = aggregate::sum("amount")
            .with_filter_expr(Expr::Literal(Value::Bool(true)))
            .distinct();
        let grouped = GroupAggregateSpec::from_aggregate_expr(aggregate.clone());

        assert_eq!(
            AggregateCacheKey::from_aggregate_expr(&aggregate),
            AggregateCacheKey::from_group_aggregate_spec(&grouped),
        );

        let different_filter = aggregate::sum("amount")
            .with_filter_expr(Expr::Literal(Value::Bool(false)))
            .distinct();
        assert_ne!(
            AggregateCacheKey::from_aggregate_expr(&aggregate),
            AggregateCacheKey::from_aggregate_expr(&different_filter),
        );
        assert_ne!(
            AggregateCacheKey::from_aggregate_expr(&aggregate),
            AggregateCacheKey::from_aggregate_expr(&aggregate::sum("other_amount").distinct()),
        );
        assert_ne!(
            AggregateCacheKey::from_aggregate_expr(&aggregate),
            AggregateCacheKey::from_aggregate_expr(&aggregate::sum("amount")),
        );
    }
}

// Exhaustive cache-retention coverage; new owned fields require accounting.
crate::retained::retained_fields!(AggregateCacheKey {
Self{kind_tag,input_expr,filter_expr,distinct} => [kind_tag,input_expr,filter_expr,distinct],
});
crate::retained::retained_copy!(BinaryOpCacheKey);
crate::retained::retained_fields!(CaseWhenArmCacheKey {
Self{condition,result} => [condition,result],
});
crate::retained::retained_copy!(ConsistencyCacheKey);
crate::retained::retained_copy!(DiagnosticCacheKey);
crate::retained::retained_fields!(GroupingCacheKey {
Self{group_fields,aggregates,having_expr,max_groups,max_group_bytes} => [group_fields,aggregates,having_expr,max_groups,max_group_bytes],
});
crate::retained::retained_copy!(OrderDirectionCacheKey);
crate::retained::retained_fields!(OrderTermCacheKey {
Self{expr,direction} => [expr,direction],
});
crate::retained::retained_fields!(ProjectionCacheKey {
Self::All => [],
Self::Fields(field_0) => [field_0],
Self::Exprs(field_0) => [field_0],
});
crate::retained::retained_fields!(ProjectionExprCacheKey {
Self::Field(field_0) => [field_0],
Self::FieldPath{root,segments} => [root,segments],
Self::Literal(field_0) => [field_0],
Self::FunctionCall{function,args} => [function,args],
Self::Unary{op,expr} => [op,expr],
Self::Case{when_then_arms,else_expr} => [when_then_arms,else_expr],
Self::Binary{op,left,right} => [op,left,right],
Self::Aggregate(field_0) => [field_0],
});
crate::retained::retained_fields!(QueryModeCacheKey {
Self::Load{limit,offset} => [limit,offset],
Self::Delete{limit,offset} => [limit,offset],
});
crate::retained::retained_fields!(StructuralQueryCacheKey {
Self(data) => [data],
});
crate::retained::retained_fields!(StructuralQueryCacheKeyData {
Self{mode,predicate,parameter_contract,filter_expr,order,distinct,projection,grouping,consistency} => [mode,predicate,parameter_contract,filter_expr,order,distinct,projection,grouping,consistency],
});
crate::retained::retained_copy!(UnaryOpCacheKey);
crate::retained::retained_fields!(ValueCacheKey {
Self::Canonical(field_0) => [field_0],
Self::HashError(field_0) => [field_0],
});