oxirs-arq 0.4.1

//! # QueryExecutor - new_group Methods
//!
//! This module contains method implementations for `QueryExecutor`.
//!
//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)

pub use super::config::{
    ExecutionContext, ParallelConfig, StreamingResultConfig, ThreadPoolConfig,
};
pub use super::dataset::{
    convert_property_path, ConcreteStoreDataset, Dataset, DatasetPathAdapter, InMemoryDataset,
};
use crate::algebra::{Aggregate, Algebra, Expression, Solution, Variable};
use anyhow::Result;
use std::collections::HashMap;
use std::sync::{Arc, RwLock};

use super::queryexecutor_type::QueryExecutor;
use super::types::{ExecutionStrategy, FunctionRegistry, UnknownFunctionError};

/// Whether an evaluated term is an `xsd:integer` (or a derived integer type),
/// used by `SUM` to decide whether it can keep exact integer typing.
fn algebra_term_is_integer(term: &crate::algebra::Term) -> bool {
    let lit = match term {
        crate::algebra::Term::Literal(lit) => lit,
        _ => return false,
    };
    let datatype = match &lit.datatype {
        Some(dt) => dt.as_str(),
        None => return false,
    };
    matches!(
        datatype,
        "http://www.w3.org/2001/XMLSchema#integer"
            | "http://www.w3.org/2001/XMLSchema#long"
            | "http://www.w3.org/2001/XMLSchema#int"
            | "http://www.w3.org/2001/XMLSchema#short"
            | "http://www.w3.org/2001/XMLSchema#byte"
            | "http://www.w3.org/2001/XMLSchema#nonNegativeInteger"
            | "http://www.w3.org/2001/XMLSchema#nonPositiveInteger"
            | "http://www.w3.org/2001/XMLSchema#negativeInteger"
            | "http://www.w3.org/2001/XMLSchema#positiveInteger"
            | "http://www.w3.org/2001/XMLSchema#unsignedLong"
            | "http://www.w3.org/2001/XMLSchema#unsignedInt"
            | "http://www.w3.org/2001/XMLSchema#unsignedShort"
            | "http://www.w3.org/2001/XMLSchema#unsignedByte"
    ) && lit.value.parse::<i64>().is_ok()
}

