linkrs-query 0.1.0

//! Implementation of container operation functions
//!
//! Provide functions for operating on lists and maps, including head, last, tail, size, range, and keys.

use crate::executor::expression::ExpressionError;
use linkrs_core::value::list::List;
use linkrs_core::value::NullType;
use linkrs_core::Value;
use std::collections::BTreeSet;

/// Container function enumeration
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ContainerFunction {
    Head,
    Last,
    Tail,
    Size,
    Range,
    Keys,
    ReverseList,
    ToSet,
    ListContains,
    ListAppend,
    ListPrepend,
    ListFilter,
    ListTransform,
    ListConcat,
    ListSort,
    ListSlice,
    ListToString,
    ListDistinct,
    ListUnique,
    ListExtract,
    ListAny,
    ListAll,
    ListSingle,
    ListReduce,
    StructPack,
    StructExtract,
    MapCreation,
    MapExtract,
    ElementAt,
    Cardinality,
    MapKeys,
    MapValues,
}

impl ContainerFunction {
    /// Obtain the function name
    pub fn name(&self) -> &str {
        match self {
            Self::Head => "head",
            Self::Last => "last",
            Self::Tail => "tail",
            Self::Size => "size",
            Self::Range => "range",
            Self::Keys => "keys",
            Self::ReverseList => "reverse",
            Self::ToSet => "toset",
            Self::ListContains => "list_contains",
            Self::ListAppend => "list_append",
            Self::ListPrepend => "list_prepend",
            Self::ListFilter => "list_filter",
            Self::ListTransform => "list_transform",
            Self::ListConcat => "list_concat",
            Self::ListSort => "list_sort",
            Self::ListSlice => "list_slice",
            Self::ListToString => "list_to_string",
            Self::ListDistinct => "list_distinct",
            Self::ListUnique => "list_unique",
            Self::ListExtract => "list_extract",
            Self::ListAny => "list_any",
            Self::ListAll => "list_all",
            Self::ListSingle => "list_single",
            Self::ListReduce => "list_reduce",
            Self::StructPack => "struct_pack",
            Self::StructExtract => "struct_extract",
            Self::MapCreation => "map",
            Self::MapExtract => "map_extract",
            Self::ElementAt => "element_at",
            Self::Cardinality => "cardinality",
            Self::MapKeys => "map_keys",
            Self::MapValues => "map_values",
        }
    }

    /// Determine the number of parameters
    pub fn arity(&self) -> usize {
        match self {
            Self::Head => 1,
            Self::Last => 1,
            Self::Tail => 1,
            Self::Size => 1,
            Self::Range => 2,
            Self::Keys => 1,
            Self::ReverseList => 1,
            Self::ToSet => 1,
            Self::ListContains => 2,
            Self::ListAppend => 2,
            Self::ListPrepend => 2,
            Self::ListFilter => 2,
            Self::ListTransform => 2,
            Self::ListConcat => 2,
            Self::ListSort => 1,
            Self::ListSlice => 3,
            Self::ListToString => 2,
            Self::ListDistinct => 1,
            Self::ListUnique => 1,
            Self::ListExtract => 2,
            Self::ListAny => 2,
            Self::ListAll => 2,
            Self::ListSingle => 2,
            Self::ListReduce => 3,
            Self::StructPack => 0,
            Self::StructExtract => 2,
            Self::MapCreation => 0,
            Self::MapExtract => 2,
            Self::ElementAt => 2,
            Self::Cardinality => 1,
            Self::MapKeys => 1,
            Self::MapValues => 1,
        }
    }

    /// Is it a function with variable parameters?
    pub fn is_variadic(&self) -> bool {
        matches!(
            self,
            Self::Range | Self::ListConcat | Self::StructPack | Self::MapCreation
        )
    }

