Skip to main content

oxirs_arq/optimizer/
mod.rs

1//! Query Optimization Module
2//!
3//! This module provides query optimization capabilities including
4//! rule-based and cost-based optimization passes.
5
6pub mod cardinality_integration;
7pub mod config;
8pub mod execution_tracking;
9pub mod index_types;
10pub mod production_tuning;
11pub mod statistics;
12
13pub mod adaptive;
14pub mod federated_plan;
15pub mod join_order;
16pub mod materialized_view;
17pub mod passes;
18pub mod view_registry;
19
20pub use adaptive::*;
21pub use join_order::*;
22pub use materialized_view::*;
23pub use passes::{
24    ConstantFoldingPass, OptimizationPass, OptimizationPipeline, PipelineResult,
25    RedundantJoinEliminationPass, UnusedVariableEliminationPass,
26};
27pub use view_registry::*;
28
29pub use cardinality_integration::*;
30pub use config::*;
31pub use execution_tracking::*;
32pub use index_types::*;
33pub use production_tuning::*;
34pub use statistics::*;
35
36use crate::algebra::{Algebra, Expression, TriplePattern, Variable};
37use crate::cost_model::{CostEstimate, CostModel, CostModelConfig};
38use crate::optimizer::federated_plan::{
39    FederatedPlanOutcome, FederatedPlanner, SourceSelectivityProvider,
40};
41use crate::plan_cache::{compute_fingerprint, PlanCache};
42use anyhow::Result;
43use std::collections::hash_map::DefaultHasher;
44use std::collections::HashSet;
45use std::hash::{Hash, Hasher};
46use std::sync::Arc;
47
48/// Query complexity metrics for adaptive optimization
49#[derive(Debug, Clone, Default)]
50struct QueryComplexity {
51    /// Number of triple patterns in the query
52    triple_patterns: usize,
53    /// Number of joins
54    joins: usize,
55    /// Number of filters
56    filters: usize,
57    /// Whether query has ordering
58    ordering: bool,
59    /// Whether query has grouping
60    grouping: bool,
61}
62
63/// Main query optimizer
64pub struct Optimizer {
65    config: OptimizerConfig,
66    statistics: Statistics,
67    execution_records: Vec<ExecutionRecord>,
68    cost_model: CostModel,
69    /// Optional federation-aware planning provider (W2-S4 deepening).
70    ///
71    /// When `Some`, [`Optimizer::optimize`] applies a [`FederatedPlanner`] pass
72    /// after the standard rule/cost-based passes, rewriting BGPs whose IRIs
73    /// resolve to known endpoints into [`crate::algebra::Algebra::Service`]
74    /// nodes.  When `None`, optimization behavior is identical to the
75    /// pre-W2-S4 baseline (no federated rewrite).
76    federated_provider: Option<Arc<dyn SourceSelectivityProvider>>,
77    /// Latency weight passed to [`FederatedPlanner`].  Defaults to 1.0.
78    federated_latency_weight: f64,
79    /// Last federated planning outcome, captured for observability.
80    last_federated_outcome: Option<FederatedPlanOutcome>,
81    /// Optional algebra-level plan cache (JIT phase a).
82    ///
83    /// When `Some`, repeated queries whose algebra fingerprint matches a cached
84    /// entry skip the rule/cost-based optimization passes entirely.  The cache
85    /// is **not** consulted when a federation provider is registered, because
86    /// the federation pass writes observable side-state
87    /// ([`Optimizer::last_federated_outcome`]) that must remain correct on
88    /// every call.
89    plan_cache: Option<PlanCache<Algebra>>,
90}
91
92impl Optimizer {
93    /// Create a new optimizer with configuration
94    pub fn new(config: OptimizerConfig) -> Self {
95        Self {
96            config,
97            statistics: Statistics::new(),
98            execution_records: Vec::new(),
99            cost_model: CostModel::new(CostModelConfig::default()),
100            federated_provider: None,
101            federated_latency_weight: 1.0,
102            last_federated_outcome: None,
103            plan_cache: None,
104        }
105    }
106
107    /// Create a new optimizer with custom cost model configuration
108    pub fn with_cost_model(config: OptimizerConfig, cost_config: CostModelConfig) -> Self {
109        Self {
110            config,
111            statistics: Statistics::new(),
112            execution_records: Vec::new(),
113            cost_model: CostModel::new(cost_config),
114            federated_provider: None,
115            federated_latency_weight: 1.0,
116            last_federated_outcome: None,
117            plan_cache: None,
118        }
119    }
120
121    /// Enable the algebra-level plan cache with the given LRU `capacity`.
122    ///
123    /// When enabled, calls to [`Self::optimize`] that match a cached fingerprint
124    /// skip the rule/cost-based passes.  The cache is bypassed when a federation
125    /// provider is registered (see [`Self::with_federated_planner`]) to preserve
126    /// [`Self::last_federated_outcome`] correctness.
127    ///
128    /// Builder pattern; chains naturally with [`Self::new`].
129    ///
130    /// ```rust
131    /// use oxirs_arq::optimizer::{Optimizer, OptimizerConfig};
132    ///
133    /// let optimizer = Optimizer::new(OptimizerConfig::default())
134    ///     .with_plan_cache_capacity(1024);
135    /// assert!(optimizer.has_plan_cache());
136    /// ```
137    pub fn with_plan_cache_capacity(mut self, capacity: usize) -> Self {
138        self.plan_cache = Some(PlanCache::new(capacity));
139        self
140    }
141
142    /// Returns `true` if a plan cache has been attached.
143    pub fn has_plan_cache(&self) -> bool {
144        self.plan_cache.is_some()
145    }
146
147    /// Return `(hits, misses, evictions)` from the attached plan cache, or
148    /// `(0, 0, 0)` when no cache is configured.
149    pub fn plan_cache_stats(&self) -> (u64, u64, u64) {
150        self.plan_cache
151            .as_ref()
152            .map(|c| c.stats())
153            .unwrap_or((0, 0, 0))
154    }
155
156    /// Invalidate all entries in the plan cache (e.g. after a schema change).
157    ///
158    /// A no-op when no cache is configured.
159    pub fn invalidate_plan_cache(&self) {
160        if let Some(ref cache) = self.plan_cache {
161            cache.invalidate_all();
162        }
163    }
164
165    /// Register a [`SourceSelectivityProvider`] so that [`Self::optimize`]
166    /// transparently applies federated planning after the standard passes.
167    ///
168    /// This is the canonical opt-in entry point for embedders that want
169    /// federation-aware rewriting.  When a provider is registered, queries
170    /// whose IRIs map to known endpoints are rewritten into
171    /// [`crate::algebra::Algebra::Service`] nodes.  When no provider is
172    /// registered the optimizer behavior is unchanged.
173    ///
174    /// Builder pattern; chains naturally with [`Self::new`].
175    ///
176    /// ```
177    /// # use std::sync::Arc;
178    /// # use oxirs_arq::optimizer::{Optimizer, OptimizerConfig};
179    /// # use oxirs_arq::optimizer::federated_plan::StaticSourceProvider;
180    /// let provider = Arc::new(StaticSourceProvider::new());
181    /// let optimizer = Optimizer::new(OptimizerConfig::default())
182    ///     .with_federated_planner(provider);
183    /// assert!(optimizer.has_federated_planner());
184    /// ```
185    pub fn with_federated_planner(mut self, provider: Arc<dyn SourceSelectivityProvider>) -> Self {
186        self.federated_provider = Some(provider);
187        self
188    }
189
190    /// Set the latency weight used by the federation planner.
191    ///
192    /// Higher values penalise slow endpoints more aggressively.  Default is
193    /// 1.0.  Has no effect unless [`Self::with_federated_planner`] is also
194    /// invoked.
195    pub fn with_federated_latency_weight(mut self, weight: f64) -> Self {
196        self.federated_latency_weight = weight;
197        self
198    }
199
200    /// Whether a federated planning provider is registered.
201    pub fn has_federated_planner(&self) -> bool {
202        self.federated_provider.is_some()
203    }
204
205    /// Outcome of the most recent federated planning pass, if any.
206    ///
207    /// Useful for observability — embedders can inspect which endpoints
208    /// were touched by the last [`Self::optimize`] call.  Returns `None`
209    /// when no provider is registered or when no query has been optimized
210    /// yet.
211    ///
212    /// **Note:** the `algebra` field of the cached
213    /// [`FederatedPlanOutcome`] is intentionally a placeholder
214    /// (`Algebra::Bgp(vec![])`); the actually rewritten algebra is moved into
215    /// the value returned by [`Self::optimize`] to avoid paying for a deep
216    /// clone of the plan tree.  Inspect `endpoints_used` and
217    /// `patterns_federated` for observability — those carry the useful state.
218    pub fn last_federated_outcome(&self) -> Option<&FederatedPlanOutcome> {
219        self.last_federated_outcome.as_ref()
220    }
221
222    /// Optimize a query algebra
223    ///
224    /// When a plan cache is configured **and** no federation provider is
225    /// registered, this first computes the fingerprint of `algebra` and
226    /// returns the cached result on a hit, skipping all optimization passes.
227    /// On a miss it runs the standard passes, stores the result, and returns
228    /// it.
229    ///
230    /// The cache is bypassed when a federation provider is registered to
231    /// preserve the correctness of [`Self::last_federated_outcome`] — the
232    /// federation pass writes observable side-state on every call.
233    pub fn optimize(&mut self, algebra: Algebra) -> Result<Algebra> {
234        // JIT plan cache — phase a.
235        // Only active when a cache is attached AND no federation provider is
236        // registered (federation writes per-call side-state that must stay fresh).
237        if self.plan_cache.is_some() && self.federated_provider.is_none() {
238            let fp = compute_fingerprint(&algebra);
239            if let Some(cached) = self.plan_cache.as_ref().and_then(|c| c.get(fp)) {
240                return Ok(cached);
241            }
242            // Cache miss — run optimization, then store.
243            let result = self.run_optimization_passes(algebra)?;
244            if let Some(ref cache) = self.plan_cache {
245                cache.insert(fp, result.clone());
246            }
247            return Ok(result);
248        }
249
250        // No cache (or cache bypassed due to federation provider).
251        self.run_optimization_passes_and_federate(algebra)
252    }
253
254    /// Internal helper: run rule/cost passes + federation pass.
255    /// Called when the plan cache is bypassed (federation or no cache).
256    fn run_optimization_passes_and_federate(&mut self, algebra: Algebra) -> Result<Algebra> {
257        let optimised = self.run_optimization_passes(algebra)?;
258
259        // Federation pass (W2-S4).
260        if let Some(provider) = self.federated_provider.clone() {
261            let planner =
262                FederatedPlanner::new(provider).with_latency_weight(self.federated_latency_weight);
263            let mut outcome = planner.plan(optimised);
264            let rewritten = std::mem::replace(&mut outcome.algebra, Algebra::Bgp(Vec::new()));
265            self.last_federated_outcome = Some(outcome);
266            Ok(rewritten)
267        } else {
268            self.last_federated_outcome = None;
269            Ok(optimised)
270        }
271    }
272
273    /// Run the rule/cost-based optimization passes only (no federation, no cache).
274    fn run_optimization_passes(&mut self, algebra: Algebra) -> Result<Algebra> {
275        // Adaptive optimization: use fast path for simple queries
276        let complexity = self.estimate_query_complexity(&algebra);
277
278        // For simple queries (≤5 triple patterns), skip cost-based optimization
279        // to avoid optimization overhead exceeding benefits
280        let use_cost_based = if complexity.triple_patterns <= 5 {
281            false // Fast path: simple heuristics only
282        } else {
283            self.config.cost_based // Complex queries benefit from cost model
284        };
285
286        let mut optimized = algebra;
287        let mut pass = 0;
288
289        // Limit passes for simple queries to minimize overhead
290        let effective_max_passes = if complexity.triple_patterns <= 5 {
291            2.min(self.config.max_passes) // Maximum 2 passes for simple queries
292        } else {
293            self.config.max_passes
294        };
295
296        // Apply optimization passes
297        while pass < effective_max_passes {
298            let before = optimized.clone();
299
300            if self.config.filter_pushdown {
301                optimized = self.apply_filter_pushdown(optimized)?;
302            }
303
304            if self.config.join_reordering {
305                optimized = if use_cost_based {
306                    self.apply_cost_based_join_reordering(optimized)?
307                } else {
308                    self.apply_join_reordering(optimized)?
309                };
310            }
311
312            if self.config.projection_pushdown {
313                optimized = self.apply_projection_pushdown(optimized)?;
314            }
315
316            if self.config.constant_folding {
317                optimized = self.apply_constant_folding(optimized)?;
318            }
319
320            if self.config.dead_code_elimination {
321                optimized = self.apply_dead_code_elimination(optimized)?;
322            }
323
324            // Check for convergence
325            if self.algebra_equal(&before, &optimized) {
326                break;
327            }
328
329            pass += 1;
330        }
331
332        Ok(optimized)
333    }
334
335    /// Add execution record for learning
336    pub fn add_execution_record(&mut self, record: ExecutionRecord) {
337        self.statistics.update_with_execution(&record);
338
339        // Update cost model with actual execution feedback
340        // Convert execution time to cost units (milliseconds)
341        let actual_cost = record.execution_time.as_millis() as f64;
342        self.cost_model
343            .update_with_feedback(&record.algebra, actual_cost, record.cardinality);
344
345        self.execution_records.push(record);
346    }
347
348    /// Get cost estimate for an algebra expression
349    pub fn estimate_cost(&mut self, algebra: &Algebra) -> Result<CostEstimate> {
350        self.cost_model.estimate_cost(algebra)
351    }
352
353    /// Clear cost model cache
354    pub fn clear_cost_cache(&mut self) {
355        self.cost_model.clear_cache();
356    }
357
358    /// Get optimizer statistics
359    pub fn statistics(&self) -> &Statistics {
360        &self.statistics
361    }
362
363    /// Apply filter pushdown optimization
364    fn apply_filter_pushdown(&self, algebra: Algebra) -> Result<Algebra> {
365        match algebra {
366            Algebra::Filter { pattern, condition } => {
367                // First, apply advanced filter optimizations
368                let optimized_conditions = self.optimize_filter_conditions(&condition)?;
369
370                // Apply each condition separately for better pushdown opportunities
371                let mut result_pattern = *pattern;
372                for cond in optimized_conditions {
373                    result_pattern = self.push_filter_down(result_pattern, &cond)?;
374                }
375                Ok(result_pattern)
376            }
377            Algebra::Join { left, right } => Ok(Algebra::Join {
378                left: Box::new(self.apply_filter_pushdown(*left)?),
379                right: Box::new(self.apply_filter_pushdown(*right)?),
380            }),
381            Algebra::Union { left, right } => Ok(Algebra::Union {
382                left: Box::new(self.apply_filter_pushdown(*left)?),
383                right: Box::new(self.apply_filter_pushdown(*right)?),
384            }),
385            other => Ok(other),
386        }
387    }
388
389    /// Push filter down into the algebra tree
390    fn push_filter_down(&self, algebra: Algebra, condition: &Expression) -> Result<Algebra> {
391        match algebra {
392            Algebra::Join { left, right } => {
393                let left_vars = self.extract_variables(&left);
394                let right_vars = self.extract_variables(&right);
395                let filter_vars = self.extract_expression_variables(condition);
396
397                if filter_vars.iter().all(|v| left_vars.contains(v)) {
398                    // Filter only uses left variables - push to left
399                    Ok(Algebra::Join {
400                        left: Box::new(Algebra::Filter {
401                            pattern: left,
402                            condition: condition.clone(),
403                        }),
404                        right,
405                    })
406                } else if filter_vars.iter().all(|v| right_vars.contains(v)) {
407                    // Filter only uses right variables - push to right
408                    Ok(Algebra::Join {
409                        left,
410                        right: Box::new(Algebra::Filter {
411                            pattern: right,
412                            condition: condition.clone(),
413                        }),
414                    })
415                } else {
416                    // Filter uses variables from both sides - keep at join level
417                    Ok(Algebra::Filter {
418                        pattern: Box::new(Algebra::Join { left, right }),
419                        condition: condition.clone(),
420                    })
421                }
422            }
423            other => Ok(Algebra::Filter {
424                pattern: Box::new(other),
425                condition: condition.clone(),
426            }),
427        }
428    }
429
430    /// Apply join reordering optimization based on selectivity
431    fn apply_join_reordering(&self, algebra: Algebra) -> Result<Algebra> {
432        match algebra {
433            Algebra::Join { left, right } => {
434                let left_cost = self.estimate_simple_cost(&left);
435                let right_cost = self.estimate_simple_cost(&right);
436
437                // Always put lower cost operation first for left-deep join trees
438                if left_cost > right_cost {
439                    Ok(Algebra::Join {
440                        left: Box::new(self.apply_join_reordering(*right)?),
441                        right: Box::new(self.apply_join_reordering(*left)?),
442                    })
443                } else {
444                    Ok(Algebra::Join {
445                        left: Box::new(self.apply_join_reordering(*left)?),
446                        right: Box::new(self.apply_join_reordering(*right)?),
447                    })
448                }
449            }
450            Algebra::Union { left, right } => Ok(Algebra::Union {
451                left: Box::new(self.apply_join_reordering(*left)?),
452                right: Box::new(self.apply_join_reordering(*right)?),
453            }),
454            other => Ok(other),
455        }
456    }
457
458    /// Apply cost-based join reordering using detailed cost model
459    fn apply_cost_based_join_reordering(&mut self, algebra: Algebra) -> Result<Algebra> {
460        match algebra {
461            Algebra::Join { left, right } => {
462                // Get detailed cost estimates for both sides
463                let left_estimate = self.cost_model.estimate_cost(&left)?;
464                let right_estimate = self.cost_model.estimate_cost(&right)?;
465
466                // Choose join order based on total cost and cardinality
467                let reordered = if self.should_reorder_join(&left_estimate, &right_estimate) {
468                    Algebra::Join {
469                        left: Box::new(self.apply_cost_based_join_reordering(*right)?),
470                        right: Box::new(self.apply_cost_based_join_reordering(*left)?),
471                    }
472                } else {
473                    Algebra::Join {
474                        left: Box::new(self.apply_cost_based_join_reordering(*left)?),
475                        right: Box::new(self.apply_cost_based_join_reordering(*right)?),
476                    }
477                };
478
479                Ok(reordered)
480            }
481            Algebra::Union { left, right } => Ok(Algebra::Union {
482                left: Box::new(self.apply_cost_based_join_reordering(*left)?),
483                right: Box::new(self.apply_cost_based_join_reordering(*right)?),
484            }),
485            Algebra::Filter { pattern, condition } => Ok(Algebra::Filter {
486                pattern: Box::new(self.apply_cost_based_join_reordering(*pattern)?),
487                condition,
488            }),
489            Algebra::Project { pattern, variables } => Ok(Algebra::Project {
490                pattern: Box::new(self.apply_cost_based_join_reordering(*pattern)?),
491                variables,
492            }),
493            other => Ok(other),
494        }
495    }
496
497    /// Determine if a join should be reordered based on cost estimates
498    fn should_reorder_join(
499        &self,
500        left_estimate: &CostEstimate,
501        right_estimate: &CostEstimate,
502    ) -> bool {
503        // Use multiple criteria for join reordering decision
504
505        // Primary criterion: smaller relation should be build side (left)
506        if left_estimate.cardinality > right_estimate.cardinality * 2 {
507            return true;
508        }
509
510        // Secondary criterion: total cost consideration
511        if left_estimate.total_cost > right_estimate.total_cost * 1.5 {
512            return true;
513        }
514
515        // Tertiary criterion: selectivity (more selective should go first)
516        if left_estimate.selectivity > right_estimate.selectivity * 2.0 {
517            return true;
518        }
519
520        false
521    }
522
523    /// Apply projection pushdown optimization
524    fn apply_projection_pushdown(&self, algebra: Algebra) -> Result<Algebra> {
525        match algebra {
526            Algebra::Project { pattern, variables } => {
527                let optimized_pattern = self.push_projection_down(*pattern, &variables)?;
528                Ok(Algebra::Project {
529                    pattern: Box::new(optimized_pattern),
530                    variables,
531                })
532            }
533            Algebra::Join { left, right } => Ok(Algebra::Join {
534                left: Box::new(self.apply_projection_pushdown(*left)?),
535                right: Box::new(self.apply_projection_pushdown(*right)?),
536            }),
537            other => Ok(other),
538        }
539    }
540
541    /// Push projection down into algebra tree
542    fn push_projection_down(&self, algebra: Algebra, needed_vars: &[Variable]) -> Result<Algebra> {
543        match algebra {
544            Algebra::Join { left, right } => {
545                let left_vars = self.extract_variables(&left);
546                let right_vars = self.extract_variables(&right);
547
548                let left_needed: Vec<Variable> = needed_vars
549                    .iter()
550                    .filter(|v| left_vars.contains(v))
551                    .cloned()
552                    .collect();
553
554                let right_needed: Vec<Variable> = needed_vars
555                    .iter()
556                    .filter(|v| right_vars.contains(v))
557                    .cloned()
558                    .collect();
559
560                let left_projected =
561                    if !left_needed.is_empty() && left_needed.len() < left_vars.len() {
562                        Algebra::Project {
563                            pattern: left,
564                            variables: left_needed,
565                        }
566                    } else {
567                        *left
568                    };
569
570                let right_projected =
571                    if !right_needed.is_empty() && right_needed.len() < right_vars.len() {
572                        Algebra::Project {
573                            pattern: right,
574                            variables: right_needed,
575                        }
576                    } else {
577                        *right
578                    };
579
580                Ok(Algebra::Join {
581                    left: Box::new(left_projected),
582                    right: Box::new(right_projected),
583                })
584            }
585            other => Ok(other),
586        }
587    }
588
589    /// Apply constant folding optimization
590    fn apply_constant_folding(&self, algebra: Algebra) -> Result<Algebra> {
591        match algebra {
592            Algebra::Filter { pattern, condition } => {
593                let folded_condition = self.fold_expression_constants(condition)?;
594
595                // Check if condition is constant true/false
596                if let Some(constant_value) = self.evaluate_constant_expression(&folded_condition) {
597                    if constant_value {
598                        // Filter is always true - remove it
599                        Ok(self.apply_constant_folding(*pattern)?)
600                    } else {
601                        // Filter is always false - return empty result
602                        Ok(Algebra::Bgp(vec![]))
603                    }
604                } else {
605                    Ok(Algebra::Filter {
606                        pattern: Box::new(self.apply_constant_folding(*pattern)?),
607                        condition: folded_condition,
608                    })
609                }
610            }
611            Algebra::Join { left, right } => Ok(Algebra::Join {
612                left: Box::new(self.apply_constant_folding(*left)?),
613                right: Box::new(self.apply_constant_folding(*right)?),
614            }),
615            other => Ok(other),
616        }
617    }
618
619    /// Apply dead code elimination
620    fn apply_dead_code_elimination(&self, algebra: Algebra) -> Result<Algebra> {
621        match algebra {
622            Algebra::Project { pattern, variables } => {
623                let used_vars = self.extract_variables(&pattern);
624                let needed_vars: Vec<Variable> = variables
625                    .into_iter()
626                    .filter(|v| used_vars.contains(v))
627                    .collect();
628
629                if needed_vars.is_empty() {
630                    Ok(Algebra::Bgp(vec![]))
631                } else {
632                    Ok(Algebra::Project {
633                        pattern: Box::new(self.apply_dead_code_elimination(*pattern)?),
634                        variables: needed_vars,
635                    })
636                }
637            }
638            Algebra::Join { left, right } => {
639                let optimized_left = self.apply_dead_code_elimination(*left)?;
640                let optimized_right = self.apply_dead_code_elimination(*right)?;
641
642                match (&optimized_left, &optimized_right) {
643                    (Algebra::Bgp(left_patterns), Algebra::Bgp(right_patterns))
644                        if left_patterns.is_empty() || right_patterns.is_empty() =>
645                    {
646                        Ok(Algebra::Bgp(vec![]))
647                    }
648                    (Algebra::Bgp(patterns), _) if patterns.is_empty() => Ok(Algebra::Bgp(vec![])),
649                    (_, Algebra::Bgp(patterns)) if patterns.is_empty() => Ok(Algebra::Bgp(vec![])),
650                    _ => Ok(Algebra::Join {
651                        left: Box::new(optimized_left),
652                        right: Box::new(optimized_right),
653                    }),
654                }
655            }
656            other => Ok(other),
657        }
658    }
659
660    /// Optimize filter conditions using advanced techniques
661    fn optimize_filter_conditions(&self, condition: &Expression) -> Result<Vec<Expression>> {
662        // Step 1: Factor AND conditions into separate filters
663        let factored_conditions = Self::factor_and_conditions(condition);
664
665        // Step 2: Remove redundant conditions
666        let deduplicated = self.remove_redundant_filters(&factored_conditions);
667
668        // Step 3: Order by estimated selectivity (most selective first)
669        let mut ordered = deduplicated;
670        ordered.sort_by(|a, b| {
671            let selectivity_a = Self::estimate_filter_selectivity(a);
672            let selectivity_b = Self::estimate_filter_selectivity(b);
673            selectivity_a
674                .partial_cmp(&selectivity_b)
675                .unwrap_or(std::cmp::Ordering::Equal)
676        });
677
678        Ok(ordered)
679    }
680
681    /// Factor AND conditions into separate expressions for better pushdown
682    fn factor_and_conditions(expr: &Expression) -> Vec<Expression> {
683        match expr {
684            Expression::Binary { op, left, right } => {
685                if let crate::algebra::BinaryOperator::And = op {
686                    let mut conditions = Vec::new();
687                    conditions.extend(Self::factor_and_conditions(left));
688                    conditions.extend(Self::factor_and_conditions(right));
689                    conditions
690                } else {
691                    vec![expr.clone()]
692                }
693            }
694            _ => vec![expr.clone()],
695        }
696    }
697
698    /// Remove redundant filter conditions
699    fn remove_redundant_filters(&self, conditions: &[Expression]) -> Vec<Expression> {
700        let mut result = Vec::new();
701        let mut seen_hashes = HashSet::new();
702
703        for condition in conditions {
704            let hash = self.hash_expression(condition);
705            if !seen_hashes.contains(&hash) {
706                // Check for logical redundancy
707                if !self.is_logically_redundant(condition, &result) {
708                    result.push(condition.clone());
709                    seen_hashes.insert(hash);
710                }
711            }
712        }
713
714        result
715    }
716
717    /// Estimate selectivity of a filter condition (lower is more selective)
718    fn estimate_filter_selectivity(expr: &Expression) -> f64 {
719        match expr {
720            Expression::Binary { op, left, right } => {
721                match op {
722                    crate::algebra::BinaryOperator::Equal => {
723                        // Equality is highly selective
724                        match (left.as_ref(), right.as_ref()) {
725                            (Expression::Variable(_), Expression::Literal(_))
726                            | (Expression::Literal(_), Expression::Variable(_)) => 0.1, // Very selective
727                            _ => 0.3,
728                        }
729                    }
730                    crate::algebra::BinaryOperator::Less
731                    | crate::algebra::BinaryOperator::LessEqual
732                    | crate::algebra::BinaryOperator::Greater
733                    | crate::algebra::BinaryOperator::GreaterEqual => 0.3, // Range conditions
734                    crate::algebra::BinaryOperator::NotEqual => 0.9, // Usually not very selective
735                    crate::algebra::BinaryOperator::And => {
736                        // Combined selectivity (product for AND)
737                        let left_sel = Self::estimate_filter_selectivity(left);
738                        let right_sel = Self::estimate_filter_selectivity(right);
739                        left_sel * right_sel
740                    }
741                    crate::algebra::BinaryOperator::Or => {
742                        // Combined selectivity for OR (higher selectivity)
743                        let left_sel = Self::estimate_filter_selectivity(left);
744                        let right_sel = Self::estimate_filter_selectivity(right);
745                        left_sel + right_sel - (left_sel * right_sel)
746                    }
747                    _ => 0.5, // Default moderate selectivity
748                }
749            }
750            Expression::Function { name, args: _ } => {
751                match name.as_str() {
752                    "bound" => 0.8, // BOUND function is often not very selective
753                    "isURI" | "isIRI" | "isLiteral" | "isBlank" => 0.4, // Type checks
754                    "regex" => 0.6, // Regular expressions - moderate selectivity
755                    "contains" | "strstarts" | "strends" => 0.5, // String functions
756                    _ => 0.5,       // Default for other functions
757                }
758            }
759            Expression::Unary {
760                op: crate::algebra::UnaryOperator::Not,
761                operand,
762            } => {
763                // Negation typically increases selectivity
764                1.0 - Self::estimate_filter_selectivity(operand)
765            }
766            Expression::Unary { op: _, operand: _ } => 0.5,
767            _ => 0.5, // Default moderate selectivity
768        }
769    }
770
771    /// Check if a condition is logically redundant given existing conditions
772    fn is_logically_redundant(&self, condition: &Expression, existing: &[Expression]) -> bool {
773        // Simple redundancy check - could be enhanced with more sophisticated logic
774        for existing_condition in existing {
775            if Self::expressions_equivalent(condition, existing_condition) {
776                return true;
777            }
778
779            // Check for simple cases like x = 1 AND x = 1
780            if let (
781                Expression::Binary {
782                    op: op1,
783                    left: left1,
784                    right: right1,
785                },
786                Expression::Binary {
787                    op: op2,
788                    left: left2,
789                    right: right2,
790                },
791            ) = (condition, existing_condition)
792            {
793                if op1 == op2
794                    && Self::expressions_equivalent(left1, left2)
795                    && Self::expressions_equivalent(right1, right2)
796                {
797                    return true;
798                }
799            }
800        }
801        false
802    }
803
804    /// Check if two expressions are equivalent
805    fn expressions_equivalent(expr1: &Expression, expr2: &Expression) -> bool {
806        match (expr1, expr2) {
807            (Expression::Variable(v1), Expression::Variable(v2)) => v1 == v2,
808            (Expression::Literal(l1), Expression::Literal(l2)) => l1 == l2,
809            (
810                Expression::Binary {
811                    op: op1,
812                    left: left1,
813                    right: right1,
814                },
815                Expression::Binary {
816                    op: op2,
817                    left: left2,
818                    right: right2,
819                },
820            ) => {
821                op1 == op2
822                    && Self::expressions_equivalent(left1, left2)
823                    && Self::expressions_equivalent(right1, right2)
824            }
825            (
826                Expression::Unary {
827                    op: op1,
828                    operand: operand1,
829                },
830                Expression::Unary {
831                    op: op2,
832                    operand: operand2,
833                },
834            ) => op1 == op2 && Self::expressions_equivalent(operand1, operand2),
835            (
836                Expression::Function {
837                    name: name1,
838                    args: args1,
839                },
840                Expression::Function {
841                    name: name2,
842                    args: args2,
843                },
844            ) => {
845                name1 == name2
846                    && args1.len() == args2.len()
847                    && args1
848                        .iter()
849                        .zip(args2.iter())
850                        .all(|(a1, a2)| Self::expressions_equivalent(a1, a2))
851            }
852            _ => false,
853        }
854    }
855
856    /// Hash an expression for deduplication
857    fn hash_expression(&self, expr: &Expression) -> u64 {
858        let mut hasher = DefaultHasher::new();
859        format!("{expr:?}").hash(&mut hasher);
860        hasher.finish()
861    }
862
863    /// Extract variables from an algebra expression
864    #[allow(clippy::only_used_in_recursion)]
865    fn extract_variables(&self, algebra: &Algebra) -> HashSet<Variable> {
866        let mut vars = HashSet::new();
867        match algebra {
868            Algebra::Bgp(patterns) => {
869                for pattern in patterns {
870                    let TriplePattern {
871                        subject,
872                        predicate,
873                        object,
874                    } = pattern;
875                    if let crate::algebra::Term::Variable(v) = subject {
876                        vars.insert(v.clone());
877                    }
878                    if let crate::algebra::Term::Variable(v) = predicate {
879                        vars.insert(v.clone());
880                    }
881                    if let crate::algebra::Term::Variable(v) = object {
882                        vars.insert(v.clone());
883                    }
884                }
885            }
886            Algebra::Join { left, right } => {
887                vars.extend(self.extract_variables(left));
888                vars.extend(self.extract_variables(right));
889            }
890            Algebra::Union { left, right } => {
891                vars.extend(self.extract_variables(left));
892                vars.extend(self.extract_variables(right));
893            }
894            Algebra::Filter { pattern, .. } => {
895                vars.extend(self.extract_variables(pattern));
896            }
897            Algebra::Project { pattern, variables } => {
898                vars.extend(self.extract_variables(pattern));
899                vars.extend(variables.iter().cloned());
900            }
901            _ => {} // Other algebra types
902        }
903        vars
904    }
905
906    /// Extract variables from an expression
907    #[allow(clippy::only_used_in_recursion)]
908    fn extract_expression_variables(&self, expr: &Expression) -> HashSet<Variable> {
909        let mut vars = HashSet::new();
910        match expr {
911            Expression::Variable(v) => {
912                vars.insert(v.clone());
913            }
914            Expression::Binary { left, right, .. } => {
915                vars.extend(self.extract_expression_variables(left));
916                vars.extend(self.extract_expression_variables(right));
917            }
918            Expression::Unary { operand, .. } => {
919                vars.extend(self.extract_expression_variables(operand));
920            }
921            Expression::Function { args, .. } => {
922                for arg in args {
923                    vars.extend(self.extract_expression_variables(arg));
924                }
925            }
926            _ => {} // Other expression types
927        }
928        vars
929    }
930
931    /// Estimate execution cost for algebra (simple heuristic version)
932    #[allow(clippy::only_used_in_recursion)]
933    fn estimate_simple_cost(&self, algebra: &Algebra) -> f64 {
934        match algebra {
935            Algebra::Bgp(patterns) => {
936                // BGP cost based on pattern count and estimated selectivity
937                patterns.len() as f64 * 10.0
938            }
939            Algebra::Join { left, right } => {
940                let left_cost = self.estimate_simple_cost(left);
941                let right_cost = self.estimate_simple_cost(right);
942                left_cost * right_cost * 0.1 // Join selectivity factor
943            }
944            Algebra::Union { left, right } => {
945                self.estimate_simple_cost(left) + self.estimate_simple_cost(right)
946            }
947            Algebra::Filter { pattern, .. } => {
948                self.estimate_simple_cost(pattern) * 0.5 // Filter selectivity
949            }
950            _ => 1.0,
951        }
952    }
953
954    /// Fold constants in expressions
955    #[allow(clippy::only_used_in_recursion)]
956    fn fold_expression_constants(&self, expr: Expression) -> Result<Expression> {
957        match expr {
958            Expression::Binary { op, left, right } => {
959                let folded_left = self.fold_expression_constants(*left)?;
960                let folded_right = self.fold_expression_constants(*right)?;
961                Ok(Expression::Binary {
962                    op,
963                    left: Box::new(folded_left),
964                    right: Box::new(folded_right),
965                })
966            }
967            Expression::Unary { op, operand } => {
968                let folded_operand = self.fold_expression_constants(*operand)?;
969                Ok(Expression::Unary {
970                    op,
971                    operand: Box::new(folded_operand),
972                })
973            }
974            other => Ok(other),
975        }
976    }
977
978    /// Evaluate constant expressions to boolean values
979    fn evaluate_constant_expression(&self, expr: &Expression) -> Option<bool> {
980        match expr {
981            Expression::Literal(literal) => {
982                // Simple boolean literal evaluation
983                if literal.value == "true" {
984                    Some(true)
985                } else if literal.value == "false" {
986                    Some(false)
987                } else {
988                    None
989                }
990            }
991            _ => None,
992        }
993    }
994
995    /// Check if two algebra expressions are equal (simplified check)
996    fn algebra_equal(&self, a: &Algebra, b: &Algebra) -> bool {
997        // Simplified equality check - should be improved
998        format!("{a:?}") == format!("{b:?}")
999    }
1000
1001    /// Hash an algebra expression for caching
1002    pub fn hash_algebra(&self, algebra: &Algebra) -> u64 {
1003        let mut hasher = DefaultHasher::new();
1004        format!("{algebra:?}").hash(&mut hasher);
1005        hasher.finish()
1006    }
1007
1008    /// Estimate query complexity for adaptive optimization
1009    fn estimate_query_complexity(&self, algebra: &Algebra) -> QueryComplexity {
1010        let mut complexity = QueryComplexity::default();
1011        self.analyze_complexity(algebra, &mut complexity);
1012        complexity
1013    }
1014
1015    /// Recursively analyze query complexity
1016    fn analyze_complexity(&self, algebra: &Algebra, complexity: &mut QueryComplexity) {
1017        match algebra {
1018            Algebra::Bgp(patterns) => {
1019                complexity.triple_patterns += patterns.len();
1020            }
1021            Algebra::Join { left, right } | Algebra::Union { left, right } => {
1022                complexity.joins += 1;
1023                self.analyze_complexity(left, complexity);
1024                self.analyze_complexity(right, complexity);
1025            }
1026            Algebra::Filter { pattern, .. } => {
1027                complexity.filters += 1;
1028                self.analyze_complexity(pattern, complexity);
1029            }
1030            Algebra::Extend { pattern, .. } => {
1031                self.analyze_complexity(pattern, complexity);
1032            }
1033            Algebra::Project { pattern, .. } => {
1034                self.analyze_complexity(pattern, complexity);
1035            }
1036            Algebra::Distinct { pattern } | Algebra::Reduced { pattern } => {
1037                self.analyze_complexity(pattern, complexity);
1038            }
1039            Algebra::OrderBy { pattern, .. } => {
1040                complexity.ordering = true;
1041                self.analyze_complexity(pattern, complexity);
1042            }
1043            Algebra::Slice { pattern, .. } => {
1044                self.analyze_complexity(pattern, complexity);
1045            }
1046            Algebra::Group { pattern, .. } => {
1047                complexity.grouping = true;
1048                self.analyze_complexity(pattern, complexity);
1049            }
1050            _ => {}
1051        }
1052    }
1053}
1054
1055impl Default for Optimizer {
1056    fn default() -> Self {
1057        Self::new(OptimizerConfig::default())
1058    }
1059}
1060
1061/// Type alias for backwards compatibility
1062pub type QueryOptimizer = Optimizer;
1063
1064#[cfg(test)]
1065mod federated_integration_tests {
1066    //! W2-S4 deepening: integration of [`FederatedPlanner`] with the main
1067    //! [`Optimizer::optimize`] entry point.
1068    //!
1069    //! These tests assert two invariants:
1070    //!
1071    //! 1. With **no** [`SourceSelectivityProvider`] registered, optimization
1072    //!    behaviour is byte-for-byte identical to the pre-W2-S4 baseline —
1073    //!    no `Algebra::Service` nodes are introduced.
1074    //! 2. With a provider registered, BGPs whose IRIs map to known endpoints
1075    //!    are transparently rewritten to `Algebra::Service` nodes with the
1076    //!    correct endpoint URL.
1077
1078    use super::*;
1079    use crate::algebra::{Term, TriplePattern, Variable};
1080    use crate::optimizer::federated_plan::{FederatedSelectivity, StaticSourceProvider};
1081    use oxirs_core::model::NamedNode;
1082
1083    fn iri_term(s: &str) -> Term {
1084        Term::Iri(NamedNode::new_unchecked(s))
1085    }
1086
1087    fn var_term(name: &str) -> Term {
1088        Term::Variable(Variable::new(name).expect("valid variable name"))
1089    }
1090
1091    fn triple(s: Term, p: Term, o: Term) -> TriplePattern {
1092        TriplePattern {
1093            subject: s,
1094            predicate: p,
1095            object: o,
1096        }
1097    }
1098
1099    fn dbpedia_provider() -> StaticSourceProvider {
1100        let mut provider = StaticSourceProvider::new();
1101        provider.register(
1102            "http://dbpedia.org/",
1103            "https://dbpedia.org/sparql",
1104            FederatedSelectivity {
1105                estimated_cardinality: 100.0,
1106                estimated_latency_ms: 80.0,
1107                confidence: 0.9,
1108            },
1109        );
1110        provider
1111    }
1112
1113    fn assert_no_service_nodes(algebra: &Algebra) {
1114        match algebra {
1115            Algebra::Service { .. } => panic!("unexpected Service node: {algebra:?}"),
1116            Algebra::Bgp(_) | Algebra::Table | Algebra::Empty | Algebra::Zero => {}
1117            Algebra::Join { left, right }
1118            | Algebra::Union { left, right }
1119            | Algebra::Minus { left, right } => {
1120                assert_no_service_nodes(left);
1121                assert_no_service_nodes(right);
1122            }
1123            Algebra::LeftJoin { left, right, .. } => {
1124                assert_no_service_nodes(left);
1125                assert_no_service_nodes(right);
1126            }
1127            Algebra::Filter { pattern, .. }
1128            | Algebra::Distinct { pattern }
1129            | Algebra::Reduced { pattern }
1130            | Algebra::Slice { pattern, .. }
1131            | Algebra::OrderBy { pattern, .. }
1132            | Algebra::Project { pattern, .. }
1133            | Algebra::Extend { pattern, .. }
1134            | Algebra::Group { pattern, .. }
1135            | Algebra::Having { pattern, .. }
1136            | Algebra::Graph { pattern, .. } => {
1137                assert_no_service_nodes(pattern);
1138            }
1139            // Other algebra variants (Values, PropertyPath, …) cannot contain
1140            // nested algebra so they're trivially Service-free.
1141            _ => {}
1142        }
1143    }
1144
1145    fn contains_service_to(algebra: &Algebra, expected_endpoint: &str) -> bool {
1146        match algebra {
1147            Algebra::Service {
1148                endpoint: Term::Iri(node),
1149                ..
1150            } => node.as_str() == expected_endpoint,
1151            Algebra::Service { .. } => false,
1152            Algebra::Join { left, right }
1153            | Algebra::Union { left, right }
1154            | Algebra::Minus { left, right } => {
1155                contains_service_to(left, expected_endpoint)
1156                    || contains_service_to(right, expected_endpoint)
1157            }
1158            Algebra::LeftJoin { left, right, .. } => {
1159                contains_service_to(left, expected_endpoint)
1160                    || contains_service_to(right, expected_endpoint)
1161            }
1162            Algebra::Filter { pattern, .. }
1163            | Algebra::Distinct { pattern }
1164            | Algebra::Reduced { pattern }
1165            | Algebra::Slice { pattern, .. }
1166            | Algebra::OrderBy { pattern, .. }
1167            | Algebra::Project { pattern, .. }
1168            | Algebra::Extend { pattern, .. }
1169            | Algebra::Group { pattern, .. }
1170            | Algebra::Having { pattern, .. }
1171            | Algebra::Graph { pattern, .. } => contains_service_to(pattern, expected_endpoint),
1172            _ => false,
1173        }
1174    }
1175
1176    #[test]
1177    fn optimizer_without_provider_skips_federation() {
1178        let mut optimizer = Optimizer::new(OptimizerConfig::default());
1179        assert!(!optimizer.has_federated_planner());
1180
1181        let alg = Algebra::Bgp(vec![triple(
1182            var_term("s"),
1183            iri_term("http://dbpedia.org/property/birthDate"),
1184            var_term("o"),
1185        )]);
1186
1187        let optimized = optimizer
1188            .optimize(alg)
1189            .expect("baseline optimize must succeed");
1190        assert_no_service_nodes(&optimized);
1191        assert!(optimizer.last_federated_outcome().is_none());
1192    }
1193
1194    #[test]
1195    fn optimizer_with_provider_emits_service_node() {
1196        let provider: Arc<dyn SourceSelectivityProvider> = Arc::new(dbpedia_provider());
1197        let mut optimizer =
1198            Optimizer::new(OptimizerConfig::default()).with_federated_planner(provider);
1199        assert!(optimizer.has_federated_planner());
1200
1201        let alg = Algebra::Bgp(vec![triple(
1202            var_term("s"),
1203            iri_term("http://dbpedia.org/property/birthDate"),
1204            var_term("o"),
1205        )]);
1206
1207        let optimized = optimizer
1208            .optimize(alg)
1209            .expect("federated optimize must succeed");
1210        assert!(
1211            contains_service_to(&optimized, "https://dbpedia.org/sparql"),
1212            "expected Service node targeting dbpedia, got {optimized:?}"
1213        );
1214
1215        let outcome = optimizer
1216            .last_federated_outcome()
1217            .expect("outcome should be recorded");
1218        assert!(outcome.touched_federation());
1219        assert_eq!(outcome.patterns_federated, 1);
1220        assert!(outcome
1221            .endpoints_used
1222            .contains_key("https://dbpedia.org/sparql"));
1223    }
1224
1225    #[test]
1226    fn optimizer_with_provider_keeps_local_only_query_unchanged() {
1227        let provider: Arc<dyn SourceSelectivityProvider> = Arc::new(dbpedia_provider());
1228        let mut optimizer =
1229            Optimizer::new(OptimizerConfig::default()).with_federated_planner(provider);
1230
1231        let alg = Algebra::Bgp(vec![triple(
1232            iri_term("http://example.org/local/alice"),
1233            iri_term("http://example.org/local/knows"),
1234            var_term("friend"),
1235        )]);
1236
1237        let optimized = optimizer
1238            .optimize(alg)
1239            .expect("optimize must succeed even with provider");
1240        assert_no_service_nodes(&optimized);
1241
1242        let outcome = optimizer
1243            .last_federated_outcome()
1244            .expect("outcome should be recorded even when nothing federates");
1245        assert!(!outcome.touched_federation());
1246    }
1247
1248    #[test]
1249    fn optimizer_emits_join_for_mixed_local_and_federated_query() {
1250        let provider: Arc<dyn SourceSelectivityProvider> = Arc::new(dbpedia_provider());
1251        let mut optimizer =
1252            Optimizer::new(OptimizerConfig::default()).with_federated_planner(provider);
1253
1254        let alg = Algebra::Bgp(vec![
1255            triple(
1256                var_term("s"),
1257                iri_term("http://example.org/local/labelOf"),
1258                var_term("label"),
1259            ),
1260            triple(
1261                var_term("s"),
1262                iri_term("http://dbpedia.org/property/birthDate"),
1263                var_term("date"),
1264            ),
1265        ]);
1266
1267        let optimized = optimizer
1268            .optimize(alg)
1269            .expect("optimize must succeed for mixed BGP");
1270        assert!(
1271            contains_service_to(&optimized, "https://dbpedia.org/sparql"),
1272            "expected Service node, got {optimized:?}"
1273        );
1274
1275        let outcome = optimizer
1276            .last_federated_outcome()
1277            .expect("outcome should be recorded");
1278        assert_eq!(outcome.patterns_federated, 1);
1279    }
1280
1281    #[test]
1282    fn with_federated_latency_weight_propagates_to_planner() {
1283        // Smoke test: setting the latency weight should not panic and should
1284        // not change optimize() success on a query with no federated patterns.
1285        let provider: Arc<dyn SourceSelectivityProvider> = Arc::new(dbpedia_provider());
1286        let mut optimizer = Optimizer::new(OptimizerConfig::default())
1287            .with_federated_planner(provider)
1288            .with_federated_latency_weight(2.5);
1289
1290        let alg = Algebra::Bgp(vec![triple(
1291            iri_term("http://example.org/local/x"),
1292            iri_term("http://example.org/local/p"),
1293            var_term("o"),
1294        )]);
1295
1296        let optimized = optimizer
1297            .optimize(alg)
1298            .expect("latency-weighted optimize must succeed");
1299        assert_no_service_nodes(&optimized);
1300    }
1301
1302    #[test]
1303    fn federation_pass_runs_after_filter_pushdown() {
1304        // FILTER on a federated BGP — the filter should sit on top of the
1305        // emitted Service node so the executor evaluates it on the joined
1306        // (local + remote) solutions, preserving SPARQL 1.1 SERVICE semantics.
1307        let provider: Arc<dyn SourceSelectivityProvider> = Arc::new(dbpedia_provider());
1308        let mut optimizer =
1309            Optimizer::new(OptimizerConfig::default()).with_federated_planner(provider);
1310
1311        let alg = Algebra::Filter {
1312            pattern: Box::new(Algebra::Bgp(vec![triple(
1313                var_term("s"),
1314                iri_term("http://dbpedia.org/property/birthDate"),
1315                var_term("o"),
1316            )])),
1317            condition: Expression::Variable(Variable::new("o").expect("valid var")),
1318        };
1319
1320        let optimized = optimizer
1321            .optimize(alg)
1322            .expect("filter+federate optimize must succeed");
1323        assert!(
1324            contains_service_to(&optimized, "https://dbpedia.org/sparql"),
1325            "expected Service node under filter, got {optimized:?}"
1326        );
1327    }
1328
1329    #[test]
1330    fn last_federated_outcome_resets_when_provider_unset_after_run() {
1331        // Sanity: building a fresh optimizer without a provider always
1332        // produces None for last_federated_outcome.
1333        let mut optimizer = Optimizer::new(OptimizerConfig::default());
1334        let alg = Algebra::Bgp(vec![triple(
1335            var_term("s"),
1336            iri_term("http://example.org/local/p"),
1337            var_term("o"),
1338        )]);
1339        let _ = optimizer.optimize(alg).expect("optimize must succeed");
1340        assert!(optimizer.last_federated_outcome().is_none());
1341    }
1342}