impl QueryExecutor {
    /// Create new query executor with default configuration
    pub fn new() -> Self {
        let context = ExecutionContext::default();
        let parallel_executor = if context.parallel {
            match crate::parallel::ParallelExecutor::new(context.parallel_config.clone()) {
                Ok(pe) => Some(Arc::new(pe)),
                Err(_) => None,
            }
        } else {
            None
        };
        Self {
            context,
            function_registry: FunctionRegistry::new(),
            parallel_executor,
            result_cache: Arc::new(RwLock::new(HashMap::new())),
            execution_strategy: ExecutionStrategy::default(),
            adaptive_stats: Arc::new(crate::optimizer::adaptive::AdaptiveStatsStore::new(1024)),
            sla_gate: None,
            execution_budget: None,
        }
    }
    /// Create executor with custom context
    pub fn with_context(context: ExecutionContext) -> Self {
        let parallel_executor = if context.parallel {
            match crate::parallel::ParallelExecutor::new(context.parallel_config.clone()) {
                Ok(pe) => Some(Arc::new(pe)),
                Err(_) => None,
            }
        } else {
            None
        };
        Self {
            context,
            function_registry: FunctionRegistry::new(),
            parallel_executor,
            result_cache: Arc::new(RwLock::new(HashMap::new())),
            execution_strategy: ExecutionStrategy::default(),
            adaptive_stats: Arc::new(crate::optimizer::adaptive::AdaptiveStatsStore::new(1024)),
            sla_gate: None,
            execution_budget: None,
        }
    }
    /// Attach an SLA gate to this executor.
    ///
    /// When set, queries can be routed through
    /// [`QueryExecutor::execute_for_tenant`] which performs admission control
    /// before delegating to [`QueryExecutor::execute`].
    pub fn with_sla_gate(mut self, gate: crate::sla_integration::ArqSlaGate) -> Self {
        self.sla_gate = Some(gate);
        self
    }
    /// Whether an SLA gate has been attached.
    pub fn has_sla_gate(&self) -> bool {
        self.sla_gate.is_some()
    }
    /// Borrow the attached SLA gate, if any.
    pub fn sla_gate(&self) -> Option<&crate::sla_integration::ArqSlaGate> {
        self.sla_gate.as_ref()
    }
    /// Attach a runtime resource budget to this executor.
    ///
    /// When set, [`QueryExecutor::execute`] will:
    ///
    /// 1. Call [`crate::query_governor::ExecutionBudget::check_time`] at
    ///    entry, returning early if the wall-time limit is already breached.
    /// 2. Call [`crate::query_governor::ExecutionBudget::record_result_row`]
    ///    once for every binding in the produced solution.
    ///
    /// Triple-scan recording via
    /// [`crate::query_governor::ExecutionBudget::record_triple_scan`] is
    /// provided by the `ExecutionBudget` API but is not yet threaded into
    /// `execute_single_pattern` — callers that need per-triple enforcement
    /// should call it directly in their BGP iterator loops.
    pub fn with_budget(
        mut self,
        budget: std::sync::Arc<crate::query_governor::ExecutionBudget>,
    ) -> Self {
        self.execution_budget = Some(budget);
        self
    }
    /// Whether a runtime budget has been attached.
    pub fn has_budget(&self) -> bool {
        self.execution_budget.is_some()
    }
    /// Borrow the attached budget, if any.
    pub fn budget(&self) -> Option<&std::sync::Arc<crate::query_governor::ExecutionBudget>> {
        self.execution_budget.as_ref()
    }
    /// Wall-clock budget check for use inside hot evaluation loops.
    ///
    /// A no-op when no budget is attached. When a budget *is* attached it
    /// forwards to [`crate::query_governor::ExecutionBudget::check_time`] and, on
    /// breach, returns the **typed** [`crate::query_governor::BudgetExceeded`]
    /// wrapped via [`anyhow::Error::new`] (not stringified) so a caller up the
    /// stack can `downcast_ref` it and map a wall-time timeout to the correct
    /// HTTP status. Call it *throttled* (e.g. once every 1024 loop iterations —
    /// see [`Self::hash_join`], [`Self::execute_minus`], [`Self::apply_left_join`])
    /// because the underlying `Instant::now()` is not free at O(N*M) scale.
    #[inline]
    pub(super) fn budget_check_time(&self) -> Result<()> {
        if let Some(ref budget) = self.execution_budget {
            budget.check_time().map_err(anyhow::Error::new)?;
        }
        Ok(())
    }
    /// Execute a query on behalf of `tenant_id`, gated by the attached
    /// [`crate::sla_integration::ArqSlaGate`].
    ///
    /// On admission, the algebra is enqueued in the priority dispatcher (for
    /// observability), then dispatched immediately via [`Self::execute`].
    /// On rejection, [`crate::sla_integration::ArqSlaError::SlaExceeded`] is
    /// returned without touching the dataset.
    ///
    /// Returns an `Err(anyhow::Error)` when no gate is attached or when the
    /// underlying execution fails; the error chain carries the SLA root cause
    /// when applicable.
    pub fn execute_for_tenant(
        &mut self,
        tenant_id: &str,
        algebra: &crate::algebra::Algebra,
        dataset: &dyn super::dataset::Dataset,
    ) -> Result<(crate::algebra::Solution, super::stats::ExecutionStats)> {
        let gate = self
            .sla_gate
            .clone()
            .ok_or_else(|| anyhow::anyhow!("no SLA gate attached to QueryExecutor"))?;
        let admitted = gate
            .admit(tenant_id, ())
            .map_err(|err| anyhow::anyhow!("{err}"))?;
        // Stick the admitted entry into the priority dispatcher for observability,
        // then immediately dequeue (single-thread mode).
        gate.enqueue(&admitted, format!("query for {tenant_id}"));
        let _ = gate.next_dispatch();
        self.execute(algebra, dataset)
    }
    /// Execute using serial strategy with index-aware optimizations
    pub(super) fn execute_serial(
        &self,
        algebra: &Algebra,
        dataset: &dyn Dataset,
    ) -> Result<Solution> {
        // Wall-time budget check at every operator boundary. `execute_serial` is
        // the single recursive dispatch point for the Serial strategy (the
        // strategy fuseki forces for every query), so a check here fires between
        // Union / LeftJoin / Minus / Filter / … sub-evaluations and guarantees a
        // deeply-nested-but-cheap-per-node tree still gets stopped. The genuinely
        // hot O(N*M) inner loops (hash_join / execute_minus / apply_left_join)
        // carry their own throttled checks on top of this.
        self.budget_check_time()?;
        match algebra {
            Algebra::Bgp(patterns) => self.execute_bgp_index_aware(patterns, dataset),
            Algebra::Join { left, right } => {
                self.execute_index_optimized_join(left, right, dataset)
            }
            Algebra::Union { left, right } => {
                let left_results = self.execute_serial(left, dataset)?;
                let right_results = self.execute_serial(right, dataset)?;
                Ok(self.union_solutions(left_results, right_results))
            }
            Algebra::Filter { pattern, condition } => {
                // FILTER(?v IN (<iri>, …)) over a BGP is answered with per-IRI
                // index lookups instead of an unconstrained scan when the
                // shape allows (see try_filter_in_pushdown's gates).
                if let Some(result) = self.try_filter_in_pushdown(pattern, condition, dataset)? {
                    return Ok(result);
                }
                let pattern_results = self.execute_serial(pattern, dataset)?;
                self.apply_filter_with_dataset(pattern_results, condition, dataset)
            }
            Algebra::Project { pattern, variables } => {
                let pattern_results = self.execute_serial(pattern, dataset)?;
                self.apply_projection(pattern_results, variables)
            }
            Algebra::Distinct { pattern } => {
                let pattern_results = self.execute_serial(pattern, dataset)?;
                Ok(self.apply_distinct(pattern_results))
            }
            Algebra::OrderBy {
                pattern,
                conditions,
            } => {
                let pattern_results = self.execute_serial(pattern, dataset)?;
                Ok(self.apply_order_by(pattern_results, conditions))
            }
            Algebra::Slice {
                pattern,
                offset,
                limit,
            } => {
                let pattern_results = self.execute_serial(pattern, dataset)?;
                Ok(self.apply_slice(pattern_results, *offset, *limit))
            }
            Algebra::Group {
                pattern,
                variables,
                aggregates,
            } => {
                let pattern_results = self.execute_serial(pattern, dataset)?;
                self.apply_group_by(pattern_results, variables, aggregates)
            }
            Algebra::LeftJoin {
                left,
                right,
                filter,
            } => {
                let left_results = self.execute_serial(left, dataset)?;
                let right_results = self.execute_serial(right, dataset)?;
                self.apply_left_join(left_results, right_results, filter)
            }
            Algebra::Minus { left, right } => self.execute_minus(left, right, dataset),
            Algebra::Reduced { pattern } => {
                // REDUCED is a hint that the engine may (but is not required to) remove
                // duplicate solutions.  The simplest correct implementation is a passthrough
                // that returns all solutions unchanged.
                self.execute_serial(pattern, dataset)
            }
            Algebra::Extend {
                pattern,
                variable,
                expr,
            } => self.execute_extend(pattern, variable, expr, dataset),
            Algebra::Values {
                variables,
                bindings,
            } => self.execute_values(variables, bindings),
            Algebra::PropertyPath {
                subject,
                path,
                object,
            } => self.execute_property_path(subject, path, object, dataset),
            Algebra::Graph { graph, pattern } => self.execute_graph(graph, pattern, dataset),
            Algebra::Having { pattern, condition } => {
                self.execute_having(pattern, condition, dataset)
            }
            Algebra::Service {
                endpoint,
                pattern,
                silent,
            } => crate::service_federation::execute_service_clause(endpoint, pattern, *silent),
            // Unit table (join identity): one solution with no bindings. This is
            // the left operand produced by the parser for leading BIND/OPTIONAL,
            // so it MUST yield a single empty row, not zero rows.
            Algebra::Table => {
                let solution: Solution = vec![crate::algebra::Binding::new()];
                Ok(solution)
            }
            // Zero / Empty are genuinely empty result sets.
            Algebra::Zero | Algebra::Empty => Ok(Solution::new()),
        }
    }
    /// Evaluate a `GRAPH` pattern with real named-graph scoping.
    ///
    /// * `GRAPH <iri> { P }` restricts evaluation of `P` to the named graph
    ///   `<iri>` by wrapping the dataset in a [`super::dataset::GraphScopedDataset`]
    ///   view, so the inner pattern code reads only that graph.
    /// * `GRAPH ?g { P }` enumerates the dataset's named graphs
    ///   ([`Dataset::named_graphs`]), evaluates `P` scoped to each, and extends
    ///   every produced row with `?g` bound to that graph. If the inner pattern
    ///   itself already binds `?g`, rows whose binding disagrees with the
    ///   enumerated graph are dropped (self-consistency); the join-time
    ///   compatibility check handles the case where an *outer* operator has
    ///   already fixed `?g`.
    pub(super) fn execute_graph(
        &self,
        graph: &crate::algebra::Term,
        pattern: &Algebra,
        dataset: &dyn Dataset,
    ) -> Result<Solution> {
        use super::dataset::{GraphScopedDataset, GraphSelector};
        use crate::algebra::Term;
        match graph {
            Term::Iri(iri) => {
                let scoped = GraphScopedDataset::new(dataset, GraphSelector::Named(iri.clone()));
                self.execute_serial(pattern, &scoped)
            }
            Term::Variable(var) => {
                let graphs = dataset.named_graphs()?;
                let mut combined = Solution::new();
                for g in graphs {
                    let iri = match &g {
                        Term::Iri(n) => n.clone(),
                        // Named graphs are always IRIs; ignore anything else.
                        _ => continue,
                    };
                    let scoped = GraphScopedDataset::new(dataset, GraphSelector::Named(iri));
                    let rows = self.execute_serial(pattern, &scoped)?;
                    for mut binding in rows {
                        match binding.get(var) {
                            Some(existing) if existing != &g => continue,
                            Some(_) => combined.push(binding),
                            None => {
                                binding.insert(var.clone(), g.clone());
                                combined.push(binding);
                            }
                        }
                    }
                }
                Ok(combined)
            }
            other => Err(anyhow::anyhow!(
                "GRAPH label must be an IRI or a variable, got: {other}"
            )),
        }
    }
    /// Execute using streaming strategy
    pub(super) fn execute_streaming(
        &self,
        algebra: &Algebra,
        dataset: &dyn Dataset,
    ) -> Result<Solution> {
        use crate::executor::streaming::{StreamingConfig as StreamConfig, StreamingSolution};
        let stream_config = StreamConfig {
            memory_limit: self.context.memory_limit.unwrap_or(1024 * 1024 * 1024),
            temp_dir: None,
            buffer_size: self.context.streaming.buffer_size,
            compress_spills: true,
            spill_strategy: crate::executor::streaming::SpillStrategy::Adaptive,
            adaptive_buffering: true,
            parallel_spilling: true,
            compression_algorithm: crate::executor::streaming::CompressionAlgorithm::Zstd,
        };
        let mut streaming_solution = StreamingSolution::new(stream_config);
        self.execute_algebra_streaming(algebra, dataset, &mut streaming_solution)?;
        streaming_solution.finish();
        let mut result = Solution::new();
        for solution_result in streaming_solution {
            result.extend(solution_result?);
        }
        Ok(result)
    }
    /// Execute algebra expression with streaming support
    pub(super) fn execute_algebra_streaming(
        &self,
        algebra: &Algebra,
        dataset: &dyn Dataset,
        streaming_solution: &mut super::streaming::StreamingSolution,
    ) -> Result<()> {
        match algebra {
            Algebra::Bgp(patterns) => {
                let bgp_algebra = Algebra::Bgp(patterns.clone());
                let solutions = self.execute_serial(&bgp_algebra, dataset)?;
                for solution in solutions {
                    streaming_solution.add_solution(vec![solution])?;
                }
                Ok(())
            }
            Algebra::Join { .. } => {
                // Delegate to Serial: execute_index_optimized_join carries the
                // bound-join pushdown and the correct per-binding join
                // granularity. The previous SpillableHashJoin call fed each
                // side's WHOLE solution as a single row with join_vars = []
                // (extract_join_variables was a stub), so any top-level Join
                // under the Streaming strategy returned at most one — possibly
                // conflicting — row. As with the Filter arm below,
                // execute_streaming materializes everything anyway, so no
                // streaming benefit is lost.
                let joined = self.execute_serial(algebra, dataset)?;
                for binding in joined {
                    streaming_solution.add_solution(vec![binding])?;
                }
                Ok(())
            }
            Algebra::Union { left, right } => {
                self.execute_algebra_streaming(left, dataset, streaming_solution)?;
                self.execute_algebra_streaming(right, dataset, streaming_solution)?;
                Ok(())
            }
            Algebra::Filter { .. } => {
                // The FILTER condition MUST be applied. The previous
                // implementation discarded it (`condition: _`) and streamed the
                // unfiltered pattern, so every Streaming-strategy FILTER — and
                // both branches of a `Filter(Union(A, B))` — returned rows that
                // Serial would have excluded (a silent wrong-answer). Because
                // `execute_streaming` already materializes the whole solution
                // before returning (see `execute_streaming`), there is no real
                // streaming benefit to forgo: evaluate the entire Filter node via
                // the Serial path so the dataset-aware condition (including
                // EXISTS / NOT EXISTS and the typed-error propagation) matches
                // Serial exactly.
                let filtered = self.execute_serial(algebra, dataset)?;
                for binding in filtered {
                    streaming_solution.add_solution(vec![binding])?;
                }
                Ok(())
            }
            _ => {
                let solution = self.execute_serial(algebra, dataset)?;
                for binding in solution {
                    streaming_solution.add_solution(vec![binding])?;
                }
                Ok(())
            }
        }
    }
    /// Execute algebra in serial mode for update operations.
    ///
    /// This helper has no access to a dataset, so it cannot evaluate a BGP (or
    /// any pattern that reads the store) against real data. Rather than
    /// fabricate bindings — which previously mapped every subject/object
    /// variable to hardcoded `http://example.org/...` constants and silently
    /// corrupted DELETE/INSERT WHERE — it fails loudly. Callers that need to
    /// evaluate WHERE clauses must run [`QueryExecutor::execute`] against a real
    /// [`Dataset`] (see `update::UpdateExecutor::evaluate_pattern`).
    pub(super) fn execute_serial_algebra(
        &self,
        algebra: &Algebra,
        _context: &mut crate::algebra::EvaluationContext,
    ) -> Result<Solution> {
        match algebra {
            // An empty/unit table is representable without a dataset.
            Algebra::Table => {
                let solution: Solution = vec![crate::algebra::Binding::new()];
                Ok(solution)
            }
            Algebra::Zero | Algebra::Empty => Ok(Solution::new()),
            other => Err(anyhow::anyhow!(
                "execute_serial_algebra cannot evaluate {:?} without a dataset; \
                 use QueryExecutor::execute against a real Dataset instead",
                std::mem::discriminant(other)
            )),
        }
    }
    /// Execute BGP with index-aware optimizations
    pub(super) fn execute_bgp_index_aware(
        &self,
        patterns: &[crate::algebra::TriplePattern],
        dataset: &dyn Dataset,
    ) -> Result<Solution> {
        if patterns.is_empty() {
            return Ok(Solution::new());
        }
        let stats = crate::optimizer::Statistics::new();
        let index_stats = crate::optimizer::IndexStatistics::default();
        let optimizer = crate::bgp_optimizer::BGPOptimizer::new(&stats, &index_stats);
        let optimized_bgp = optimizer.optimize_bgp(patterns.to_vec())?;
        let mut current_solution =
            self.execute_single_pattern(&optimized_bgp.patterns[0], dataset)?;
        for pattern in optimized_bgp.patterns.iter().skip(1) {
            let pattern_results = self.execute_single_pattern(pattern, dataset)?;
            // Hash-join on the shared variables rather than a nested-loop scan:
            // a high-cardinality BGP such as `?c a :C ; :label ?l` (tens of
            // thousands x hundreds of thousands of rows) is otherwise O(|L|x|R|)
            // and effectively hangs. Build the hash table on the SMALLER side.
            current_solution = if current_solution.len() <= pattern_results.len() {
                self.hash_join(current_solution, pattern_results)?
            } else {
                self.hash_join(pattern_results, current_solution)?
            };
            if current_solution.is_empty() {
                break;
            }
        }
        Ok(current_solution)
    }
    /// Apply GROUP BY with aggregation
    pub(super) fn apply_group_by(
        &self,
        solution: Solution,
        variables: &[crate::algebra::GroupCondition],
        aggregates: &[(crate::algebra::Variable, crate::algebra::Aggregate)],
    ) -> Result<Solution> {
        use std::collections::HashMap;
        // Each grouping condition is a full expression (`GROUP BY ?v`,
        // `GROUP BY (LANG(?l))`, `GROUP BY (STR(?x) AS ?k)`), so the group key is
        // the ordered tuple of the *evaluated* condition values. Evaluating the
        // expression per binding — rather than only reading plain variables — is
        // what makes `GROUP BY (expr)` split into the correct groups instead of
        // collapsing every row into one. An unevaluable condition (unbound
        // variable, type error) contributes `None`, a distinct key slot.
        let mut groups: HashMap<Vec<Option<crate::algebra::Term>>, Vec<&crate::algebra::Binding>> =
            HashMap::new();
        for binding in &solution {
            let mut group_key = Vec::with_capacity(variables.len());
            for group_condition in variables {
                let value = match &group_condition.expr {
                    crate::algebra::Expression::Variable(var) => binding.get(var).cloned(),
                    other => self.evaluate_expression(other, binding).ok(),
                };
                group_key.push(value);
            }
            groups.entry(group_key).or_default().push(binding);
        }
        // When there are no GROUP BY variables, ensure we always have exactly one group
        // (even for empty solutions, so COUNT(*) returns 0 rather than no rows)
        if variables.is_empty() {
            groups.entry(Vec::new()).or_default();
        }
        let mut result = Solution::new();
        for (group_key, group_bindings) in groups {
            let mut group_result = crate::algebra::Binding::new();
            // Expose each grouping key under its bound name: a plain `?v`
            // grouping binds `?v`, and an aliased `(expr AS ?k)` binds `?k`, so
            // the key value is visible to the projection / ORDER BY. A bare
            // `GROUP BY (expr)` with no alias binds nothing (SPARQL requires an
            // alias to project such a key).
            for (group_condition, value) in variables.iter().zip(group_key.iter()) {
                let bind_var = match &group_condition.expr {
                    crate::algebra::Expression::Variable(var) => Some(var.clone()),
                    _ => group_condition.alias.clone(),
                };
                if let (Some(var), Some(term)) = (bind_var, value) {
                    group_result.insert(var, term.clone());
                }
            }
            for (agg_var, aggregate) in aggregates {
                let agg_value = self.calculate_aggregate(aggregate, &group_bindings)?;
                group_result.insert(agg_var.clone(), agg_value);
            }
            result.push(group_result);
        }
        Ok(result)
    }
    /// Calculate aggregate value
    pub(super) fn calculate_aggregate(
        &self,
        aggregate: &crate::algebra::Aggregate,
        bindings: &[&crate::algebra::Binding],
    ) -> Result<crate::algebra::Term> {
        use std::collections::HashSet;
        match aggregate {
            crate::algebra::Aggregate::Count { distinct, expr } => {
                if let Some(expr) = expr {
                    let mut values = Vec::new();
                    for binding in bindings {
                        if let Ok(value) = self.evaluate_expression(expr, binding) {
                            values.push(value);
                        }
                    }
                    let count = if *distinct {
                        let unique_values: HashSet<_> = values.into_iter().collect();
                        unique_values.len()
                    } else {
                        values.len()
                    };
                    Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                        value: count.to_string(),
                        language: None,
                        datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                            "http://www.w3.org/2001/XMLSchema#integer",
                        )),
                    }))
                } else {
                    let count = bindings.len();
                    Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                        value: count.to_string(),
                        language: None,
                        datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                            "http://www.w3.org/2001/XMLSchema#integer",
                        )),
                    }))
                }
            }
            crate::algebra::Aggregate::Sum { distinct, expr } => {
                // Collect each operand's (numeric value, integer-ness) so SUM
                // over xsd:integer operands keeps xsd:integer typing and full
                // i64 precision (SPARQL type promotion), only widening to
                // xsd:decimal when a non-integer operand appears.
                let mut values: Vec<(f64, bool, String)> = Vec::new();
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        if let Ok(num) = self.extract_numeric_value(&value) {
                            let is_int = algebra_term_is_integer(&value);
                            let lexical = match &value {
                                crate::algebra::Term::Literal(lit) => lit.value.clone(),
                                _ => num.to_string(),
                            };
                            values.push((num, is_int, lexical));
                        }
                    }
                }
                if *distinct {
                    values
                        .sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal));
                    values.dedup_by(|a, b| a.0 == b.0);
                }
                let all_integer = !values.is_empty() && values.iter().all(|(_, is_int, _)| *is_int);
                if all_integer {
                    // Exact integer accumulation with overflow fallback to f64.
                    let mut acc: i64 = 0;
                    let mut overflowed = false;
                    for (_, _, lexical) in &values {
                        match lexical.parse::<i64>() {
                            Ok(i) => match acc.checked_add(i) {
                                Some(v) => acc = v,
                                None => {
                                    overflowed = true;
                                    break;
                                }
                            },
                            Err(_) => {
                                overflowed = true;
                                break;
                            }
                        }
                    }
                    if !overflowed {
                        return Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                            value: acc.to_string(),
                            language: None,
                            datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                                "http://www.w3.org/2001/XMLSchema#integer",
                            )),
                        }));
                    }
                }
                let sum: f64 = values.iter().map(|(n, _, _)| *n).sum();
                Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                    value: sum.to_string(),
                    language: None,
                    datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                        "http://www.w3.org/2001/XMLSchema#decimal",
                    )),
                }))
            }
            crate::algebra::Aggregate::Min { distinct: _, expr } => {
                // SPARQL 1.1 §18.5.1.9: MIN returns one of the input terms
                // verbatim (datatype included, e.g. xsd:gYear) — never a
                // synthesized xsd:decimal like SUM/AVG.
                let mut min_entry: Option<(f64, crate::algebra::Term)> = None;
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        if let Ok(num) = self.extract_numeric_value(&value) {
                            let replace = match &min_entry {
                                Some((best, _)) => num < *best,
                                None => true,
                            };
                            if replace {
                                min_entry = Some((num, value));
                            }
                        }
                    }
                }
                match min_entry {
                    Some((_, term)) => Ok(term),
                    None => Err(anyhow::anyhow!("No numeric values found for MIN aggregate")),
                }
            }
            crate::algebra::Aggregate::Max { distinct: _, expr } => {
                // SPARQL 1.1 §18.5.1.10: MAX returns one of the input terms
                // verbatim (datatype included) — see the MIN arm above.
                let mut max_entry: Option<(f64, crate::algebra::Term)> = None;
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        if let Ok(num) = self.extract_numeric_value(&value) {
                            let replace = match &max_entry {
                                Some((best, _)) => num > *best,
                                None => true,
                            };
                            if replace {
                                max_entry = Some((num, value));
                            }
                        }
                    }
                }
                match max_entry {
                    Some((_, term)) => Ok(term),
                    None => Err(anyhow::anyhow!("No numeric values found for MAX aggregate")),
                }
            }
            crate::algebra::Aggregate::Avg { distinct, expr } => {
                let mut values = Vec::new();
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        if let Ok(num) = self.extract_numeric_value(&value) {
                            values.push(num);
                        }
                    }
                }
                if *distinct {
                    values.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
                    values.dedup_by(|a, b| a == b);
                }
                if values.is_empty() {
                    Err(anyhow::anyhow!("No numeric values found for AVG aggregate"))
                } else {
                    let sum: f64 = values.iter().sum();
                    let avg = sum / values.len() as f64;
                    Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                        value: avg.to_string(),
                        language: None,
                        datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                            "http://www.w3.org/2001/XMLSchema#decimal",
                        )),
                    }))
                }
            }
            crate::algebra::Aggregate::Sample { distinct: _, expr } => {
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        return Ok(value);
                    }
                }
                Err(anyhow::anyhow!("No values found for SAMPLE aggregate"))
            }
            crate::algebra::Aggregate::GroupConcat {
                distinct,
                expr,
                separator,
            } => {
                let mut values = Vec::new();
                for binding in bindings {
                    if let Ok(value) = self.evaluate_expression(expr, binding) {
                        let string_value = match value {
                            crate::algebra::Term::Literal(lit) => lit.value,
                            crate::algebra::Term::Iri(iri) => iri.to_string(),
                            crate::algebra::Term::BlankNode(bn) => format!("_{bn}"),
                            _ => value.to_string(),
                        };
                        values.push(string_value);
                    }
                }
                if *distinct {
                    let unique_values: HashSet<_> = values.into_iter().collect();
                    values = unique_values.into_iter().collect();
                    values.sort();
                }
                let sep = separator.as_deref().unwrap_or(" ");
                let concatenated = values.join(sep);
                Ok(crate::algebra::Term::Literal(crate::algebra::Literal {
                    value: concatenated,
                    language: None,
                    datatype: Some(oxirs_core::model::NamedNode::new_unchecked(
                        "http://www.w3.org/2001/XMLSchema#string",
                    )),
                }))
            }
        }
    }
    /// Apply LEFT JOIN (OPTIONAL)
    pub(super) fn apply_left_join(
        &self,
        left: Solution,
        right: Solution,
        conditions: &Option<crate::algebra::Expression>,
    ) -> Result<Solution> {
        use std::collections::{HashMap, HashSet};
        // A no-op right side keeps every left row unbound.
        if right.is_empty() {
            return Ok(left);
        }
        // Hash the right side on the variables it shares with the left, so an
        // OPTIONAL against a high-cardinality right pattern is O(|L|+|R|) rather
        // than the O(|L|x|R|) nested loop that hangs at scale (the `?c :label ?l`
        // OPTIONAL shape). Left rows with no compatible right row are kept with
        // their unbound optional variables (left-join semantics).
        let left_vars: HashSet<crate::algebra::Variable> =
            left.iter().flat_map(|b| b.keys().cloned()).collect();
        let right_vars: HashSet<crate::algebra::Variable> =
            right.iter().flat_map(|b| b.keys().cloned()).collect();
        let shared_vars: Vec<crate::algebra::Variable> =
            left_vars.intersection(&right_vars).cloned().collect();

        let mut hash_table: HashMap<
            Vec<(crate::algebra::Variable, crate::algebra::Term)>,
            Vec<&crate::algebra::Binding>,
        > = HashMap::new();
        for binding in &right {
            let key: Vec<_> = shared_vars
                .iter()
                .filter_map(|var| binding.get(var).map(|term| (var.clone(), term.clone())))
                .collect();
            hash_table.entry(key).or_default().push(binding);
        }

        let mut result = Solution::new();
        // Persistent throttled wall-time check (see [`Self::hash_join`]): an
        // OPTIONAL against a high-cardinality right side is O(|L|+|R|) in the
        // common case but degrades toward O(|L|*|R|) when many left rows collide
        // on the same shared-variable key, so both loops are instrumented.
        let mut budget_ticks: u64 = 0;
        for left_binding in &left {
            if budget_ticks & 0x3FF == 0 {
                self.budget_check_time()?;
            }
            budget_ticks += 1;
            let key: Vec<_> = shared_vars
                .iter()
                .filter_map(|var| {
                    left_binding
                        .get(var)
                        .map(|term| (var.clone(), term.clone()))
                })
                .collect();
            let mut has_join = false;
            if let Some(matching) = hash_table.get(&key) {
                for &right_binding in matching {
                    if budget_ticks & 0x3FF == 0 {
                        self.budget_check_time()?;
                    }
                    budget_ticks += 1;
                    let mut is_compatible = true;
                    let mut merged = left_binding.clone();
                    for (var, term) in right_binding {
                        if let Some(existing_term) = merged.get(var) {
                            if existing_term != term {
                                is_compatible = false;
                                break;
                            }
                        } else {
                            merged.insert(var.clone(), term.clone());
                        }
                    }
                    if is_compatible {
                        // SPARQL 1.1 §18.5 LeftJoin: the OPTIONAL FILTER (the
                        // condition carried on the LeftJoin node) is evaluated
                        // over the MERGED (left+right) binding. A merged row that
                        // fails the filter does not count as a match — so a left
                        // row whose only compatible right rows all fail the filter
                        // is still emitted with its optional variables unbound.
                        // This mirrors perform_parallel_left_join so the Serial
                        // and Parallel strategies agree.
                        if let Some(condition) = conditions {
                            if self.left_join_filter_passes(condition, &merged)? {
                                result.push(merged);
                                has_join = true;
                            }
                        } else {
                            result.push(merged);
                            has_join = true;
                        }
                    }
                }
            }
            if !has_join {
                result.push(left_binding.clone());
            }
        }
        Ok(result)
    }

    /// Evaluate a LeftJoin (OPTIONAL) FILTER condition over a merged binding.
    ///
    /// Truthiness rules mirror [`Self::apply_filter`]: a whole-query fault
    /// (unknown function / runtime-budget breach) propagates as `Err`, while an
    /// ordinary per-row evaluation error (unbound variable, type error) yields
    /// `Ok(false)` so that merged row is simply not treated as a match.
    fn left_join_filter_passes(
        &self,
        condition: &crate::algebra::Expression,
        merged: &crate::algebra::Binding,
    ) -> Result<bool> {
        match self.evaluate_expression(condition, merged) {
            Ok(crate::algebra::Term::Literal(lit)) => Ok(self.is_truthy(&lit)),
            Ok(_) => Ok(true),
            Err(err) => {
                if err.downcast_ref::<UnknownFunctionError>().is_some()
                    || err
                        .downcast_ref::<crate::query_governor::BudgetExceeded>()
                        .is_some()
                {
                    Err(err)
                } else {
                    Ok(false)
                }
            }
        }
    }
    /// Hash join implementation
    pub(super) fn hash_join(&self, build_side: Solution, probe_side: Solution) -> Result<Solution> {
        use std::collections::{HashMap, HashSet};
        let build_vars: HashSet<_> = if let Some(first_binding) = build_side.first() {
            first_binding.keys().collect()
        } else {
            return Ok(Solution::new());
        };
        let probe_vars: HashSet<_> = if let Some(first_binding) = probe_side.first() {
            first_binding.keys().collect()
        } else {
            return Ok(Solution::new());
        };
        let shared_vars: Vec<_> = build_vars.intersection(&probe_vars).cloned().collect();
        let mut hash_table: HashMap<
            Vec<(crate::algebra::Variable, crate::algebra::Term)>,
            Vec<&crate::algebra::Binding>,
        > = HashMap::new();
        for binding in &build_side {
            let key: Vec<_> = shared_vars
                .iter()
                .filter_map(|var| binding.get(var).map(|term| ((*var).clone(), term.clone())))
                .collect();
            hash_table.entry(key).or_default().push(binding);
        }
        let mut result = Solution::new();
        // Persistent (not per-probe-row) counter so the throttled wall-time check
        // fires across the whole join, including the pathological cross join where
        // every build row lands in a single hash bucket and the inner loop runs
        // |build|*|probe| times. Checking `& 0x3FF` fires at 0 (immediately) then
        // every 1024 iterations; the check is incremented in BOTH loops so a
        // "many probes, few matches" shape (large outer, empty inner) is covered
        // too. See [`Self::budget_check_time`] for why this is throttled.
        let mut budget_ticks: u64 = 0;
        for probe_binding in &probe_side {
            if budget_ticks & 0x3FF == 0 {
                self.budget_check_time()?;
            }
            budget_ticks += 1;
            let probe_key: Vec<_> = shared_vars
                .iter()
                .filter_map(|var| {
                    probe_binding
                        .get(var)
                        .map(|term| ((*var).clone(), term.clone()))
                })
                .collect();
            if let Some(matching_bindings) = hash_table.get(&probe_key) {
                for &build_binding in matching_bindings {
                    if budget_ticks & 0x3FF == 0 {
                        self.budget_check_time()?;
                    }
                    budget_ticks += 1;
                    let mut is_compatible = true;
                    let mut merged = probe_binding.clone();
                    for (var, term) in build_binding {
                        if let Some(existing_term) = merged.get(var) {
                            if existing_term != term {
                                is_compatible = false;
                                break;
                            }
                        } else {
                            merged.insert(var.clone(), term.clone());
                        }
                    }
                    if is_compatible {
                        result.push(merged);
                    }
                }
            }
        }
        Ok(result)
    }
    /// Apply filter to solution
    pub(super) fn apply_filter(
        &self,
        solution: Solution,
        condition: &crate::algebra::Expression,
    ) -> Result<Solution> {
        let mut filtered = Solution::new();
        for binding in solution {
            match self.evaluate_expression(condition, &binding) {
                Ok(crate::algebra::Term::Literal(lit)) => {
                    if self.is_truthy(&lit) {
                        filtered.push(binding);
                    }
                }
                Ok(_) => {
                    filtered.push(binding);
                }
                Err(err) => {
                    // An unknown function is a whole-query fault: fail the entire
                    // filter loudly rather than silently shrinking the result set
                    // (no-silent-empty contract). A runtime budget breach (raised
                    // by a FILTER (NOT) EXISTS subquery that timed out or blew its
                    // scan budget) is likewise a whole-query fault and MUST NOT be
                    // swallowed — doing so would drop the row and return a wrong
                    // 200 instead of a timeout. Every OTHER error class is a
                    // per-row evaluation error (unbound variable, type error, ...)
                    // which SPARQL 1.1 §17.3 treats as excluding that one row, so
                    // it is swallowed and the row is dropped.
                    if err.downcast_ref::<UnknownFunctionError>().is_some()
                        || err
                            .downcast_ref::<crate::query_governor::BudgetExceeded>()
                            .is_some()
                    {
                        return Err(err);
                    }
                }
            }
        }
        Ok(filtered)
    }

    /// Apply filter with dataset access (for EXISTS/NOT EXISTS evaluation)
    pub(super) fn apply_filter_with_dataset(
        &self,
        solution: Solution,
        condition: &crate::algebra::Expression,
        dataset: &dyn super::dataset::Dataset,
    ) -> Result<Solution> {
        use super::queryexecutor_queries::EXISTS_DATASET;
        // Set the dataset pointer in thread-local storage so that EXISTS/NOT EXISTS
        // subquery evaluation can access it
        // Store the dataset as a (data_ptr, vtable_ptr) usize pair in thread-local storage
        // to avoid lifetime issues with storing references in thread-locals.
        let dataset_ptr: *const dyn super::dataset::Dataset = dataset;
        let (data_addr, vtable_addr): (usize, usize) = unsafe { std::mem::transmute(dataset_ptr) };
        // SAVE the previous value and RESTORE it afterwards, rather than clearing
        // to `None`. A FILTER can itself be evaluated inside an EXISTS/NOT EXISTS
        // subquery whose inner pattern also contains a FILTER; clearing here
        // would strip the dataset the OUTER EXISTS loop still depends on and
        // silently drop every subsequent row into the (dataset-less, incorrect)
        // syntactic fallback.
        let prev = EXISTS_DATASET.with(|cell| cell.replace(Some((data_addr, vtable_addr))));

        let result = self.apply_filter(solution, condition);

        EXISTS_DATASET.with(|cell| {
            *cell.borrow_mut() = prev;
        });

        result
    }

    /// Evaluate a `HAVING` clause over a grouped solution.
    ///
    /// `HAVING` is a post-aggregation `FILTER`, but its condition may itself
    /// contain aggregate function calls (`HAVING (COUNT(?s) > 1)`,
    /// `HAVING (SUM(?a * ?b) >= 10)`). A plain post-group filter cannot evaluate
    /// those, because the per-group source bindings have already been collapsed
    /// into one row per group. This method therefore hoists every aggregate call
    /// found in the condition into a synthetic group aggregate, evaluates it per
    /// group *alongside* the declared aggregates, rewrites the condition to
    /// reference the synthetic alias, applies the filter, and finally strips the
    /// synthetic columns so they do not leak into downstream operators.
    ///
    /// Any caller that builds `Group` + `Having` straight from parsed SPARQL thus
    /// gets correct semantics without pre-rewriting the algebra.
    ///
    /// A condition with no aggregate calls (a plain-variable `HAVING` over
    /// grouping keys / declared aggregates) takes the original fast path: execute
    /// the pattern and filter it directly, behavior-identical to before.
    pub(super) fn execute_having(
        &self,
        pattern: &Algebra,
        condition: &Expression,
        dataset: &dyn Dataset,
    ) -> Result<Solution> {
        // Detect and hoist aggregate calls in the condition.
        let mut hoisted: Vec<(Variable, Aggregate)> = Vec::new();
        let mut counter = 0usize;
        let rewritten = rewrite_having_aggregates(condition, &mut hoisted, &mut counter)?;

        if hoisted.is_empty() {
            // Plain-variable HAVING: a straight post-aggregation FILTER.
            let grouped = self.execute_serial(pattern, dataset)?;
            return self.apply_filter_with_dataset(grouped, &rewritten, dataset);
        }

        let synthetic_vars: Vec<Variable> = hoisted.iter().map(|(var, _)| var.clone()).collect();

        // Evaluate the hoisted aggregates per group. When the HAVING pattern is a
        // Group (the normal case), append them to its aggregate list so they are
        // computed over the same groups as the declared aggregates. When there is
        // no explicit Group (an aggregate HAVING with neither GROUP BY nor a
        // projected aggregate), the aggregate implies the single implicit group.
        let augmented = match pattern {
            Algebra::Group {
                pattern: inner,
                variables,
                aggregates,
            } => {
                let mut aggregates = aggregates.clone();
                aggregates.extend(hoisted);
                Algebra::Group {
                    pattern: inner.clone(),
                    variables: variables.clone(),
                    aggregates,
                }
            }
            other => Algebra::Group {
                pattern: Box::new(other.clone()),
                variables: Vec::new(),
                aggregates: hoisted,
            },
        };

        let grouped = self.execute_serial(&augmented, dataset)?;
        let filtered = self.apply_filter_with_dataset(grouped, &rewritten, dataset)?;

        // Strip the synthetic aggregate columns so downstream operators
        // (projection, ORDER BY, ...) never observe them.
        let cleaned = filtered
            .into_iter()
            .map(|mut binding| {
                for var in &synthetic_vars {
                    binding.remove(var);
                }
                binding
            })
            .collect();
        Ok(cleaned)
    }
}