    /// Obtain the function description
    pub fn description(&self) -> &str {
        match self {
            Self::Head => "Get the first element of the list",
            Self::Last => "Get the last element of the list",
            Self::Tail => "Get all elements of the list except the first one",
            Self::Size => "Get the size of a string, list, map, or set",
            Self::Range => "Generate a list of integer ranges",
            Self::Keys => "Get all the keys for vertices, edges, or mappings.",
            Self::ReverseList => "inversion list",
            Self::ToSet => "Convert the list to a set.",
            Self::ListContains => "Check if list contains a value",
            Self::ListAppend => "Append element to list",
            Self::ListPrepend => "Prepend element to list",
            Self::ListFilter => "Filter list elements by predicate",
            Self::ListTransform => "Transform list elements by function",
            Self::ListConcat => "Concatenate multiple lists into one",
            Self::ListSort => "Sort list elements",
            Self::ListSlice => "Extract a slice from a list",
            Self::ListToString => "Convert a list to a string with delimiter",
            Self::ListDistinct => "Remove duplicate elements from list",
            Self::ListUnique => "Remove duplicate elements from list",
            Self::ListExtract => "Extract element from list at specified index",
            Self::ListAny => "Check if any element in mask list is true",
            Self::ListAll => "Check if all elements in mask list are true",
            Self::ListSingle => "Check if exactly one element in mask list is true",
            Self::ListReduce => "Reduce list to a single value using accumulator",
            Self::StructPack => "Pack values into a struct",
            Self::StructExtract => "Extract field from struct",
            Self::MapCreation => "Create a map from key-value pairs",
            Self::MapExtract => "Extract value from map by key",
            Self::ElementAt => "Get element at index from list or map",
            Self::Cardinality => "Get number of elements in map or set",
            Self::MapKeys => "Get all keys from a map",
            Self::MapValues => "Get all values from a map",
        }
    }

    pub fn execute(&self, args: &[Value]) -> Result<Value, ExpressionError> {
        if !self.is_variadic() && args.len() != self.arity() {
            return Err(ExpressionError::invalid_arity(
                self.name(),
                self.arity(),
                args.len(),
            ));
        }
        match self {
            Self::Head => execute_head(args),
            Self::Last => execute_last(args),
            Self::Tail => execute_tail(args),
            Self::Size => execute_size(args),
            Self::Range => execute_range(args),
            Self::Keys => execute_keys(args),
            Self::ReverseList => execute_reverse_list(args),
            Self::ToSet => execute_toset(args),
            Self::ListContains => execute_list_contains(args),
            Self::ListAppend => execute_list_append(args),
            Self::ListPrepend => execute_list_prepend(args),
            Self::ListFilter => execute_list_filter(args),
            Self::ListTransform => execute_list_transform(args),
            Self::ListConcat => execute_list_concat(args),
            Self::ListSort => execute_list_sort(args),
            Self::ListSlice => execute_list_slice(args),
            Self::ListToString => execute_list_to_string(args),
            Self::ListDistinct => execute_list_distinct(args),
            Self::ListUnique => execute_list_distinct(args),
            Self::ListExtract => execute_list_extract(args),
            Self::ListAny => execute_list_any(args),
            Self::ListAll => execute_list_all(args),
            Self::ListSingle => execute_list_single(args),
            Self::ListReduce => execute_list_reduce(args),
            Self::StructPack => execute_struct_pack(args),
            Self::StructExtract => execute_struct_extract(args),
            Self::MapCreation => execute_map_creation(args),
            Self::MapExtract => execute_map_extract(args),
            Self::ElementAt => execute_element_at(args),
            Self::Cardinality => execute_cardinality(args),
            Self::MapKeys => execute_map_keys(args),
            Self::MapValues => execute_map_values(args),
        }
    }
}

fn execute_head(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => Ok(list
            .values
            .first()
            .cloned()
            .unwrap_or(Value::Null(NullType::Null))),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("head requires a list type")),
    }
}

fn execute_last(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => Ok(list
            .values
            .last()
            .cloned()
            .unwrap_or(Value::Null(NullType::Null))),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("last requires a list type")),
    }
}

fn execute_tail(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            if list.values.is_empty() {
                Ok(Value::list(List { values: vec![] }))
            } else {
                Ok(Value::list(List {
                    values: list.values[1..].to_vec(),
                }))
            }
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("tail requires a list type")),
    }
}

