donadb-rel 0.1.2

// predicate.rs — FieldOp and Predicate for scan_where().
//
// A Predicate names a field and applies a comparison operator to filter records.
// Predicates are evaluated in-process after the storage-layer scan returns raw bytes.
// When a secondary index exists for the named field, scan_where() uses the index
// instead of a full scan — the predicate is then used only for final filtering.
//
// Multiple predicates: all must be satisfied (implicit AND).
// OR semantics require calling scan_where() multiple times and merging results.

use crate::schema::FieldValue;

/// Comparison operator for a field predicate.
#[derive(Debug, Clone)]
pub enum FieldOp {
    Eq(FieldValue),
    Gt(FieldValue),
    Gte(FieldValue),
    Lt(FieldValue),
    Lte(FieldValue),
    /// Byte-prefix match. Only valid for Bytes/VarBytes/Address/Hash fields.
    Prefix(bytes::Bytes),
    /// Membership test. Field value must be one of the listed values.
    In(Vec<FieldValue>),
    /// Field is not null.
    IsNotNull,
}

/// A single field filter: field name + comparison operator.
#[derive(Debug, Clone)]
pub struct Predicate {
    pub field: String,
    pub op: FieldOp,
}

impl Predicate {
    pub fn eq(field: impl Into<String>, value: FieldValue) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Eq(value),
        }
    }
    pub fn gt(field: impl Into<String>, value: FieldValue) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Gt(value),
        }
    }
    pub fn gte(field: impl Into<String>, value: FieldValue) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Gte(value),
        }
    }
    pub fn lt(field: impl Into<String>, value: FieldValue) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Lt(value),
        }
    }
    pub fn lte(field: impl Into<String>, value: FieldValue) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Lte(value),
        }
    }
    pub fn prefix(field: impl Into<String>, prefix: impl Into<bytes::Bytes>) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::Prefix(prefix.into()),
        }
    }
    pub fn in_values(field: impl Into<String>, values: Vec<FieldValue>) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::In(values),
        }
    }
    pub fn is_not_null(field: impl Into<String>) -> Self {
        Self {
            field: field.into(),
            op: FieldOp::IsNotNull,
        }
    }
}

impl Predicate {
    /// Evaluate this predicate against a record field value.
    /// Returns true if the record satisfies the predicate.
    pub fn evaluate(&self, value: &FieldValue) -> bool {
        match &self.op {
            FieldOp::Eq(expected) => value == expected,
            FieldOp::IsNotNull => value != &FieldValue::Null,
            FieldOp::In(list) => list.iter().any(|v| v == value),
            FieldOp::Prefix(pfx) => {
                if let Some(b) = value.as_bytes() {
                    b.starts_with(pfx)
                } else {
                    false
                }
            }
            FieldOp::Gt(bound) => compare_field(value, bound) == Some(std::cmp::Ordering::Greater),
            FieldOp::Gte(bound) => matches!(
                compare_field(value, bound),
                Some(std::cmp::Ordering::Greater) | Some(std::cmp::Ordering::Equal)
            ),
            FieldOp::Lt(bound) => compare_field(value, bound) == Some(std::cmp::Ordering::Less),
            FieldOp::Lte(bound) => matches!(
                compare_field(value, bound),
                Some(std::cmp::Ordering::Less) | Some(std::cmp::Ordering::Equal)
            ),
        }
    }
}

/// Numeric/bytes comparison between two FieldValues of the same type.
/// Returns None if types are incomparable.
fn compare_field(a: &FieldValue, b: &FieldValue) -> Option<std::cmp::Ordering> {
    match (a, b) {
        (FieldValue::U8(x), FieldValue::U8(y)) => Some(x.cmp(y)),
        (FieldValue::U16(x), FieldValue::U16(y)) => Some(x.cmp(y)),
        (FieldValue::U32(x), FieldValue::U32(y)) => Some(x.cmp(y)),
        (FieldValue::U64(x), FieldValue::U64(y)) => Some(x.cmp(y)),
        (FieldValue::U128(x), FieldValue::U128(y)) => Some(x.cmp(y)),
        (FieldValue::I64(x), FieldValue::I64(y)) => Some(x.cmp(y)),
        (FieldValue::Bytes(x), FieldValue::Bytes(y)) => {
            Some(x.as_ref() as &[u8]).map(|xr| xr.cmp(y.as_ref()))
        }
        _ => None,
    }
}