/// Rewrite a `HAVING` condition so every aggregate function call is replaced by a
/// reference to a freshly-allocated synthetic group aggregate.
///
/// Aggregate calls (`COUNT(?s)`, `SUM(?a * ?b)`, ...) parse as
/// [`Expression::Function`]; each is hoisted into `aggregates` under a synthetic
/// `__having_agg_N` alias and replaced in the condition with
/// [`Expression::Variable`], so the resulting condition references only
/// grouped / aggregate variables that the post-group filter can evaluate.
/// Non-aggregate function calls and composite expressions are walked recursively;
/// leaves are returned unchanged.
fn rewrite_having_aggregates(
    expr: &Expression,
    aggregates: &mut Vec<(Variable, Aggregate)>,
    counter: &mut usize,
) -> Result<Expression> {
    match expr {
        Expression::Function { name, args } => {
            if let Some(aggregate) = function_to_aggregate(name, args)? {
                let alias = Variable::new(format!("__having_agg_{counter}"))
                    .map_err(|e| anyhow::anyhow!("invalid synthetic HAVING variable: {e}"))?;
                *counter += 1;
                aggregates.push((alias.clone(), aggregate));
                Ok(Expression::Variable(alias))
            } else {
                let rewritten_args = args
                    .iter()
                    .map(|a| rewrite_having_aggregates(a, aggregates, counter))
                    .collect::<Result<Vec<_>>>()?;
                Ok(Expression::Function {
                    name: name.clone(),
                    args: rewritten_args,
                })
            }
        }
        Expression::Binary { op, left, right } => Ok(Expression::Binary {
            op: op.clone(),
            left: Box::new(rewrite_having_aggregates(left, aggregates, counter)?),
            right: Box::new(rewrite_having_aggregates(right, aggregates, counter)?),
        }),
        Expression::Unary { op, operand } => Ok(Expression::Unary {
            op: op.clone(),
            operand: Box::new(rewrite_having_aggregates(operand, aggregates, counter)?),
        }),
        Expression::Conditional {
            condition,
            then_expr,
            else_expr,
        } => Ok(Expression::Conditional {
            condition: Box::new(rewrite_having_aggregates(condition, aggregates, counter)?),
            then_expr: Box::new(rewrite_having_aggregates(then_expr, aggregates, counter)?),
            else_expr: Box::new(rewrite_having_aggregates(else_expr, aggregates, counter)?),
        }),
        other => Ok(other.clone()),
    }
}

