#![forbid(unsafe_code)]
#![warn(rustdoc::broken_intra_doc_links)]

//! Expression engine for **frankenpandas** — implements pandas
//! `DataFrame.eval()` / `DataFrame.query()` semantics for Rust
//! callers and provides the underlying [`Expr`] AST + parser /
//! evaluator that the rest of the workspace can build on.
//!
//! ## Why
//!
//! Pandas users write things like:
//! ```python
//! df.query("price > 100 and category == 'A'")
//! df.eval("margin = revenue - cost")
//! ```
//!
//! This crate gives Rust callers the equivalent: `df.query(...)`,
//! `df.eval(...)`, with a parsed [`Expr`] AST you can build
//! programmatically or via the string parser.
//!
//! ## Public surface
//!
//! - [`Expr`]: the expression AST. Built either by hand
//!   ([`Expr::Series`], [`Expr::Add`], [`Expr::Sub`],
//!   [`Expr::Local`], etc.) or by parsing a pandas-style string
//!   with [`parse_expr`].
//! - [`SeriesRef`]: a typed column-name reference. Used by `Expr`
//!   variants and by callers building expressions programmatically.
//! - [`EvalContext`]: the evaluation environment — bindings for
//!   local variables (the `@local_var` syntax in pandas eval) plus
//!   the [`RuntimePolicy`] / [`EvidenceLedger`] from fp-runtime
//!   for decision recording.
//! - [`ExprError`]: failure modes (parse error, unknown column,
//!   type mismatch, division by zero, ...).
//!
//! ## Evaluation entry points
//!
//! Direct AST eval:
//! - [`evaluate`]: evaluate an [`Expr`] against a Series-like input.
//! - [`evaluate_on_dataframe`]: bind columns from a DataFrame as
//!   identifiers and evaluate.
//! - [`evaluate_on_dataframe_with_locals`]: same, plus an external
//!   `@local` binding map.
//! - [`filter_dataframe_on_expr`] /
//!   [`filter_dataframe_on_expr_with_locals`]: shortcut for
//!   `df[df.eval(expr)]` — the boolean-mask filter pattern.
//!
//! String entry points (parse-then-eval):
//! - [`eval_str`] / [`eval_str_with_locals`]: pandas
//!   `df.eval(string)` — returns a new Series / DataFrame column.
//! - [`query_str`] / [`query_str_with_locals`]: pandas
//!   `df.query(string)` — returns the row-filtered DataFrame.
//! - [`parse_expr`]: the standalone parser if you only want the
//!   AST.
//!
//! ## DataFrame extension trait
//!
//! [`DataFrameExprExt`] adds `df.eval(expr)` / `df.query(expr)`
//! method-style entry points on `DataFrame` so users can call them
//! fluently after `use fp_expr::DataFrameExprExt;`.
//!
//! ## Incremental views
//!
//! [`MaterializedView`] + [`Delta`]: the foundation for incremental
//! `eval`-derived columns. A `MaterializedView` caches the last
//! result and the input fingerprint; on next call, only re-evaluates
//! when inputs change. `Delta` records what changed for downstream
//! consumers.
//!
//! ## Cross-crate relationships
//!
//! - **fp-types** (`Scalar`), **fp-columnar** (`ComparisonOp`),
//!   **fp-index** (`Index`), **fp-frame** (`Series`, `DataFrame`,
//!   `FrameError`) are all consumed.
//! - **fp-runtime** (`RuntimePolicy`, `EvidenceLedger`) provides
//!   the optional decision-policy hook threaded through
//!   `EvalContext`.

use std::collections::BTreeMap;