fn execute_size(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::String(s) => Ok(Value::BigInt(s.chars().count() as i64)),
        Value::List(list) => Ok(Value::BigInt(list.values.len() as i64)),
        Value::Map(map) => Ok(Value::BigInt(map.len() as i64)),
        Value::Set(set) => Ok(Value::BigInt(set.len() as i64)),
        Value::Json(j) => {
            if let Ok(v) = j.to_value() {
                match v {
                    serde_json::Value::Object(m) => Ok(Value::BigInt(m.len() as i64)),
                    serde_json::Value::Array(a) => Ok(Value::BigInt(a.len() as i64)),
                    _ => Ok(Value::BigInt(1)),
                }
            } else {
                Ok(Value::Null(NullType::Null))
            }
        }
        Value::JsonB(j) => match j.as_value() {
            serde_json::Value::Object(m) => Ok(Value::BigInt(m.len() as i64)),
            serde_json::Value::Array(a) => Ok(Value::BigInt(a.len() as i64)),
            _ => Ok(Value::BigInt(1)),
        },
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "size requires string, list, map, set, or json type",
        )),
    }
}

fn execute_range(args: &[Value]) -> Result<Value, ExpressionError> {
    if args.len() < 2 || args.len() > 3 {
        return Err(ExpressionError::type_error(
            "range requires 2 or 3 arguments",
        ));
    }
    let to_i64 = |value: &Value| -> Option<i64> {
        match value {
            Value::SmallInt(i) => Some(*i as i64),
            Value::Int(i) => Some(*i as i64),
            Value::BigInt(i) => Some(*i),
            _ => None,
        }
    };
    let start = match to_i64(&args[0]) {
        Some(v) => v,
        None if args[0].is_null() => return Ok(Value::Null(NullType::Null)),
        None => {
            return Err(ExpressionError::type_error(
                "range requires integer arguments",
            ))
        }
    };
    let end = match to_i64(&args[1]) {
        Some(v) => v,
        None if args[1].is_null() => return Ok(Value::Null(NullType::Null)),
        None => {
            return Err(ExpressionError::type_error(
                "The range function takes integer arguments",
            ))
        }
    };
    let step = if args.len() > 2 {
        match to_i64(&args[2]) {
            Some(v) => v,
            None if args[2].is_null() => return Ok(Value::Null(NullType::Null)),
            None => return Err(ExpressionError::type_error("range step must be an integer")),
        }
    } else {
        1
    };

    if step == 0 {
        return Err(ExpressionError::new(
            crate::executor::expression::ExpressionErrorType::InvalidOperation,
            "range step cannot be 0".to_string(),
        ));
    }

    let mut result = Vec::new();
    if step > 0 {
        let mut i = start;
        while i <= end {
            result.push(Value::BigInt(i));
            i += step;
        }
    } else {
        let mut i = start;
        while i >= end {
            result.push(Value::BigInt(i));
            i += step;
        }
    }

    Ok(Value::list(List { values: result }))
}

fn execute_keys(args: &[Value]) -> Result<Value, ExpressionError> {
    let mut keys: BTreeSet<String> = BTreeSet::new();

    match &args[0] {
        Value::Vertex(v) => {
            for key in v.tag.properties.keys() {
                keys.insert(key.to_string());
            }
        }
        Value::Edge(e) => {
            for key in e.props.keys() {
                keys.insert(key.to_string());
            }
        }
        Value::Map(m) => {
            for key in m.keys() {
                // Map keys are generalized Values; stringify for the keys()
                // result list.
                keys.insert(format!("{}", key));
            }
        }
        Value::Json(j) => {
            if let Ok(serde_json::Value::Object(m)) = j.to_value() {
                for key in m.keys() {
                    keys.insert(key.to_string());
                }
            }
        }
        Value::JsonB(j) => {
            if let serde_json::Value::Object(m) = j.as_value() {
                for key in m.keys() {
                    keys.insert(key.to_string());
                }
            }
        }
        Value::Null(_) => return Ok(Value::Null(NullType::Null)),
        _ => {
            return Err(ExpressionError::type_error(
                "keys requires vertex, edge, map or json type",
            ))
        }
    }

    let result: Vec<Value> = keys.into_iter().map(Value::string).collect();
    Ok(Value::list(List { values: result }))
}