/// Recognize a SPARQL aggregate function call by name and build the corresponding
/// [`Aggregate`], or return `Ok(None)` for a non-aggregate function.
///
/// Name recognition and arity validation are delegated to the shared helpers
/// [`crate::algebra::aggregate_function_name`] and
/// [`crate::algebra::check_aggregate_arity`], so this executor-side check and the
/// parse-time `HAVING` validator agree on both which calls are aggregates and the
/// exact wrong-arity error texts. This remains as defense-in-depth: callers that
/// build the algebra directly (bypassing the parser) still get a loud, correct
/// rejection of wrong-arity aggregate calls.
fn function_to_aggregate(name: &str, args: &[Expression]) -> Result<Option<Aggregate>> {
    let Some(canonical) = crate::algebra::aggregate_function_name(name) else {
        return Ok(None);
    };
    crate::algebra::check_aggregate_arity(name, args.len()).map_err(|msg| anyhow::anyhow!(msg))?;
    // Arity is validated above: `COUNT` has 0 or 1 argument, every other
    // aggregate exactly one, so the indexing below cannot go out of bounds.
    let aggregate = match canonical {
        "COUNT" => Aggregate::Count {
            distinct: false,
            expr: args.first().cloned(),
        },
        "SUM" => Aggregate::Sum {
            distinct: false,
            expr: args[0].clone(),
        },
        "MIN" => Aggregate::Min {
            distinct: false,
            expr: args[0].clone(),
        },
        "MAX" => Aggregate::Max {
            distinct: false,
            expr: args[0].clone(),
        },
        "AVG" => Aggregate::Avg {
            distinct: false,
            expr: args[0].clone(),
        },
        "SAMPLE" => Aggregate::Sample {
            distinct: false,
            expr: args[0].clone(),
        },
        "GROUP_CONCAT" => Aggregate::GroupConcat {
            distinct: false,
            expr: args[0].clone(),
            separator: None,
        },
        // `aggregate_function_name` only returns the seven names handled above.
        _ => return Ok(None),
    };
    Ok(Some(aggregate))
}

