icydb-core 0.69.9

//! Module: executor::group::key
//! Responsibility: canonical grouped/distinct key materialization and set semantics.
//! Does not own: aggregation fold logic or planner-level grouped query validation.
//! Boundary: canonical equality/hash substrate for grouped execution.

use crate::{
    db::executor::{
        aggregate::GroupError,
        group::{StableHash, stable_hash_value},
    },
    error::InternalError,
    value::{MapValueError, Value},
};
use std::{collections::BTreeMap, fmt};

///
/// KeyCanonicalError
///
/// KeyCanonicalError reports canonicalization failures while materializing one
/// grouping/distinct key from a runtime value.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) enum KeyCanonicalError {
    InvalidMapValue(MapValueError),
    HashingFailed { reason: String },
}

impl KeyCanonicalError {
    // Build the canonical grouped-key invariant for invalid map payloads.
    fn invalid_map_value(err: &MapValueError) -> InternalError {
        InternalError::executor_invariant(format!(
            "group key canonicalization rejected map value: {err}"
        ))
    }

    /// Convert one key-canonicalization failure into the executor error surface.
    pub(in crate::db) fn into_internal_error(self) -> InternalError {
        match self {
            Self::InvalidMapValue(err) => Self::invalid_map_value(&err),
            Self::HashingFailed { reason } => {
                InternalError::executor_internal(format!("group key hashing failed: {reason}"))
            }
        }
    }

    /// Convert one key-canonicalization failure into the grouped execution
    /// error surface while preserving grouped runtime ownership.
    pub(in crate::db::executor) fn into_group_error(self) -> GroupError {
        GroupError::from(self.into_internal_error())
    }
}

impl fmt::Display for KeyCanonicalError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::InvalidMapValue(err) => write!(f, "{err}"),
            Self::HashingFailed { reason } => write!(f, "{reason}"),
        }
    }
}

impl std::error::Error for KeyCanonicalError {}

///
/// CanonicalValue
///
/// CanonicalValue wraps one recursively normalized value used by grouping and
/// distinct semantics.
///

#[derive(Clone, Debug, Eq, PartialEq)]
struct CanonicalValue(Value);

///
/// GroupKey
///
/// GroupKey is the canonical equality/hash substrate for grouping and distinct
/// execution paths.
///

#[derive(Clone, Debug, Eq, PartialEq)]
pub(in crate::db) struct GroupKey {
    raw: CanonicalValue,
    hash: StableHash,
}

/// Compare two grouped keys with canonical grouped-equality semantics.
#[must_use]
pub(in crate::db) fn canonical_group_key_equals(left: &GroupKey, right: &GroupKey) -> bool {
    left == right
}

impl GroupKey {
    fn from_raw(raw: Value) -> Result<Self, KeyCanonicalError> {
        let hash = stable_hash_value(&raw).map_err(|err| KeyCanonicalError::HashingFailed {
            reason: err.display_with_class(),
        })?;

        Ok(Self {
            raw: CanonicalValue(raw),
            hash,
        })
    }

    #[must_use]
    pub(in crate::db) const fn hash(&self) -> StableHash {
        self.hash
    }

    #[must_use]
    pub(in crate::db) const fn canonical_value(&self) -> &Value {
        &self.raw.0
    }

    // Consume one grouped key and return the owned canonical grouped value
    // so grouped fast paths can keep moving owned key payloads without clones.
    pub(in crate::db::executor) fn into_canonical_value(self) -> Value {
        self.raw.0
    }

    // Materialize one grouped key from owned grouped slot values without
    // cloning them back through the borrowed canonicalization path.
    pub(in crate::db::executor) fn from_group_values(
        group_values: Vec<Value>,
    ) -> Result<Self, KeyCanonicalError> {
        let canonical = canonicalize_owned_value(Value::List(group_values))?;

        Self::from_raw(canonical)
    }

    #[cfg(test)]
    #[must_use]
    pub(in crate::db) const fn raw(&self) -> &Value {
        &self.raw.0
    }
}

///
/// CanonicalKey
///
/// CanonicalKey materializes one opaque canonical grouping key from a value.
///

pub(in crate::db) trait CanonicalKey {
    /// Materialize one canonical grouped key from this value.
    fn canonical_key(&self) -> Result<GroupKey, KeyCanonicalError>;
}

impl CanonicalKey for Value {
    fn canonical_key(&self) -> Result<GroupKey, KeyCanonicalError> {
        let canonical = canonicalize_value(self)?;
        GroupKey::from_raw(canonical)
    }
}

impl CanonicalKey for &Value {
    fn canonical_key(&self) -> Result<GroupKey, KeyCanonicalError> {
        (*self).canonical_key()
    }
}

///
/// GroupKeySet
///
/// GroupKeySet tracks canonical distinct keys by stable-hash bucket while
/// preserving canonical-value equality checks inside each bucket.
///

#[derive(Debug)]
pub(in crate::db) struct GroupKeySet {
    buckets: BTreeMap<StableHash, Vec<GroupKey>>,
}