fn execute_reverse_list(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let mut reversed = list.values.clone();
            reversed.reverse();
            Ok(Value::list(List { values: reversed }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("reverse requires a list type")),
    }
}

fn execute_toset(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let set: std::collections::HashSet<Value> = list.values.iter().cloned().collect();
            Ok(Value::set(set))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("toset requires a list type")),
    }
}

fn execute_list_contains(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), value) => {
            let contains = list.values.iter().any(|v| v == value);
            Ok(Value::Bool(contains))
        }
        (Value::Null(_), _) | (_, Value::Null(_)) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_contains requires a list as first argument",
        )),
    }
}

fn execute_list_append(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let mut new_values = list.values.clone();
            new_values.push(args[1].clone());
            Ok(Value::list(List { values: new_values }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_append requires a list as first argument",
        )),
    }
}

fn execute_list_prepend(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let mut new_values = vec![args[1].clone()];
            new_values.extend(list.values.clone());
            Ok(Value::list(List { values: new_values }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_prepend requires a list as first argument",
        )),
    }
}

fn execute_list_filter(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::List(mask)) => {
            let min_len = list.values.len().min(mask.values.len());
            let filtered: Vec<Value> = list.values[..min_len]
                .iter()
                .zip(&mask.values[..min_len])
                .filter(|(_, m)| match m {
                    Value::Bool(b) => *b,
                    Value::Null(_) => false,
                    _ => true,
                })
                .map(|(v, _)| v.clone())
                .collect();
            Ok(Value::list(List { values: filtered }))
        }
        (Value::List(_), _) => Err(ExpressionError::type_error(
            "list_filter second argument must be a boolean list",
        )),
        (Value::Null(_), _) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_filter requires a list as first argument",
        )),
    }
}

fn execute_list_transform(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::List(replacements)) => {
            let mut result = list.values.clone();
            for (item, replacement) in result.iter_mut().zip(replacements.values.iter()) {
                if !replacement.is_null() {
                    *item = replacement.clone();
                }
            }
            Ok(Value::list(List { values: result }))
        }
        (Value::List(_), _) => Err(ExpressionError::type_error(
            "list_transform second argument must be a list",
        )),
        (Value::Null(_), _) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_transform requires lists as arguments",
        )),
    }
}

fn execute_list_concat(args: &[Value]) -> Result<Value, ExpressionError> {
    if args.len() < 2 {
        return Err(ExpressionError::type_error(
            "list_concat requires at least 2 arguments",
        ));
    }
    let mut result = Vec::new();
    for arg in args {
        match arg {
            Value::List(list) => result.extend(list.values.clone()),
            Value::Null(_) => return Ok(Value::Null(NullType::Null)),
            _ => {
                return Err(ExpressionError::type_error(
                    "list_concat requires list arguments",
                ))
            }
        }
    }
    Ok(Value::list(List { values: result }))
}

fn execute_list_sort(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let mut sorted = list.values.clone();
            sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
            Ok(Value::list(List { values: sorted }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_sort requires a list type",
        )),
    }
}

fn execute_list_slice(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1], &args[2]) {
        (Value::List(list), Value::Int(start), Value::Int(end)) => {
            let len = list.values.len() as i32;
            let start = if *start < 0 {
                (len + *start).max(0)
            } else {
                (*start).min(len)
            } as usize;
            let end = if *end < 0 {
                (len + *end).max(0)
            } else {
                (*end).min(len)
            } as usize;
            if start >= end {
                Ok(Value::list(List { values: vec![] }))
            } else {
                Ok(Value::list(List {
                    values: list.values[start..end].to_vec(),
                }))
            }
        }
        (Value::Null(_), _, _) | (_, Value::Null(_), _) | (_, _, Value::Null(_)) => {
            Ok(Value::Null(NullType::Null))
        }
        _ => Err(ExpressionError::type_error(
            "list_slice requires list, integer, and integer arguments",
        )),
    }
}

fn execute_list_to_string(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::String(delimiter)) => {
            let strings: Vec<String> = list.values.iter().map(value_to_string).collect();
            Ok(Value::string(strings.join(delimiter)))
        }
        (Value::Null(_), _) | (_, Value::Null(_)) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_to_string requires list and string arguments",
        )),
    }
}