#[cfg(test)]
mod serial_executor_tests {
    use crate::algebra::{
        Algebra, BinaryOperator, Expression, GroupCondition, Literal, Term, TriplePattern, Variable,
    };
    use crate::executor::dataset::InMemoryDataset;
    use crate::executor::QueryExecutor;
    use oxirs_core::model::NamedNode;

    fn v(name: &str) -> Term {
        Term::Variable(Variable::new_unchecked(name))
    }

    fn iri(s: &str) -> Term {
        Term::Iri(NamedNode::new_unchecked(s))
    }

    fn ev(name: &str) -> Expression {
        Expression::Variable(Variable::new_unchecked(name))
    }

    fn int_expr(n: i64) -> Expression {
        Expression::Literal(Literal {
            value: n.to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        })
    }

    fn sample_dataset() -> InMemoryDataset {
        InMemoryDataset::from_triples(vec![
            (iri("http://ex/s1"), iri("http://ex/p"), iri("http://ex/o1")),
            (iri("http://ex/s2"), iri("http://ex/p"), iri("http://ex/o2")),
        ])
    }

    fn bgp_s_p_o() -> Algebra {
        Algebra::Bgp(vec![TriplePattern {
            subject: v("s"),
            predicate: iri("http://ex/p"),
            object: v("o"),
        }])
    }

    /// A dataset with one default-graph triple and two triples in the named
    /// graph `<http://ex/g>`.
    fn graph_scoped_dataset() -> InMemoryDataset {
        let g = NamedNode::new_unchecked("http://ex/g");
        let mut ds = InMemoryDataset::new();
        ds.add_triple(iri("http://ex/sd"), iri("http://ex/p"), iri("http://ex/od"));
        ds.add_triple_in_graph(
            g.clone(),
            iri("http://ex/s1"),
            iri("http://ex/p"),
            iri("http://ex/o1"),
        );
        ds.add_triple_in_graph(
            g,
            iri("http://ex/s2"),
            iri("http://ex/p"),
            iri("http://ex/o2"),
        );
        ds
    }

    #[test]
    fn execute_serial_graph_iri_scopes_to_that_graph() {
        let exec = QueryExecutor::new();
        let ds = graph_scoped_dataset();
        let algebra = Algebra::Graph {
            graph: iri("http://ex/g"),
            pattern: Box::new(bgp_s_p_o()),
        };
        let sol = exec.execute_serial(&algebra, &ds).expect("graph serial");
        assert_eq!(
            sol.len(),
            2,
            "GRAPH <g> must return only the two triples in <g>, not the default graph"
        );
    }

    #[test]
    fn execute_serial_plain_bgp_reads_default_graph_only() {
        let exec = QueryExecutor::new();
        let ds = graph_scoped_dataset();
        // Plain BGP (no GRAPH) must see only the single default-graph triple,
        // NOT the union with the named graph.
        let sol = exec.execute_serial(&bgp_s_p_o(), &ds).expect("bgp serial");
        assert_eq!(
            sol.len(),
            1,
            "plain BGP must read the default graph only (regression for union bug)"
        );
    }

    #[test]
    fn execute_serial_graph_variable_enumerates_and_binds() {
        let exec = QueryExecutor::new();
        let ds = graph_scoped_dataset();
        let var_g = Variable::new_unchecked("g");
        let algebra = Algebra::Graph {
            graph: Term::Variable(var_g.clone()),
            pattern: Box::new(bgp_s_p_o()),
        };
        let sol = exec
            .execute_serial(&algebra, &ds)
            .expect("graph var serial");
        assert_eq!(
            sol.len(),
            2,
            "GRAPH ?g must enumerate the one named graph and bind both its rows"
        );
        for binding in &sol {
            assert_eq!(
                binding.get(&var_g),
                Some(&iri("http://ex/g")),
                "?g must be bound to the enumerated named graph"
            );
        }
    }