use fp_columnar::ComparisonOp;
use fp_frame::{self, FrameError, Series};
use fp_index::{DuplicateKeep, Index, IndexLabel};
use fp_runtime::{EvidenceLedger, RuntimePolicy};
use fp_types::{DType, Scalar};
use serde::{Deserialize, Serialize};
use thiserror::Error;

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct SeriesRef(pub String);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BetweenInclusive {
    Both,
    Left,
    Right,
    Neither,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExprDuplicateKeep {
    First,
    Last,
    None,
}

impl ExprDuplicateKeep {
    fn parse(value: Scalar, context: &str) -> Result<Self, ExprError> {
        match value {
            Scalar::Utf8(value) => match value.as_str() {
                "first" => Ok(Self::First),
                "last" => Ok(Self::Last),
                other => Err(ExprError::ParseError(format!(
                    "{context} keep must be 'first', 'last', or False, got {other:?}"
                ))),
            },
            Scalar::Bool(false) | Scalar::Int64(0) => Ok(Self::None),
            Scalar::Float64(0.0) => Ok(Self::None),
            other => Err(ExprError::ParseError(format!(
                "{context} keep must be 'first', 'last', or False, got {other:?}"
            ))),
        }
    }

    fn as_frame_keep(self) -> DuplicateKeep {
        match self {
            Self::First => DuplicateKeep::First,
            Self::Last => DuplicateKeep::Last,
            Self::None => DuplicateKeep::None,
        }
    }
}

impl BetweenInclusive {
    fn parse(value: &str) -> Result<Self, ExprError> {
        match value {
            "both" => Ok(Self::Both),
            "left" => Ok(Self::Left),
            "right" => Ok(Self::Right),
            "neither" => Ok(Self::Neither),
            other => Err(ExprError::ParseError(format!(
                "between() inclusive must be one of 'both', 'left', 'right', or 'neither', got {other:?}"
            ))),
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            Self::Both => "both",
            Self::Left => "left",
            Self::Right => "right",
            Self::Neither => "neither",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Expr {
    Series {
        name: SeriesRef,
    },
    Local {
        name: String,
    },
    Add {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Sub {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Mul {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Div {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Modulo {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    FloorDiv {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Pow {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    And {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Or {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Not {
        expr: Box<Expr>,
    },
    Abs {
        expr: Box<Expr>,
    },
    Round {
        expr: Box<Expr>,
        decimals: i32,
    },
    IsNull {
        expr: Box<Expr>,
        negated: bool,
    },
    FillNa {
        expr: Box<Expr>,
        value: Scalar,
    },
    DropNa {
        expr: Box<Expr>,
    },
    SortValues {
        expr: Box<Expr>,
        ascending: bool,
        na_position: String,
    },
    SortIndex {
        expr: Box<Expr>,
        ascending: bool,
        ignore_index: bool,
    },
    ArgSort {
        expr: Box<Expr>,
    },
    Mode {
        expr: Box<Expr>,
        dropna: bool,
    },
    Duplicated {
        expr: Box<Expr>,
        keep: ExprDuplicateKeep,
    },
    DropDuplicates {
        expr: Box<Expr>,
        keep: ExprDuplicateKeep,
    },
    HeadTail {
        expr: Box<Expr>,
        n: i64,
        tail: bool,
    },
    TopN {
        expr: Box<Expr>,
        n: usize,
        keep: String,
        largest: bool,
    },
    Replace {
        expr: Box<Expr>,
        to_replace: Scalar,
        value: Scalar,
    },
    Astype {
        expr: Box<Expr>,
        dtype: DType,
    },
    CombineFirst {
        left: Box<Expr>,
        right: Box<Expr>,
    },
    Rank {
        expr: Box<Expr>,
        method: String,
        ascending: bool,
        na_option: String,
        pct: bool,
    },
    Where {
        expr: Box<Expr>,
        cond: Box<Expr>,
        other: Option<Box<Expr>>,
        mask: bool,
    },
    Between {
        expr: Box<Expr>,
        left: Scalar,
        right: Scalar,
        inclusive: BetweenInclusive,
    },
    Clip {
        expr: Box<Expr>,
        lower: Option<f64>,
        upper: Option<f64>,
    },
    Shift {
        expr: Box<Expr>,
        periods: i64,
    },
    Diff {
        expr: Box<Expr>,
        periods: i64,
    },
    CumSum {
        expr: Box<Expr>,
    },
    CumProd {
        expr: Box<Expr>,
    },
    CumMin {
        expr: Box<Expr>,
    },
    CumMax {
        expr: Box<Expr>,
    },
    PctChange {
        expr: Box<Expr>,
        periods: usize,
    },
    Compare {
        left: Box<Expr>,
        right: Box<Expr>,
        op: ComparisonOp,
    },
    IsIn {
        left: Box<Expr>,
        values: Vec<Scalar>,
        negated: bool,
    },
    Literal {
        value: Scalar,
    },
}

#[derive(Debug, Clone, Default)]
pub struct EvalContext {
    series: BTreeMap<String, Series>,
    locals: BTreeMap<String, Scalar>,
    anchor_index: Option<Index>,
}

impl EvalContext {
    #[must_use]
    pub fn new() -> Self {
        Self {
            series: BTreeMap::new(),
            locals: BTreeMap::new(),
            anchor_index: None,
        }
    }

    pub fn insert_series(&mut self, series: Series) {
        if self.anchor_index.is_none() {
            self.anchor_index = Some(series.index().clone());
        }
        self.series.insert(series.name().to_owned(), series);
    }

    pub fn from_dataframe(frame: &fp_frame::DataFrame) -> Result<Self, ExprError> {
        Self::from_dataframe_with_locals(frame, &BTreeMap::new())
    }

    pub fn from_dataframe_with_locals(
        frame: &fp_frame::DataFrame,
        locals: &BTreeMap<String, Scalar>,
    ) -> Result<Self, ExprError> {
        let mut context = Self {
            series: BTreeMap::new(),
            locals: locals.clone(),
            anchor_index: Some(frame.index().clone()),
        };
        context.insert_index_series("index", frame.index())?;
        context.insert_index_series("ilevel_0", frame.index())?;
        if let Some(name) = frame.index().name() {
            context.insert_index_series(name, frame.index())?;
        }
        for (name, column) in frame.columns() {
            let series = Series::new(name.clone(), frame.index().clone(), column.clone())?;
            context.insert_series(series);
        }
        Ok(context)
    }

    #[must_use]
    pub fn get_series(&self, name: &str) -> Option<&Series> {
        self.series.get(name)
    }

    pub fn insert_local(&mut self, name: impl Into<String>, value: Scalar) {
        self.locals.insert(name.into(), value);
    }

    fn insert_index_series(&mut self, name: &str, index: &Index) -> Result<(), ExprError> {
        let values = index.labels().iter().map(index_label_to_scalar).collect();
        let series = Series::from_values(name, index.labels().to_vec(), values)?;
        self.insert_series(series);
        Ok(())
    }

    #[must_use]
    pub fn get_local(&self, name: &str) -> Option<&Scalar> {
        self.locals.get(name)
    }

    fn broadcast_local(&self, name: &str, value: &Scalar) -> Result<Series, ExprError> {
        let index = self
            .anchor_index
            .as_ref()
            .ok_or_else(|| ExprError::UnanchoredLocal(name.to_owned()))?;
        Series::broadcast(name, value.clone(), index.labels().to_vec()).map_err(ExprError::from)
    }
}

fn index_label_to_scalar(label: &IndexLabel) -> Scalar {
    match label {
        IndexLabel::Int64(value) => Scalar::Int64(*value),
        IndexLabel::Float64(value) => Scalar::Float64(value.0),
        IndexLabel::Bool(value) => Scalar::Bool(*value),
        IndexLabel::Utf8(value) => Scalar::Utf8(value.clone()),
        IndexLabel::Timedelta64(value) => Scalar::Timedelta64(*value),
        IndexLabel::Datetime64(value) => Scalar::Datetime64(*value),
        // Typed-null label round-trips to the same-kind missing scalar.
        IndexLabel::Null(kind) => Scalar::Null(*kind),
    }
}

#[derive(Debug, Error)]
#[non_exhaustive]
pub enum ExprError {
    #[error("unknown series reference: {0}")]
    UnknownSeries(String),
    #[error("unknown local reference: @{0}")]
    UnknownLocal(String),
    #[error("cannot evaluate a pure literal expression without an index anchor")]
    UnanchoredLiteral,
    #[error("cannot evaluate local reference @{0} without an index anchor")]
    UnanchoredLocal(String),
    #[error("parse error: {0}")]
    ParseError(String),
    #[error(transparent)]
    Frame(#[from] FrameError),
}

pub fn evaluate(
    expr: &Expr,
    context: &EvalContext,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    match expr {
        Expr::Series { name } => context
            .get_series(&name.0)
            .cloned()
            .ok_or_else(|| ExprError::UnknownSeries(name.0.clone())),
        Expr::Local { name } => {
            let value = context
                .get_local(name)
                .ok_or_else(|| ExprError::UnknownLocal(name.clone()))?;
            context.broadcast_local(name, value)
        }
        Expr::Add { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.add_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Sub { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.sub_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Mul { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.mul_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Div { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.div_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Modulo { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.modulo_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::FloorDiv { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.floordiv_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Pow { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.pow_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::And { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.and(&rhs).map_err(ExprError::from)
        }
        Expr::Or { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.or(&rhs).map_err(ExprError::from)
        }
        Expr::Not { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.not().map_err(ExprError::from)
        }
        Expr::Abs { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.abs().map_err(ExprError::from)
        }
        Expr::Round { expr, decimals } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.round(*decimals).map_err(ExprError::from)
        }
        Expr::IsNull { expr, negated } => {
            let input = evaluate(expr, context, policy, ledger)?;
            if *negated {
                input.notna().map_err(ExprError::from)
            } else {
                input.isna().map_err(ExprError::from)
            }
        }
        Expr::FillNa { expr, value } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.fillna(value).map_err(ExprError::from)
        }
        Expr::DropNa { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.dropna().map_err(ExprError::from)
        }
        Expr::SortValues {
            expr,
            ascending,
            na_position,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .sort_values_na(*ascending, na_position)
                .map_err(ExprError::from)
        }
        Expr::SortIndex {
            expr,
            ascending,
            ignore_index,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            sort_index_series(input, *ascending, *ignore_index)
        }
        Expr::ArgSort { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.argsort(true).map_err(ExprError::from)
        }
        Expr::Mode { expr, dropna } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.mode_with_dropna(*dropna).map_err(ExprError::from)
        }
        Expr::Duplicated { expr, keep } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .duplicated_keep(keep.as_frame_keep())
                .map_err(ExprError::from)
        }
        Expr::DropDuplicates { expr, keep } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .drop_duplicates_keep(keep.as_frame_keep())
                .map_err(ExprError::from)
        }
        Expr::HeadTail { expr, n, tail } => {
            let input = evaluate(expr, context, policy, ledger)?;
            if *tail {
                input.tail(*n).map_err(ExprError::from)
            } else {
                input.head(*n).map_err(ExprError::from)
            }
        }
        Expr::TopN {
            expr,
            n,
            keep,
            largest,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            if *largest {
                input.nlargest_keep(*n, keep).map_err(ExprError::from)
            } else {
                input.nsmallest_keep(*n, keep).map_err(ExprError::from)
            }
        }
        Expr::Replace {
            expr,
            to_replace,
            value,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .replace(&[(to_replace.clone(), value.clone())])
                .map_err(ExprError::from)
        }
        Expr::Astype { expr, dtype } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.astype(*dtype).map_err(ExprError::from)
        }
        Expr::CombineFirst { left, right } => {
            let lhs = evaluate(left, context, policy, ledger)?;
            let rhs = evaluate(right, context, policy, ledger)?;
            lhs.combine_first(&rhs).map_err(ExprError::from)
        }
        Expr::Rank {
            expr,
            method,
            ascending,
            na_option,
            pct,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .rank_with_pct(method, *ascending, na_option, *pct)
                .map_err(ExprError::from)
        }
        Expr::Where {
            expr,
            cond,
            other,
            mask,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            let condition = evaluate(cond, context, policy, ledger)?;
            match other.as_deref() {
                None => {
                    if *mask {
                        input.mask(&condition, None).map_err(ExprError::from)
                    } else {
                        input.r#where(&condition, None).map_err(ExprError::from)
                    }
                }
                Some(Expr::Literal { value }) => {
                    if *mask {
                        input.mask(&condition, Some(value)).map_err(ExprError::from)
                    } else {
                        input
                            .r#where(&condition, Some(value))
                            .map_err(ExprError::from)
                    }
                }
                Some(other_expr) => {
                    let replacement = evaluate(other_expr, context, policy, ledger)?;
                    if *mask {
                        input
                            .mask_series(&condition, &replacement)
                            .map_err(ExprError::from)
                    } else {
                        input
                            .where_cond_series(&condition, &replacement)
                            .map_err(ExprError::from)
                    }
                }
            }
        }
        Expr::Between {
            expr,
            left,
            right,
            inclusive,
        } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input
                .between(left, right, inclusive.as_str())
                .map_err(ExprError::from)
        }
        Expr::Clip { expr, lower, upper } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.clip(*lower, *upper).map_err(ExprError::from)
        }
        Expr::Shift { expr, periods } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.shift(*periods).map_err(ExprError::from)
        }
        Expr::Diff { expr, periods } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.diff(*periods).map_err(ExprError::from)
        }
        Expr::CumSum { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.cumsum().map_err(ExprError::from)
        }
        Expr::CumProd { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.cumprod().map_err(ExprError::from)
        }
        Expr::CumMin { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.cummin().map_err(ExprError::from)
        }
        Expr::CumMax { expr } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.cummax().map_err(ExprError::from)
        }
        Expr::PctChange { expr, periods } => {
            let input = evaluate(expr, context, policy, ledger)?;
            input.pct_change(*periods as i64).map_err(ExprError::from)
        }
        Expr::Compare { left, right, op } => {
            evaluate_comparison(left, right, *op, context, policy, ledger)
        }
        Expr::IsIn {
            left,
            values,
            negated,
        } => {
            let out = evaluate(left, context, policy, ledger)?
                .isin(values)
                .map_err(ExprError::from)?;
            if *negated {
                out.not().map_err(ExprError::from)
            } else {
                Ok(out)
            }
        }
        Expr::Literal { value } => {
            let index = context
                .anchor_index
                .as_ref()
                .ok_or(ExprError::UnanchoredLiteral)?;
            Series::broadcast("_literal", value.clone(), index.labels().to_vec())
                .map_err(ExprError::from)
        }
    }
}

fn sort_index_series(
    input: Series,
    ascending: bool,
    ignore_index: bool,
) -> Result<Series, ExprError> {
    let sorted = input.sort_index(ascending).map_err(ExprError::from)?;
    if !ignore_index {
        return Ok(sorted);
    }

    let mut labels = Vec::with_capacity(sorted.len());
    for position in 0..sorted.len() {
        let label = i64::try_from(position).map_err(|_| {
            ExprError::Frame(FrameError::CompatibilityRejected(format!(
                "sort_index(ignore_index=True) cannot materialize RangeIndex label for position {position}"
            )))
        })?;
        labels.push(IndexLabel::Int64(label));
    }
    Series::from_values(sorted.name().to_owned(), labels, sorted.values().to_vec())
        .map_err(ExprError::from)
}

pub fn evaluate_on_dataframe(
    expr: &Expr,
    frame: &fp_frame::DataFrame,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    let context = EvalContext::from_dataframe(frame)?;
    evaluate(expr, &context, policy, ledger)
}

pub fn evaluate_on_dataframe_with_locals(
    expr: &Expr,
    frame: &fp_frame::DataFrame,
    locals: &BTreeMap<String, Scalar>,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    let context = EvalContext::from_dataframe_with_locals(frame, locals)?;
    evaluate(expr, &context, policy, ledger)
}

fn is_pure_boolean_literal_filter(expr: &Expr) -> bool {
    match expr {
        Expr::Literal {
            value: Scalar::Bool(_),
        } => true,
        Expr::And { left, right } | Expr::Or { left, right } => {
            is_pure_boolean_literal_filter(left) && is_pure_boolean_literal_filter(right)
        }
        Expr::Not { expr } => is_pure_boolean_literal_filter(expr),
        _ => false,
    }
}

pub fn filter_dataframe_on_expr(
    expr: &Expr,
    frame: &fp_frame::DataFrame,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<fp_frame::DataFrame, ExprError> {
    if is_pure_boolean_literal_filter(expr) {
        return Err(ExprError::Frame(FrameError::CompatibilityRejected(
            "scalar boolean query expressions are not valid row filters".to_string(),
        )));
    }
    let mask = evaluate_on_dataframe(expr, frame, policy, ledger)?;
    if let Some(offending) = mask
        .values()
        .iter()
        .find(|value| !matches!(value, Scalar::Bool(_) | Scalar::Null(_)))
    {
        return Err(ExprError::Frame(FrameError::CompatibilityRejected(
            format!(
                "boolean mask required for query-style filter; found dtype {:?}",
                offending.dtype()
            ),
        )));
    }
    frame.filter_rows(&mask).map_err(ExprError::from)
}

pub fn filter_dataframe_on_expr_with_locals(
    expr: &Expr,
    frame: &fp_frame::DataFrame,
    locals: &BTreeMap<String, Scalar>,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<fp_frame::DataFrame, ExprError> {
    if is_pure_boolean_literal_filter(expr) {
        return Err(ExprError::Frame(FrameError::CompatibilityRejected(
            "scalar boolean query expressions are not valid row filters".to_string(),
        )));
    }
    let mask = evaluate_on_dataframe_with_locals(expr, frame, locals, policy, ledger)?;
    if let Some(offending) = mask
        .values()
        .iter()
        .find(|value| !matches!(value, Scalar::Bool(_) | Scalar::Null(_)))
    {
        return Err(ExprError::Frame(FrameError::CompatibilityRejected(
            format!(
                "boolean mask required for query-style filter; found dtype {:?}",
                offending.dtype()
            ),
        )));
    }
    frame.filter_rows(&mask).map_err(ExprError::from)
}

/// Evaluate a string expression against a DataFrame and return a Series.
///
/// Analogous to `pandas.DataFrame.eval(expr_str)`. Parses the string
/// expression and evaluates it using the DataFrame's columns as variables.
pub fn eval_str(
    expr_str: &str,
    frame: &fp_frame::DataFrame,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    eval_str_with_locals(expr_str, frame, &BTreeMap::new(), policy, ledger)
}

pub fn eval_str_with_locals(
    expr_str: &str,
    frame: &fp_frame::DataFrame,
    locals: &BTreeMap<String, Scalar>,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    let expr = parse_expr(expr_str)?;
    evaluate_on_dataframe_with_locals(&expr, frame, locals, policy, ledger)
}

/// Filter a DataFrame using a string expression.
///
/// Analogous to `pandas.DataFrame.query(expr_str)`. Parses the string
/// expression, evaluates it to a boolean mask, then filters the DataFrame.
pub fn query_str(
    expr_str: &str,
    frame: &fp_frame::DataFrame,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<fp_frame::DataFrame, ExprError> {
    query_str_with_locals(expr_str, frame, &BTreeMap::new(), policy, ledger)
}

pub fn query_str_with_locals(
    expr_str: &str,
    frame: &fp_frame::DataFrame,
    locals: &BTreeMap<String, Scalar>,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<fp_frame::DataFrame, ExprError> {
    let expr = parse_expr(expr_str)?;
    filter_dataframe_on_expr_with_locals(&expr, frame, locals, policy, ledger)
}

// ── Extension trait for DataFrame eval/query convenience ─────────────

/// Extension trait that adds `eval()` and `query()` methods directly to DataFrame.
///
/// Import this trait to call `df.eval("a + b")` and `df.query("a > 1")` directly.
/// Uses a default hardened policy and fresh evidence ledger.
pub trait DataFrameExprExt {
    /// Evaluate an expression string in the context of this DataFrame.
    ///
    /// Matches `pd.DataFrame.eval(expr)`.
    fn eval(&self, expr_str: &str) -> Result<Series, ExprError>;

    /// Matches `pd.DataFrame.eval(expr)` with explicit `@local` scalar bindings.
    fn eval_with_locals(
        &self,
        expr_str: &str,
        locals: &BTreeMap<String, Scalar>,
    ) -> Result<Series, ExprError>;

    /// Filter rows by a boolean expression string.
    ///
    /// Matches `pd.DataFrame.query(expr)`.
    fn query(&self, expr_str: &str) -> Result<fp_frame::DataFrame, ExprError>;

    /// Matches `pd.DataFrame.query(expr)` with explicit `@local` scalar bindings.
    fn query_with_locals(
        &self,
        expr_str: &str,
        locals: &BTreeMap<String, Scalar>,
    ) -> Result<fp_frame::DataFrame, ExprError>;
}

impl DataFrameExprExt for fp_frame::DataFrame {
    fn eval(&self, expr_str: &str) -> Result<Series, ExprError> {
        self.eval_with_locals(expr_str, &BTreeMap::new())
    }

    fn eval_with_locals(
        &self,
        expr_str: &str,
        locals: &BTreeMap<String, Scalar>,
    ) -> Result<Series, ExprError> {
        let policy = RuntimePolicy::hardened(Some(100_000));
        let mut ledger = EvidenceLedger::new();
        eval_str_with_locals(expr_str, self, locals, &policy, &mut ledger)
    }

    fn query(&self, expr_str: &str) -> Result<fp_frame::DataFrame, ExprError> {
        self.query_with_locals(expr_str, &BTreeMap::new())
    }

    fn query_with_locals(
        &self,
        expr_str: &str,
        locals: &BTreeMap<String, Scalar>,
    ) -> Result<fp_frame::DataFrame, ExprError> {
        let policy = RuntimePolicy::hardened(Some(100_000));
        let mut ledger = EvidenceLedger::new();
        query_str_with_locals(expr_str, self, locals, &policy, &mut ledger)
    }
}

fn apply_series_comparison(
    left: &Series,
    right: &Series,
    op: ComparisonOp,
) -> Result<Series, ExprError> {
    match op {
        ComparisonOp::Gt => left.gt(right),
        ComparisonOp::Lt => left.lt(right),
        ComparisonOp::Eq => left.eq_series(right),
        ComparisonOp::Ne => left.ne_series(right),
        ComparisonOp::Ge => left.ge(right),
        ComparisonOp::Le => left.le(right),
    }
    .map_err(ExprError::from)
}

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

fn evaluate_comparison(
    left: &Expr,
    right: &Expr,
    op: ComparisonOp,
    context: &EvalContext,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    match (left, right) {
        (Expr::Literal { value }, right_expr) => {
            let rhs = evaluate(right_expr, context, policy, ledger)?;
            rhs.compare_scalar(value, reverse_comparison_op(op))
                .map_err(ExprError::from)
        }
        (left_expr, Expr::Literal { value }) => {
            let lhs = evaluate(left_expr, context, policy, ledger)?;
            lhs.compare_scalar(value, op).map_err(ExprError::from)
        }
        (left_expr, right_expr) => {
            let lhs = evaluate(left_expr, context, policy, ledger)?;
            let rhs = evaluate(right_expr, context, policy, ledger)?;
            apply_series_comparison(&lhs, &rhs, op)
        }
    }
}

// ── AG-15: Incremental View Maintenance ────────────────────────────────

/// A delta represents new rows appended to a base series.
#[derive(Debug, Clone)]
pub struct Delta {
    pub series_name: String,
    pub new_labels: Vec<fp_index::IndexLabel>,
    pub new_values: Vec<Scalar>,
}

/// Cached result of a previous full evaluation, used as base for incremental updates.
#[derive(Debug, Clone)]
pub struct MaterializedView {
    pub expr: Expr,
    pub result: Series,
    pub base_snapshot: EvalContext,
}

impl MaterializedView {
    pub fn from_full_eval(
        expr: &Expr,
        context: &EvalContext,
        policy: &RuntimePolicy,
        ledger: &mut EvidenceLedger,
    ) -> Result<Self, ExprError> {
        let result = evaluate(expr, context, policy, ledger)?;
        Ok(Self {
            expr: expr.clone(),
            result,
            base_snapshot: context.clone(),
        })
    }

    /// Apply a delta (appended rows) incrementally.
    ///
    /// For linear expressions (series refs, arithmetic, anchored comparisons), only the new rows
    /// are computed and concatenated to the existing result. Falls back to
    /// full re-evaluation for expressions that cannot be incrementally maintained.
    pub fn apply_delta(
        &mut self,
        delta: &Delta,
        context: &EvalContext,
        policy: &RuntimePolicy,
        ledger: &mut EvidenceLedger,
    ) -> Result<&Series, ExprError> {
        // Build a context containing only the delta rows
        let delta_series = Series::from_values(
            &delta.series_name,
            delta.new_labels.clone(),
            delta.new_values.clone(),
        )
        .map_err(ExprError::from)?;

        let mut delta_ctx = context.clone();
        delta_ctx.insert_series(delta_series);

        // Check if the expression can be incrementally maintained
        if Self::is_linear(&self.expr) {
            // Evaluate only the delta portion
            let delta_result = evaluate_delta(&self.expr, &delta_ctx, delta, policy, ledger)?;

            // Concatenate: old result + delta result
            let combined =
                fp_frame::concat_series(&[&self.result, &delta_result]).map_err(ExprError::from)?;
            self.result = combined;
            self.base_snapshot = context.clone();
        } else {
            // Fallback: full re-evaluation
            self.result = evaluate(&self.expr, context, policy, ledger)?;
            self.base_snapshot = context.clone();
        }

        Ok(&self.result)
    }

    fn extract_series(expr: &Expr, series_set: &mut std::collections::BTreeSet<String>) {
        match expr {
            Expr::Series { name } => {
                series_set.insert(name.0.clone());
            }
            Expr::Local { .. } => {}
            Expr::Add { left, right }
            | Expr::Sub { left, right }
            | Expr::Mul { left, right }
            | Expr::Div { left, right }
            | Expr::Modulo { left, right }
            | Expr::FloorDiv { left, right }
            | Expr::Pow { left, right }
            | Expr::And { left, right }
            | Expr::Or { left, right }
            | Expr::Compare { left, right, .. }
            | Expr::CombineFirst { left, right } => {
                Self::extract_series(left, series_set);
                Self::extract_series(right, series_set);
            }
            Expr::IsIn { left, .. } => Self::extract_series(left, series_set),
            Expr::Not { expr }
            | Expr::Abs { expr }
            | Expr::Round { expr, .. }
            | Expr::Rank { expr, .. } => {
                Self::extract_series(expr, series_set);
            }
            Expr::IsNull { expr, .. } => Self::extract_series(expr, series_set),
            Expr::FillNa { expr, .. } => Self::extract_series(expr, series_set),
            Expr::DropNa { expr } => Self::extract_series(expr, series_set),
            Expr::SortValues { expr, .. } => Self::extract_series(expr, series_set),
            Expr::SortIndex { expr, .. } => Self::extract_series(expr, series_set),
            Expr::ArgSort { expr } => Self::extract_series(expr, series_set),
            Expr::Mode { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Duplicated { expr, .. } => Self::extract_series(expr, series_set),
            Expr::DropDuplicates { expr, .. } => Self::extract_series(expr, series_set),
            Expr::HeadTail { expr, .. } => Self::extract_series(expr, series_set),
            Expr::TopN { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Replace { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Astype { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Where {
                expr, cond, other, ..
            } => {
                Self::extract_series(expr, series_set);
                Self::extract_series(cond, series_set);
                if let Some(other) = other {
                    Self::extract_series(other, series_set);
                }
            }
            Expr::Between { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Clip { expr, .. }
            | Expr::Shift { expr, .. }
            | Expr::Diff { expr, .. }
            | Expr::CumSum { expr }
            | Expr::CumProd { expr }
            | Expr::CumMin { expr }
            | Expr::CumMax { expr }
            | Expr::PctChange { expr, .. } => Self::extract_series(expr, series_set),
            Expr::Literal { .. } => {}
        }
    }

    fn is_linear(expr: &Expr) -> bool {
        let mut series_set = std::collections::BTreeSet::new();
        Self::extract_series(expr, &mut series_set);
        series_set.len() == 1 && Self::is_append_local(expr)
    }

    fn is_append_local(expr: &Expr) -> bool {
        match expr {
            Expr::Series { .. } | Expr::Local { .. } | Expr::Literal { .. } => true,
            Expr::Add { left, right }
            | Expr::Sub { left, right }
            | Expr::Mul { left, right }
            | Expr::Div { left, right }
            | Expr::Modulo { left, right }
            | Expr::FloorDiv { left, right }
            | Expr::Pow { left, right }
            | Expr::And { left, right }
            | Expr::Or { left, right }
            | Expr::Compare { left, right, .. } => {
                Self::is_append_local(left) && Self::is_append_local(right)
            }
            Expr::Not { expr }
            | Expr::Abs { expr }
            | Expr::Round { expr, .. }
            | Expr::IsNull { expr, .. }
            | Expr::FillNa { expr, .. }
            | Expr::Replace { expr, .. }
            | Expr::Astype { expr, .. }
            | Expr::Between { expr, .. }
            | Expr::Clip { expr, .. }
            | Expr::IsIn { left: expr, .. } => Self::is_append_local(expr),
            Expr::Where {
                expr, cond, other, ..
            } => {
                Self::is_append_local(expr)
                    && Self::is_append_local(cond)
                    && other.as_deref().is_none_or(Self::is_append_local)
            }
            Expr::DropNa { .. }
            | Expr::SortValues { .. }
            | Expr::SortIndex { .. }
            | Expr::ArgSort { .. }
            | Expr::Mode { .. }
            | Expr::Duplicated { .. }
            | Expr::DropDuplicates { .. }
            | Expr::HeadTail { .. }
            | Expr::TopN { .. }
            | Expr::CombineFirst { .. }
            | Expr::Rank { .. }
            | Expr::Shift { .. }
            | Expr::Diff { .. }
            | Expr::CumSum { .. }
            | Expr::CumProd { .. }
            | Expr::CumMin { .. }
            | Expr::CumMax { .. }
            | Expr::PctChange { .. } => false,
        }
    }
}

/// Evaluate only the delta rows of an expression.
///
/// For `Expr::Series`, returns the delta rows for the named series.
/// For arithmetic/logical/comparison operators, recursively evaluates delta
/// operands and applies the operation only on appended rows.
fn evaluate_delta(
    expr: &Expr,
    delta_ctx: &EvalContext,
    delta: &Delta,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    match expr {
        Expr::Series { name } => {
            if name.0 == delta.series_name {
                // This is the series that has the delta
                Series::from_values(&name.0, delta.new_labels.clone(), delta.new_values.clone())
                    .map_err(ExprError::from)
            } else {
                // Other series: extract the corresponding labels from the full series
                let full = delta_ctx
                    .get_series(&name.0)
                    .ok_or_else(|| ExprError::UnknownSeries(name.0.clone()))?;
                // For non-delta series in an add, we need the values at the delta labels.
                // If the other series doesn't have those labels, alignment will fill nulls.
                let reindexed = full
                    .reindex(delta.new_labels.clone())
                    .map_err(ExprError::from)?;
                Ok(reindexed)
            }
        }
        Expr::Local { name } => {
            let value = delta_ctx
                .get_local(name)
                .cloned()
                .ok_or_else(|| ExprError::UnknownLocal(name.clone()))?;
            Series::broadcast(name.as_str(), value, delta.new_labels.clone())
                .map_err(ExprError::from)
        }
        Expr::Add { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.add_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Sub { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.sub_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Mul { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.mul_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Div { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.div_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Modulo { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.modulo_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::FloorDiv { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.floordiv_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::Pow { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.pow_with_policy(&rhs, policy, ledger)
                .map_err(ExprError::from)
        }
        Expr::And { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.and(&rhs).map_err(ExprError::from)
        }
        Expr::Or { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.or(&rhs).map_err(ExprError::from)
        }
        Expr::Not { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.not().map_err(ExprError::from)
        }
        Expr::Abs { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.abs().map_err(ExprError::from)
        }
        Expr::Round { expr, decimals } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.round(*decimals).map_err(ExprError::from)
        }
        Expr::IsNull { expr, negated } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            if *negated {
                input.notna().map_err(ExprError::from)
            } else {
                input.isna().map_err(ExprError::from)
            }
        }
        Expr::FillNa { expr, value } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.fillna(value).map_err(ExprError::from)
        }
        Expr::DropNa { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.dropna().map_err(ExprError::from)
        }
        Expr::SortValues {
            expr,
            ascending,
            na_position,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .sort_values_na(*ascending, na_position)
                .map_err(ExprError::from)
        }
        Expr::SortIndex {
            expr,
            ascending,
            ignore_index,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            sort_index_series(input, *ascending, *ignore_index)
        }
        Expr::ArgSort { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.argsort(true).map_err(ExprError::from)
        }
        Expr::Mode { expr, dropna } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.mode_with_dropna(*dropna).map_err(ExprError::from)
        }
        Expr::Duplicated { expr, keep } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .duplicated_keep(keep.as_frame_keep())
                .map_err(ExprError::from)
        }
        Expr::DropDuplicates { expr, keep } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .drop_duplicates_keep(keep.as_frame_keep())
                .map_err(ExprError::from)
        }
        Expr::HeadTail { expr, n, tail } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            if *tail {
                input.tail(*n).map_err(ExprError::from)
            } else {
                input.head(*n).map_err(ExprError::from)
            }
        }
        Expr::TopN {
            expr,
            n,
            keep,
            largest,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            if *largest {
                input.nlargest_keep(*n, keep).map_err(ExprError::from)
            } else {
                input.nsmallest_keep(*n, keep).map_err(ExprError::from)
            }
        }
        Expr::Replace {
            expr,
            to_replace,
            value,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .replace(&[(to_replace.clone(), value.clone())])
                .map_err(ExprError::from)
        }
        Expr::Astype { expr, dtype } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.astype(*dtype).map_err(ExprError::from)
        }
        Expr::CombineFirst { left, right } => {
            let lhs = evaluate_delta(left, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right, delta_ctx, delta, policy, ledger)?;
            lhs.combine_first(&rhs).map_err(ExprError::from)
        }
        Expr::Rank {
            expr,
            method,
            ascending,
            na_option,
            pct,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .rank_with_pct(method, *ascending, na_option, *pct)
                .map_err(ExprError::from)
        }
        Expr::Where {
            expr,
            cond,
            other,
            mask,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            let condition = evaluate_delta(cond, delta_ctx, delta, policy, ledger)?;
            match other.as_deref() {
                None => {
                    if *mask {
                        input.mask(&condition, None).map_err(ExprError::from)
                    } else {
                        input.r#where(&condition, None).map_err(ExprError::from)
                    }
                }
                Some(Expr::Literal { value }) => {
                    if *mask {
                        input.mask(&condition, Some(value)).map_err(ExprError::from)
                    } else {
                        input
                            .r#where(&condition, Some(value))
                            .map_err(ExprError::from)
                    }
                }
                Some(other_expr) => {
                    let replacement = evaluate_delta(other_expr, delta_ctx, delta, policy, ledger)?;
                    if *mask {
                        input
                            .mask_series(&condition, &replacement)
                            .map_err(ExprError::from)
                    } else {
                        input
                            .where_cond_series(&condition, &replacement)
                            .map_err(ExprError::from)
                    }
                }
            }
        }
        Expr::Between {
            expr,
            left,
            right,
            inclusive,
        } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input
                .between(left, right, inclusive.as_str())
                .map_err(ExprError::from)
        }
        Expr::Clip { expr, lower, upper } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.clip(*lower, *upper).map_err(ExprError::from)
        }
        Expr::Shift { expr, periods } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.shift(*periods).map_err(ExprError::from)
        }
        Expr::Diff { expr, periods } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.diff(*periods).map_err(ExprError::from)
        }
        Expr::CumSum { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.cumsum().map_err(ExprError::from)
        }
        Expr::CumProd { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.cumprod().map_err(ExprError::from)
        }
        Expr::CumMin { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.cummin().map_err(ExprError::from)
        }
        Expr::CumMax { expr } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.cummax().map_err(ExprError::from)
        }
        Expr::PctChange { expr, periods } => {
            let input = evaluate_delta(expr, delta_ctx, delta, policy, ledger)?;
            input.pct_change(*periods as i64).map_err(ExprError::from)
        }
        Expr::Compare { left, right, op } => {
            evaluate_delta_comparison(left, right, *op, delta_ctx, delta, policy, ledger)
        }
        Expr::IsIn {
            left,
            values,
            negated,
        } => {
            let out = evaluate_delta(left, delta_ctx, delta, policy, ledger)?
                .isin(values)
                .map_err(ExprError::from)?;
            if *negated {
                out.not().map_err(ExprError::from)
            } else {
                Ok(out)
            }
        }
        Expr::Literal { value } => {
            Series::broadcast("_literal", value.clone(), delta.new_labels.clone())
                .map_err(ExprError::from)
        }
    }
}

fn evaluate_delta_comparison(
    left: &Expr,
    right: &Expr,
    op: ComparisonOp,
    delta_ctx: &EvalContext,
    delta: &Delta,
    policy: &RuntimePolicy,
    ledger: &mut EvidenceLedger,
) -> Result<Series, ExprError> {
    match (left, right) {
        (Expr::Literal { .. }, Expr::Literal { .. }) => Err(ExprError::UnanchoredLiteral),
        (Expr::Literal { value }, right_expr) => {
            let rhs = evaluate_delta(right_expr, delta_ctx, delta, policy, ledger)?;
            rhs.compare_scalar(value, reverse_comparison_op(op))
                .map_err(ExprError::from)
        }
        (left_expr, Expr::Literal { value }) => {
            let lhs = evaluate_delta(left_expr, delta_ctx, delta, policy, ledger)?;
            lhs.compare_scalar(value, op).map_err(ExprError::from)
        }
        (left_expr, right_expr) => {
            let lhs = evaluate_delta(left_expr, delta_ctx, delta, policy, ledger)?;
            let rhs = evaluate_delta(right_expr, delta_ctx, delta, policy, ledger)?;
            apply_series_comparison(&lhs, &rhs, op)
        }
    }
}

// ── Expression Parser ───────────────────────────────────────────────────
//
// A simple recursive-descent parser for pandas-style query/eval expressions.
// Supports:
//   - Column references (identifiers)
//   - Numeric literals (integer and float)
//   - String literals ('...' or "...")
//   - Boolean literals (`True` / `False`)
//   - Comparison operators: ==, !=, >, >=, <, <=
//   - Membership operators: in, not in (with scalar list literals)
//   - Logical operators: and, or, not
//   - Unary function calls: abs(expr)
//   - Series method calls: .isin([...]), .between(left, right, inclusive=...),
//     .abs(), .fillna(value), .add(other), .sub(other), .mul(other),
//     .div(other), .truediv(other), .floordiv(other), .mod(other),
//     .pow(other), reflected arithmetic variants, .eq(other), .ne(other),
//     .gt(other), .ge(other), .lt(other), .le(other), .clip(lower, upper),
//     .shift(periods), .diff(periods), .cumsum(), .cumprod(), .cummin(),
//     .cummax(), .pct_change(periods), .round(decimals), .dropna(),
//     .sort_values(ascending=..., na_position=...), .sort_index(ascending=...),
//     .argsort(axis, kind, order, stable), .mode(dropna),
//     .duplicated(keep), .drop_duplicates(keep), .head(n), .tail(n),
//     .replace(to_replace, value), .nlargest(n, keep), .nsmallest(n, keep),
//     .astype(dtype), .combine_first(other), .rank(method=..., ascending=...,
//     na_option=...), .where(cond, other), .mask(cond, other), .isna(),
//     .notna(), .isnull(), .notnull()
//   - Arithmetic operators: +, -, *, /, //, %, **
//   - Parenthesized sub-expressions

/// Parse a string expression into an `Expr` AST.
///
/// Syntax:
///   expr       → or_expr
///   or_expr    → and_expr ( "or" and_expr )*
///   and_expr   → not_expr ( "and" not_expr )*
///   not_expr   → "not" not_expr | comparison
///   comparison → add_expr ( ("==" | "!=" | ">" | ">=" | "<" | "<=") add_expr | ("in" | "not" "in") list_literal )?
///   add_expr   → mul_expr ( ("+" | "-") mul_expr )*
///   mul_expr   → unary_expr ( ("*" | "/" | "//" | "%") unary_expr )*
///   unary_expr → ("+" | "-") unary_expr | pow_expr
///   pow_expr   → atom ( "**" unary_expr )?
///   atom       → primary ( "." METHOD_CALL )*
///   primary    → NUMBER | STRING | BOOL | IDENT | LOCAL | "abs" "(" expr ")" | "(" expr ")"
pub fn parse_expr(input: &str) -> Result<Expr, ExprError> {
    let tokens = tokenize(input)?;
    let mut pos = 0;
    let result = parse_or(&tokens, &mut pos)?;
    if pos < tokens.len() {
        return Err(ExprError::ParseError(format!(
            "unexpected token at position {pos}: {:?}",
            tokens[pos]
        )));
    }
    Ok(result)
}

#[derive(Debug, Clone, PartialEq)]
enum Token {
    Ident(String),
    Local(String),
    Int(i64),
    Float(f64),
    Str(String),
    Bool(bool),
    // Comparison
    EqEq,
    NotEq,
    Gt,
    Ge,
    Lt,
    Le,
    Assign,
    // Arithmetic
    Plus,
    Minus,
    Star,
    Slash,
    FloorDiv,
    Percent,
    Pow,
    // Grouping
    LParen,
    RParen,
    LBracket,
    RBracket,
    Comma,
    Dot,
    // Logical (keywords)
    And,
    Or,
    Not,
    In,
}

fn tokenize(input: &str) -> Result<Vec<Token>, ExprError> {
    let chars: Vec<char> = input.chars().collect();
    let mut tokens = Vec::new();
    let mut i = 0;
    while i < chars.len() {
        let c = chars[i];
        if c.is_whitespace() {
            i += 1;
            continue;
        }
        match c {
            '+' => {
                tokens.push(Token::Plus);
                i += 1;
            }
            '-' => {
                // Check for negative number literal
                if i + 1 < chars.len()
                    && (chars[i + 1].is_ascii_digit() || chars[i + 1] == '.')
                    && (tokens.is_empty()
                        || matches!(
                            tokens.last(),
                            Some(
                                Token::LParen
                                    | Token::EqEq
                                    | Token::Assign
                                    | Token::NotEq
                                    | Token::Gt
                                    | Token::Ge
                                    | Token::Lt
                                    | Token::Le
                                    | Token::Plus
                                    | Token::Minus
                                    | Token::Star
                                    | Token::Slash
                                    | Token::FloorDiv
                                    | Token::Percent
                                    | Token::Pow
                                    | Token::And
                                    | Token::Or
                                    | Token::Not
                                    | Token::In
                                    | Token::LBracket
                                    | Token::Comma
                            )
                        ))
                {
                    let start = i;
                    let mut scan = i + 1;
                    let mut dot_count = 0;
                    while scan < chars.len() && (chars[scan].is_ascii_digit() || chars[scan] == '.')
                    {
                        if chars[scan] == '.' {
                            dot_count += 1;
                        }
                        scan += 1;
                    }
                    let mut after = scan;
                    while after < chars.len() && chars[after].is_whitespace() {
                        after += 1;
                    }
                    if after + 1 < chars.len() && chars[after] == '*' && chars[after + 1] == '*' {
                        tokens.push(Token::Minus);
                        i += 1;
                        continue;
                    }
                    i = scan;
                    let num_str: String = chars[start..i].iter().collect();
                    if dot_count > 1 {
                        return Err(ExprError::ParseError(format!(
                            "invalid number literal with multiple decimal points: {num_str}"
                        )));
                    }
                    if dot_count == 1 {
                        tokens.push(Token::Float(num_str.parse::<f64>().map_err(|_| {
                            ExprError::ParseError(format!("invalid float: {num_str}"))
                        })?));
                    } else {
                        tokens.push(Token::Int(num_str.parse::<i64>().map_err(|_| {
                            ExprError::ParseError(format!("invalid integer: {num_str}"))
                        })?));
                    }
                } else {
                    tokens.push(Token::Minus);
                    i += 1;
                }
            }
            '.' if i + 1 < chars.len() && chars[i + 1].is_ascii_digit() => {
                let start = i;
                i += 1;
                let mut extra_dots = 0;
                while i < chars.len() && (chars[i].is_ascii_digit() || chars[i] == '.') {
                    if chars[i] == '.' {
                        extra_dots += 1;
                    }
                    i += 1;
                }
                let num_str: String = chars[start..i].iter().collect();
                if extra_dots > 0 {
                    return Err(ExprError::ParseError(format!(
                        "invalid number literal with multiple decimal points: {num_str}"
                    )));
                }
                tokens.push(Token::Float(num_str.parse::<f64>().map_err(|_| {
                    ExprError::ParseError(format!("invalid float: {num_str}"))
                })?));
            }
            '.' => {
                tokens.push(Token::Dot);
                i += 1;
            }
            '*' => {
                if i + 1 < chars.len() && chars[i + 1] == '*' {
                    tokens.push(Token::Pow);
                    i += 2;
                } else {
                    tokens.push(Token::Star);
                    i += 1;
                }
            }
            '/' => {
                if i + 1 < chars.len() && chars[i + 1] == '/' {
                    tokens.push(Token::FloorDiv);
                    i += 2;
                } else {
                    tokens.push(Token::Slash);
                    i += 1;
                }
            }
            '%' => {
                tokens.push(Token::Percent);
                i += 1;
            }
            '(' => {
                tokens.push(Token::LParen);
                i += 1;
            }
            ')' => {
                tokens.push(Token::RParen);
                i += 1;
            }
            '[' => {
                tokens.push(Token::LBracket);
                i += 1;
            }
            ']' => {
                tokens.push(Token::RBracket);
                i += 1;
            }
            ',' => {
                tokens.push(Token::Comma);
                i += 1;
            }
            '=' => {
                if i + 1 < chars.len() && chars[i + 1] == '=' {
                    tokens.push(Token::EqEq);
                    i += 2;
                } else {
                    tokens.push(Token::Assign);
                    i += 1;
                }
            }
            '!' => {
                if i + 1 < chars.len() && chars[i + 1] == '=' {
                    tokens.push(Token::NotEq);
                    i += 2;
                } else {
                    return Err(ExprError::ParseError(
                        "expected '!=' but found single '!'".into(),
                    ));
                }
            }
            '>' => {
                if i + 1 < chars.len() && chars[i + 1] == '=' {
                    tokens.push(Token::Ge);
                    i += 2;
                } else {
                    tokens.push(Token::Gt);
                    i += 1;
                }
            }
            '<' => {
                if i + 1 < chars.len() && chars[i + 1] == '=' {
                    tokens.push(Token::Le);
                    i += 2;
                } else {
                    tokens.push(Token::Lt);
                    i += 1;
                }
            }
            '\'' | '"' => {
                let quote = c;
                i += 1;
                let mut s = String::new();
                while i < chars.len() && chars[i] != quote {
                    if chars[i] == '\\' && i + 1 < chars.len() {
                        i += 1;
                        match chars[i] {
                            'n' => s.push('\n'),
                            't' => s.push('\t'),
                            'r' => s.push('\r'),
                            '\\' => s.push('\\'),
                            c if c == quote => s.push(c),
                            other => {
                                s.push('\\');
                                s.push(other);
                            }
                        }
                    } else {
                        s.push(chars[i]);
                    }
                    i += 1;
                }
                if i >= chars.len() {
                    return Err(ExprError::ParseError("unterminated string literal".into()));
                }
                tokens.push(Token::Str(s));
                i += 1; // skip closing quote
            }
            '`' => {
                i += 1;
                let mut name = String::new();
                while i < chars.len() && chars[i] != '`' {
                    name.push(chars[i]);
                    i += 1;
                }
                if i >= chars.len() {
                    return Err(ExprError::ParseError(
                        "unterminated backtick identifier".into(),
                    ));
                }
                if name.is_empty() {
                    return Err(ExprError::ParseError("empty backtick identifier".into()));
                }
                tokens.push(Token::Ident(name));
                i += 1; // skip closing backtick
            }
            '@' => {
                i += 1;
                if i >= chars.len() || !(chars[i].is_alphabetic() || chars[i] == '_') {
                    return Err(ExprError::ParseError(
                        "expected identifier after '@'".into(),
                    ));
                }
                let start = i;
                while i < chars.len() && (chars[i].is_alphanumeric() || chars[i] == '_') {
                    i += 1;
                }
                let word: String = chars[start..i].iter().collect();
                tokens.push(Token::Local(word));
            }
            '&' => {
                tokens.push(Token::And);
                i += 1;
            }
            '|' => {
                tokens.push(Token::Or);
                i += 1;
            }
            '~' => {
                tokens.push(Token::Not);
                i += 1;
            }
            _ if c.is_ascii_digit() => {
                let start = i;
                let mut dot_count = 0;
                while i < chars.len() && (chars[i].is_ascii_digit() || chars[i] == '.') {
                    if chars[i] == '.' {
                        dot_count += 1;
                    }
                    i += 1;
                }
                let num_str: String = chars[start..i].iter().collect();
                if dot_count > 1 {
                    return Err(ExprError::ParseError(format!(
                        "invalid number literal with multiple decimal points: {num_str}"
                    )));
                }
                if dot_count == 1 {
                    tokens.push(Token::Float(num_str.parse::<f64>().map_err(|_| {
                        ExprError::ParseError(format!("invalid float: {num_str}"))
                    })?));
                } else {
                    tokens.push(Token::Int(num_str.parse::<i64>().map_err(|_| {
                        ExprError::ParseError(format!("invalid integer: {num_str}"))
                    })?));
                }
            }
            _ if c.is_alphabetic() || c == '_' => {
                let start = i;
                while i < chars.len() && (chars[i].is_alphanumeric() || chars[i] == '_') {
                    i += 1;
                }
                let word: String = chars[start..i].iter().collect();
                match word.as_str() {
                    "and" => tokens.push(Token::And),
                    "or" => tokens.push(Token::Or),
                    "not" => tokens.push(Token::Not),
                    "in" => tokens.push(Token::In),
                    "True" => tokens.push(Token::Bool(true)),
                    "False" => tokens.push(Token::Bool(false)),
                    _ => tokens.push(Token::Ident(word)),
                }
            }
            _ => {
                return Err(ExprError::ParseError(format!(
                    "unexpected character: '{c}'"
                )));
            }
        }
    }
    Ok(tokens)
}

fn parse_or(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let mut left = parse_and(tokens, pos)?;
    while *pos < tokens.len() && tokens[*pos] == Token::Or {
        *pos += 1;
        let right = parse_and(tokens, pos)?;
        left = Expr::Or {
            left: Box::new(left),
            right: Box::new(right),
        };
    }
    Ok(left)
}

fn parse_and(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let mut left = parse_not(tokens, pos)?;
    while *pos < tokens.len() && tokens[*pos] == Token::And {
        *pos += 1;
        let right = parse_not(tokens, pos)?;
        left = Expr::And {
            left: Box::new(left),
            right: Box::new(right),
        };
    }
    Ok(left)
}

fn parse_not(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    if *pos < tokens.len() && tokens[*pos] == Token::Not {
        *pos += 1;
        let inner = parse_not(tokens, pos)?;
        return Ok(Expr::Not {
            expr: Box::new(inner),
        });
    }
    parse_comparison(tokens, pos)
}

fn parse_comparison(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let mut left = parse_add(tokens, pos)?;
    let mut chained = None;
    while *pos < tokens.len() {
        let membership = if tokens[*pos] == Token::In {
            *pos += 1;
            Some(false)
        } else if *pos + 1 < tokens.len()
            && tokens[*pos] == Token::Not
            && tokens[*pos + 1] == Token::In
        {
            *pos += 2;
            Some(true)
        } else {
            None
        };
        if let Some(negated) = membership {
            let comparison = Expr::IsIn {
                left: Box::new(left.clone()),
                values: parse_list_literal(tokens, pos)?,
                negated,
            };
            chained = Some(match chained {
                Some(previous) => Expr::And {
                    left: Box::new(previous),
                    right: Box::new(comparison),
                },
                None => comparison,
            });
            break;
        }

        let op = match &tokens[*pos] {
            Token::EqEq => Some(ComparisonOp::Eq),
            Token::NotEq => Some(ComparisonOp::Ne),
            Token::Gt => Some(ComparisonOp::Gt),
            Token::Ge => Some(ComparisonOp::Ge),
            Token::Lt => Some(ComparisonOp::Lt),
            Token::Le => Some(ComparisonOp::Le),
            _ => None,
        };
        if let Some(op) = op {
            *pos += 1;
            let right = parse_add(tokens, pos)?;
            let comparison = Expr::Compare {
                left: Box::new(left.clone()),
                right: Box::new(right.clone()),
                op,
            };
            chained = Some(match chained {
                Some(previous) => Expr::And {
                    left: Box::new(previous),
                    right: Box::new(comparison),
                },
                None => comparison,
            });
            left = right;
        } else {
            break;
        }
    }
    Ok(chained.unwrap_or(left))
}

fn parse_list_literal(tokens: &[Token], pos: &mut usize) -> Result<Vec<Scalar>, ExprError> {
    if *pos >= tokens.len() || tokens[*pos] != Token::LBracket {
        return Err(ExprError::ParseError(
            "expected list literal after membership operator".into(),
        ));
    }
    *pos += 1;

    let mut values = Vec::new();
    if *pos < tokens.len() && tokens[*pos] == Token::RBracket {
        *pos += 1;
        return Ok(values);
    }

    loop {
        if *pos >= tokens.len() {
            return Err(ExprError::ParseError("unterminated list literal".into()));
        }

        let value = match &tokens[*pos] {
            Token::Int(value) => Scalar::Int64(*value),
            Token::Float(value) => Scalar::Float64(*value),
            Token::Str(value) => Scalar::Utf8(value.clone()),
            Token::Bool(value) => Scalar::Bool(*value),
            other => {
                return Err(ExprError::ParseError(format!(
                    "membership list values must be scalar literals, got {other:?}"
                )));
            }
        };
        values.push(value);
        *pos += 1;

        if *pos >= tokens.len() {
            return Err(ExprError::ParseError("unterminated list literal".into()));
        }
        match tokens[*pos] {
            Token::Comma => {
                *pos += 1;
                if *pos < tokens.len() && tokens[*pos] == Token::RBracket {
                    *pos += 1;
                    return Ok(values);
                }
            }
            Token::RBracket => {
                *pos += 1;
                return Ok(values);
            }
            ref other => {
                return Err(ExprError::ParseError(format!(
                    "expected ',' or ']' in list literal, got {other:?}"
                )));
            }
        }
    }
}

fn parse_scalar_literal(tokens: &[Token], pos: &mut usize) -> Result<Scalar, ExprError> {
    if *pos >= tokens.len() {
        return Err(ExprError::ParseError("expected scalar literal".into()));
    }
    let scalar = match &tokens[*pos] {
        Token::Int(value) => Scalar::Int64(*value),
        Token::Float(value) => Scalar::Float64(*value),
        Token::Str(value) => Scalar::Utf8(value.clone()),
        Token::Bool(value) => Scalar::Bool(*value),
        Token::Minus => match tokens.get(*pos + 1) {
            Some(Token::Int(value)) => {
                *pos += 1;
                Scalar::Int64(value.saturating_neg())
            }
            Some(Token::Float(value)) => {
                *pos += 1;
                Scalar::Float64(-value)
            }
            other => {
                return Err(ExprError::ParseError(format!(
                    "expected numeric literal after '-', got {other:?}"
                )));
            }
        },
        other => {
            return Err(ExprError::ParseError(format!(
                "expected scalar literal, got {other:?}"
            )));
        }
    };
    *pos += 1;
    Ok(scalar)
}

fn parse_numeric_literal(tokens: &[Token], pos: &mut usize) -> Result<f64, ExprError> {
    match parse_scalar_literal(tokens, pos)? {
        Scalar::Int64(value) => Ok(value as f64),
        Scalar::Float64(value) => Ok(value),
        other => Err(ExprError::ParseError(format!(
            "expected numeric literal, got {other:?}"
        ))),
    }
}

fn parse_optional_numeric_literal(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<Option<f64>, ExprError> {
    if matches!(tokens.get(*pos), Some(Token::Ident(value)) if value == "None") {
        *pos += 1;
        return Ok(None);
    }

    parse_numeric_literal(tokens, pos)
        .map(Some)
        .map_err(|err| ExprError::ParseError(format!("{context}: {err}")))
}

fn parse_i32_literal(tokens: &[Token], pos: &mut usize, context: &str) -> Result<i32, ExprError> {
    let value = parse_scalar_literal(tokens, pos)?;
    let Scalar::Int64(value) = value else {
        return Err(ExprError::ParseError(format!(
            "{context} must be an integer literal, got {value:?}"
        )));
    };
    i32::try_from(value)
        .map_err(|_| ExprError::ParseError(format!("{context} is outside the i32 range: {value}")))
}

fn parse_i64_literal_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<i64, ExprError> {
    let value = parse_scalar_literal(tokens, pos)?;
    let Scalar::Int64(value) = value else {
        return Err(ExprError::ParseError(format!(
            "{context} must be an integer literal, got {value:?}"
        )));
    };
    Ok(value)
}

fn parse_usize_literal_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<usize, ExprError> {
    let value = parse_i64_literal_argument(tokens, pos, context)?;
    usize::try_from(value).map_err(|_| {
        ExprError::ParseError(format!(
            "{context} must be a non-negative integer, got {value}"
        ))
    })
}

fn parse_top_n_literal_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<usize, ExprError> {
    let value = parse_i64_literal_argument(tokens, pos, context)?;
    if value < 0 {
        return Ok(0);
    }
    usize::try_from(value).map_err(|_| {
        ExprError::ParseError(format!(
            "{context} is too large to represent on this platform: {value}"
        ))
    })
}

fn parse_dtype_alias(value: &str) -> Result<DType, ExprError> {
    match value.to_ascii_lowercase().as_str() {
        "null" | "none" => Ok(DType::Null),
        "bool" | "boolean" | "?" => Ok(DType::Bool),
        "int" | "integer" | "int64" | "i8" => Ok(DType::Int64),
        "float" | "floating" | "float64" | "f8" => Ok(DType::Float64),
        "object" | "string" | "str" | "utf8" | "o" => Ok(DType::Utf8),
        "category" | "categorical" => Ok(DType::Categorical),
        "timedelta" | "timedelta64" | "timedelta64[ns]" | "m8" | "m8[ns]" => Ok(DType::Timedelta64),
        "datetime" | "datetime64" | "datetime64[ns]" => Ok(DType::Datetime64),
        "period" => Ok(DType::Period),
        "interval" => Ok(DType::Interval),
        "sparse" => Ok(DType::Sparse),
        other => Err(ExprError::ParseError(format!(
            "astype() dtype is not supported in expressions: {other:?}"
        ))),
    }
}

fn parse_dtype_literal(tokens: &[Token], pos: &mut usize) -> Result<DType, ExprError> {
    let dtype = parse_scalar_literal(tokens, pos)?;
    let Scalar::Utf8(value) = dtype else {
        return Err(ExprError::ParseError(format!(
            "astype() dtype must be a string literal, got {dtype:?}"
        )));
    };
    parse_dtype_alias(&value)
}

fn parse_string_literal_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<String, ExprError> {
    let value = parse_scalar_literal(tokens, pos)?;
    let Scalar::Utf8(value) = value else {
        return Err(ExprError::ParseError(format!(
            "{context} must be a string literal, got {value:?}"
        )));
    };
    Ok(value)
}

fn parse_bool_literal_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<bool, ExprError> {
    let value = parse_scalar_literal(tokens, pos)?;
    let Scalar::Bool(value) = value else {
        return Err(ExprError::ParseError(format!(
            "{context} must be a boolean literal, got {value:?}"
        )));
    };
    Ok(value)
}

fn parse_duplicate_keep_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<ExprDuplicateKeep, ExprError> {
    let value = parse_scalar_literal(tokens, pos)?;
    ExprDuplicateKeep::parse(value, context)
}

fn parse_axis_zero_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<(), ExprError> {
    let axis = parse_scalar_literal(tokens, pos)?;
    match axis {
        Scalar::Int64(0) => Ok(()),
        Scalar::Utf8(value) if value == "index" || value == "rows" => Ok(()),
        other => Err(ExprError::ParseError(format!(
            "{context} axis must be 0, 'index', or 'rows', got {other:?}"
        ))),
    }
}

fn parse_sort_kind_argument(
    tokens: &[Token],
    pos: &mut usize,
    context: &str,
) -> Result<(), ExprError> {
    let kind = parse_string_literal_argument(tokens, pos, context)?;
    match kind.as_str() {
        "quicksort" | "mergesort" | "heapsort" | "stable" => Ok(()),
        other => Err(ExprError::ParseError(format!(
            "{context} must be one of 'quicksort', 'mergesort', 'heapsort', or 'stable', got {other:?}"
        ))),
    }
}

fn parse_none_or_scalar_argument(tokens: &[Token], pos: &mut usize) -> Result<(), ExprError> {
    if matches!(tokens.get(*pos), Some(Token::Ident(value)) if value == "None") {
        *pos += 1;
        return Ok(());
    }
    parse_scalar_literal(tokens, pos).map(|_| ())
}

fn parse_add(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let mut left = parse_mul(tokens, pos)?;
    while *pos < tokens.len() {
        match &tokens[*pos] {
            Token::Plus => {
                *pos += 1;
                let right = parse_mul(tokens, pos)?;
                left = Expr::Add {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            Token::Minus => {
                *pos += 1;
                let right = parse_mul(tokens, pos)?;
                left = Expr::Sub {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            _ => break,
        }
    }
    Ok(left)
}

fn parse_mul(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let mut left = parse_unary(tokens, pos)?;
    while *pos < tokens.len() {
        match &tokens[*pos] {
            Token::Star => {
                *pos += 1;
                let right = parse_unary(tokens, pos)?;
                left = Expr::Mul {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            Token::Slash => {
                *pos += 1;
                let right = parse_unary(tokens, pos)?;
                left = Expr::Div {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            Token::FloorDiv => {
                *pos += 1;
                let right = parse_unary(tokens, pos)?;
                left = Expr::FloorDiv {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            Token::Percent => {
                *pos += 1;
                let right = parse_unary(tokens, pos)?;
                left = Expr::Modulo {
                    left: Box::new(left),
                    right: Box::new(right),
                };
            }
            _ => break,
        }
    }
    Ok(left)
}

fn parse_unary(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    if *pos < tokens.len() {
        match tokens[*pos] {
            Token::Plus => {
                *pos += 1;
                return parse_unary(tokens, pos);
            }
            Token::Minus => {
                *pos += 1;
                let inner = parse_unary(tokens, pos)?;
                return Ok(Expr::Sub {
                    left: Box::new(Expr::Literal {
                        value: Scalar::Int64(0),
                    }),
                    right: Box::new(inner),
                });
            }
            _ => {}
        }
    }
    parse_pow(tokens, pos)
}

fn parse_pow(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    let left = parse_atom(tokens, pos)?;
    if *pos < tokens.len() && tokens[*pos] == Token::Pow {
        *pos += 1;
        let right = parse_unary(tokens, pos)?;
        Ok(Expr::Pow {
            left: Box::new(left),
            right: Box::new(right),
        })
    } else {
        Ok(left)
    }
}

fn parse_atom(tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    if *pos >= tokens.len() {
        return Err(ExprError::ParseError("unexpected end of expression".into()));
    }
    let expr = match &tokens[*pos] {
        Token::Int(n) => {
            let val = *n;
            *pos += 1;
            Ok(Expr::Literal {
                value: Scalar::Int64(val),
            })
        }
        Token::Float(f) => {
            let val = *f;
            *pos += 1;
            Ok(Expr::Literal {
                value: Scalar::Float64(val),
            })
        }
        Token::Str(s) => {
            let val = s.clone();
            *pos += 1;
            Ok(Expr::Literal {
                value: Scalar::Utf8(val),
            })
        }
        Token::Bool(value) => {
            let val = *value;
            *pos += 1;
            Ok(Expr::Literal {
                value: Scalar::Bool(val),
            })
        }
        Token::Ident(name)
            if name == "abs" && *pos + 1 < tokens.len() && tokens[*pos + 1] == Token::LParen =>
        {
            *pos += 2; // skip function name and opening '('
            let inner = parse_or(tokens, pos)?;
            if *pos >= tokens.len() || tokens[*pos] != Token::RParen {
                return Err(ExprError::ParseError(
                    "expected closing ')' after abs argument".into(),
                ));
            }
            *pos += 1; // skip ')'
            Ok(Expr::Abs {
                expr: Box::new(inner),
            })
        }
        Token::Ident(name) => {
            let name = name.clone();
            *pos += 1;
            Ok(Expr::Series {
                name: SeriesRef(name),
            })
        }
        Token::Local(name) => {
            let name = name.clone();
            *pos += 1;
            Ok(Expr::Local { name })
        }
        Token::LParen => {
            *pos += 1; // skip '('
            let inner = parse_or(tokens, pos)?;
            if *pos >= tokens.len() || tokens[*pos] != Token::RParen {
                return Err(ExprError::ParseError("expected closing ')'".into()));
            }
            *pos += 1; // skip ')'
            Ok(inner)
        }
        other => Err(ExprError::ParseError(format!(
            "unexpected token: {other:?}"
        ))),
    }?;
    parse_postfix(expr, tokens, pos)
}

fn build_arithmetic_method_expr(
    method: &str,
    receiver: Expr,
    other: Expr,
) -> Result<Expr, ExprError> {
    let reflected = matches!(
        method,
        "radd" | "rsub" | "rmul" | "rdiv" | "rtruediv" | "rfloordiv" | "rmod" | "rpow"
    );
    let (left, right) = if reflected {
        (other, receiver)
    } else {
        (receiver, other)
    };
    let left = Box::new(left);
    let right = Box::new(right);

    match method {
        "add" | "radd" => Ok(Expr::Add { left, right }),
        "sub" | "subtract" | "rsub" => Ok(Expr::Sub { left, right }),
        "mul" | "multiply" | "rmul" => Ok(Expr::Mul { left, right }),
        "div" | "divide" | "truediv" | "rdiv" | "rtruediv" => Ok(Expr::Div { left, right }),
        "floordiv" | "rfloordiv" => Ok(Expr::FloorDiv { left, right }),
        "mod" | "rmod" => Ok(Expr::Modulo { left, right }),
        "pow" | "rpow" => Ok(Expr::Pow { left, right }),
        other => Err(ExprError::ParseError(format!(
            "unsupported arithmetic method: {other}"
        ))),
    }
}

fn parse_postfix(mut expr: Expr, tokens: &[Token], pos: &mut usize) -> Result<Expr, ExprError> {
    while *pos < tokens.len() && tokens[*pos] == Token::Dot {
        let Some(Token::Ident(method)) = tokens.get(*pos + 1) else {
            return Err(ExprError::ParseError(
                "expected method name after '.'".into(),
            ));
        };
        if tokens.get(*pos + 2) != Some(&Token::LParen) {
            return Err(ExprError::ParseError(format!(
                "expected '(' after method name {method}"
            )));
        }

        match method.as_str() {
            "abs" => {
                if tokens.get(*pos + 3) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "method abs does not accept arguments in expressions".into(),
                    ));
                }
                expr = Expr::Abs {
                    expr: Box::new(expr),
                };
                *pos += 4;
            }
            "isna" | "isnull" | "notna" | "notnull" => {
                if tokens.get(*pos + 3) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(format!(
                        "method {method} does not accept arguments in expressions"
                    )));
                }
                expr = Expr::IsNull {
                    expr: Box::new(expr),
                    negated: matches!(method.as_str(), "notna" | "notnull"),
                };
                *pos += 4;
            }
            "fillna" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "value" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected fillna() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let value = parse_scalar_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after fillna() value".into(),
                    ));
                }
                expr = Expr::FillNa {
                    expr: Box::new(expr),
                    value,
                };
                *pos = arg_pos + 1;
            }
            "dropna" => {
                if tokens.get(*pos + 3) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "method dropna does not accept arguments in expressions".into(),
                    ));
                }
                expr = Expr::DropNa {
                    expr: Box::new(expr),
                };
                *pos += 4;
            }
            "sort_values" => {
                let mut arg_pos = *pos + 3;
                let mut ascending = true;
                let mut na_position = "last".to_owned();
                let mut ascending_seen = false;
                let mut na_position_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated sort_values() arguments".into(),
                        ));
                    }

                    let Some(Token::Ident(keyword)) = tokens.get(arg_pos) else {
                        return Err(ExprError::ParseError(
                            "sort_values() arguments are keyword-only in expressions".into(),
                        ));
                    };
                    if tokens.get(arg_pos + 1) != Some(&Token::Assign) {
                        return Err(ExprError::ParseError(
                            "sort_values() arguments are keyword-only in expressions".into(),
                        ));
                    }
                    let keyword = keyword.clone();
                    arg_pos += 2;

                    match keyword.as_str() {
                        "ascending" => {
                            if ascending_seen {
                                return Err(ExprError::ParseError(
                                    "sort_values() ascending argument was provided more than once"
                                        .into(),
                                ));
                            }
                            ascending = parse_bool_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_values() ascending",
                            )?;
                            ascending_seen = true;
                        }
                        "na_position" => {
                            if na_position_seen {
                                return Err(ExprError::ParseError(
                                    "sort_values() na_position argument was provided more than once"
                                        .into(),
                                ));
                            }
                            na_position = parse_string_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_values() na_position",
                            )?;
                            na_position_seen = true;
                        }
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "unexpected sort_values() keyword argument: {other}"
                            )));
                        }
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "sort_values() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in sort_values() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::SortValues {
                    expr: Box::new(expr),
                    ascending,
                    na_position,
                };
                *pos = arg_pos + 1;
            }
            "sort_index" => {
                let mut arg_pos = *pos + 3;
                let mut ascending = true;
                let mut ignore_index = false;
                let mut axis_seen = false;
                let mut ascending_seen = false;
                let mut kind_seen = false;
                let mut na_position_seen = false;
                let mut level_seen = false;
                let mut sort_remaining_seen = false;
                let mut ignore_index_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated sort_index() arguments".into(),
                        ));
                    }

                    let Some(Token::Ident(keyword)) = tokens.get(arg_pos) else {
                        return Err(ExprError::ParseError(
                            "sort_index() arguments are keyword-only in expressions".into(),
                        ));
                    };
                    if tokens.get(arg_pos + 1) != Some(&Token::Assign) {
                        return Err(ExprError::ParseError(
                            "sort_index() arguments are keyword-only in expressions".into(),
                        ));
                    }
                    let keyword = keyword.clone();
                    arg_pos += 2;

                    match keyword.as_str() {
                        "axis" => {
                            if axis_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() axis argument was provided more than once".into(),
                                ));
                            }
                            let axis = parse_scalar_literal(tokens, &mut arg_pos)?;
                            match axis {
                                Scalar::Int64(0) => {}
                                Scalar::Utf8(value) if value == "index" || value == "rows" => {}
                                other => {
                                    return Err(ExprError::ParseError(format!(
                                        "sort_index() axis must be 0, 'index', or 'rows', got {other:?}"
                                    )));
                                }
                            }
                            axis_seen = true;
                        }
                        "ascending" => {
                            if ascending_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() ascending argument was provided more than once"
                                        .into(),
                                ));
                            }
                            ascending = parse_bool_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_index() ascending",
                            )?;
                            ascending_seen = true;
                        }
                        "kind" => {
                            if kind_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() kind argument was provided more than once".into(),
                                ));
                            }
                            let kind = parse_string_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_index() kind",
                            )?;
                            match kind.as_str() {
                                "quicksort" | "mergesort" | "heapsort" | "stable" => {}
                                other => {
                                    return Err(ExprError::ParseError(format!(
                                        "sort_index() kind must be one of 'quicksort', 'mergesort', 'heapsort', or 'stable', got {other:?}"
                                    )));
                                }
                            }
                            kind_seen = true;
                        }
                        "na_position" => {
                            if na_position_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() na_position argument was provided more than once"
                                        .into(),
                                ));
                            }
                            let na_position = parse_string_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_index() na_position",
                            )?;
                            match na_position.as_str() {
                                "first" | "last" => {}
                                other => {
                                    return Err(ExprError::ParseError(format!(
                                        "sort_index() na_position must be 'first' or 'last', got {other:?}"
                                    )));
                                }
                            }
                            na_position_seen = true;
                        }
                        "level" => {
                            if level_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() level argument was provided more than once"
                                        .into(),
                                ));
                            }
                            let _ = parse_scalar_literal(tokens, &mut arg_pos)?;
                            level_seen = true;
                        }
                        "sort_remaining" => {
                            if sort_remaining_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() sort_remaining argument was provided more than once"
                                        .into(),
                                ));
                            }
                            let _ = parse_scalar_literal(tokens, &mut arg_pos)?;
                            sort_remaining_seen = true;
                        }
                        "ignore_index" => {
                            if ignore_index_seen {
                                return Err(ExprError::ParseError(
                                    "sort_index() ignore_index argument was provided more than once"
                                        .into(),
                                ));
                            }
                            ignore_index = parse_bool_literal_argument(
                                tokens,
                                &mut arg_pos,
                                "sort_index() ignore_index",
                            )?;
                            ignore_index_seen = true;
                        }
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "unexpected sort_index() keyword argument: {other}"
                            )));
                        }
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "sort_index() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in sort_index() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::SortIndex {
                    expr: Box::new(expr),
                    ascending,
                    ignore_index,
                };
                *pos = arg_pos + 1;
            }
            "argsort" => {
                let mut arg_pos = *pos + 3;
                let mut axis_seen = false;
                let mut kind_seen = false;
                let mut order_seen = false;
                let mut stable_seen = false;
                let mut positional_count = 0_usize;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated argsort() arguments".into(),
                        ));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "axis" => {
                                if axis_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() axis argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_axis_zero_argument(tokens, &mut arg_pos, "argsort()")?;
                                axis_seen = true;
                            }
                            "kind" => {
                                if kind_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() kind argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_sort_kind_argument(tokens, &mut arg_pos, "argsort() kind")?;
                                kind_seen = true;
                            }
                            "order" => {
                                if order_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() order argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_none_or_scalar_argument(tokens, &mut arg_pos)?;
                                order_seen = true;
                            }
                            "stable" => {
                                if stable_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() stable argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_none_or_scalar_argument(tokens, &mut arg_pos)?;
                                stable_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected argsort() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if keyword_seen {
                            return Err(ExprError::ParseError(
                                "argsort() positional arguments cannot follow keyword arguments"
                                    .into(),
                            ));
                        }
                        match positional_count {
                            0 => {
                                if axis_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() axis argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_axis_zero_argument(tokens, &mut arg_pos, "argsort()")?;
                                axis_seen = true;
                            }
                            1 => {
                                if kind_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() kind argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_sort_kind_argument(tokens, &mut arg_pos, "argsort() kind")?;
                                kind_seen = true;
                            }
                            2 => {
                                if order_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() order argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_none_or_scalar_argument(tokens, &mut arg_pos)?;
                                order_seen = true;
                            }
                            3 => {
                                if stable_seen {
                                    return Err(ExprError::ParseError(
                                        "argsort() stable argument was provided more than once"
                                            .into(),
                                    ));
                                }
                                parse_none_or_scalar_argument(tokens, &mut arg_pos)?;
                                stable_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(
                                    "argsort() accepts at most axis, kind, order, and stable arguments"
                                        .into(),
                                ));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "argsort() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in argsort() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::ArgSort {
                    expr: Box::new(expr),
                };
                *pos = arg_pos + 1;
            }
            "mode" => {
                let mut arg_pos = *pos + 3;
                let mut dropna = true;
                let mut dropna_seen = false;
                let mut positional_count = 0_usize;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated mode() arguments".into(),
                        ));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "dropna" => {
                                if dropna_seen {
                                    return Err(ExprError::ParseError(
                                        "mode() dropna argument was provided more than once".into(),
                                    ));
                                }
                                dropna = parse_bool_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "mode() dropna",
                                )?;
                                dropna_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected mode() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if keyword_seen {
                            return Err(ExprError::ParseError(
                                "mode() positional arguments cannot follow keyword arguments"
                                    .into(),
                            ));
                        }
                        match positional_count {
                            0 => {
                                if dropna_seen {
                                    return Err(ExprError::ParseError(
                                        "mode() dropna argument was provided more than once".into(),
                                    ));
                                }
                                dropna = parse_bool_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "mode() dropna",
                                )?;
                                dropna_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(
                                    "mode() accepts at most one dropna argument".into(),
                                ));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "mode() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in mode() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::Mode {
                    expr: Box::new(expr),
                    dropna,
                };
                *pos = arg_pos + 1;
            }
            "duplicated" | "drop_duplicates" => {
                let mut arg_pos = *pos + 3;
                let mut keep = ExprDuplicateKeep::First;
                let mut keep_seen = false;
                let mut positional_count = 0_usize;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(format!(
                            "unterminated {method}() arguments"
                        )));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "keep" => {
                                if keep_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() keep argument was provided more than once"
                                    )));
                                }
                                keep = parse_duplicate_keep_argument(tokens, &mut arg_pos, method)?;
                                keep_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected {method}() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if method == "drop_duplicates" {
                            return Err(ExprError::ParseError(
                                "drop_duplicates() arguments are keyword-only in expressions"
                                    .into(),
                            ));
                        }
                        if keyword_seen {
                            return Err(ExprError::ParseError(format!(
                                "{method}() positional arguments cannot follow keyword arguments"
                            )));
                        }
                        match positional_count {
                            0 => {
                                if keep_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() keep argument was provided more than once"
                                    )));
                                }
                                keep = parse_duplicate_keep_argument(tokens, &mut arg_pos, method)?;
                                keep_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() accepts at most one keep argument"
                                )));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() arguments cannot end with ','"
                                )));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in {method}() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = if method == "duplicated" {
                    Expr::Duplicated {
                        expr: Box::new(expr),
                        keep,
                    }
                } else {
                    Expr::DropDuplicates {
                        expr: Box::new(expr),
                        keep,
                    }
                };
                *pos = arg_pos + 1;
            }
            "head" | "tail" => {
                let mut arg_pos = *pos + 3;
                let mut n = 5_i64;
                let mut n_seen = false;
                let mut positional_count = 0_usize;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(format!(
                            "unterminated {method}() arguments"
                        )));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "n" => {
                                if n_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() n argument was provided more than once"
                                    )));
                                }
                                n = parse_i64_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "head/tail n",
                                )?;
                                n_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected {method}() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if keyword_seen {
                            return Err(ExprError::ParseError(format!(
                                "{method}() positional arguments cannot follow keyword arguments"
                            )));
                        }
                        match positional_count {
                            0 => {
                                if n_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() n argument was provided more than once"
                                    )));
                                }
                                n = parse_i64_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "head/tail n",
                                )?;
                                n_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() accepts at most one n argument"
                                )));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() arguments cannot end with ','"
                                )));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in {method}() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::HeadTail {
                    expr: Box::new(expr),
                    n,
                    tail: method == "tail",
                };
                *pos = arg_pos + 1;
            }
            "nlargest" | "nsmallest" => {
                let mut arg_pos = *pos + 3;
                let mut n = 5_usize;
                let mut keep = "first".to_owned();
                let mut n_seen = false;
                let mut keep_seen = false;
                let mut positional_count = 0_usize;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(format!(
                            "unterminated {method}() arguments"
                        )));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "n" => {
                                if n_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() n argument was provided more than once"
                                    )));
                                }
                                n = parse_top_n_literal_argument(tokens, &mut arg_pos, "top-n n")?;
                                n_seen = true;
                            }
                            "keep" => {
                                if keep_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() keep argument was provided more than once"
                                    )));
                                }
                                keep = parse_string_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "top-n keep",
                                )?;
                                keep_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected {method}() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if keyword_seen {
                            return Err(ExprError::ParseError(format!(
                                "{method}() positional arguments cannot follow keyword arguments"
                            )));
                        }
                        match positional_count {
                            0 => {
                                if n_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() n argument was provided more than once"
                                    )));
                                }
                                n = parse_top_n_literal_argument(tokens, &mut arg_pos, "top-n n")?;
                                n_seen = true;
                            }
                            1 => {
                                if keep_seen {
                                    return Err(ExprError::ParseError(format!(
                                        "{method}() keep argument was provided more than once"
                                    )));
                                }
                                keep = parse_string_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "top-n keep",
                                )?;
                                keep_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() accepts at most n and keep arguments"
                                )));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() arguments cannot end with ','"
                                )));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in {method}() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::TopN {
                    expr: Box::new(expr),
                    n,
                    keep,
                    largest: method == "nlargest",
                };
                *pos = arg_pos + 1;
            }
            "replace" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "to_replace" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected replace() first keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let to_replace = parse_scalar_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::Comma) {
                    return Err(ExprError::ParseError(
                        "expected ',' between replace() arguments".into(),
                    ));
                }
                arg_pos += 1;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "value" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected replace() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let value = parse_scalar_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after replace() arguments".into(),
                    ));
                }
                expr = Expr::Replace {
                    expr: Box::new(expr),
                    to_replace,
                    value,
                };
                *pos = arg_pos + 1;
            }
            "astype" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "dtype" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected astype() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let dtype = parse_dtype_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after astype() dtype".into(),
                    ));
                }
                expr = Expr::Astype {
                    expr: Box::new(expr),
                    dtype,
                };
                *pos = arg_pos + 1;
            }
            "combine_first" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "other" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected combine_first() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let right = parse_or(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after combine_first() argument".into(),
                    ));
                }
                expr = Expr::CombineFirst {
                    left: Box::new(expr),
                    right: Box::new(right),
                };
                *pos = arg_pos + 1;
            }
            "rank" => {
                let mut arg_pos = *pos + 3;
                let mut method = "average".to_owned();
                let mut ascending = true;
                let mut na_option = "keep".to_owned();
                let mut pct = false;
                let mut positional_method_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated rank() arguments".into(),
                        ));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "method" => {
                                method = parse_string_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "rank() method",
                                )?;
                            }
                            "ascending" => {
                                ascending = parse_bool_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "rank() ascending",
                                )?;
                            }
                            "na_option" => {
                                na_option = parse_string_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "rank() na_option",
                                )?;
                            }
                            "pct" => {
                                pct = parse_bool_literal_argument(
                                    tokens,
                                    &mut arg_pos,
                                    "rank() pct",
                                )?;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected rank() keyword argument: {other}"
                                )));
                            }
                        }
                    } else if !positional_method_seen {
                        method =
                            parse_string_literal_argument(tokens, &mut arg_pos, "rank() method")?;
                        positional_method_seen = true;
                    } else {
                        return Err(ExprError::ParseError(
                            "rank() only accepts one positional method argument".into(),
                        ));
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "rank() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in rank() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::Rank {
                    expr: Box::new(expr),
                    method,
                    ascending,
                    na_option,
                    pct,
                };
                *pos = arg_pos + 1;
            }
            "add" | "radd" | "sub" | "subtract" | "rsub" | "mul" | "multiply" | "rmul" | "div"
            | "divide" | "truediv" | "rdiv" | "rtruediv" | "floordiv" | "rfloordiv" | "mod"
            | "rmod" | "pow" | "rpow" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "other" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected {method}() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let other = parse_or(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(format!(
                        "expected ')' after {method}() argument"
                    )));
                }
                expr = build_arithmetic_method_expr(method, expr, other)?;
                *pos = arg_pos + 1;
            }
            "where" | "mask" => {
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "cond" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected {method}() first keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let cond = parse_or(tokens, &mut arg_pos)?;
                let other = if tokens.get(arg_pos) == Some(&Token::Comma) {
                    arg_pos += 1;
                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "other" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected {method}() keyword argument: {keyword}"
                            )));
                        }
                        arg_pos += 2;
                    }
                    Some(Box::new(parse_or(tokens, &mut arg_pos)?))
                } else {
                    None
                };
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(format!(
                        "expected ')' after {method}() arguments"
                    )));
                }
                expr = Expr::Where {
                    expr: Box::new(expr),
                    cond: Box::new(cond),
                    other,
                    mask: method == "mask",
                };
                *pos = arg_pos + 1;
            }
            "eq" | "ne" | "gt" | "ge" | "lt" | "le" => {
                let op = match method.as_str() {
                    "eq" => ComparisonOp::Eq,
                    "ne" => ComparisonOp::Ne,
                    "gt" => ComparisonOp::Gt,
                    "ge" => ComparisonOp::Ge,
                    "lt" => ComparisonOp::Lt,
                    "le" => ComparisonOp::Le,
                    other => {
                        return Err(ExprError::ParseError(format!(
                            "unsupported comparison method: {other}"
                        )));
                    }
                };
                let mut arg_pos = *pos + 3;
                if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                    && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                {
                    if keyword != "other" {
                        return Err(ExprError::ParseError(format!(
                            "unexpected {method}() keyword argument: {keyword}"
                        )));
                    }
                    arg_pos += 2;
                }
                let right = parse_or(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(format!(
                        "expected ')' after {method}() argument"
                    )));
                }
                expr = Expr::Compare {
                    left: Box::new(expr),
                    right: Box::new(right),
                    op,
                };
                *pos = arg_pos + 1;
            }
            "isin" => {
                let mut arg_pos = *pos + 3;
                let values = parse_list_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after isin list literal".into(),
                    ));
                }
                expr = Expr::IsIn {
                    left: Box::new(expr),
                    values,
                    negated: false,
                };
                *pos = arg_pos + 1;
            }
            "between" => {
                let mut arg_pos = *pos + 3;
                let left = parse_scalar_literal(tokens, &mut arg_pos)?;
                if tokens.get(arg_pos) != Some(&Token::Comma) {
                    return Err(ExprError::ParseError(
                        "expected ',' between between() bounds".into(),
                    ));
                }
                arg_pos += 1;
                let right = parse_scalar_literal(tokens, &mut arg_pos)?;
                let mut inclusive = BetweenInclusive::Both;
                if tokens.get(arg_pos) == Some(&Token::Comma) {
                    arg_pos += 1;
                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "inclusive" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected between() keyword argument: {keyword}"
                            )));
                        }
                        arg_pos += 2;
                    }
                    let inclusive_arg = parse_scalar_literal(tokens, &mut arg_pos)?;
                    let Scalar::Utf8(value) = inclusive_arg else {
                        return Err(ExprError::ParseError(format!(
                            "between() inclusive must be a string literal, got {inclusive_arg:?}"
                        )));
                    };
                    inclusive = BetweenInclusive::parse(&value)?;
                }
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after between() arguments".into(),
                    ));
                }
                expr = Expr::Between {
                    expr: Box::new(expr),
                    left,
                    right,
                    inclusive,
                };
                *pos = arg_pos + 1;
            }
            "clip" => {
                let mut arg_pos = *pos + 3;
                let mut lower = None;
                let mut upper = None;
                let mut lower_seen = false;
                let mut upper_seen = false;
                let mut positional_count = 0_u8;
                let mut keyword_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated clip() arguments".into(),
                        ));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        keyword_seen = true;
                        let keyword = keyword.clone();
                        arg_pos += 2;
                        match keyword.as_str() {
                            "lower" => {
                                if lower_seen {
                                    return Err(ExprError::ParseError(
                                        "clip() lower argument was provided more than once".into(),
                                    ));
                                }
                                lower = parse_optional_numeric_literal(
                                    tokens,
                                    &mut arg_pos,
                                    "clip() lower",
                                )?;
                                lower_seen = true;
                            }
                            "upper" => {
                                if upper_seen {
                                    return Err(ExprError::ParseError(
                                        "clip() upper argument was provided more than once".into(),
                                    ));
                                }
                                upper = parse_optional_numeric_literal(
                                    tokens,
                                    &mut arg_pos,
                                    "clip() upper",
                                )?;
                                upper_seen = true;
                            }
                            other => {
                                return Err(ExprError::ParseError(format!(
                                    "unexpected clip() keyword argument: {other}"
                                )));
                            }
                        }
                    } else {
                        if keyword_seen {
                            return Err(ExprError::ParseError(
                                "clip() positional arguments cannot follow keyword arguments"
                                    .into(),
                            ));
                        }
                        match positional_count {
                            0 => {
                                lower = parse_optional_numeric_literal(
                                    tokens,
                                    &mut arg_pos,
                                    "clip() lower",
                                )?;
                                lower_seen = true;
                            }
                            1 => {
                                upper = parse_optional_numeric_literal(
                                    tokens,
                                    &mut arg_pos,
                                    "clip() upper",
                                )?;
                                upper_seen = true;
                            }
                            _ => {
                                return Err(ExprError::ParseError(
                                    "clip() accepts at most lower and upper bounds".into(),
                                ));
                            }
                        }
                        positional_count += 1;
                    }

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "clip() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in clip() arguments, got {other:?}"
                            )));
                        }
                    }
                }
                expr = Expr::Clip {
                    expr: Box::new(expr),
                    lower,
                    upper,
                };
                *pos = arg_pos + 1;
            }
            "cumsum" | "cumprod" | "cummin" | "cummax" => {
                let mut arg_pos = *pos + 3;
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "skipna" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected {method}() keyword argument: {keyword}"
                            )));
                        }
                        arg_pos += 2;
                        let skipna = parse_bool_literal_argument(
                            tokens,
                            &mut arg_pos,
                            &format!("{method}() skipna"),
                        )?;
                        if !skipna {
                            return Err(ExprError::ParseError(format!(
                                "{method}() skipna=False is not supported in expressions"
                            )));
                        }
                    } else {
                        return Err(ExprError::ParseError(format!(
                            "{method}() only accepts skipna=True in expressions"
                        )));
                    }

                    if tokens.get(arg_pos) != Some(&Token::RParen) {
                        return Err(ExprError::ParseError(format!(
                            "expected ')' after {method}() arguments"
                        )));
                    }
                }

                expr = match method.as_str() {
                    "cumsum" => Expr::CumSum {
                        expr: Box::new(expr),
                    },
                    "cumprod" => Expr::CumProd {
                        expr: Box::new(expr),
                    },
                    "cummin" => Expr::CumMin {
                        expr: Box::new(expr),
                    },
                    "cummax" => Expr::CumMax {
                        expr: Box::new(expr),
                    },
                    other => {
                        return Err(ExprError::ParseError(format!(
                            "unsupported cumulative method: {other}"
                        )));
                    }
                };
                *pos = arg_pos + 1;
            }
            "pct_change" => {
                let mut arg_pos = *pos + 3;
                let mut periods = 1_usize;
                let mut periods_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(
                            "unterminated pct_change() arguments".into(),
                        ));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "periods" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected pct_change() keyword argument: {keyword}"
                            )));
                        }
                        if periods_seen {
                            return Err(ExprError::ParseError(
                                "pct_change() periods argument was provided more than once".into(),
                            ));
                        }
                        arg_pos += 2;
                    } else if periods_seen {
                        return Err(ExprError::ParseError(
                            "pct_change() accepts only one periods argument".into(),
                        ));
                    }

                    periods =
                        parse_usize_literal_argument(tokens, &mut arg_pos, "pct_change() periods")?;
                    periods_seen = true;

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(
                                    "pct_change() arguments cannot end with ','".into(),
                                ));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in pct_change() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = Expr::PctChange {
                    expr: Box::new(expr),
                    periods,
                };
                *pos = arg_pos + 1;
            }
            "shift" | "diff" => {
                let mut arg_pos = *pos + 3;
                let mut periods = 1_i64;
                let mut periods_seen = false;

                while tokens.get(arg_pos) != Some(&Token::RParen) {
                    if arg_pos >= tokens.len() {
                        return Err(ExprError::ParseError(format!(
                            "unterminated {method}() arguments"
                        )));
                    }

                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "periods" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected {method}() keyword argument: {keyword}"
                            )));
                        }
                        if periods_seen {
                            return Err(ExprError::ParseError(format!(
                                "{method}() periods argument was provided more than once"
                            )));
                        }
                        arg_pos += 2;
                    } else if periods_seen {
                        return Err(ExprError::ParseError(format!(
                            "{method}() accepts only one periods argument"
                        )));
                    }

                    periods = parse_i64_literal_argument(
                        tokens,
                        &mut arg_pos,
                        &format!("{method}() periods"),
                    )?;
                    periods_seen = true;

                    match tokens.get(arg_pos) {
                        Some(Token::Comma) => {
                            arg_pos += 1;
                            if tokens.get(arg_pos) == Some(&Token::RParen) {
                                return Err(ExprError::ParseError(format!(
                                    "{method}() arguments cannot end with ','"
                                )));
                            }
                        }
                        Some(Token::RParen) => {}
                        other => {
                            return Err(ExprError::ParseError(format!(
                                "expected ',' or ')' in {method}() arguments, got {other:?}"
                            )));
                        }
                    }
                }

                expr = if method == "shift" {
                    Expr::Shift {
                        expr: Box::new(expr),
                        periods,
                    }
                } else {
                    Expr::Diff {
                        expr: Box::new(expr),
                        periods,
                    }
                };
                *pos = arg_pos + 1;
            }
            "round" => {
                let mut arg_pos = *pos + 3;
                let decimals = if tokens.get(arg_pos) == Some(&Token::RParen) {
                    0
                } else {
                    if let Some(Token::Ident(keyword)) = tokens.get(arg_pos)
                        && tokens.get(arg_pos + 1) == Some(&Token::Assign)
                    {
                        if keyword != "decimals" {
                            return Err(ExprError::ParseError(format!(
                                "unexpected round() keyword argument: {keyword}"
                            )));
                        }
                        arg_pos += 2;
                    }
                    parse_i32_literal(tokens, &mut arg_pos, "round() decimals")?
                };
                if tokens.get(arg_pos) != Some(&Token::RParen) {
                    return Err(ExprError::ParseError(
                        "expected ')' after round() arguments".into(),
                    ));
                }
                expr = Expr::Round {
                    expr: Box::new(expr),
                    decimals,
                };
                *pos = arg_pos + 1;
            }
            _ => {
                return Err(ExprError::ParseError(format!(
                    "unsupported expression method: {method}"
                )));
            }
        };
    }
    Ok(expr)
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use fp_columnar::ComparisonOp;
    use fp_frame::{FrameError, Series};
    use fp_runtime::{EvidenceLedger, RuntimePolicy};
    use fp_types::{DType, NullKind, Scalar};

    use super::{
        BetweenInclusive, Delta, EvalContext, Expr, ExprError, MaterializedView, SeriesRef,
        evaluate,
    };

    #[test]
    fn expression_add_works_through_series_refs() {
        let a = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into()],
            vec![Scalar::Int64(1), Scalar::Int64(2)],
        )
        .expect("a");
        let b = Series::from_values(
            "b",
            vec![2_i64.into(), 3_i64.into()],
            vec![Scalar::Int64(10), Scalar::Int64(20)],
        )
        .expect("b");

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);

        let expr = Expr::Add {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".to_owned()),
            }),
        };

        let mut ledger = EvidenceLedger::new();
        let out = evaluate(
            &expr,
            &ctx,
            &RuntimePolicy::hardened(Some(10_000)),
            &mut ledger,
        )
        .expect("eval");
        // Alignment introduces gaps → promotes to Float64 for NaN sentinel.
        assert_eq!(out.values()[1], Scalar::Float64(12.0));
    }

    #[test]
    fn expression_sub_mul_div_work_through_series_refs() {
        let a = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into()],
            vec![Scalar::Int64(8), Scalar::Int64(6)],
        )
        .expect("a");
        let b = Series::from_values(
            "b",
            vec![1_i64.into(), 2_i64.into()],
            vec![Scalar::Int64(2), Scalar::Int64(3)],
        )
        .expect("b");

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut ledger = EvidenceLedger::new();
        let sub_out = evaluate(
            &Expr::Sub {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("b".to_owned()),
                }),
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("sub eval");
        assert_eq!(sub_out.values(), &[Scalar::Int64(6), Scalar::Int64(3)]);

        let mut ledger = EvidenceLedger::new();
        let mul_out = evaluate(
            &Expr::Mul {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("b".to_owned()),
                }),
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("mul eval");
        assert_eq!(mul_out.values(), &[Scalar::Int64(16), Scalar::Int64(18)]);

        let mut ledger = EvidenceLedger::new();
        let div_out = evaluate(
            &Expr::Div {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("b".to_owned()),
                }),
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("div eval");
        assert_eq!(
            div_out.values(),
            &[Scalar::Float64(4.0), Scalar::Float64(2.0)]
        );
    }

    #[test]
    fn expression_compare_work_through_series_refs() {
        let a = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into(), 3_i64.into()],
            vec![Scalar::Int64(1), Scalar::Int64(3), Scalar::Int64(2)],
        )
        .expect("a");
        let b = Series::from_values(
            "b",
            vec![1_i64.into(), 2_i64.into(), 3_i64.into()],
            vec![Scalar::Int64(2), Scalar::Int64(2), Scalar::Int64(2)],
        )
        .expect("b");

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut ledger = EvidenceLedger::new();
        let gt_out = evaluate(
            &Expr::Compare {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("b".to_owned()),
                }),
                op: ComparisonOp::Gt,
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("gt eval");
        assert_eq!(
            gt_out.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(false)]
        );

        let mut ledger = EvidenceLedger::new();
        let eq_out = evaluate(
            &Expr::Compare {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("b".to_owned()),
                }),
                op: ComparisonOp::Eq,
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("eq eval");
        assert_eq!(
            eq_out.values(),
            &[Scalar::Bool(false), Scalar::Bool(false), Scalar::Bool(true)]
        );
    }

    #[test]
    fn expression_compare_supports_series_scalar_and_scalar_series() {
        let a = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into(), 3_i64.into()],
            vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
        )
        .expect("a");

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut ledger = EvidenceLedger::new();
        let series_gt_scalar = evaluate(
            &Expr::Compare {
                left: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                right: Box::new(Expr::Literal {
                    value: Scalar::Int64(2),
                }),
                op: ComparisonOp::Gt,
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("series > scalar");
        assert_eq!(
            series_gt_scalar.values(),
            &[Scalar::Bool(false), Scalar::Bool(false), Scalar::Bool(true)]
        );

        let mut ledger = EvidenceLedger::new();
        let scalar_ge_series = evaluate(
            &Expr::Compare {
                left: Box::new(Expr::Literal {
                    value: Scalar::Int64(2),
                }),
                right: Box::new(Expr::Series {
                    name: SeriesRef("a".to_owned()),
                }),
                op: ComparisonOp::Ge,
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("scalar >= series");
        assert_eq!(
            scalar_ge_series.values(),
            &[Scalar::Bool(true), Scalar::Bool(true), Scalar::Bool(false)]
        );
    }

    #[test]
    fn expression_compare_supports_scalar_scalar_with_anchor() {
        let anchor = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into()],
            vec![Scalar::Int64(10), Scalar::Int64(20)],
        )
        .expect("anchor");

        let mut ctx = EvalContext::new();
        ctx.insert_series(anchor);
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut ledger = EvidenceLedger::new();
        let out = evaluate(
            &Expr::Compare {
                left: Box::new(Expr::Literal {
                    value: Scalar::Int64(1),
                }),
                right: Box::new(Expr::Literal {
                    value: Scalar::Int64(0),
                }),
                op: ComparisonOp::Gt,
            },
            &ctx,
            &policy,
            &mut ledger,
        )
        .expect("scalar > scalar");
        assert_eq!(out.values(), &[Scalar::Bool(true), Scalar::Bool(true)]);
    }

    #[test]
    fn expression_logical_ops_support_boolean_masks() {
        let a = Series::from_values(
            "a",
            vec![1_i64.into(), 2_i64.into(), 3_i64.into()],
            vec![
                Scalar::Bool(true),
                Scalar::Bool(false),
                Scalar::Null(fp_types::NullKind::Null),
            ],
        )
        .expect("a");
        let b = Series::from_values(
            "b",
            vec![2_i64.into(), 3_i64.into(), 4_i64.into()],
            vec![
                Scalar::Bool(true),
                Scalar::Null(fp_types::NullKind::Null),
                Scalar::Bool(false),
            ],
        )
        .expect("b");

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);
        let policy = RuntimePolicy::hardened(Some(10_000));

        let and_expr = Expr::And {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".to_owned()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let and_out = evaluate(&and_expr, &ctx, &policy, &mut ledger).expect("and eval");
        assert_eq!(
            and_out.values(),
            &[
                Scalar::Null(fp_types::NullKind::Null),
                Scalar::Bool(false),
                Scalar::Null(fp_types::NullKind::Null),
                Scalar::Bool(false)
            ]
        );

        let or_expr = Expr::Or {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".to_owned()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let or_out = evaluate(&or_expr, &ctx, &policy, &mut ledger).expect("or eval");
        assert_eq!(
            or_out.values(),
            &[
                Scalar::Bool(true),
                Scalar::Bool(true),
                Scalar::Null(fp_types::NullKind::Null),
                Scalar::Null(fp_types::NullKind::Null)
            ]
        );

        let not_expr = Expr::Not {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let not_out = evaluate(&not_expr, &ctx, &policy, &mut ledger).expect("not eval");
        assert_eq!(
            not_out.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Null(fp_types::NullKind::Null)
            ]
        );
    }

    #[test]
    fn eval_context_from_dataframe_builds_series_bindings() {
        let frame = fp_frame::DataFrame::from_dict(
            &["a", "b"],
            vec![
                ("a", vec![Scalar::Int64(1), Scalar::Int64(2)]),
                ("b", vec![Scalar::Int64(10), Scalar::Int64(20)]),
            ],
        )
        .expect("frame");

        let context = EvalContext::from_dataframe(&frame).expect("context");
        let a = context.get_series("a").expect("a series present");
        let b = context.get_series("b").expect("b series present");

        assert_eq!(a.values(), &[Scalar::Int64(1), Scalar::Int64(2)]);
        assert_eq!(b.values(), &[Scalar::Int64(10), Scalar::Int64(20)]);
        assert_eq!(a.index().labels(), frame.index().labels());
        assert_eq!(b.index().labels(), frame.index().labels());
    }

    #[test]
    fn evaluate_on_dataframe_matches_manual_context_eval() {
        let frame = fp_frame::DataFrame::from_dict(
            &["a", "b"],
            vec![
                ("a", vec![Scalar::Int64(3), Scalar::Int64(4)]),
                ("b", vec![Scalar::Int64(30), Scalar::Int64(40)]),
            ],
        )
        .expect("frame");

        let expr = Expr::Add {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".to_owned()),
            }),
        };
        let policy = RuntimePolicy::hardened(Some(10_000));
        let mut ledger = EvidenceLedger::new();
        let via_frame =
            super::evaluate_on_dataframe(&expr, &frame, &policy, &mut ledger).expect("frame eval");

        let mut manual = EvalContext::new();
        manual.insert_series(
            Series::new(
                "a",
                frame.index().clone(),
                frame.column("a").expect("a column").clone(),
            )
            .expect("a series"),
        );
        manual.insert_series(
            Series::new(
                "b",
                frame.index().clone(),
                frame.column("b").expect("b column").clone(),
            )
            .expect("b series"),
        );
        let mut ledger = EvidenceLedger::new();
        let manual_out = evaluate(&expr, &manual, &policy, &mut ledger).expect("manual eval");

        assert_eq!(via_frame.values(), manual_out.values());
        assert_eq!(via_frame.index().labels(), manual_out.index().labels());
    }

    #[test]
    fn filter_dataframe_on_expr_applies_boolean_mask() {
        let frame = fp_frame::DataFrame::from_dict(
            &["a", "b"],
            vec![
                (
                    "a",
                    vec![
                        Scalar::Int64(1),
                        Scalar::Int64(2),
                        Scalar::Int64(3),
                        Scalar::Int64(4),
                    ],
                ),
                (
                    "b",
                    vec![
                        Scalar::Int64(10),
                        Scalar::Int64(20),
                        Scalar::Int64(30),
                        Scalar::Int64(40),
                    ],
                ),
            ],
        )
        .expect("frame");

        let expr = Expr::Compare {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".to_owned()),
            }),
            right: Box::new(Expr::Literal {
                value: Scalar::Int64(2),
            }),
            op: ComparisonOp::Gt,
        };
        let policy = RuntimePolicy::hardened(Some(10_000));
        let mut ledger = EvidenceLedger::new();
        let filtered = super::filter_dataframe_on_expr(&expr, &frame, &policy, &mut ledger)
            .expect("filter via expr");

        assert_eq!(filtered.len(), 2);
        assert_eq!(
            filtered.column("a").expect("a").values(),
            &[Scalar::Int64(3), Scalar::Int64(4)]
        );
        assert_eq!(
            filtered.column("b").expect("b").values(),
            &[Scalar::Int64(30), Scalar::Int64(40)]
        );
    }

    #[test]
    fn filter_dataframe_on_expr_rejects_non_boolean_mask() {
        let frame = fp_frame::DataFrame::from_dict(
            &["a"],
            vec![("a", vec![Scalar::Int64(1), Scalar::Int64(2)])],
        )
        .expect("frame");

        let expr = Expr::Series {
            name: SeriesRef("a".to_owned()),
        };
        let policy = RuntimePolicy::hardened(Some(10_000));
        let mut ledger = EvidenceLedger::new();
        let err = super::filter_dataframe_on_expr(&expr, &frame, &policy, &mut ledger).unwrap_err();

        assert!(matches!(
            err,
            ExprError::Frame(FrameError::CompatibilityRejected(msg))
                if msg.contains("boolean mask required for query-style filter")
        ));
    }

    // === AG-15: Incremental View Maintenance Tests ===

    fn make_series(name: &str, labels: Vec<i64>, values: Vec<Scalar>) -> Series {
        Series::from_values(
            name,
            labels.into_iter().map(fp_index::IndexLabel::from).collect(),
            values,
        )
        .expect("series")
    }

    #[test]
    fn materialized_view_from_full_eval() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(10), Scalar::Int64(20)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::Series {
            name: SeriesRef("a".into()),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("full eval");
        assert_eq!(view.result.values().len(), 2);
    }

    #[test]
    fn ivm_append_delta_series_ref() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(10), Scalar::Int64(20)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::Series {
            name: SeriesRef("a".into()),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("full eval");
        assert_eq!(view.result.values().len(), 2);

        // Append 2 new rows
        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into(), 3_i64.into()],
            new_values: vec![Scalar::Int64(30), Scalar::Int64(40)],
        };

        // Update context with full new series
        let a_full = make_series(
            "a",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(10),
                Scalar::Int64(20),
                Scalar::Int64(30),
                Scalar::Int64(40),
            ],
        );
        ctx.insert_series(a_full);

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");
        assert_eq!(view.result.values().len(), 4);
        assert_eq!(view.result.values()[2], Scalar::Int64(30));
        assert_eq!(view.result.values()[3], Scalar::Int64(40));
    }

    #[test]
    fn ivm_append_delta_add_expression() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(1), Scalar::Int64(2)]);
        let b = make_series("b", vec![0, 1], vec![Scalar::Int64(10), Scalar::Int64(20)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b.clone());

        let expr = Expr::Add {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("full eval");
        assert_eq!(view.result.values().len(), 2);
        assert_eq!(view.result.values()[0], Scalar::Int64(11));
        assert_eq!(view.result.values()[1], Scalar::Int64(22));

        // Append rows to "a" — "b" needs corresponding rows at labels 2,3
        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into(), 3_i64.into()],
            new_values: vec![Scalar::Int64(3), Scalar::Int64(4)],
        };

        let a_full = make_series(
            "a",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(1),
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Int64(4),
            ],
        );
        let b_full = make_series(
            "b",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(10),
                Scalar::Int64(20),
                Scalar::Int64(30),
                Scalar::Int64(40),
            ],
        );
        ctx.insert_series(a_full);
        ctx.insert_series(b_full);

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");
        assert_eq!(view.result.values().len(), 4);
        // New rows: 3+30=33, 4+40=44
        assert_eq!(view.result.values()[2], Scalar::Int64(33));
        assert_eq!(view.result.values()[3], Scalar::Int64(44));
    }

    #[test]
    fn ivm_append_delta_mul_expression() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(2), Scalar::Int64(3)]);
        let b = make_series("b", vec![0, 1], vec![Scalar::Int64(4), Scalar::Int64(5)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);

        let expr = Expr::Mul {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");
        assert_eq!(view.result.values(), &[Scalar::Int64(8), Scalar::Int64(15)]);

        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into(), 3_i64.into()],
            new_values: vec![Scalar::Int64(4), Scalar::Int64(6)],
        };
        let a_full = make_series(
            "a",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Int64(4),
                Scalar::Int64(6),
            ],
        );
        let b_full = make_series(
            "b",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(4),
                Scalar::Int64(5),
                Scalar::Int64(6),
                Scalar::Int64(7),
            ],
        );
        ctx.insert_series(a_full);
        ctx.insert_series(b_full);

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");
        assert_eq!(
            view.result.values(),
            &[
                Scalar::Int64(8),
                Scalar::Int64(15),
                Scalar::Int64(24),
                Scalar::Int64(42)
            ]
        );
    }

    #[test]
    fn ivm_append_delta_comparison_expression() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(1), Scalar::Int64(2)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::Compare {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Literal {
                value: Scalar::Int64(1),
            }),
            op: ComparisonOp::Gt,
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");
        assert_eq!(
            view.result.values(),
            &[Scalar::Bool(false), Scalar::Bool(true)]
        );

        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into(), 3_i64.into()],
            new_values: vec![Scalar::Int64(3), Scalar::Int64(0)],
        };
        ctx.insert_series(make_series(
            "a",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(1),
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Int64(0),
            ],
        ));

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");
        assert_eq!(
            view.result.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(true),
                Scalar::Bool(false)
            ]
        );
    }

    #[test]
    fn ivm_append_delta_logical_expression() {
        let a = make_series(
            "a",
            vec![0, 1],
            vec![Scalar::Bool(false), Scalar::Bool(true)],
        );
        let b = make_series(
            "b",
            vec![0, 1],
            vec![Scalar::Bool(true), Scalar::Bool(true)],
        );
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);

        let expr = Expr::And {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");
        assert_eq!(
            view.result.values(),
            &[Scalar::Bool(false), Scalar::Bool(true)]
        );

        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into()],
            new_values: vec![Scalar::Bool(true)],
        };
        let a_full = make_series(
            "a",
            vec![0, 1, 2],
            vec![Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(true)],
        );
        let b_full = make_series(
            "b",
            vec![0, 1, 2],
            vec![Scalar::Bool(true), Scalar::Bool(true), Scalar::Bool(false)],
        );
        ctx.insert_series(a_full);
        ctx.insert_series(b_full);

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");
        assert_eq!(
            view.result.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(false)]
        );
    }

    #[test]
    fn ivm_isomorphism_incremental_matches_full() {
        // The key correctness property: incremental result must equal full re-eval.
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(5), Scalar::Int64(10)]);
        let b = make_series(
            "b",
            vec![0, 1],
            vec![Scalar::Int64(100), Scalar::Int64(200)],
        );
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);

        let expr = Expr::Add {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");

        // Apply delta
        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into()],
            new_values: vec![Scalar::Int64(15)],
        };
        let a_full = make_series(
            "a",
            vec![0, 1, 2],
            vec![Scalar::Int64(5), Scalar::Int64(10), Scalar::Int64(15)],
        );
        let b_full = make_series(
            "b",
            vec![0, 1, 2],
            vec![Scalar::Int64(100), Scalar::Int64(200), Scalar::Int64(300)],
        );
        ctx.insert_series(a_full);
        ctx.insert_series(b_full);

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("incremental");

        // Full re-evaluation for comparison
        let full_result = evaluate(&expr, &ctx, &policy, &mut ledger).expect("full");

        // Compare: incremental result must match full
        assert_eq!(view.result.values().len(), full_result.values().len());
        for (i, (inc, full)) in view
            .result
            .values()
            .iter()
            .zip(full_result.values().iter())
            .enumerate()
        {
            assert!(
                inc.semantic_eq(full),
                "mismatch at position {i}: incremental={inc:?} full={full:?}"
            );
        }
    }

    #[test]
    fn ivm_multiple_deltas() {
        let a = make_series("a", vec![0], vec![Scalar::Int64(1)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::Series {
            name: SeriesRef("a".into()),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");

        // First delta
        let delta1 = Delta {
            series_name: "a".into(),
            new_labels: vec![1_i64.into()],
            new_values: vec![Scalar::Int64(2)],
        };
        ctx.insert_series(make_series(
            "a",
            vec![0, 1],
            vec![Scalar::Int64(1), Scalar::Int64(2)],
        ));
        view.apply_delta(&delta1, &ctx, &policy, &mut ledger)
            .expect("delta1");
        assert_eq!(view.result.values().len(), 2);

        // Second delta
        let delta2 = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into(), 3_i64.into()],
            new_values: vec![Scalar::Int64(3), Scalar::Int64(4)],
        };
        ctx.insert_series(make_series(
            "a",
            vec![0, 1, 2, 3],
            vec![
                Scalar::Int64(1),
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Int64(4),
            ],
        ));
        view.apply_delta(&delta2, &ctx, &policy, &mut ledger)
            .expect("delta2");
        assert_eq!(view.result.values().len(), 4);
        assert_eq!(view.result.values()[3], Scalar::Int64(4));
    }

    #[test]
    fn ivm_falls_back_for_cumulative_expressions() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(1), Scalar::Int64(2)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::CumSum {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");
        assert_eq!(view.result.values(), &[Scalar::Int64(1), Scalar::Int64(3)]);

        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into()],
            new_values: vec![Scalar::Int64(3)],
        };
        ctx.insert_series(make_series(
            "a",
            vec![0, 1, 2],
            vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
        ));

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");

        let full_result = evaluate(&expr, &ctx, &policy, &mut ledger).expect("full");
        assert_eq!(view.result.values(), full_result.values());
        assert_eq!(
            view.result.values(),
            &[Scalar::Int64(1), Scalar::Int64(3), Scalar::Int64(6)]
        );
    }

    #[test]
    fn ivm_falls_back_for_order_dependent_expressions() {
        let a = make_series("a", vec![0, 1], vec![Scalar::Int64(2), Scalar::Int64(1)]);
        let mut ctx = EvalContext::new();
        ctx.insert_series(a);

        let expr = Expr::SortValues {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            ascending: true,
            na_position: "last".into(),
        };
        let mut ledger = EvidenceLedger::new();
        let policy = RuntimePolicy::hardened(Some(10_000));

        let mut view =
            MaterializedView::from_full_eval(&expr, &ctx, &policy, &mut ledger).expect("base");
        assert_eq!(view.result.values(), &[Scalar::Int64(1), Scalar::Int64(2)]);

        let delta = Delta {
            series_name: "a".into(),
            new_labels: vec![2_i64.into()],
            new_values: vec![Scalar::Int64(0)],
        };
        ctx.insert_series(make_series(
            "a",
            vec![0, 1, 2],
            vec![Scalar::Int64(2), Scalar::Int64(1), Scalar::Int64(0)],
        ));

        view.apply_delta(&delta, &ctx, &policy, &mut ledger)
            .expect("delta");

        let full_result = evaluate(&expr, &ctx, &policy, &mut ledger).expect("full");
        assert_eq!(view.result.values(), full_result.values());
        assert_eq!(
            view.result.values(),
            &[Scalar::Int64(0), Scalar::Int64(1), Scalar::Int64(2)]
        );
    }

    #[test]
    fn ivm_is_linear_detects_expressions() {
        assert!(MaterializedView::is_linear(&Expr::Series {
            name: SeriesRef("a".into()),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Add {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Sub {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Mul {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Div {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Modulo {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::FloorDiv {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Pow {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::And {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Or {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
        }));
        assert!(MaterializedView::is_linear(&Expr::Not {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::Compare {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Series {
                name: SeriesRef("b".into()),
            }),
            op: ComparisonOp::Ge,
        }));
        assert!(MaterializedView::is_linear(&Expr::Compare {
            left: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            right: Box::new(Expr::Literal {
                value: Scalar::Int64(1),
            }),
            op: ComparisonOp::Gt,
        }));
        assert!(!MaterializedView::is_linear(&Expr::Compare {
            left: Box::new(Expr::Literal {
                value: Scalar::Int64(1),
            }),
            right: Box::new(Expr::Literal {
                value: Scalar::Int64(2),
            }),
            op: ComparisonOp::Lt,
        }));
        assert!(!MaterializedView::is_linear(&Expr::Literal {
            value: Scalar::Int64(42),
        }));
        assert!(!MaterializedView::is_linear(&Expr::CumSum {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
        }));
        assert!(!MaterializedView::is_linear(&Expr::SortValues {
            expr: Box::new(Expr::Series {
                name: SeriesRef("a".into()),
            }),
            ascending: true,
            na_position: "last".into(),
        }));
    }

    // ── Parser tests ──

    #[test]
    fn parse_simple_comparison() {
        let expr = super::parse_expr("col_a > 5").unwrap();
        assert_eq!(
            expr,
            Expr::Compare {
                left: Box::new(Expr::Series {
                    name: SeriesRef("col_a".into()),
                }),
                right: Box::new(Expr::Literal {
                    value: Scalar::Int64(5),
                }),
                op: ComparisonOp::Gt,
            }
        );
    }

    #[test]
    fn parse_chained_comparison_lowers_to_pairwise_and() {
        let expr = super::parse_expr("a < b <= c").unwrap();
        let Expr::And { left, right } = expr else {
            panic!("expected And, got {expr:?}");
        };

        assert!(matches!(
            left.as_ref(),
            Expr::Compare {
                op: ComparisonOp::Lt,
                ..
            }
        ));
        assert!(matches!(
            right.as_ref(),
            Expr::Compare {
                op: ComparisonOp::Le,
                ..
            }
        ));

        if let Expr::Compare { left, right, .. } = left.as_ref() {
            assert_eq!(
                left.as_ref(),
                &Expr::Series {
                    name: SeriesRef("a".into())
                }
            );
            assert_eq!(
                right.as_ref(),
                &Expr::Series {
                    name: SeriesRef("b".into())
                }
            );
        }
        if let Expr::Compare { left, right, .. } = right.as_ref() {
            assert_eq!(
                left.as_ref(),
                &Expr::Series {
                    name: SeriesRef("b".into())
                }
            );
            assert_eq!(
                right.as_ref(),
                &Expr::Series {
                    name: SeriesRef("c".into())
                }
            );
        }
    }

    #[test]
    fn parse_and_or_expression() {
        let expr = super::parse_expr("a > 1 and b < 2").unwrap();
        assert!(
            matches!(&expr, Expr::And { .. }),
            "expected And, got {expr:?}"
        );
        if let Expr::And { left, right } = expr {
            assert!(matches!(*left, Expr::Compare { .. }));
            assert!(matches!(*right, Expr::Compare { .. }));
        }
    }

    #[test]
    fn parse_not_expression() {
        let expr = super::parse_expr("not a == 1").unwrap();
        assert!(matches!(expr, Expr::Not { .. }));
    }

    #[test]
    fn parse_arithmetic() {
        let expr = super::parse_expr("a + b * 2").unwrap();
        // Should parse as a + (b * 2) due to precedence
        assert!(
            matches!(&expr, Expr::Add { .. }),
            "expected Add, got {expr:?}"
        );
        if let Expr::Add { left, right } = expr {
            assert!(matches!(*left, Expr::Series { .. }));
            assert!(matches!(*right, Expr::Mul { .. }));
        }
    }

    #[test]
    fn parse_arithmetic_advanced() {
        let expr = super::parse_expr("a ** b // 2 % 3").unwrap();
        assert!(
            matches!(&expr, Expr::Modulo { .. }),
            "expected Modulo, got {expr:?}"
        );
        if let Expr::Modulo { left, .. } = expr {
            assert!(
                matches!(*left, Expr::FloorDiv { .. }),
                "expected FloorDiv, got {left:?}"
            );
            if let Expr::FloorDiv {
                left: inner_left, ..
            } = *left
            {
                assert!(
                    matches!(*inner_left, Expr::Pow { .. }),
                    "expected Pow, got {inner_left:?}"
                );
            }
        }
    }

    #[test]
    fn parse_unary_minus_identifier() {
        let expr = super::parse_expr("-a").unwrap();
        assert!(
            matches!(&expr, Expr::Sub { .. }),
            "expected Sub, got {expr:?}"
        );
    }

    #[test]
    fn parse_unary_plus_identifier() {
        let expr = super::parse_expr("+a").unwrap();
        assert!(
            matches!(&expr, Expr::Series { .. }),
            "expected Series, got {expr:?}"
        );
    }

    #[test]
    fn parse_parentheses() {
        let expr = super::parse_expr("(a + b) * 2").unwrap();
        assert!(
            matches!(&expr, Expr::Mul { .. }),
            "expected Mul, got {expr:?}"
        );
        if let Expr::Mul { left, right } = expr {
            assert!(matches!(*left, Expr::Add { .. }));
            assert!(matches!(
                *right,
                Expr::Literal {
                    value: Scalar::Int64(2)
                }
            ));
        }
    }

    #[test]
    fn parse_string_literal() {
        let expr = super::parse_expr("name == 'alice'").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, op, .. } = expr {
            assert_eq!(op, ComparisonOp::Eq);
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Utf8("alice".into())
                }
            );
        }
    }

    #[test]
    fn parse_pandas_boolean_literals() {
        let true_expr = super::parse_expr("flag == True").unwrap();
        let true_right = match &true_expr {
            Expr::Compare { right, .. } => Some(right.as_ref()),
            _ => None,
        };
        assert_eq!(
            true_right,
            Some(&Expr::Literal {
                value: Scalar::Bool(true)
            })
        );

        let false_expr = super::parse_expr("flag != False").unwrap();
        let false_right = match &false_expr {
            Expr::Compare { right, .. } => Some(right.as_ref()),
            _ => None,
        };
        assert_eq!(
            false_right,
            Some(&Expr::Literal {
                value: Scalar::Bool(false)
            })
        );
    }

    #[test]
    fn parse_float_literal() {
        let expr = super::parse_expr("x > 4.56").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Float64(4.56)
                }
            );
        }
    }

    #[test]
    fn parse_negative_literal() {
        let expr = super::parse_expr("x > -5").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Int64(-5)
                }
            );
        }
    }

    #[test]
    fn parse_leading_dot_float_literal() {
        let expr = super::parse_expr("x > .5").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Float64(0.5)
                }
            );
        }
    }

    #[test]
    fn parse_negative_leading_dot_float_literal() {
        let expr = super::parse_expr("x > -.5").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Float64(-0.5)
                }
            );
        }
    }

    #[test]
    fn parse_local_reference() {
        let expr = super::parse_expr("x > @threshold").unwrap();
        assert!(
            matches!(&expr, Expr::Compare { .. }),
            "expected Compare, got {expr:?}"
        );
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Local {
                    name: "threshold".into(),
                }
            );
        }
    }

    #[test]
    fn parse_membership_list_literal() {
        let expr = super::parse_expr("a in [1, 3,]").unwrap();
        let Expr::IsIn {
            left,
            values,
            negated,
        } = expr
        else {
            panic!("expected IsIn expression");
        };
        assert_eq!(
            left.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert!(!negated);
        assert_eq!(values, vec![Scalar::Int64(1), Scalar::Int64(3)]);
    }

    #[test]
    fn parse_not_in_membership_list_literal() {
        let expr = super::parse_expr("name not in ['x', 'y']").unwrap();
        let Expr::IsIn {
            left,
            values,
            negated,
        } = expr
        else {
            panic!("expected IsIn expression");
        };
        assert_eq!(
            left.as_ref(),
            &Expr::Series {
                name: SeriesRef("name".into())
            }
        );
        assert!(negated);
        assert_eq!(
            values,
            vec![Scalar::Utf8("x".into()), Scalar::Utf8("y".into())]
        );
    }

    #[test]
    fn parse_abs_function_call() {
        let expr = super::parse_expr("abs(a - 3)").unwrap();
        let Expr::Abs { expr: inner } = expr else {
            panic!("expected Abs expression");
        };
        assert!(matches!(inner.as_ref(), Expr::Sub { .. }));
    }

    #[test]
    fn parse_abs_method_call() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.abs()")?;
        let Expr::Abs { expr: inner } = expr else {
            return Err(ExprError::ParseError("expected Abs expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        Ok(())
    }

    #[test]
    fn parse_null_predicate_method_calls() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.isna()")?;
        let Expr::IsNull {
            expr: inner,
            negated,
        } = expr
        else {
            return Err(ExprError::ParseError("expected IsNull expression".into()));
        };
        assert!(!negated);
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );

        let expr = super::parse_expr("a.notnull()")?;
        let Expr::IsNull { negated, .. } = expr else {
            return Err(ExprError::ParseError("expected IsNull expression".into()));
        };
        assert!(negated);
        Ok(())
    }

    #[test]
    fn parse_fillna_method_call() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.fillna(value=-3)")?;
        let Expr::FillNa { expr: inner, value } = expr else {
            return Err(ExprError::ParseError("expected FillNa expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(value, Scalar::Int64(-3));
        Ok(())
    }

    #[test]
    fn parse_comparison_method_call() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.ge(other=b + 1)")?;
        let Expr::Compare { left, right, op } = expr else {
            return Err(ExprError::ParseError("expected Compare expression".into()));
        };
        assert_eq!(op, ComparisonOp::Ge);
        assert_eq!(
            left.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert!(matches!(right.as_ref(), Expr::Add { .. }));
        Ok(())
    }

    #[test]
    fn parse_arithmetic_method_call() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.rsub(other=b + 10)")?;
        let Expr::Sub { left, right } = expr else {
            return Err(ExprError::ParseError("expected Sub expression".into()));
        };
        assert!(matches!(left.as_ref(), Expr::Add { .. }));
        assert_eq!(
            right.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        Ok(())
    }

    #[test]
    fn parse_where_mask_method_call() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.mask(cond=a.gt(1), other=b + 10)")?;
        let Expr::Where {
            expr: inner,
            cond,
            other,
            mask,
        } = expr
        else {
            return Err(ExprError::ParseError("expected Where expression".into()));
        };
        assert!(mask);
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert!(matches!(cond.as_ref(), Expr::Compare { .. }));
        assert!(matches!(other.as_deref(), Some(Expr::Add { .. })));
        Ok(())
    }

    #[test]
    fn parse_isin_method_call_list_literal() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.isin([1, 3,])")?;
        let Expr::IsIn {
            left,
            values,
            negated,
        } = expr
        else {
            return Err(ExprError::ParseError("expected IsIn expression".into()));
        };
        assert_eq!(
            left.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert!(!negated);
        assert_eq!(values, vec![Scalar::Int64(1), Scalar::Int64(3)]);
        Ok(())
    }

    #[test]
    fn parse_between_method_call_scalar_bounds() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.between(2, 8)")?;
        let Expr::Between {
            expr: inner,
            left,
            right,
            inclusive,
        } = expr
        else {
            return Err(ExprError::ParseError("expected Between expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(left, Scalar::Int64(2));
        assert_eq!(right, Scalar::Int64(8));
        assert_eq!(inclusive, BetweenInclusive::Both);
        Ok(())
    }

    #[test]
    fn parse_between_method_call_inclusive_argument() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.between(2, 8, inclusive=\"left\")")?;
        let Expr::Between { inclusive, .. } = expr else {
            return Err(ExprError::ParseError("expected Between expression".into()));
        };
        assert_eq!(inclusive, BetweenInclusive::Left);

        let positional = super::parse_expr("a.between(2, 8, \"right\")")?;
        let Expr::Between { inclusive, .. } = positional else {
            return Err(ExprError::ParseError("expected Between expression".into()));
        };
        assert_eq!(inclusive, BetweenInclusive::Right);
        Ok(())
    }

    #[test]
    fn parse_clip_method_call_numeric_bounds() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.clip(2, 8)")?;
        let Expr::Clip {
            expr: inner,
            lower,
            upper,
        } = expr
        else {
            return Err(ExprError::ParseError("expected Clip expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(lower, Some(2.0));
        assert_eq!(upper, Some(8.0));

        let one_sided = super::parse_expr("a.clip(upper=8)")?;
        let Expr::Clip { lower, upper, .. } = one_sided else {
            return Err(ExprError::ParseError("expected Clip expression".into()));
        };
        assert_eq!(lower, None);
        assert_eq!(upper, Some(8.0));

        let mixed = super::parse_expr("a.clip(2, upper=None)")?;
        let Expr::Clip { lower, upper, .. } = mixed else {
            return Err(ExprError::ParseError("expected Clip expression".into()));
        };
        assert_eq!(lower, Some(2.0));
        assert_eq!(upper, None);
        Ok(())
    }

    #[test]
    fn parse_shift_and_diff_method_call_periods() -> Result<(), ExprError> {
        let shift_default = super::parse_expr("a.shift()")?;
        let Expr::Shift { periods, .. } = shift_default else {
            return Err(ExprError::ParseError("expected Shift expression".into()));
        };
        assert_eq!(periods, 1);

        let shift_keyword = super::parse_expr("a.shift(periods=-1)")?;
        let Expr::Shift { periods, .. } = shift_keyword else {
            return Err(ExprError::ParseError("expected Shift expression".into()));
        };
        assert_eq!(periods, -1);

        let diff_positional = super::parse_expr("a.diff(2)")?;
        let Expr::Diff {
            expr: inner,
            periods,
        } = diff_positional
        else {
            return Err(ExprError::ParseError("expected Diff expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(periods, 2);
        Ok(())
    }

    #[test]
    fn parse_cumulative_method_calls() -> Result<(), ExprError> {
        let cumsum = super::parse_expr("a.cumsum()")?;
        let Expr::CumSum { expr: inner } = cumsum else {
            return Err(ExprError::ParseError("expected CumSum expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );

        assert!(matches!(
            super::parse_expr("a.cumprod(skipna=True)")?,
            Expr::CumProd { .. }
        ));
        assert!(matches!(
            super::parse_expr("a.cummin()")?,
            Expr::CumMin { .. }
        ));
        assert!(matches!(
            super::parse_expr("a.cummax()")?,
            Expr::CumMax { .. }
        ));
        assert!(super::parse_expr("a.cumsum(skipna=False)").is_err());
        Ok(())
    }

    #[test]
    fn parse_pct_change_method_call_periods() -> Result<(), ExprError> {
        let default = super::parse_expr("a.pct_change()")?;
        let Expr::PctChange { periods, .. } = default else {
            return Err(ExprError::ParseError(
                "expected PctChange expression".into(),
            ));
        };
        assert_eq!(periods, 1);

        let positional = super::parse_expr("a.pct_change(2)")?;
        let Expr::PctChange {
            expr: inner,
            periods,
        } = positional
        else {
            return Err(ExprError::ParseError(
                "expected PctChange expression".into(),
            ));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(periods, 2);

        let keyword = super::parse_expr("a.pct_change(periods=3)")?;
        let Expr::PctChange { periods, .. } = keyword else {
            return Err(ExprError::ParseError(
                "expected PctChange expression".into(),
            ));
        };
        assert_eq!(periods, 3);
        assert!(super::parse_expr("a.pct_change(-1)").is_err());
        Ok(())
    }

    #[test]
    fn parse_round_method_call_optional_decimals() -> Result<(), ExprError> {
        let expr = super::parse_expr("a.round()")?;
        let Expr::Round {
            expr: inner,
            decimals,
        } = expr
        else {
            return Err(ExprError::ParseError("expected Round expression".into()));
        };
        assert_eq!(
            inner.as_ref(),
            &Expr::Series {
                name: SeriesRef("a".into())
            }
        );
        assert_eq!(decimals, 0);

        let positional = super::parse_expr("a.round(-1)")?;
        let Expr::Round { decimals, .. } = positional else {
            return Err(ExprError::ParseError("expected Round expression".into()));
        };
        assert_eq!(decimals, -1);

        let keyword = super::parse_expr("a.round(decimals=1)")?;
        let Expr::Round { decimals, .. } = keyword else {
            return Err(ExprError::ParseError("expected Round expression".into()));
        };
        assert_eq!(decimals, 1);
        Ok(())
    }

    #[test]
    fn parse_backtick_identifier() {
        let expr = super::parse_expr("`gross margin` + normal").unwrap();
        assert!(matches!(&expr, Expr::Add { .. }));
        if let Expr::Add { left, right } = expr {
            assert_eq!(
                *left,
                Expr::Series {
                    name: SeriesRef("gross margin".into()),
                }
            );
            assert_eq!(
                *right,
                Expr::Series {
                    name: SeriesRef("normal".into()),
                }
            );
        }
    }

    #[test]
    fn parse_error_single_equals() {
        assert!(super::parse_expr("x = 1").is_err());
    }

    #[test]
    fn parse_error_unterminated_backtick_identifier() {
        let err = super::parse_expr("`gross margin + normal").unwrap_err();
        assert!(
            matches!(err, ExprError::ParseError(msg) if msg.contains("unterminated backtick identifier"))
        );
    }

    #[test]
    fn parse_string_with_escaped_quote() {
        let expr = super::parse_expr(r#"name == 'O\'Brien'"#).unwrap();
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Utf8("O'Brien".into())
                }
            );
        } else {
            panic!("expected Compare");
        }
    }

    #[test]
    fn parse_string_with_escape_sequences() {
        let expr = super::parse_expr(r#"msg == "hello\nworld""#).unwrap();
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Utf8("hello\nworld".into())
                }
            );
        } else {
            panic!("expected Compare");
        }
    }

    #[test]
    fn parse_string_with_escaped_backslash() {
        let expr = super::parse_expr(r#"path == "c:\\users""#).unwrap();
        if let Expr::Compare { right, .. } = expr {
            assert_eq!(
                *right,
                Expr::Literal {
                    value: Scalar::Utf8("c:\\users".into())
                }
            );
        } else {
            panic!("expected Compare");
        }
    }

    #[test]
    fn parse_error_multiple_decimal_points() {
        let err = super::parse_expr("x > 1.2.3").unwrap_err();
        assert!(
            matches!(err, ExprError::ParseError(msg) if msg.contains("multiple decimal points"))
        );
    }

    #[test]
    fn parse_error_negative_multiple_decimal_points() {
        let err = super::parse_expr("x > -1.2.3").unwrap_err();
        assert!(
            matches!(err, ExprError::ParseError(msg) if msg.contains("multiple decimal points"))
        );
    }

    #[test]
    fn parse_error_leading_dot_multiple_decimal_points() {
        let err = super::parse_expr("x > .1.2").unwrap_err();
        assert!(
            matches!(err, ExprError::ParseError(msg) if msg.contains("multiple decimal points"))
        );
    }

    #[test]
    fn parse_eval_integration() {
        // Parse, then evaluate against a context
        let expr = super::parse_expr("a + b").unwrap();
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let a = fp_frame::Series::from_values(
            "a",
            vec![0_i64.into(), 1_i64.into()],
            vec![Scalar::Int64(10), Scalar::Int64(20)],
        )
        .unwrap();
        let b = fp_frame::Series::from_values(
            "b",
            vec![0_i64.into(), 1_i64.into()],
            vec![Scalar::Int64(3), Scalar::Int64(7)],
        )
        .unwrap();

        let mut ctx = EvalContext::new();
        ctx.insert_series(a);
        ctx.insert_series(b);

        let result = evaluate(&expr, &ctx, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(13));
        assert_eq!(result.values()[1], Scalar::Int64(27));
    }

    #[test]
    fn eval_str_arithmetic() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(3), Scalar::Int64(7)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::eval_str("a + b", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(13));
        assert_eq!(result.values()[1], Scalar::Int64(27));
    }

    #[test]
    fn eval_str_reflected_arithmetic_methods_pizig() {
        // br-frankenpandas-pizig: reflected method-call arithmetic (rsub/rfloordiv)
        // swaps operands: a.rsub(b)==b-a. Seeded LCG, no mocks.
        let mut st: u64 = 0x4ef1_e7a1_2b3c_4d5e;
        let mut next = || {
            st = st
                .wrapping_mul(6_364_136_223_846_793_005)
                .wrapping_add(1_442_695_040_888_963_407);
            (st >> 33) as u32
        };
        let policy = RuntimePolicy::hardened(Some(100));
        for iter in 0..400u32 {
            let n = (next() % 5) as usize + 1;
            let a: Vec<i64> = (0..n).map(|_| (next() % 40) as i64 - 20).collect();
            let b: Vec<i64> = (0..n).map(|_| (next() % 40) as i64 - 20).collect();
            let mut ledger = EvidenceLedger::new();
            let frame = fp_frame::DataFrame::from_series(vec![
                fp_frame::Series::from_values(
                    "a",
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    a.iter().map(|&v| Scalar::Int64(v)).collect::<Vec<_>>(),
                )
                .unwrap(),
                fp_frame::Series::from_values(
                    "b",
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    b.iter().map(|&v| Scalar::Int64(v)).collect::<Vec<_>>(),
                )
                .unwrap(),
            ])
            .unwrap();
            let geti = |s: &Scalar| -> i64 {
                match s {
                    Scalar::Int64(v) => *v,
                    Scalar::Float64(v) => *v as i64,
                    _ => i64::MIN,
                }
            };
            let rsub = super::eval_str("a.rsub(b)", &frame, &policy, &mut ledger).unwrap();
            for i in 0..n {
                assert_eq!(
                    geti(&rsub.values()[i]),
                    b[i] - a[i],
                    "rsub iter={iter} i={i}"
                );
            }
        }
    }

    #[test]
    fn eval_str_true_division_float_8bnl8() {
        // br-frankenpandas-8bnl8: expression-level true division a/b -> Float64
        // (3j187 used %,// not /). Seeded LCG, no mocks.
        let mut st: u64 = 0xd1f0_e7a1_2b3c_4d5e;
        let mut next = || {
            st = st
                .wrapping_mul(6_364_136_223_846_793_005)
                .wrapping_add(1_442_695_040_888_963_407);
            (st >> 33) as u32
        };
        let policy = RuntimePolicy::hardened(Some(100));
        for iter in 0..400u32 {
            let n = (next() % 5) as usize + 1;
            let a: Vec<i64> = (0..n).map(|_| (next() % 40) as i64 - 20).collect();
            let b: Vec<i64> = (0..n).map(|_| ((next() % 5) as i64) + 1).collect(); // non-zero
            let mut ledger = EvidenceLedger::new();
            let frame = fp_frame::DataFrame::from_series(vec![
                fp_frame::Series::from_values(
                    "a",
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    a.iter().map(|&v| Scalar::Int64(v)).collect::<Vec<_>>(),
                )
                .unwrap(),
                fp_frame::Series::from_values(
                    "b",
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    b.iter().map(|&v| Scalar::Int64(v)).collect::<Vec<_>>(),
                )
                .unwrap(),
            ])
            .unwrap();
            let out = super::eval_str("a / b", &frame, &policy, &mut ledger).unwrap();
            for i in 0..n {
                let got = match &out.values()[i] {
                    Scalar::Float64(x) => *x,
                    Scalar::Int64(x) => *x as f64,
                    _ => f64::NAN,
                };
                assert!(
                    (got - a[i] as f64 / b[i] as f64).abs() < 1e-9,
                    "div iter={iter} i={i} {}/{}",
                    a[i],
                    b[i]
                );
            }
        }
    }

    #[test]
    fn eval_str_arithmetic_precedence_3j187() {
        // End-to-end correctness oracle (br-frankenpandas-3j187): parse + evaluate
        // arithmetic over a small Int64 frame, asserting per-row results computed
        // by hand. Int operands keep Int64 (exact — no float fragility).
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let col = |name: &str, vals: Vec<i64>| {
            fp_frame::Series::from_values(
                name,
                (0..vals.len() as i64).map(|i| i.into()).collect::<Vec<_>>(),
                vals.into_iter().map(Scalar::Int64).collect::<Vec<_>>(),
            )
            .unwrap()
        };
        // Row0: a=10,b=3,c=2  Row1: a=20,b=4,c=5
        let frame = fp_frame::DataFrame::from_series(vec![
            col("a", vec![10, 20]),
            col("b", vec![3, 4]),
            col("c", vec![2, 5]),
        ])
        .unwrap();

        let i = |x: i64, y: i64| vec![Scalar::Int64(x), Scalar::Int64(y)];
        let mut ev = |e: &str| {
            super::eval_str(e, &frame, &policy, &mut ledger)
                .unwrap()
                .values()
                .to_vec()
        };

        assert_eq!(ev("a + b * c"), i(16, 40), "* binds tighter than +");
        assert_eq!(
            ev("(a + b) * c"),
            i(26, 120),
            "parentheses override precedence"
        );
        assert_eq!(ev("a - b - c"), i(5, 11), "- is left-associative");
        assert_eq!(ev("a * b + c"), i(32, 85), "* before +");
        assert_eq!(ev("a % b"), i(1, 0), "modulo");
        assert_eq!(ev("a // b"), i(3, 5), "floor division");
        assert_eq!(ev("-a"), i(-10, -20), "unary minus");
        assert_eq!(
            ev("a > b"),
            vec![Scalar::Bool(true), Scalar::Bool(true)],
            "comparison yields bool"
        );
    }

    #[test]
    fn query_str_parenthesized_boolean_rnyb3() {
        // br-frankenpandas-rnyb3: parenthesized boolean grouping '(a>2) and (b<8)'.
        let mut st: u64 = 0x4c02_e7a1_2b3c_4d5e;
        let mut next = || {
            st = st
                .wrapping_mul(6_364_136_223_846_793_005)
                .wrapping_add(1_442_695_040_888_963_407);
            (st >> 33) as u32
        };
        let policy = RuntimePolicy::hardened(Some(100));
        for iter in 0..300u32 {
            let n = (next() % 8) as usize + 1;
            let a: Vec<i64> = (0..n).map(|_| (next() % 8) as i64).collect();
            let b: Vec<i64> = (0..n).map(|_| (next() % 12) as i64).collect();
            let mut ledger = EvidenceLedger::new();
            let col = |name: &str, v: &[i64]| {
                fp_frame::Series::from_values(
                    name,
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    v.iter().map(|&x| Scalar::Int64(x)).collect::<Vec<_>>(),
                )
                .unwrap()
            };
            let frame = fp_frame::DataFrame::from_series(vec![col("a", &a), col("b", &b)]).unwrap();
            let out =
                super::query_str("(a > 2) and (b < 8)", &frame, &policy, &mut ledger).unwrap();
            let got: Vec<i64> = out
                .column("a")
                .unwrap()
                .values()
                .iter()
                .map(|s| match s {
                    Scalar::Int64(x) => *x,
                    _ => i64::MIN,
                })
                .collect();
            let exp: Vec<i64> = (0..n)
                .filter(|&i| a[i] > 2 && b[i] < 8)
                .map(|i| a[i])
                .collect();
            assert_eq!(got, exp, "paren query iter={iter}");
        }
    }

    #[test]
    fn query_str_compound_arith_comparison_1l96h() {
        // br-frankenpandas-1l96h: query 'a + b > c' (arithmetic sub-expr then
        // comparison). Seeded LCG, no mocks.
        let mut st: u64 = 0x4c01_e7a1_2b3c_4d5e;
        let mut next = || {
            st = st
                .wrapping_mul(6_364_136_223_846_793_005)
                .wrapping_add(1_442_695_040_888_963_407);
            (st >> 33) as u32
        };
        let policy = RuntimePolicy::hardened(Some(100));
        for iter in 0..300u32 {
            let n = (next() % 8) as usize + 1;
            let a: Vec<i64> = (0..n).map(|_| (next() % 20) as i64).collect();
            let b: Vec<i64> = (0..n).map(|_| (next() % 20) as i64).collect();
            let c: Vec<i64> = (0..n).map(|_| (next() % 30) as i64).collect();
            let mut ledger = EvidenceLedger::new();
            let col = |name: &str, v: &[i64]| {
                fp_frame::Series::from_values(
                    name,
                    (0..n as i64).map(Into::into).collect::<Vec<_>>(),
                    v.iter().map(|&x| Scalar::Int64(x)).collect::<Vec<_>>(),
                )
                .unwrap()
            };
            let frame =
                fp_frame::DataFrame::from_series(vec![col("a", &a), col("b", &b), col("c", &c)])
                    .unwrap();
            let out = super::query_str("a + b > c", &frame, &policy, &mut ledger).unwrap();
            let got_a: Vec<i64> = out
                .column("a")
                .unwrap()
                .values()
                .iter()
                .map(|s| match s {
                    Scalar::Int64(x) => *x,
                    _ => i64::MIN,
                })
                .collect();
            let exp_a: Vec<i64> = (0..n)
                .filter(|&i| a[i] + b[i] > c[i])
                .map(|i| a[i])
                .collect();
            assert_eq!(got_a, exp_a, "compound query iter={iter}");
        }
    }

    #[test]
    fn query_str_string_equality_x2nfn() {
        // br-frankenpandas-x2nfn: query string-literal equality on a Utf8 column.
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let s = |v: &str| Scalar::Utf8(v.to_owned());
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "s",
                (0..4i64).map(Into::into).collect::<Vec<_>>(),
                vec![s("a"), s("b"), s("b"), s("c")],
            )
            .unwrap(),
        ])
        .unwrap();
        let kept = |q: &str, ledger: &mut EvidenceLedger| -> Vec<String> {
            super::query_str(q, &frame, &policy, ledger)
                .unwrap()
                .column("s")
                .unwrap()
                .values()
                .iter()
                .map(|v| match v {
                    Scalar::Utf8(x) => x.clone(),
                    _ => String::new(),
                })
                .collect()
        };
        assert_eq!(
            kept("s == 'b'", &mut ledger),
            vec!["b".to_string(), "b".to_string()],
            "eq"
        );
        assert_eq!(
            kept("s != 'b'", &mut ledger),
            vec!["a".to_string(), "c".to_string()],
            "ne"
        );
    }

    #[test]
    fn query_str_boolean_not_negation_flrzd() {
        // br-frankenpandas-flrzd: query 'not' negation.
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                (0..10).map(|i| (i as i64).into()).collect::<Vec<_>>(),
                (0..10).map(Scalar::Int64).collect::<Vec<_>>(),
            )
            .unwrap(),
        ])
        .unwrap();
        let kept = |q: &str, ledger: &mut EvidenceLedger| -> Vec<i64> {
            super::query_str(q, &frame, &policy, ledger)
                .unwrap()
                .column("x")
                .unwrap()
                .values()
                .iter()
                .map(|s| match s {
                    Scalar::Int64(v) => *v,
                    _ => i64::MIN,
                })
                .collect()
        };
        assert_eq!(
            kept("not (x > 5)", &mut ledger),
            vec![0, 1, 2, 3, 4, 5],
            "not gt"
        );
        assert_eq!(
            kept("not (x < 3 or x > 7)", &mut ledger),
            vec![3, 4, 5, 6, 7],
            "not (or)"
        );
    }

    #[test]
    fn query_str_boolean_filter_correctness_wo4wi() {
        // End-to-end correctness oracle (br-frankenpandas-wo4wi): query_str must
        // keep exactly the rows matching the boolean predicate. No rand, no mocks.
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                (0..10).map(|i| (i as i64).into()).collect::<Vec<_>>(),
                (0..10).map(Scalar::Int64).collect::<Vec<_>>(),
            )
            .unwrap(),
        ])
        .unwrap();
        let kept = |q: &str, ledger: &mut EvidenceLedger| -> Vec<i64> {
            super::query_str(q, &frame, &policy, ledger)
                .unwrap()
                .column("x")
                .unwrap()
                .values()
                .iter()
                .map(|s| match s {
                    Scalar::Int64(v) => *v,
                    // Sentinel: any non-Int64 would fail the equality assert below.
                    _ => i64::MIN,
                })
                .collect()
        };

        assert_eq!(kept("x < 2 or x > 7", &mut ledger), vec![0, 1, 8, 9], "or");
        assert_eq!(kept("x > 2 and x < 6", &mut ledger), vec![3, 4, 5], "and");
        assert_eq!(
            kept("x >= 5 and x <= 5", &mut ledger),
            vec![5],
            "single row"
        );
    }

    #[test]
    fn query_str_membership_filter_correctness_8e5vp() {
        // End-to-end correctness (br-frankenpandas-8e5vp): in / not in membership.
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                (0..7).map(|i| (i as i64).into()).collect::<Vec<_>>(),
                (0..7).map(Scalar::Int64).collect::<Vec<_>>(),
            )
            .unwrap(),
        ])
        .unwrap();
        let kept = |q: &str, ledger: &mut EvidenceLedger| -> Vec<i64> {
            super::query_str(q, &frame, &policy, ledger)
                .unwrap()
                .column("a")
                .unwrap()
                .values()
                .iter()
                .map(|s| match s {
                    Scalar::Int64(v) => *v,
                    _ => i64::MIN,
                })
                .collect()
        };
        assert_eq!(kept("a in [1, 3, 5]", &mut ledger), vec![1, 3, 5], "in");
        assert_eq!(
            kept("a not in [0, 2, 4, 6]", &mut ledger),
            vec![1, 3, 5],
            "not in"
        );
    }

    #[test]
    fn eval_query_with_locals_substitution_0qyro() {
        // End-to-end (br-frankenpandas-0qyro): @local scalar bindings substitute
        // into eval/query (the BTreeMap key omits the leading '@').
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        let mut offset = std::collections::BTreeMap::new();
        offset.insert("offset".to_string(), Scalar::Int64(100));
        let r = super::eval_str_with_locals("a + @offset", &frame, &offset, &policy, &mut ledger)
            .unwrap();
        assert_eq!(
            r.values(),
            &[Scalar::Int64(110), Scalar::Int64(120), Scalar::Int64(130)],
            "a + @offset"
        );

        let mut lo = std::collections::BTreeMap::new();
        lo.insert("lo".to_string(), Scalar::Int64(15));
        let q = super::query_str_with_locals("a > @lo", &frame, &lo, &policy, &mut ledger).unwrap();
        let kept: Vec<_> = q.column("a").unwrap().values().to_vec();
        assert_eq!(
            kept,
            vec![Scalar::Int64(20), Scalar::Int64(30)],
            "a > @lo keeps {{20,30}}"
        );
    }

    #[test]
    fn query_str_chained_comparison_and_ops_match_pandas() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                (0..5).map(|i| (i as i64).into()).collect(),
                (1..=5).map(Scalar::Int64).collect(),
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                (0..5).map(|i| (i as i64).into()).collect(),
                vec![Scalar::Int64(2); 5],
            )
            .unwrap(),
        ])
        .unwrap();

        // Python chained comparison: `1 < a < 4` == `(1 < a) and (a < 4)` -> a in {2,3}.
        let q = super::query_str("1 < a < 4", &frame, &policy, &mut ledger).unwrap();
        let qa: Vec<_> = q.column("a").unwrap().values().to_vec();
        assert_eq!(
            qa,
            vec![Scalar::Int64(2), Scalar::Int64(3)],
            "chained comparison 1<a<4 should keep a in {{2,3}}; got {:?}",
            qa
        );

        // Floor-division and modulo on int operands: pandas keeps int64.
        // (verified vs pandas 2.2.3: a//b == [0,1,1,2,2], a%b == [1,0,1,0,1])
        let fdiv = super::eval_str("a // b", &frame, &policy, &mut ledger).unwrap();
        let fv: Vec<_> = fdiv.values().to_vec();
        assert_eq!(
            fv[3],
            Scalar::Int64(2),
            "4 // 2 should be Int64(2); got {:?}",
            fv[3]
        );
        let md = super::eval_str("a % b", &frame, &policy, &mut ledger).unwrap();
        let mv: Vec<_> = md.values().to_vec();
        assert_eq!(
            mv[0],
            Scalar::Int64(1),
            "1 % 2 should be Int64(1); got {:?}",
            mv[0]
        );

        // NOTE: `a ** 2` (int**int) currently returns Float64 in FrankenPandas
        // (fp-columnar Pow kernel uses powf), diverging from pandas's int64.
        // Tracked separately; not asserted here.
    }

    fn surviving_a(frame: &fp_frame::DataFrame, expr: &str) -> Result<Vec<i64>, String> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();
        let r =
            super::query_str(expr, frame, &policy, &mut ledger).map_err(|e| format!("{e:?}"))?;
        Ok(r.column("a")
            .unwrap()
            .values()
            .iter()
            .filter_map(|s| {
                if let Scalar::Int64(v) = s {
                    Some(*v)
                } else {
                    None
                }
            })
            .collect())
    }

    #[test]
    fn query_str_operators_probe_vs_pandas() {
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                (0..5).map(|i| (i as i64).into()).collect(),
                (1..=5).map(Scalar::Int64).collect(),
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "s",
                (0..5).map(|i| (i as i64).into()).collect(),
                ["x", "y", "x", "z", "y"]
                    .iter()
                    .map(|s| Scalar::Utf8((*s).into()))
                    .collect(),
            )
            .unwrap(),
        ])
        .unwrap();

        // Each verified vs pandas 2.2.3 df.query(...)["a"].tolist().
        assert_eq!(
            surviving_a(&frame, "a in [1, 3, 5]"),
            Ok(vec![1, 3, 5]),
            "a in list"
        );
        assert_eq!(
            surviving_a(&frame, "a not in [1, 3, 5]"),
            Ok(vec![2, 4]),
            "a not in list"
        );
        assert_eq!(surviving_a(&frame, "s == 'x'"), Ok(vec![1, 3]), "str eq");
        assert_eq!(
            surviving_a(&frame, "s in ['x', 'z']"),
            Ok(vec![1, 3, 4]),
            "str in list"
        );
        assert_eq!(
            surviving_a(&frame, "a > 1 and a < 4"),
            Ok(vec![2, 3]),
            "logical and"
        );
        assert_eq!(
            surviving_a(&frame, "not (a > 3)"),
            Ok(vec![1, 2, 3]),
            "unary not"
        );
        assert_eq!(
            surviving_a(&frame, "a + 1 > 3"),
            Ok(vec![3, 4, 5]),
            "arith precedence"
        );
        // pandas treats &/| in query as element-wise and/or with
        // comparison-level precedence: `a > 1 & a < 4` == `(a>1) & (a<4)`.
        assert_eq!(
            surviving_a(&frame, "a > 1 & a < 4"),
            Ok(vec![2, 3]),
            "bitwise &"
        );
        assert_eq!(
            surviving_a(&frame, "a == 1 | a == 5"),
            Ok(vec![1, 5]),
            "bitwise |"
        );
    }

    #[test]
    fn eval_str_with_locals_broadcasts_scalar_bindings() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20)],
            )
            .unwrap(),
        ])
        .unwrap();

        let locals = BTreeMap::from([("offset".to_owned(), Scalar::Int64(5))]);
        let result =
            super::eval_str_with_locals("a + @offset * 2", &frame, &locals, &policy, &mut ledger)
                .unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(20));
        assert_eq!(result.values()[1], Scalar::Int64(30));
    }

    #[test]
    fn eval_str_accepts_backtick_column_names() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "gross margin",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "and",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result =
            super::eval_str("`gross margin` + `and`", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(11));
        assert_eq!(result.values()[1], Scalar::Int64(22));
    }

    #[test]
    fn eval_str_unary_minus_respects_pow_precedence() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values("anchor", vec![0_i64.into()], vec![Scalar::Int64(1)])
                .unwrap(),
        ])
        .unwrap();

        // Verified vs live pandas 2.2.3 (br-frankenpandas-k75hq):
        // pd.eval('-2 ** 2') -> -4 (int) and pd.eval('(-2) ** 2') -> 4 (int).
        // int ** int stays integer (numpy/pandas semantics); the old Float64
        // expectations codified FP's pre-typed-pow behavior. The precedence
        // property under test is unchanged: unary minus binds LOOSER than **.
        let result = super::eval_str("-2 ** 2", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(-4));

        let result = super::eval_str("(-2) ** 2", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(4));
    }

    #[test]
    fn query_str_with_locals_filters_rows() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(5), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        let locals = BTreeMap::from([
            ("threshold".to_owned(), Scalar::Int64(2)),
            ("limit".to_owned(), Scalar::Int64(25)),
        ]);
        let result = super::query_str_with_locals(
            "a > @threshold and b < @limit",
            &frame,
            &locals,
            &policy,
            &mut ledger,
        )
        .unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result.columns()["a"].values()[0], Scalar::Int64(5));
        assert_eq!(result.columns()["b"].values()[0], Scalar::Int64(20));
    }

    #[test]
    fn query_str_accepts_backtick_column_names() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "gross margin",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "and",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str(
            "`gross margin` > 15 and `and` < 3",
            &frame,
            &policy,
            &mut ledger,
        )
        .unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(
            result.columns()["gross margin"].values()[0],
            Scalar::Int64(20)
        );
        assert_eq!(result.columns()["and"].values()[0], Scalar::Int64(2));
    }

    #[test]
    fn eval_str_with_locals_errors_on_unknown_binding() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values("a", vec![0_i64.into()], vec![Scalar::Int64(10)])
                .unwrap(),
        ])
        .unwrap();

        let err = super::eval_str_with_locals(
            "a > @threshold",
            &frame,
            &BTreeMap::new(),
            &policy,
            &mut ledger,
        )
        .unwrap_err();
        assert!(matches!(err, ExprError::UnknownLocal(name) if name == "threshold"));
    }

    #[test]
    fn query_str_filters_rows() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(5), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "y",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("x > 2", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 2); // rows where x=5 and x=3
        assert_eq!(result.columns()["x"].values()[0], Scalar::Int64(5));
        assert_eq!(result.columns()["x"].values()[1], Scalar::Int64(3));
    }

    #[test]
    fn query_str_filters_with_list_membership() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Utf8("x".into()),
                    Scalar::Utf8("y".into()),
                    Scalar::Utf8("x".into()),
                ],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("a in [1, 3]", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(
            result.columns()["a"].values(),
            &[Scalar::Int64(1), Scalar::Int64(3)]
        );

        let result = super::query_str("b not in ['x']", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result.columns()["a"].values(), &[Scalar::Int64(2)]);
        assert_eq!(result.columns()["b"].values(), &[Scalar::Utf8("y".into())]);
    }

    #[test]
    fn query_str_exposes_index_namespace() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![5_i64.into(), 6_i64.into(), 7_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("index > 5", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.index().labels(), &[6_i64.into(), 7_i64.into()]);
        assert_eq!(
            result.columns()["a"].values(),
            &[Scalar::Int64(20), Scalar::Int64(30)]
        );

        let result = super::query_str("ilevel_0 > 5", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.index().labels(), &[6_i64.into(), 7_i64.into()]);
    }

    #[test]
    fn query_str_exposes_named_index_without_shadowing_columns() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "idx",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(5), Scalar::Int64(6), Scalar::Int64(7)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "value",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap()
        .set_index("idx", true)
        .unwrap();

        let result = super::query_str("idx > 5", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.index().labels(), &[6_i64.into(), 7_i64.into()]);
        assert_eq!(
            result.columns()["value"].values(),
            &[Scalar::Int64(20), Scalar::Int64(30)]
        );

        let shadowing_frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "index",
                vec![5_i64.into(), 6_i64.into(), 7_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(10), Scalar::Int64(1)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "value",
                vec![5_i64.into(), 6_i64.into(), 7_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("index > 5", &shadowing_frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.index().labels(), &[6_i64.into()]);
        assert_eq!(result.columns()["index"].values(), &[Scalar::Int64(10)]);
    }

    #[test]
    fn eval_and_query_str_accept_abs_function() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![
                    0_i64.into(),
                    1_i64.into(),
                    2_i64.into(),
                    3_i64.into(),
                    4_i64.into(),
                ],
                vec![
                    Scalar::Int64(-2),
                    Scalar::Int64(-1),
                    Scalar::Int64(0),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                ],
            )
            .unwrap(),
        ])
        .unwrap();

        let evaluated = super::eval_str("abs(a)", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(
            evaluated.values(),
            &[
                Scalar::Int64(2),
                Scalar::Int64(1),
                Scalar::Int64(0),
                Scalar::Int64(1),
                Scalar::Int64(2),
            ]
        );

        let filtered = super::query_str("abs(a) > 1", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 4_i64.into()]);
        assert_eq!(
            filtered.columns()["a"].values(),
            &[Scalar::Int64(-2), Scalar::Int64(2)]
        );
    }

    #[test]
    fn eval_and_query_str_accept_abs_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(-2), Scalar::Int64(-1), Scalar::Int64(3)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let evaluated = super::eval_str("a.abs()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            evaluated.values(),
            &[Scalar::Int64(2), Scalar::Int64(1), Scalar::Int64(3)]
        );

        let filtered = super::query_str("a.abs() >= 2", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 2_i64.into()]);
        assert_eq!(
            filtered.columns()["a"].values(),
            &[Scalar::Int64(-2), Scalar::Int64(3)]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_null_predicate_methods() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Null(fp_types::NullKind::Null),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let isna = super::eval_str("a.isna()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            isna.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(false)]
        );

        let isnull = super::query_str("a.isnull()", &frame, &policy, &mut ledger)?;
        assert_eq!(isnull.index().labels(), &[1_i64.into()]);
        assert_eq!(
            isnull.columns()["a"].values(),
            &[Scalar::Null(fp_types::NullKind::Null)]
        );

        let notna = super::query_str("a.notna()", &frame, &policy, &mut ledger)?;
        assert_eq!(notna.index().labels(), &[0_i64.into(), 2_i64.into()]);

        let notnull = super::eval_str("a.notnull()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            notnull.values(),
            &[Scalar::Bool(true), Scalar::Bool(false), Scalar::Bool(true)]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_fillna_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Null(fp_types::NullKind::Null),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let filled = super::eval_str("a.fillna(0)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            filled.values(),
            &[Scalar::Int64(1), Scalar::Int64(0), Scalar::Int64(3)]
        );

        let filtered = super::query_str("a.fillna(0) < 2", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 1_i64.into()]);

        let keyword_filtered =
            super::query_str("a.fillna(value=5) >= 3", &frame, &policy, &mut ledger)?;
        assert_eq!(
            keyword_filtered.index().labels(),
            &[1_i64.into(), 2_i64.into()]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_dropna_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Null(fp_types::NullKind::Null),
                    Scalar::Int64(3),
                    Scalar::Int64(4),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(30),
                    Scalar::Int64(10),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let dropped = super::eval_str("a.dropna()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            dropped.index().labels(),
            &[0_i64.into(), 2_i64.into(), 3_i64.into()]
        );
        assert_eq!(
            dropped.values(),
            &[Scalar::Int64(1), Scalar::Int64(3), Scalar::Int64(4)]
        );

        let predicate = super::eval_str("a.dropna().gt(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            predicate.index().labels(),
            &[0_i64.into(), 2_i64.into(), 3_i64.into()]
        );
        assert_eq!(
            predicate.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(true)]
        );

        let filtered = super::query_str(
            "a.dropna().gt(2) and b.gt(20)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[2_i64.into()]);

        assert!(super::parse_expr("a.dropna(0)").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_sort_values_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let sorted = super::eval_str("a.sort_values()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            sorted.index().labels(),
            &[2_i64.into(), 3_i64.into(), 0_i64.into(), 1_i64.into()]
        );
        assert_eq!(
            sorted.values(),
            &[
                Scalar::Int64(1),
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Null(fp_types::NullKind::NaN)
            ]
        );

        let descending_first = super::eval_str(
            "a.sort_values(ascending=False, na_position='first')",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            descending_first.index().labels(),
            &[1_i64.into(), 0_i64.into(), 3_i64.into(), 2_i64.into()]
        );

        let filtered = super::query_str(
            "a.sort_values().gt(1) and b.gt(10)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 3_i64.into()]);

        assert!(super::parse_expr("a.sort_values(False)").is_err());
        assert!(super::parse_expr("a.sort_values(kind='mergesort')").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_sort_index_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let labels = vec![
            10_i64.into(),
            7_i64.into(),
            12_i64.into(),
            11_i64.into(),
            8_i64.into(),
        ];
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                labels.clone(),
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                labels,
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let sorted = super::eval_str("a.sort_index()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            sorted.index().labels(),
            &[
                7_i64.into(),
                8_i64.into(),
                10_i64.into(),
                11_i64.into(),
                12_i64.into()
            ]
        );
        assert_eq!(
            sorted.values(),
            &[
                Scalar::Null(fp_types::NullKind::NaN),
                Scalar::Int64(3),
                Scalar::Int64(3),
                Scalar::Int64(2),
                Scalar::Int64(1)
            ]
        );

        let descending = super::eval_str(
            "a.sort_index(ascending=False)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            descending.index().labels(),
            &[
                12_i64.into(),
                11_i64.into(),
                10_i64.into(),
                8_i64.into(),
                7_i64.into()
            ]
        );

        let stable = super::eval_str(
            "a.sort_index(axis='rows', kind='stable', na_position='last', sort_remaining=True)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            stable.index().labels(),
            &[
                7_i64.into(),
                8_i64.into(),
                10_i64.into(),
                11_i64.into(),
                12_i64.into()
            ]
        );

        let ignored_index = super::eval_str(
            "a.sort_index(ascending=False, ignore_index=True)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            ignored_index.index().labels(),
            &[
                0_i64.into(),
                1_i64.into(),
                2_i64.into(),
                3_i64.into(),
                4_i64.into()
            ]
        );
        assert_eq!(
            ignored_index.values(),
            &[
                Scalar::Int64(1),
                Scalar::Int64(2),
                Scalar::Int64(3),
                Scalar::Int64(3),
                Scalar::Null(fp_types::NullKind::NaN)
            ]
        );

        let filtered = super::query_str(
            "a.sort_index(ascending=False).gt(2) and b.gt(40)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[8_i64.into()]);

        assert!(super::parse_expr("a.sort_index(False)").is_err());
        assert!(super::parse_expr("a.sort_index(axis=1)").is_err());
        assert!(super::parse_expr("a.sort_index(kind='bogus')").is_err());
        assert!(super::parse_expr("a.sort_index(na_position='bogus')").is_err());
        assert!(super::parse_expr("a.sort_index(ignore_index='yes')").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_argsort_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let labels = vec![
            10_i64.into(),
            7_i64.into(),
            12_i64.into(),
            11_i64.into(),
            8_i64.into(),
        ];
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                labels.clone(),
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                labels,
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let argsorted = super::eval_str("a.argsort()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            argsorted.index().labels(),
            &[
                10_i64.into(),
                7_i64.into(),
                12_i64.into(),
                11_i64.into(),
                8_i64.into()
            ]
        );
        assert_eq!(
            argsorted.values(),
            &[
                Scalar::Int64(1),
                Scalar::Int64(-1),
                Scalar::Int64(2),
                Scalar::Int64(0),
                Scalar::Int64(3)
            ]
        );

        let stable = super::eval_str("a.argsort(0, 'stable')", &frame, &policy, &mut ledger)?;
        assert_eq!(stable.values(), argsorted.values());

        let keyword = super::eval_str(
            "a.argsort(axis='rows', kind='mergesort', order=None, stable=True)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(keyword.values(), argsorted.values());

        let filtered = super::query_str(
            "a.argsort().ge(0) and b.gt(20)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[10_i64.into(), 8_i64.into()]);

        assert!(super::parse_expr("a.argsort(axis=1)").is_err());
        assert!(super::parse_expr("a.argsort(kind='bogus')").is_err());
        assert!(super::parse_expr("a.argsort(0, kind='stable', 'extra')").is_err());
        assert!(super::parse_expr("a.argsort(ascending=False)").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_mode_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![
                    10_i64.into(),
                    7_i64.into(),
                    12_i64.into(),
                    11_i64.into(),
                    8_i64.into(),
                ],
                vec![
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(3),
                    Scalar::Int64(3),
                    Scalar::Int64(1),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![
                    10_i64.into(),
                    7_i64.into(),
                    12_i64.into(),
                    11_i64.into(),
                    8_i64.into(),
                ],
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let mode = super::eval_str("a.mode()", &frame, &policy, &mut ledger)?;
        assert_eq!(mode.index().labels(), &[0_i64.into()]);
        assert_eq!(mode.values(), &[Scalar::Int64(3)]);

        let with_na = super::eval_str("a.mode(False)", &frame, &policy, &mut ledger)?;
        assert_eq!(with_na.index().labels(), &[0_i64.into(), 1_i64.into()]);
        assert_eq!(
            with_na.values(),
            &[Scalar::Int64(3), Scalar::Null(fp_types::NullKind::NaN)]
        );

        let keyword = super::eval_str("a.mode(dropna=False)", &frame, &policy, &mut ledger)?;
        assert_eq!(keyword.values(), with_na.values());

        let filtered =
            super::query_str("a.mode().ge(3) and b.gt(40)", &frame, &policy, &mut ledger)?;
        assert!(filtered.index().labels().is_empty());

        assert!(super::parse_expr("a.mode(dropna=False, True)").is_err());
        assert!(super::parse_expr("a.mode(skipna=False)").is_err());
        assert!(super::parse_expr("a.mode(dropna='no')").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_head_tail_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let labels = vec![
            10_i64.into(),
            7_i64.into(),
            12_i64.into(),
            11_i64.into(),
            8_i64.into(),
        ];
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                labels.clone(),
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                labels,
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let head = super::eval_str("a.head(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(head.index().labels(), &[10_i64.into(), 7_i64.into()]);
        assert_eq!(
            head.values(),
            &[Scalar::Int64(3), Scalar::Null(fp_types::NullKind::NaN)]
        );

        let head_negative = super::eval_str("a.head(n=-1)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            head_negative.index().labels(),
            &[10_i64.into(), 7_i64.into(), 12_i64.into(), 11_i64.into()]
        );

        let tail = super::eval_str("a.tail(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(tail.index().labels(), &[11_i64.into(), 8_i64.into()]);
        assert_eq!(tail.values(), &[Scalar::Int64(2), Scalar::Int64(3)]);

        let tail_negative = super::eval_str("a.tail(n=-1)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            tail_negative.index().labels(),
            &[7_i64.into(), 12_i64.into(), 11_i64.into(), 8_i64.into()]
        );

        let filtered =
            super::query_str("a.head().gt(1) and b.gt(20)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[10_i64.into(), 8_i64.into()]);

        assert!(super::parse_expr("a.head(1, 2)").is_err());
        assert!(super::parse_expr("a.tail(n=1, 2)").is_err());
        assert!(super::parse_expr("a.head(skipna=True)").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_deduplicate_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let labels = vec![
            10_i64.into(),
            7_i64.into(),
            12_i64.into(),
            11_i64.into(),
            8_i64.into(),
        ];
        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                labels.clone(),
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                labels,
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let duplicated = super::eval_str("a.duplicated()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            duplicated.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(false),
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(true),
            ]
        );

        let duplicated_last =
            super::eval_str("a.duplicated(keep='last')", &frame, &policy, &mut ledger)?;
        assert_eq!(
            duplicated_last.values(),
            &[
                Scalar::Bool(true),
                Scalar::Bool(true),
                Scalar::Bool(false),
                Scalar::Bool(false),
                Scalar::Bool(false),
            ]
        );

        let duplicated_none =
            super::eval_str("a.duplicated(keep=False)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            duplicated_none.values(),
            &[
                Scalar::Bool(true),
                Scalar::Bool(true),
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(true),
            ]
        );

        let dropped = super::eval_str("a.drop_duplicates()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            dropped.index().labels(),
            &[10_i64.into(), 7_i64.into(), 12_i64.into()]
        );
        assert_eq!(
            dropped.values(),
            &[
                Scalar::Int64(3),
                Scalar::Null(fp_types::NullKind::NaN),
                Scalar::Int64(1),
            ]
        );

        let dropped_last = super::eval_str(
            "a.drop_duplicates(keep='last')",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            dropped_last.index().labels(),
            &[12_i64.into(), 11_i64.into(), 8_i64.into()]
        );
        assert_eq!(
            dropped_last.values(),
            &[
                Scalar::Int64(1),
                Scalar::Int64(3),
                Scalar::Null(fp_types::NullKind::NaN),
            ]
        );

        let dropped_none =
            super::eval_str("a.drop_duplicates(keep=0)", &frame, &policy, &mut ledger)?;
        assert_eq!(dropped_none.index().labels(), &[12_i64.into()]);
        assert_eq!(dropped_none.values(), &[Scalar::Int64(1)]);

        let filtered = super::query_str(
            "not a.duplicated(keep=False) and b.lt(40)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[12_i64.into()]);

        assert!(super::parse_expr("a.duplicated(keep=True)").is_err());
        assert!(super::parse_expr("a.duplicated(keep='middle')").is_err());
        assert!(super::parse_expr("a.drop_duplicates('last')").is_err());
        assert!(super::parse_expr("a.drop_duplicates('first', 'last')").is_err());
        assert!(super::parse_expr("a.duplicated(skipna=True)").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_top_n_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![
                    0_i64.into(),
                    1_i64.into(),
                    2_i64.into(),
                    3_i64.into(),
                    4_i64.into(),
                ],
                vec![
                    Scalar::Int64(3),
                    Scalar::Null(fp_types::NullKind::NaN),
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![
                    0_i64.into(),
                    1_i64.into(),
                    2_i64.into(),
                    3_i64.into(),
                    4_i64.into(),
                ],
                vec![
                    Scalar::Int64(30),
                    Scalar::Int64(40),
                    Scalar::Int64(10),
                    Scalar::Int64(20),
                    Scalar::Int64(50),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let largest = super::eval_str("a.nlargest(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(largest.index().labels(), &[0_i64.into(), 4_i64.into()]);
        assert_eq!(largest.values(), &[Scalar::Int64(3), Scalar::Int64(3)]);

        let largest_last =
            super::eval_str("a.nlargest(2, keep='last')", &frame, &policy, &mut ledger)?;
        assert_eq!(largest_last.index().labels(), &[4_i64.into(), 0_i64.into()]);

        let smallest = super::eval_str("a.nsmallest(n=2)", &frame, &policy, &mut ledger)?;
        assert_eq!(smallest.index().labels(), &[2_i64.into(), 3_i64.into()]);
        assert_eq!(smallest.values(), &[Scalar::Int64(1), Scalar::Int64(2)]);

        let negative = super::eval_str("a.nlargest(n=-1)", &frame, &policy, &mut ledger)?;
        assert!(negative.is_empty());

        let filtered = super::query_str(
            "a.nlargest(2).gt(2) and b.gt(40)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[4_i64.into()]);

        assert!(super::parse_expr("a.nlargest(2, keep='first', 'last')").is_err());
        assert!(super::parse_expr("a.nsmallest(1, 'first', 'extra')").is_err());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_replace_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(1)],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "label",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Utf8("old".to_owned()),
                    Scalar::Utf8("keep".to_owned()),
                    Scalar::Utf8("old".to_owned()),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let replaced = super::eval_str("a.replace(1, 9)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            replaced.values(),
            &[Scalar::Int64(9), Scalar::Int64(2), Scalar::Int64(9)]
        );

        let keyword = super::eval_str(
            "label.replace(to_replace=\"old\", value=\"new\")",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            keyword.values(),
            &[
                Scalar::Utf8("new".to_owned()),
                Scalar::Utf8("keep".to_owned()),
                Scalar::Utf8("new".to_owned()),
            ]
        );

        let filtered = super::query_str("a.replace(1, 9) == 9", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_astype_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(0), Scalar::Int64(1)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let as_float = super::eval_str("a.astype(\"float64\")", &frame, &policy, &mut ledger)?;
        assert_eq!(as_float.dtype(), DType::Float64);
        assert_eq!(
            as_float.values(),
            &[
                Scalar::Float64(1.0),
                Scalar::Float64(0.0),
                Scalar::Float64(1.0),
            ]
        );

        let as_string = super::eval_str("a.astype(dtype=\"str\")", &frame, &policy, &mut ledger)?;
        assert_eq!(as_string.dtype(), DType::Utf8);
        assert_eq!(
            as_string.values(),
            &[
                Scalar::Utf8("1".to_owned()),
                Scalar::Utf8("0".to_owned()),
                Scalar::Utf8("1".to_owned()),
            ]
        );

        let filtered = super::query_str("a.astype(\"bool\")", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_combine_first_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Null(NullKind::NaN),
                    Scalar::Int64(3),
                ],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(9), Scalar::Int64(8), Scalar::Int64(7)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let combined = super::eval_str("a.combine_first(b)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            combined.values(),
            &[Scalar::Int64(1), Scalar::Int64(8), Scalar::Int64(3)]
        );

        let keyword = super::eval_str("a.combine_first(other=b)", &frame, &policy, &mut ledger)?;
        assert_eq!(keyword.values(), combined.values());

        let filtered = super::query_str("a.combine_first(b).gt(5)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_rank_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(10)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let average = super::eval_str("a.rank()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            average.values(),
            &[
                Scalar::Float64(1.5),
                Scalar::Float64(3.0),
                Scalar::Float64(1.5),
            ]
        );

        let dense = super::eval_str("a.rank(\"dense\")", &frame, &policy, &mut ledger)?;
        assert_eq!(
            dense.values(),
            &[
                Scalar::Float64(1.0),
                Scalar::Float64(2.0),
                Scalar::Float64(1.0),
            ]
        );

        let descending = super::eval_str(
            "a.rank(method=\"dense\", ascending=False, na_option=\"keep\", pct=True)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            descending.values(),
            &[
                Scalar::Float64(1.0),
                Scalar::Float64(0.5),
                Scalar::Float64(1.0),
            ]
        );

        let filtered = super::query_str("a.rank().gt(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_comparison_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(2), Scalar::Int64(2), Scalar::Int64(2)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let eq_scalar = super::eval_str("a.eq(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            eq_scalar.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(false)]
        );

        let ne_scalar = super::eval_str("a.ne(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            ne_scalar.values(),
            &[Scalar::Bool(true), Scalar::Bool(false), Scalar::Bool(true)]
        );

        let eq_column = super::eval_str("a.eq(b)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            eq_column.values(),
            &[Scalar::Bool(false), Scalar::Bool(true), Scalar::Bool(false)]
        );

        let filtered = super::query_str("a.gt(1) and a.le(3)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into(), 2_i64.into()]);

        let keyword_filtered = super::query_str(
            "a.ge(other=2) and a.lt(other=3)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(keyword_filtered.index().labels(), &[1_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_arithmetic_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let added = super::eval_str("a.add(1)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            added.values(),
            &[Scalar::Int64(2), Scalar::Int64(3), Scalar::Int64(4)]
        );

        let multiplied = super::eval_str("a.multiply(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            multiplied.values(),
            &[Scalar::Int64(2), Scalar::Int64(4), Scalar::Int64(6)]
        );

        let divided = super::eval_str("a.truediv(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            divided.values(),
            &[
                Scalar::Float64(0.5),
                Scalar::Float64(1.0),
                Scalar::Float64(1.5)
            ]
        );

        let reflected = super::eval_str("a.rsub(other=10)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            reflected.values(),
            &[Scalar::Int64(9), Scalar::Int64(8), Scalar::Int64(7)]
        );

        let filtered = super::query_str(
            "a.pow(2).ge(4) and a.mod(2).eq(1)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(filtered.index().labels(), &[2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_where_mask_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .map_err(ExprError::from)?,
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(9), Scalar::Int64(9), Scalar::Int64(9)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let where_scalar = super::eval_str("a.where(a > 1, 0)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            where_scalar.values(),
            &[Scalar::Int64(0), Scalar::Int64(2), Scalar::Int64(3)]
        );

        let mask_scalar = super::eval_str("a.mask(a > 1, 0)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            mask_scalar.values(),
            &[Scalar::Int64(1), Scalar::Int64(0), Scalar::Int64(0)]
        );

        let where_series = super::eval_str(
            "a.where(cond=a.gt(1), other=b)",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            where_series.values(),
            &[Scalar::Int64(9), Scalar::Int64(2), Scalar::Int64(3)]
        );

        let filtered = super::query_str("a.mask(a > 1, 0).eq(0)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into(), 2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_isin_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(4), Scalar::Int64(9)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let mask = super::eval_str("a.isin([1, 9])", &frame, &policy, &mut ledger)?;
        assert_eq!(
            mask.values(),
            &[Scalar::Bool(true), Scalar::Bool(false), Scalar::Bool(true)]
        );

        let filtered = super::query_str("a.isin([1, 9])", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[0_i64.into(), 2_i64.into()]);
        assert_eq!(
            filtered.columns()["a"].values(),
            &[Scalar::Int64(1), Scalar::Int64(9)]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_between_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Int64(2),
                    Scalar::Int64(8),
                    Scalar::Int64(9),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let mask = super::eval_str("a.between(2, 8)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            mask.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(true),
                Scalar::Bool(false)
            ]
        );

        let filtered = super::query_str("a.between(2, 8)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into(), 2_i64.into()]);
        assert_eq!(
            filtered.columns()["a"].values(),
            &[Scalar::Int64(2), Scalar::Int64(8)]
        );

        let left_inclusive = super::eval_str(
            "a.between(2, 8, inclusive=\"left\")",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert_eq!(
            left_inclusive.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(false),
                Scalar::Bool(false)
            ]
        );

        let right_inclusive =
            super::eval_str("a.between(2, 8, \"right\")", &frame, &policy, &mut ledger)?;
        assert_eq!(
            right_inclusive.values(),
            &[
                Scalar::Bool(false),
                Scalar::Bool(false),
                Scalar::Bool(true),
                Scalar::Bool(false)
            ]
        );

        let neither = super::query_str(
            "a.between(2, 8, inclusive=\"neither\")",
            &frame,
            &policy,
            &mut ledger,
        )?;
        assert!(neither.is_empty());
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_clip_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(4), Scalar::Int64(9)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let clipped = super::eval_str("a.clip(2, 8)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            clipped.values(),
            &[Scalar::Int64(2), Scalar::Int64(4), Scalar::Int64(8)]
        );

        let lower_only = super::eval_str("a.clip(lower=2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            lower_only.values(),
            &[Scalar::Int64(2), Scalar::Int64(4), Scalar::Int64(9)]
        );

        let upper_only = super::eval_str("a.clip(upper=8)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            upper_only.values(),
            &[Scalar::Int64(1), Scalar::Int64(4), Scalar::Int64(8)]
        );

        let lower_positional = super::eval_str("a.clip(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            lower_positional.values(),
            &[Scalar::Int64(2), Scalar::Int64(4), Scalar::Int64(9)]
        );

        let unchanged = super::eval_str("a.clip()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            unchanged.values(),
            &[Scalar::Int64(1), Scalar::Int64(4), Scalar::Int64(9)]
        );

        let filtered = super::query_str("a.clip(2, 8) == 8", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[2_i64.into()]);
        assert_eq!(filtered.columns()["a"].values(), &[Scalar::Int64(9)]);

        let keyword_filtered =
            super::query_str("a.clip(upper=8).ge(8)", &frame, &policy, &mut ledger)?;
        assert_eq!(keyword_filtered.index().labels(), &[2_i64.into()]);
        assert_eq!(
            keyword_filtered.columns()["a"].values(),
            &[Scalar::Int64(9)]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_shift_and_diff_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(4), Scalar::Int64(9)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let shifted = super::eval_str("a.shift()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            shifted.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Int64(1),
                Scalar::Int64(4)
            ]
        );

        let shifted_two = super::eval_str("a.shift(periods=2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            shifted_two.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Null(NullKind::NaN),
                Scalar::Int64(1)
            ]
        );

        let diffed = super::eval_str("a.diff()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            diffed.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Float64(3.0),
                Scalar::Float64(5.0)
            ]
        );

        let diffed_two = super::eval_str("a.diff(2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            diffed_two.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Null(NullKind::NaN),
                Scalar::Float64(8.0)
            ]
        );

        let shifted_filtered = super::query_str("a.shift().isna()", &frame, &policy, &mut ledger)?;
        assert_eq!(shifted_filtered.index().labels(), &[0_i64.into()]);

        let diff_filtered = super::query_str("a.diff().gt(4)", &frame, &policy, &mut ledger)?;
        assert_eq!(diff_filtered.index().labels(), &[2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_cumulative_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(2), Scalar::Int64(3), Scalar::Int64(1)],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let cumsum = super::eval_str("a.cumsum()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            cumsum.values(),
            &[Scalar::Int64(2), Scalar::Int64(5), Scalar::Int64(6)]
        );

        let cumprod = super::eval_str("a.cumprod(skipna=True)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            cumprod.values(),
            &[Scalar::Int64(2), Scalar::Int64(6), Scalar::Int64(6)]
        );

        let cummin = super::eval_str("a.cummin()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            cummin.values(),
            &[Scalar::Int64(2), Scalar::Int64(2), Scalar::Int64(1)]
        );

        let cummax = super::eval_str("a.cummax()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            cummax.values(),
            &[Scalar::Int64(2), Scalar::Int64(3), Scalar::Int64(3)]
        );

        let filtered = super::query_str("a.cumsum().gt(4)", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[1_i64.into(), 2_i64.into()]);
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_pct_change_method_calls() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(2),
                    Scalar::Int64(4),
                    Scalar::Int64(8),
                    Scalar::Int64(16),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let pct = super::eval_str("a.pct_change()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            pct.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Float64(1.0),
                Scalar::Float64(1.0),
                Scalar::Float64(1.0)
            ]
        );

        let pct_two = super::eval_str("a.pct_change(periods=2)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            pct_two.values(),
            &[
                Scalar::Null(NullKind::NaN),
                Scalar::Null(NullKind::NaN),
                Scalar::Float64(3.0),
                Scalar::Float64(3.0)
            ]
        );

        let filtered = super::query_str("a.pct_change().eq(1)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            filtered.index().labels(),
            &[1_i64.into(), 2_i64.into(), 3_i64.into()]
        );
        Ok(())
    }

    #[test]
    fn eval_and_query_str_accept_round_method_call() -> Result<(), ExprError> {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![
                    Scalar::Float64(1.25),
                    Scalar::Float64(2.75),
                    Scalar::Float64(8.25),
                ],
            )
            .map_err(ExprError::from)?,
        ])
        .map_err(ExprError::from)?;

        let rounded = super::eval_str("a.round()", &frame, &policy, &mut ledger)?;
        assert_eq!(
            rounded.values(),
            &[
                Scalar::Float64(1.0),
                Scalar::Float64(3.0),
                Scalar::Float64(8.0)
            ]
        );

        let one_decimal = super::eval_str("a.round(decimals=1)", &frame, &policy, &mut ledger)?;
        assert_eq!(
            one_decimal.values(),
            &[
                Scalar::Float64(1.2),
                Scalar::Float64(2.8),
                Scalar::Float64(8.2)
            ]
        );

        let filtered = super::query_str("a.round(-1) >= 10", &frame, &policy, &mut ledger)?;
        assert_eq!(filtered.index().labels(), &[2_i64.into()]);
        assert_eq!(filtered.columns()["a"].values(), &[Scalar::Float64(8.25)]);
        Ok(())
    }

    #[test]
    fn query_str_compound_filter() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(5), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )
            .unwrap(),
        ])
        .unwrap();

        // Query for a > 2 and b < 25
        let result = super::query_str("a > 2 and b < 25", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 1); // only row 1 (a=5, b=20)
        assert_eq!(result.columns()["a"].values()[0], Scalar::Int64(5));
        assert_eq!(result.columns()["b"].values()[0], Scalar::Int64(20));
    }

    #[test]
    fn query_str_chained_comparison_matches_pandas_pairwise_semantics() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(2), Scalar::Int64(2), Scalar::Int64(4)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "c",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(3), Scalar::Int64(1), Scalar::Int64(5)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("a < b < c", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(
            result.columns()["a"].values(),
            &[Scalar::Int64(1), Scalar::Int64(3)]
        );
    }

    // ── eval/query parity tests (frankenpandas-xmp) ──

    #[test]
    fn eval_str_subtraction() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(3), Scalar::Int64(7)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::eval_str("a - b", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(7));
        assert_eq!(result.values()[1], Scalar::Int64(13));
    }

    #[test]
    fn eval_str_multiplication() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(3), Scalar::Int64(5)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(4), Scalar::Int64(6)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::eval_str("a * b", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(12));
        assert_eq!(result.values()[1], Scalar::Int64(30));
    }

    #[test]
    fn eval_str_division() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "a",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Float64(10.0), Scalar::Float64(21.0)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "b",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Float64(2.0), Scalar::Float64(7.0)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::eval_str("a / b", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Float64(5.0));
        assert_eq!(result.values()[1], Scalar::Float64(3.0));
    }

    #[test]
    fn eval_str_parenthesized_expression() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values("a", vec![0_i64.into()], vec![Scalar::Int64(2)]).unwrap(),
            fp_frame::Series::from_values("b", vec![0_i64.into()], vec![Scalar::Int64(3)]).unwrap(),
            fp_frame::Series::from_values("c", vec![0_i64.into()], vec![Scalar::Int64(4)]).unwrap(),
        ])
        .unwrap();

        // (a + b) * c = (2 + 3) * 4 = 20
        let result = super::eval_str("(a + b) * c", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(20));
    }

    #[test]
    fn eval_str_comparison() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(5), Scalar::Int64(3)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::eval_str("x >= 3", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.values()[0], Scalar::Bool(false));
        assert_eq!(result.values()[1], Scalar::Bool(true));
        assert_eq!(result.values()[2], Scalar::Bool(true));
    }

    #[test]
    fn query_str_equality() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "name",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(10)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("name == 10", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn query_str_or_filter() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into(), 3_i64.into()],
                vec![
                    Scalar::Int64(1),
                    Scalar::Int64(5),
                    Scalar::Int64(3),
                    Scalar::Int64(8),
                ],
            )
            .unwrap(),
        ])
        .unwrap();

        // x < 2 or x > 7 → rows where x=1 or x=8
        let result = super::query_str("x < 2 or x > 7", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn query_str_accepts_pandas_boolean_literals_in_compound_filters() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2), Scalar::Int64(3)],
            )
            .unwrap(),
            fp_frame::Series::from_values(
                "flag",
                vec![0_i64.into(), 1_i64.into(), 2_i64.into()],
                vec![Scalar::Bool(true), Scalar::Bool(false), Scalar::Bool(true)],
            )
            .unwrap(),
        ])
        .unwrap();

        let true_rows = super::query_str("flag == True", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(true_rows.len(), 2);
        assert_eq!(
            true_rows.column("x").expect("x").values(),
            &[Scalar::Int64(1), Scalar::Int64(3)]
        );

        let gated = super::query_str("x > 1 and True", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(gated.len(), 2);
        assert_eq!(
            gated.column("x").expect("x").values(),
            &[Scalar::Int64(2), Scalar::Int64(3)]
        );

        let false_or = super::query_str("False or x > 1", &frame, &policy, &mut ledger).unwrap();
        assert_eq!(
            false_or.column("x").expect("x").values(),
            gated.column("x").expect("x").values()
        );
    }

    #[test]
    fn query_str_rejects_scalar_boolean_literal_filters() {
        let policy = RuntimePolicy::hardened(Some(100));
        let mut ledger = EvidenceLedger::new();

        let frame = fp_frame::DataFrame::from_series(vec![
            fp_frame::Series::from_values(
                "x",
                vec![0_i64.into(), 1_i64.into()],
                vec![Scalar::Int64(1), Scalar::Int64(2)],
            )
            .unwrap(),
        ])
        .unwrap();

        let result = super::query_str("True", &frame, &policy, &mut ledger);
        assert!(matches!(
            result,
            Err(ExprError::Frame(FrameError::CompatibilityRejected(message)))
                if message.contains("scalar boolean query")
        ));
    }

    #[test]
    fn eval_extension_trait() {
        use super::DataFrameExprExt;

        let frame = fp_frame::DataFrame::from_dict(
            &["a", "b"],
            vec![
                ("a", vec![Scalar::Int64(1), Scalar::Int64(2)]),
                ("b", vec![Scalar::Int64(10), Scalar::Int64(20)]),
            ],
        )
        .unwrap();

        let result = frame.eval("a + b").unwrap();
        assert_eq!(result.values()[0], Scalar::Int64(11));
        assert_eq!(result.values()[1], Scalar::Int64(22));
    }

    #[test]
    fn query_extension_trait() {
        use super::DataFrameExprExt;

        let frame = fp_frame::DataFrame::from_dict(
            &["x", "y"],
            vec![
                (
                    "x",
                    vec![Scalar::Int64(1), Scalar::Int64(5), Scalar::Int64(3)],
                ),
                (
                    "y",
                    vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
                ),
            ],
        )
        .unwrap();

        let result = frame.query("x > 2 and y < 25").unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result.columns()["x"].values()[0], Scalar::Int64(5));
    }

    #[test]
    fn eval_with_locals_extension_trait() {
        use super::DataFrameExprExt;

        let frame = fp_frame::DataFrame::from_dict(
            &["val"],
            vec![(
                "val",
                vec![Scalar::Int64(10), Scalar::Int64(20), Scalar::Int64(30)],
            )],
        )
        .unwrap();

        let locals = BTreeMap::from([("threshold".to_owned(), Scalar::Int64(15))]);
        let result = frame
            .query_with_locals("val > @threshold", &locals)
            .unwrap();
        assert_eq!(result.len(), 2);
    }
}