fn execute_list_distinct(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::List(list) => {
            let mut seen = std::collections::HashSet::new();
            let mut result = Vec::new();
            for item in &list.values {
                if seen.insert(item.clone()) {
                    result.push(item.clone());
                }
            }
            Ok(Value::list(List { values: result }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_distinct requires a list type",
        )),
    }
}

fn execute_list_extract(args: &[Value]) -> Result<Value, ExpressionError> {
    let len = match &args[0] {
        Value::List(list) => list.values.len() as i64,
        Value::Null(_) => return Ok(Value::Null(NullType::Null)),
        _ => {
            return Err(ExpressionError::type_error(
                "list_extract requires a list as first argument",
            ))
        }
    };
    let idx = match &args[1] {
        Value::Int(i) => {
            let i = *i as i64;
            if i < 0 {
                len + i
            } else {
                i
            }
        }
        Value::BigInt(i) => {
            let i = *i;
            if i < 0 {
                len + i
            } else {
                i
            }
        }
        Value::Null(_) => return Ok(Value::Null(NullType::Null)),
        _ => {
            return Err(ExpressionError::type_error(
                "list_extract requires an integer as second argument",
            ))
        }
    };
    let list = match &args[0] {
        Value::List(list) => list,
        _ => unreachable!(),
    };
    if idx < 0 || idx >= len {
        Ok(Value::Null(NullType::Null))
    } else {
        Ok(list.values[idx as usize].clone())
    }
}

fn execute_list_any(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::List(mask)) => {
            let min_len = list.values.len().min(mask.values.len());
            let any_true = mask.values[..min_len]
                .iter()
                .any(|m| matches!(m, Value::Bool(true)));
            Ok(Value::Bool(any_true))
        }
        (Value::List(_), _) => Err(ExpressionError::type_error(
            "list_any second argument must be a boolean list",
        )),
        (Value::Null(_), _) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_any requires a list as first argument",
        )),
    }
}

fn execute_list_all(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::List(mask)) => {
            let min_len = list.values.len().min(mask.values.len());
            if min_len == 0 {
                return Ok(Value::Bool(true));
            }
            let all_true = mask.values[..min_len]
                .iter()
                .all(|m| matches!(m, Value::Bool(true)));
            Ok(Value::Bool(all_true))
        }
        (Value::List(_), _) => Err(ExpressionError::type_error(
            "list_all second argument must be a boolean list",
        )),
        (Value::Null(_), _) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_all requires a list as first argument",
        )),
    }
}

fn execute_list_single(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::List(mask)) => {
            let min_len = list.values.len().min(mask.values.len());
            let count = mask.values[..min_len]
                .iter()
                .filter(|m| matches!(m, Value::Bool(true)))
                .count();
            Ok(Value::Bool(count == 1))
        }
        (Value::List(_), _) => Err(ExpressionError::type_error(
            "list_single second argument must be a boolean list",
        )),
        (Value::Null(_), _) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "list_single requires a list as first argument",
        )),
    }
}

fn execute_list_reduce(args: &[Value]) -> Result<Value, ExpressionError> {
    // Lambda evaluation lives in the expression evaluator, which binds
    // (acc, item) per element and runs the lambda body. Eager values here
    // can never represent a lambda, so any direct registry call is a
    // non-lambda misuse rather than a second reduce implementation.
    if matches!(&args[0], Value::Null(_)) {
        return Ok(Value::Null(NullType::Null));
    }
    Err(ExpressionError::type_error(
        "list_reduce requires (source, lambda, initial) with a lambda as second argument",
    ))
}

fn value_to_string(v: &Value) -> String {
    match v {
        Value::Null(_) => String::new(),
        Value::Bool(b) => b.to_string(),
        Value::SmallInt(i) => i.to_string(),
        Value::Int(i) => i.to_string(),
        Value::BigInt(i) => i.to_string(),
        Value::Float(f) => f.to_string(),
        Value::Double(f) => f.to_string(),
        Value::String(s) => s.to_string(),
        _ => format!("{:?}", v),
    }
}