    #[test]
    fn execute_serial_having_filters_grouped_solution() {
        let exec = QueryExecutor::new();
        let ds = sample_dataset();
        let condition = Expression::Binary {
            op: BinaryOperator::Equal,
            left: Box::new(Expression::Variable(Variable::new_unchecked("o"))),
            right: Box::new(Expression::Iri(NamedNode::new_unchecked("http://ex/o1"))),
        };
        let algebra = Algebra::Having {
            pattern: Box::new(bgp_s_p_o()),
            condition,
        };
        let sol = exec.execute_serial(&algebra, &ds).expect("having serial");
        assert_eq!(
            sol.len(),
            1,
            "HAVING must filter, not silently drop to empty"
        );
    }

    /// `?s <dept> ?d` with three subjects in two departments (eng x2, sales x1).
    fn dept_dataset() -> InMemoryDataset {
        InMemoryDataset::from_triples(vec![
            (
                iri("http://ex/s1"),
                iri("http://ex/dept"),
                iri("http://ex/eng"),
            ),
            (
                iri("http://ex/s2"),
                iri("http://ex/dept"),
                iri("http://ex/eng"),
            ),
            (
                iri("http://ex/s3"),
                iri("http://ex/dept"),
                iri("http://ex/sales"),
            ),
        ])
    }

    fn bgp_s_dept_d() -> Algebra {
        Algebra::Bgp(vec![TriplePattern {
            subject: v("s"),
            predicate: iri("http://ex/dept"),
            object: v("d"),
        }])
    }

    #[test]
    fn having_count_aggregate_filters_groups_natively() {
        // GROUP BY ?d HAVING (COUNT(?s) > 1): the engine must hoist COUNT(?s) into
        // a per-group aggregate and keep only the eng group (size 2), dropping
        // sales (size 1) — with no caller-side rewrite of the algebra.
        let exec = QueryExecutor::new();
        let ds = dept_dataset();
        let group = Algebra::Group {
            pattern: Box::new(bgp_s_dept_d()),
            variables: vec![GroupCondition {
                expr: ev("d"),
                alias: None,
            }],
            aggregates: Vec::new(),
        };
        let condition = Expression::Binary {
            op: BinaryOperator::Greater,
            left: Box::new(Expression::Function {
                name: "COUNT".to_string(),
                args: vec![ev("s")],
            }),
            right: Box::new(int_expr(1)),
        };
        let algebra = Algebra::Having {
            pattern: Box::new(group),
            condition,
        };
        let sol = exec.execute_serial(&algebra, &ds).expect("having count");
        assert_eq!(
            sol.len(),
            1,
            "HAVING(COUNT(?s) > 1) must keep exactly the eng group"
        );
        let row = &sol[0];
        assert_eq!(
            row.get(&Variable::new_unchecked("d")),
            Some(&iri("http://ex/eng")),
            "surviving group key must be eng"
        );
        // The synthetic aggregate column must not leak downstream.
        assert!(
            row.keys().all(|k| !k.name().starts_with("__having_agg_")),
            "synthetic HAVING aggregate variable must be stripped"
        );
    }

    /// Two subjects, each with an <a> and <b> integer value.
    fn product_dataset() -> InMemoryDataset {
        let a = iri("http://ex/a");
        let b = iri("http://ex/b");
        let two = Term::Literal(Literal {
            value: "2".to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        });
        let three = Term::Literal(Literal {
            value: "3".to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        });
        let four = Term::Literal(Literal {
            value: "4".to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        });
        let five = Term::Literal(Literal {
            value: "5".to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        });
        InMemoryDataset::from_triples(vec![
            (iri("http://ex/s1"), a.clone(), two),
            (iri("http://ex/s1"), b.clone(), three),
            (iri("http://ex/s2"), a, four),
            (iri("http://ex/s2"), b, five),
        ])
    }

    fn product_group_having(threshold: i64) -> Algebra {
        // Implicit single group over `?s <a> ?a . ?s <b> ?b`, HAVING(SUM(?a*?b) >= k).
        let bgp = Algebra::Bgp(vec![
            TriplePattern {
                subject: v("s"),
                predicate: iri("http://ex/a"),
                object: v("a"),
            },
            TriplePattern {
                subject: v("s"),
                predicate: iri("http://ex/b"),
                object: v("b"),
            },
        ]);
        let group = Algebra::Group {
            pattern: Box::new(bgp),
            variables: Vec::new(),
            aggregates: Vec::new(),
        };
        let sum_product = Expression::Function {
            name: "SUM".to_string(),
            args: vec![Expression::Binary {
                op: BinaryOperator::Multiply,
                left: Box::new(ev("a")),
                right: Box::new(ev("b")),
            }],
        };
        Algebra::Having {
            pattern: Box::new(group),
            condition: Expression::Binary {
                op: BinaryOperator::GreaterEqual,
                left: Box::new(sum_product),
                right: Box::new(int_expr(threshold)),
            },
        }
    }

    #[test]
    fn having_sum_of_product_aggregate_kept_and_dropped() {
        // SUM(?a*?b) = 2*3 + 4*5 = 26.
        let exec = QueryExecutor::new();
        let ds = product_dataset();
        let kept = exec
            .execute_serial(&product_group_having(20), &ds)
            .expect("having sum kept");
        assert_eq!(kept.len(), 1, "SUM(?a*?b)=26 >= 20 must keep the group");
        let dropped = exec
            .execute_serial(&product_group_having(30), &ds)
            .expect("having sum dropped");
        assert!(
            dropped.is_empty(),
            "SUM(?a*?b)=26 >= 30 must drop the group"
        );
    }

    #[test]
    fn having_wrong_arity_aggregate_fails_loud() {
        // SUM with two arguments is a wrong-arity aggregate call; it must fail
        // loud with the arity error rather than silently mis-evaluating.
        let exec = QueryExecutor::new();
        let ds = product_dataset();
        let group = Algebra::Group {
            pattern: Box::new(bgp_s_dept_d()),
            variables: Vec::new(),
            aggregates: Vec::new(),
        };
        let algebra = Algebra::Having {
            pattern: Box::new(group),
            condition: Expression::Binary {
                op: BinaryOperator::GreaterEqual,
                left: Box::new(Expression::Function {
                    name: "SUM".to_string(),
                    args: vec![ev("a"), ev("b")],
                }),
                right: Box::new(int_expr(1)),
            },
        };
        let result = exec.execute_serial(&algebra, &ds);
        let err = result.expect_err("wrong-arity SUM in HAVING must error");
        assert!(
            err.to_string()
                .contains("SUM in HAVING expects exactly one argument"),
            "arity error message must be preserved, got: {err}"
        );
    }

    #[test]
    fn filter_unknown_function_propagates_loudly() {
        // A genuinely unknown function inside FILTER must fail the whole query
        // loud (typed UnknownFunctionError), never silently drop rows to a
        // 200-with-empty result.
        use crate::executor::UnknownFunctionError;
        let exec = QueryExecutor::new();
        let ds = sample_dataset();
        let algebra = Algebra::Filter {
            pattern: Box::new(bgp_s_p_o()),
            condition: Expression::Function {
                name: "no_such_fn".to_string(),
                args: vec![ev("o")],
            },
        };
        let err = exec
            .execute_serial(&algebra, &ds)
            .expect_err("unknown function in FILTER must fail loud, not silently drop rows");
        let unknown = err
            .downcast_ref::<UnknownFunctionError>()
            .expect("error must be a typed UnknownFunctionError");
        assert_eq!(unknown.0, "no_such_fn");
    }

    #[test]
    fn having_unknown_function_propagates_loudly() {
        // The same fail-loud contract applies to HAVING row evaluation.
        use crate::executor::UnknownFunctionError;
        let exec = QueryExecutor::new();
        let ds = dept_dataset();
        let group = Algebra::Group {
            pattern: Box::new(bgp_s_dept_d()),
            variables: vec![GroupCondition {
                expr: ev("d"),
                alias: None,
            }],
            aggregates: Vec::new(),
        };
        let algebra = Algebra::Having {
            pattern: Box::new(group),
            condition: Expression::Function {
                name: "no_such_fn".to_string(),
                args: vec![ev("d")],
            },
        };
        let err = exec
            .execute_serial(&algebra, &ds)
            .expect_err("unknown function in HAVING must fail loud, not silently drop groups");
        assert!(
            err.downcast_ref::<UnknownFunctionError>().is_some(),
            "HAVING unknown-function error must be a typed UnknownFunctionError, got: {err}"
        );
    }

    #[test]
    fn filter_per_row_type_error_excludes_row_not_propagated() {
        // Two rows: one numerically comparable, one whose object is a blank node
        // (a genuine type error under `>`). SPARQL 1.1 §17.3 excludes only the
        // erroring row; the query must still succeed (guard against
        // over-propagating non-UnknownFunctionError errors).
        let exec = QueryExecutor::new();
        let mut ds = InMemoryDataset::new();
        let int5 = Term::Literal(Literal {
            value: "5".to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        });
        ds.add_triple(iri("http://ex/s1"), iri("http://ex/p"), int5);
        ds.add_triple(
            iri("http://ex/s2"),
            iri("http://ex/p"),
            Term::BlankNode("b".to_string()),
        );

        let algebra = Algebra::Filter {
            pattern: Box::new(bgp_s_p_o()),
            condition: Expression::Binary {
                op: BinaryOperator::Greater,
                left: Box::new(ev("o")),
                right: Box::new(int_expr(3)),
            },
        };
        let sol = exec
            .execute_serial(&algebra, &ds)
            .expect("a per-row type error must not fail the whole query");
        assert_eq!(
            sol.len(),
            1,
            "only the numeric row survives; the blank-node row is excluded, not propagated"
        );
        assert_eq!(
            sol[0].get(&Variable::new_unchecked("s")),
            Some(&iri("http://ex/s1"))
        );
    }

    #[test]
    fn execute_serial_table_is_join_identity() {
        let exec = QueryExecutor::new();
        let ds = InMemoryDataset::new();
        let sol = exec
            .execute_serial(&Algebra::Table, &ds)
            .expect("table serial");
        assert_eq!(sol.len(), 1, "unit table must yield one empty binding");
        assert!(sol[0].is_empty());
    }