impl GroupKeySet {
    /// Construct one empty canonical grouped-key set.
    #[must_use]
    pub(in crate::db) const fn new() -> Self {
        Self {
            buckets: BTreeMap::new(),
        }
    }

    /// Return true when this canonical key is already present.
    #[must_use]
    pub(in crate::db) fn contains_key(&self, key: &GroupKey) -> bool {
        self.buckets.get(&key.hash()).is_some_and(|bucket| {
            bucket
                .iter()
                .any(|existing| canonical_group_key_equals(existing, key))
        })
    }

    /// Return the total number of canonical keys tracked by this set.
    #[must_use]
    pub(in crate::db) fn len(&self) -> usize {
        self.buckets
            .values()
            .fold(0usize, |count, bucket| count.saturating_add(bucket.len()))
    }

    /// Insert one canonical key and return true if it was newly observed.
    pub(in crate::db) fn insert_key(&mut self, key: GroupKey) -> bool {
        let bucket = self.buckets.entry(key.hash()).or_default();
        if bucket
            .iter()
            .any(|existing| canonical_group_key_equals(existing, &key))
        {
            return false;
        }

        bucket.push(key);
        true
    }

    /// Canonicalize+insert one raw value and return true when it is new.
    pub(in crate::db) fn insert_value(&mut self, value: &Value) -> Result<bool, KeyCanonicalError> {
        let key = value.canonical_key()?;
        Ok(self.insert_key(key))
    }
}

impl Default for GroupKeySet {
    fn default() -> Self {
        Self::new()
    }
}

// Canonicalize one runtime value into grouped-key equality form.
fn canonicalize_value(value: &Value) -> Result<Value, KeyCanonicalError> {
    match value {
        Value::Decimal(decimal) => Ok(Value::Decimal(decimal.normalize())),
        Value::List(items) => items
            .iter()
            .map(canonicalize_value)
            .collect::<Result<Vec<_>, _>>()
            .map(Value::List),
        Value::Map(entries) => canonicalize_map_entries(entries),
        _ => Ok(value.clone()),
    }
}

// Canonicalize map entries recursively and normalize key ordering.
fn canonicalize_map_entries(entries: &[(Value, Value)]) -> Result<Value, KeyCanonicalError> {
    let mut canonical_entries = Vec::with_capacity(entries.len());
    for (key, value) in entries {
        canonical_entries.push((canonicalize_value(key)?, canonicalize_value(value)?));
    }

    let normalized = Value::normalize_map_entries(canonical_entries)
        .map_err(KeyCanonicalError::InvalidMapValue)?;

    Ok(Value::Map(normalized))
}

// Canonicalize one owned runtime value into grouped-key equality form while
// preserving ownership of already-materialized grouped slot payloads.
fn canonicalize_owned_value(value: Value) -> Result<Value, KeyCanonicalError> {
    match value {
        Value::Decimal(decimal) => Ok(Value::Decimal(decimal.normalize())),
        Value::List(items) => items
            .into_iter()
            .map(canonicalize_owned_value)
            .collect::<Result<Vec<_>, _>>()
            .map(Value::List),
        Value::Map(entries) => canonicalize_owned_map_entries(entries),
        value => Ok(value),
    }
}

// Canonicalize one owned map payload recursively while preserving stable
// grouped-key map normalization.
fn canonicalize_owned_map_entries(
    entries: Vec<(Value, Value)>,
) -> Result<Value, KeyCanonicalError> {
    let mut canonical_entries = Vec::with_capacity(entries.len());
    for (key, value) in entries {
        canonical_entries.push((
            canonicalize_owned_value(key)?,
            canonicalize_owned_value(value)?,
        ));
    }

    let normalized = Value::normalize_map_entries(canonical_entries)
        .map_err(KeyCanonicalError::InvalidMapValue)?;

    Ok(Value::Map(normalized))
}

///
/// TESTS
///

#[cfg(test)]
mod tests {
    use crate::{
        db::executor::group::{
            CanonicalKey, GroupKey, GroupKeySet, KeyCanonicalError, canonical_group_key_equals,
        },
        types::Decimal,
        value::{MapValueError, Value, with_test_hash_override},
    };

    fn map_value(entries: Vec<(Value, Value)>) -> Value {
        Value::Map(entries)
    }

    #[test]
    fn canonical_key_normalizes_decimal_scale() {
        let key = Value::Decimal(Decimal::new(100, 2))
            .canonical_key()
            .expect("canonical key");

        let Value::Decimal(normalized) = key.raw() else {
            panic!("canonical decimal value expected");
        };
        assert_eq!(normalized.scale(), 0);
    }