fn execute_struct_pack(args: &[Value]) -> Result<Value, ExpressionError> {
    if !args.len().is_multiple_of(2) {
        return Err(ExpressionError::type_error(
            "struct_pack requires an even number of arguments (key-value pairs)",
        ));
    }
    let mut fields = std::collections::HashMap::new();
    let mut i = 0;
    while i < args.len() {
        let key = match &args[i] {
            Value::String(s) => s.to_string(),
            Value::Null(_) => return Ok(Value::Null(NullType::Null)),
            _ => {
                return Err(ExpressionError::type_error(
                    "struct_pack keys must be strings",
                ))
            }
        };
        fields.insert(key, args[i + 1].clone());
        i += 2;
    }
    Ok(Value::string_map(fields))
}

fn execute_struct_extract(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::Map(map), Value::String(key)) => {
            let key_val = Value::string(key);
            Ok(map
                .get(&key_val)
                .cloned()
                .unwrap_or(Value::Null(NullType::Null)))
        }
        (Value::Null(_), _) | (_, Value::Null(_)) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "struct_extract requires a map and a string key",
        )),
    }
}

fn execute_map_creation(args: &[Value]) -> Result<Value, ExpressionError> {
    if !args.len().is_multiple_of(2) {
        return Err(ExpressionError::type_error(
            "map requires an even number of arguments (key-value pairs)",
        ));
    }
    let mut fields = std::collections::HashMap::new();
    let mut i = 0;
    while i < args.len() {
        let key = match &args[i] {
            Value::String(s) => s.to_string(),
            Value::Null(_) => return Ok(Value::Null(NullType::Null)),
            _ => return Err(ExpressionError::type_error("map keys must be strings")),
        };
        fields.insert(key, args[i + 1].clone());
        i += 2;
    }
    Ok(Value::string_map(fields))
}

fn execute_map_extract(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::Map(map), Value::String(key)) => {
            let key_val = Value::string(key);
            Ok(map
                .get(&key_val)
                .cloned()
                .unwrap_or(Value::Null(NullType::Null)))
        }
        (Value::Null(_), _) | (_, Value::Null(_)) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "map_extract requires a map and a string key",
        )),
    }
}

fn execute_element_at(args: &[Value]) -> Result<Value, ExpressionError> {
    match (&args[0], &args[1]) {
        (Value::List(list), Value::Int(idx)) => {
            let idx = *idx;
            let len = list.values.len() as i32;
            let actual_idx = if idx < 0 { len + idx } else { idx };
            if actual_idx < 0 || actual_idx >= len {
                Ok(Value::Null(NullType::Null))
            } else {
                Ok(list.values[actual_idx as usize].clone())
            }
        }
        (Value::List(list), Value::BigInt(idx)) => {
            let idx = *idx;
            let len = list.values.len() as i64;
            let actual_idx = if idx < 0 { len + idx } else { idx };
            if actual_idx < 0 || actual_idx >= len {
                Ok(Value::Null(NullType::Null))
            } else {
                Ok(list.values[actual_idx as usize].clone())
            }
        }
        (Value::Map(map), Value::String(key)) => {
            let key_val = Value::string(key);
            Ok(map
                .get(&key_val)
                .cloned()
                .unwrap_or(Value::Null(NullType::Null)))
        }
        (Value::Null(_), _) | (_, Value::Null(_)) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "element_at requires a list or map as first argument",
        )),
    }
}

fn execute_cardinality(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Map(map) => Ok(Value::BigInt(map.len() as i64)),
        Value::Set(set) => Ok(Value::BigInt(set.len() as i64)),
        Value::List(list) => Ok(Value::BigInt(list.values.len() as i64)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "cardinality requires a map, set, or list type",
        )),
    }
}

fn execute_map_keys(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Map(map) => {
            let keys: Vec<Value> = map.keys().cloned().collect();
            Ok(Value::list(List { values: keys }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("map_keys requires a map type")),
    }
}

fn execute_map_values(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Map(map) => {
            let values: Vec<Value> = map.values().cloned().collect();
            Ok(Value::list(List { values }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "map_values requires a map type",
        )),
    }
}

#[cfg(test)]
mod tests;