    #[test]
    fn execute_serial_filter_in_and_not_in() {
        let exec = QueryExecutor::new();
        let ds = sample_dataset();

        // FILTER(?o IN (<o1>, <o2>)) keeps both rows.
        let in_list = Expression::Function {
            name: "list".to_string(),
            args: vec![
                Expression::Iri(NamedNode::new_unchecked("http://ex/o1")),
                Expression::Iri(NamedNode::new_unchecked("http://ex/o2")),
            ],
        };
        let in_filter = Algebra::Filter {
            pattern: Box::new(bgp_s_p_o()),
            condition: Expression::Binary {
                op: BinaryOperator::In,
                left: Box::new(Expression::Variable(Variable::new_unchecked("o"))),
                right: Box::new(in_list),
            },
        };
        let sol = exec.execute_serial(&in_filter, &ds).expect("filter in");
        assert_eq!(sol.len(), 2, "IN over both objects must keep both rows");

        // FILTER(?o NOT IN (<o1>)) keeps only the o2 row.
        let not_in_filter = Algebra::Filter {
            pattern: Box::new(bgp_s_p_o()),
            condition: Expression::Binary {
                op: BinaryOperator::NotIn,
                left: Box::new(Expression::Variable(Variable::new_unchecked("o"))),
                right: Box::new(Expression::Iri(NamedNode::new_unchecked("http://ex/o1"))),
            },
        };
        let sol2 = exec
            .execute_serial(&not_in_filter, &ds)
            .expect("filter not in");
        assert_eq!(sol2.len(), 1, "NOT IN must exclude only the listed value");
    }

    #[test]
    fn execution_budget_aborts_scan_over_triple_limit() {
        use crate::query_governor::{ExecutionBudget, ResourceBudget};

        let ds = InMemoryDataset::from_triples(vec![
            (iri("http://ex/s1"), iri("http://ex/p"), iri("http://ex/o1")),
            (iri("http://ex/s2"), iri("http://ex/p"), iri("http://ex/o2")),
            (iri("http://ex/s3"), iri("http://ex/p"), iri("http://ex/o3")),
        ]);
        let budget = ExecutionBudget::new(ResourceBudget {
            max_wall_time: None,
            max_result_rows: None,
            max_triples_scanned: Some(1),
        });
        let exec = QueryExecutor::new().with_budget(budget);
        // Scanning 3 triples under a 1-triple budget must abort in the scan hot
        // path (record_triple_scan wired into execute_single_pattern).
        let result = exec.execute_serial(&bgp_s_p_o(), &ds);
        assert!(
            result.is_err(),
            "triple-scan budget must be enforced during the BGP scan"
        );
    }

    fn binding(pairs: &[(&str, Term)]) -> crate::algebra::Binding {
        let mut b = crate::algebra::Binding::new();
        for (name, term) in pairs {
            b.insert(Variable::new_unchecked(*name), term.clone());
        }
        b
    }

    #[test]
    fn regression_apply_left_join_honors_optional_filter() {
        // OPTIONAL { ?s :age ?age } FILTER(?age > 25): the LeftJoin FILTER must
        // be evaluated over the merged binding by the Serial strategy (it was
        // silently dropped before), matching perform_parallel_left_join.
        let exec = QueryExecutor::new();
        let left = vec![binding(&[
            ("s", iri("http://ex/s1")),
            ("name", iri("http://ex/alice")),
        ])];
        let right = vec![binding(&[
            ("s", iri("http://ex/s1")),
            ("age", int_term(30)),
        ])];

        // Passing filter (?age > 25): merged row with ?age bound is kept.
        let cond_pass = Some(Expression::Binary {
            op: BinaryOperator::Greater,
            left: Box::new(ev("age")),
            right: Box::new(int_expr(25)),
        });
        let sol = exec
            .apply_left_join(left.clone(), right.clone(), &cond_pass)
            .expect("left join ok");
        assert_eq!(sol.len(), 1);
        assert!(sol[0].contains_key(&Variable::new_unchecked("age")));

        // Failing filter (?age > 40): the compatible right row does not count as
        // a match, so the left row is emitted with ?age UNBOUND.
        let cond_fail = Some(Expression::Binary {
            op: BinaryOperator::Greater,
            left: Box::new(ev("age")),
            right: Box::new(int_expr(40)),
        });
        let sol = exec
            .apply_left_join(left, right, &cond_fail)
            .expect("left join ok");
        assert_eq!(sol.len(), 1);
        assert!(!sol[0].contains_key(&Variable::new_unchecked("age")));
    }

    fn int_term(n: i64) -> Term {
        Term::Literal(Literal {
            value: n.to_string(),
            language: None,
            datatype: Some(NamedNode::new_unchecked(
                "http://www.w3.org/2001/XMLSchema#integer",
            )),
        })
    }

    fn str_of(var: &str) -> Expression {
        Expression::Function {
            name: "str".to_string(),
            args: vec![ev(var)],
        }
    }

    #[test]
    fn regression_order_by_expression_key_is_not_a_noop() {
        // ORDER BY STR(?s) used to evaluate every key to None (only
        // Expression::Variable was implemented), making the comparator a
        // no-op that preserved whatever order the pipeline produced.
        let exec = QueryExecutor::new();
        let solution = vec![
            binding(&[("s", iri("http://ex/current-account-gdp"))]),
            binding(&[("s", iri("http://ex/zeta"))]),
            binding(&[("s", iri("http://ex/current-account"))]),
        ];
        let sorted = exec.apply_order_by(
            solution,
            &[crate::algebra::OrderCondition {
                expr: str_of("s"),
                ascending: true,
            }],
        );
        let keys: Vec<&str> = sorted
            .iter()
            .map(|b| match &b[&Variable::new_unchecked("s")] {
                Term::Iri(n) => n.as_str(),
                other => panic!("expected IRI, got {other:?}"),
            })
            .collect();
        assert_eq!(
            keys,
            vec![
                "http://ex/current-account",
                "http://ex/current-account-gdp",
                "http://ex/zeta",
            ],
            "ORDER BY STR(?s) must actually sort by the evaluated key"
        );
    }

    #[test]
    fn regression_order_by_unbound_key_sorts_lowest_ascending() {
        // SPARQL 1.1 §15.1: unbound/errored keys rank lowest, so they come
        // FIRST ascending (and last descending via cmp.reverse()).
        let exec = QueryExecutor::new();
        let solution = vec![
            binding(&[("s", iri("http://ex/a")), ("k", int_term(1))]),
            binding(&[("s", iri("http://ex/b"))]), // ?k unbound
        ];
        let sorted = exec.apply_order_by(
            solution,
            &[crate::algebra::OrderCondition {
                expr: ev("k"),
                ascending: true,
            }],
        );
        assert!(
            !sorted[0].contains_key(&Variable::new_unchecked("k")),
            "the unbound-key row must sort first in ascending order"
        );
    }

    #[test]
    fn regression_order_by_mixed_numeric_and_lexical_keys_total_order() {
        // Keys 5, 10 (xsd:integer) and plain "3abc": the old per-pair
        // parse-as-f64 branch produced the cycle 3abc < 5 < 10 < 3abc; the
        // composite key puts the numeric partition first, in numeric order.
        let exec = QueryExecutor::new();
        let plain = Term::Literal(Literal {
            value: "3abc".to_string(),
            language: None,
            datatype: None,
        });
        let solution = vec![
            binding(&[("k", int_term(10))]),
            binding(&[("k", plain)]),
            binding(&[("k", int_term(5))]),
        ];
        let sorted = exec.apply_order_by(
            solution,
            &[crate::algebra::OrderCondition {
                expr: ev("k"),
                ascending: true,
            }],
        );
        let keys: Vec<String> = sorted
            .iter()
            .map(|b| match &b[&Variable::new_unchecked("k")] {
                Term::Literal(l) => l.value.clone(),
                other => panic!("expected literal, got {other:?}"),
            })
            .collect();
        assert_eq!(keys, vec!["5", "10", "3abc"]);
    }

    #[test]
    fn regression_max_over_gyear_preserves_input_datatype() {
        // MAX must return one of the input terms verbatim (SPARQL 1.1
        // §18.5.1.10); it used to synthesize "2025"^^xsd:decimal instead of
        // returning "2025"^^xsd:gYear.
        let gyear = |y: &str| {
            Term::Literal(Literal {
                value: y.to_string(),
                language: None,
                datatype: Some(NamedNode::new_unchecked(
                    "http://www.w3.org/2001/XMLSchema#gYear",
                )),
            })
        };
        let ds = InMemoryDataset::from_triples(vec![
            (iri("http://ex/o1"), iri("http://ex/y"), gyear("2023")),
            (iri("http://ex/o2"), iri("http://ex/y"), gyear("2024")),
            (iri("http://ex/o3"), iri("http://ex/y"), gyear("2025")),
        ]);
        let algebra = Algebra::Group {
            pattern: Box::new(Algebra::Bgp(vec![TriplePattern {
                subject: v("o"),
                predicate: iri("http://ex/y"),
                object: v("y"),
            }])),
            variables: vec![],
            aggregates: vec![(
                Variable::new_unchecked("m"),
                crate::algebra::Aggregate::Max {
                    distinct: false,
                    expr: ev("y"),
                },
            )],
        };
        let exec = QueryExecutor::new();
        let sol = exec.execute_serial(&algebra, &ds).expect("group serial");
        assert_eq!(sol.len(), 1);
        match &sol[0][&Variable::new_unchecked("m")] {
            Term::Literal(lit) => {
                assert_eq!(lit.value, "2025");
                assert_eq!(
                    lit.datatype.as_ref().map(|d| d.as_str()),
                    Some("http://www.w3.org/2001/XMLSchema#gYear"),
                    "MAX must preserve the input datatype, not fabricate xsd:decimal"
                );
            }
            other => panic!("expected literal, got {other:?}"),
        }
    }

    fn values_join_bgp(target: &str) -> Algebra {
        Algebra::Join {
            left: Box::new(Algebra::Values {
                variables: vec![Variable::new_unchecked("s")],
                bindings: vec![binding(&[("s", iri(target))])],
            }),
            right: Box::new(Algebra::Bgp(vec![TriplePattern {
                subject: v("s"),
                predicate: v("p"),
                object: v("o"),
            }])),
        }
    }