    #[test]
    fn canonical_key_normalizes_map_order() {
        let left = map_value(vec![
            (Value::Text("z".to_string()), Value::Uint(9)),
            (Value::Text("a".to_string()), Value::Uint(1)),
        ]);
        let right = map_value(vec![
            (Value::Text("a".to_string()), Value::Uint(1)),
            (Value::Text("z".to_string()), Value::Uint(9)),
        ]);

        let left_key = left.canonical_key().expect("left canonical key");
        let right_key = right.canonical_key().expect("right canonical key");

        assert_eq!(left_key, right_key);
        assert_eq!(left_key.hash(), right_key.hash());
    }

    #[test]
    fn canonical_key_rejects_duplicate_map_keys_after_normalization() {
        let value = map_value(vec![
            (Value::Text("a".to_string()), Value::Uint(1)),
            (Value::Text("a".to_string()), Value::Uint(2)),
        ]);

        let err = value
            .canonical_key()
            .expect_err("duplicate map keys should fail");
        assert!(matches!(
            err,
            KeyCanonicalError::InvalidMapValue(MapValueError::DuplicateKey { .. })
        ));
    }

    #[test]
    fn group_key_set_deduplicates_canonical_equivalents() {
        let mut set = GroupKeySet::default();
        let first = Value::Decimal(Decimal::new(100, 2));
        let second = Value::Decimal(Decimal::new(1, 0));

        assert!(
            set.insert_value(&first).expect("insert"),
            "first insert should be new"
        );
        assert!(
            !set.insert_value(&second).expect("insert"),
            "second insert should be deduplicated by canonical key equality"
        );
    }

    #[test]
    fn canonical_equal_keys_always_share_stable_hash() {
        let equivalent_pairs = vec![
            (
                Value::Decimal(Decimal::new(1000, 3)),
                Value::Decimal(Decimal::new(1, 0)),
            ),
            (
                Value::Map(vec![
                    (Value::Text("z".to_string()), Value::Uint(9)),
                    (Value::Text("a".to_string()), Value::Uint(1)),
                ]),
                Value::Map(vec![
                    (Value::Text("a".to_string()), Value::Uint(1)),
                    (Value::Text("z".to_string()), Value::Uint(9)),
                ]),
            ),
            (
                Value::List(vec![Value::Decimal(Decimal::new(10, 1)), Value::Uint(4)]),
                Value::List(vec![Value::Decimal(Decimal::new(1, 0)), Value::Uint(4)]),
            ),
            (
                Value::List(vec![
                    Value::Map(vec![
                        (Value::Text("z".to_string()), Value::Uint(9)),
                        (Value::Text("a".to_string()), Value::Uint(1)),
                    ]),
                    Value::Decimal(Decimal::new(2500, 2)),
                ]),
                Value::List(vec![
                    Value::Map(vec![
                        (Value::Text("a".to_string()), Value::Uint(1)),
                        (Value::Text("z".to_string()), Value::Uint(9)),
                    ]),
                    Value::Decimal(Decimal::new(25, 0)),
                ]),
            ),
        ];

        for (left_value, right_value) in equivalent_pairs {
            let left_key = left_value.canonical_key().expect("left canonical key");
            let right_key = right_value.canonical_key().expect("right canonical key");
            assert!(
                canonical_group_key_equals(&left_key, &right_key),
                "pair should be canonical-equal under group key contract",
            );
            assert_eq!(
                left_key.hash(),
                right_key.hash(),
                "canonical-equal keys must hash to the same stable hash",
            );
        }
    }

    #[test]
    fn group_key_set_handles_hash_collisions_with_equality_check() {
        with_test_hash_override([0xAB; 16], || {
            let mut set = GroupKeySet::default();
            let first = Value::Text("alpha".to_string())
                .canonical_key()
                .expect("first canonical key");
            let second = Value::Text("beta".to_string())
                .canonical_key()
                .expect("second canonical key");

            assert_eq!(
                first.hash(),
                second.hash(),
                "test setup requires an artificial hash collision",
            );
            assert!(
                !canonical_group_key_equals(&first, &second),
                "collision pair must remain distinct by canonical equality",
            );
            assert!(
                set.insert_key(first.clone()),
                "first colliding key should insert as new",
            );
            assert!(
                set.insert_key(second.clone()),
                "second colliding key must not be dropped on hash match alone",
            );
            assert!(
                !set.insert_key(first),
                "re-inserting first key should dedupe by canonical equality",
            );
            assert!(
                !set.insert_key(second),
                "re-inserting second key should dedupe by canonical equality",
            );
        });
    }

    #[test]
    fn group_key_from_group_values_matches_borrowed_canonical_key_path() {
        let group_values = vec![
            Value::Decimal(Decimal::new(100, 2)),
            Value::Text("alpha".to_string()),
            map_value(vec![(Value::Text("z".to_string()), Value::Uint(9))]),
        ];
        let borrowed = Value::List(group_values.clone())
            .canonical_key()
            .expect("borrowed canonical key");
        let owned = GroupKey::from_group_values(group_values).expect("owned canonical key");

        assert_eq!(borrowed, owned);
        assert_eq!(borrowed.hash(), owned.hash());
    }
}