    #[test]
    fn regression_values_join_uses_index_not_full_scan() {
        use crate::query_governor::{ExecutionBudget, ResourceBudget};

        // 300 triples, exactly 2 with the target subject. The bound-join path
        // substitutes the VALUES row into the BGP so the store lookup is
        // subject-indexed; the old shape scanned all 300 triples and would
        // blow this 10-triple budget (a result-only assertion passes both
        // before and after — the budget is the point of this test).
        let mut triples = Vec::new();
        for i in 0..149 {
            triples.push((
                iri(&format!("http://ex/other{i}")),
                iri("http://ex/p"),
                int_term(i),
            ));
            triples.push((
                iri(&format!("http://ex/other{i}")),
                iri("http://ex/q"),
                int_term(i),
            ));
        }
        triples.push((iri("http://ex/target"), iri("http://ex/p"), int_term(1)));
        triples.push((iri("http://ex/target"), iri("http://ex/q"), int_term(2)));
        let ds = InMemoryDataset::from_triples(triples);

        let budget = ExecutionBudget::new(ResourceBudget {
            max_wall_time: None,
            max_result_rows: None,
            max_triples_scanned: Some(10),
        });
        let exec = QueryExecutor::new().with_budget(budget);
        let sol = exec
            .execute_serial(&values_join_bgp("http://ex/target"), &ds)
            .expect("bound join must stay within the scan budget");
        assert_eq!(sol.len(), 2, "both target triples must be returned");
        for row in &sol {
            assert_eq!(
                row[&Variable::new_unchecked("s")],
                iri("http://ex/target"),
                "?s must stay pinned to the VALUES row in the merged result"
            );
            assert!(row.contains_key(&Variable::new_unchecked("p")));
            assert!(row.contains_key(&Variable::new_unchecked("o")));
        }
    }

    #[test]
    fn regression_values_join_multi_row_and_unmatched_rows() {
        // Two VALUES rows match, one matches nothing: the bound join must
        // return exactly the union of per-row matches, with no phantom row
        // for the unmatched binding.
        let ds = InMemoryDataset::from_triples(vec![
            (iri("http://ex/a"), iri("http://ex/p"), int_term(1)),
            (iri("http://ex/b"), iri("http://ex/p"), int_term(2)),
            (iri("http://ex/c"), iri("http://ex/p"), int_term(3)),
        ]);
        let join = Algebra::Join {
            left: Box::new(Algebra::Values {
                variables: vec![Variable::new_unchecked("s")],
                bindings: vec![
                    binding(&[("s", iri("http://ex/a"))]),
                    binding(&[("s", iri("http://ex/b"))]),
                    binding(&[("s", iri("http://ex/missing"))]),
                ],
            }),
            right: Box::new(Algebra::Bgp(vec![TriplePattern {
                subject: v("s"),
                predicate: iri("http://ex/p"),
                object: v("o"),
            }])),
        };
        let exec = QueryExecutor::new();
        let sol = exec.execute_serial(&join, &ds).expect("values join");
        assert_eq!(sol.len(), 2, "one row per matched VALUES entry");
        let objects: Vec<&str> = sol
            .iter()
            .map(|b| match &b[&Variable::new_unchecked("o")] {
                Term::Literal(l) => l.value.as_str(),
                other => panic!("expected literal, got {other:?}"),
            })
            .collect();
        assert!(objects.contains(&"1") && objects.contains(&"2"));
    }

    #[test]
    fn regression_bound_join_falls_back_when_filter_uses_condition_only_var() {
        // Join(Values(?s ?x), Filter(?o > ?x, Bgp(?s ?p ?o))): ?x occurs only
        // in the FILTER condition, not in the BGP. On the fallback path the
        // inner group evaluates alone, ?x is unbound, the condition errors and
        // every row drops — the join is empty. The bound path would substitute
        // ?x=5 and let rows THROUGH; Gate 4 must force the fallback so the
        // result cannot depend on the VALUES row count.
        let ds = InMemoryDataset::from_triples(vec![(
            iri("http://ex/a"),
            iri("http://ex/p"),
            int_term(10),
        )]);
        let join = Algebra::Join {
            left: Box::new(Algebra::Values {
                variables: vec![Variable::new_unchecked("s"), Variable::new_unchecked("x")],
                bindings: vec![binding(&[("s", iri("http://ex/a")), ("x", int_term(5))])],
            }),
            right: Box::new(Algebra::Filter {
                pattern: Box::new(Algebra::Bgp(vec![TriplePattern {
                    subject: v("s"),
                    predicate: v("p"),
                    object: v("o"),
                }])),
                condition: Expression::Binary {
                    op: BinaryOperator::Greater,
                    left: Box::new(ev("o")),
                    right: Box::new(ev("x")),
                },
            }),
        };
        let exec = QueryExecutor::new();
        let sol = exec.execute_serial(&join, &ds).expect("join executes");
        assert!(
            sol.is_empty(),
            "condition-only variable must take the fallback path (same result \
             regardless of VALUES size), got {sol:?}"
        );
    }

    #[test]
    fn regression_order_by_integers_beyond_f64_precision() {
        // 9999999999999999999 and 10000000000000000001 collide as f64 (1e19);
        // the exact-value tie-break must still order them numerically.
        let exec = QueryExecutor::new();
        let big = |s: &str| {
            Term::Literal(Literal {
                value: s.to_string(),
                language: None,
                datatype: Some(NamedNode::new_unchecked(
                    "http://www.w3.org/2001/XMLSchema#integer",
                )),
            })
        };
        let solution = vec![
            binding(&[("k", big("10000000000000000001"))]),
            binding(&[("k", big("9999999999999999999"))]),
        ];
        let sorted = exec.apply_order_by(
            solution,
            &[crate::algebra::OrderCondition {
                expr: ev("k"),
                ascending: true,
            }],
        );
        let keys: Vec<String> = sorted
            .iter()
            .map(|b| match &b[&Variable::new_unchecked("k")] {
                Term::Literal(l) => l.value.clone(),
                other => panic!("expected literal, got {other:?}"),
            })
            .collect();
        assert_eq!(
            keys,
            vec!["9999999999999999999", "10000000000000000001"],
            "integers that collide in f64 must still sort by exact value"
        );
    }

    #[test]
    fn regression_filter_in_iri_list_uses_index_not_full_scan() {
        use crate::query_governor::{ExecutionBudget, ResourceBudget};

        // 300 triples, 2 subjects in the IN list with one triple each. The
        // pushdown answers via per-IRI index lookups; the old path scanned all
        // 300 rows and would blow the 10-triple budget.
        let mut triples = Vec::new();
        for i in 0..298 {
            triples.push((
                iri(&format!("http://ex/other{i}")),
                iri("http://ex/p"),
                int_term(i),
            ));
        }
        triples.push((iri("http://ex/a"), iri("http://ex/p"), int_term(1)));
        triples.push((iri("http://ex/b"), iri("http://ex/p"), int_term(2)));
        let ds = InMemoryDataset::from_triples(triples);

        let filter = Algebra::Filter {
            pattern: Box::new(Algebra::Bgp(vec![TriplePattern {
                subject: v("s"),
                predicate: iri("http://ex/p"),
                object: v("o"),
            }])),
            condition: Expression::Binary {
                op: BinaryOperator::In,
                left: Box::new(ev("s")),
                right: Box::new(Expression::Function {
                    name: "list".to_string(),
                    args: vec![
                        Expression::Iri(NamedNode::new_unchecked("http://ex/a")),
                        Expression::Iri(NamedNode::new_unchecked("http://ex/b")),
                        // Duplicate entry: IN is a membership test, the row
                        // for <a> must not appear twice.
                        Expression::Iri(NamedNode::new_unchecked("http://ex/a")),
                    ],
                }),
            },
        };
        let budget = ExecutionBudget::new(ResourceBudget {
            max_wall_time: None,
            max_result_rows: None,
            max_triples_scanned: Some(10),
        });
        let exec = QueryExecutor::new().with_budget(budget);
        let sol = exec
            .execute_serial(&filter, &ds)
            .expect("IN pushdown must stay within the scan budget");
        assert_eq!(sol.len(), 2, "one row per distinct matching IRI: {sol:?}");
        for row in &sol {
            assert!(row.contains_key(&Variable::new_unchecked("s")));
            assert!(row.contains_key(&Variable::new_unchecked("o")));
        }
    }

    #[test]
    fn regression_filter_in_literal_list_falls_back_to_value_equality() {
        // "1"^^xsd:integer IN ("01"^^xsd:integer) is TRUE under IN's value
        // equality but the terms differ — literal lists must therefore take
        // the row-by-row path, never the term-substitution pushdown.
        let ds = InMemoryDataset::from_triples(vec![(
            iri("http://ex/a"),
            iri("http://ex/p"),
            int_term(1),
        )]);
        let filter = Algebra::Filter {
            pattern: Box::new(Algebra::Bgp(vec![TriplePattern {
                subject: v("s"),
                predicate: iri("http://ex/p"),
                object: v("o"),
            }])),
            condition: Expression::Binary {
                op: BinaryOperator::In,
                left: Box::new(ev("o")),
                right: Box::new(Expression::Literal(Literal {
                    value: "01".to_string(),
                    language: None,
                    datatype: Some(NamedNode::new_unchecked(
                        "http://www.w3.org/2001/XMLSchema#integer",
                    )),
                })),
            },
        };
        let exec = QueryExecutor::new();
        let sol = exec.execute_serial(&filter, &ds).expect("filter in");
        assert_eq!(
            sol.len(),
            1,
            "value equality (1 = 01) must be preserved via the fallback path"
        );
    }

    #[test]
    fn regression_distinct_dedupes_independently_built_bindings() {
        // apply_distinct used to key off `HashMap::iter()` order, which is
        // per-instance random — two independently-constructed bindings with
        // identical content produced different keys ~50% of the time each.
        // 45 pairs x 2 copies makes a spurious pass astronomically unlikely
        // (~2^-45); a 1-2 row test would pass half the time regardless.
        let exec = QueryExecutor::new();
        let mut solution = crate::algebra::Solution::new();
        for _copy in 0..2 {
            for i in 0..45 {
                // Each copy built as its own HashMap (fresh RandomState).
                solution.push(binding(&[
                    ("c", iri(&format!("http://ex/c{}", i % 9))),
                    ("i", iri(&format!("http://ex/i{}", i / 9))),
                ]));
            }
        }
        let deduped = exec.apply_distinct(solution);
        assert_eq!(
            deduped.len(),
            45,
            "DISTINCT must collapse structurally-identical bindings regardless \
             of each HashMap's iteration order"
        );
    }
}