oxirs_arq/

integrated_query_planner.rs

//! Integrated Query Planner
//!
//! This module provides unified integration of all optimization components:
//! - Index-aware BGP optimization
//! - Statistics-based cost estimation
//! - Streaming optimization
//! - Machine learning-enhanced planning
//! - Adaptive query execution

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use anyhow::Result;
use tracing::{debug, info, span, Level};

use crate::advanced_optimizer::{AdvancedOptimizer, AdvancedOptimizerConfig};
use crate::algebra::{Algebra, Expression, Term, TriplePattern, Variable};
// use crate::bgp_optimizer::BGPOptimizer;
use crate::bgp_optimizer_types::{
    IndexAssignment, IndexUsagePlan, OptimizedBGP, PatternSelectivity, SelectivityFactors,
    SelectivityInfo,
};
use crate::cost_model::{CostEstimate, CostModel};
use crate::optimizer::{IndexStatistics, IndexType, Statistics};
use crate::statistics_collector::StatisticsCollector;
use crate::streaming::{StreamingConfig, StreamingExecutor};

/// Integrated query planner combining all optimization techniques
pub struct IntegratedQueryPlanner {
    config: IntegratedPlannerConfig,
    cost_model: Arc<Mutex<CostModel>>,
    #[allow(dead_code)]
    statistics_collector: Arc<Mutex<StatisticsCollector>>,
    #[allow(dead_code)]
    statistics: Statistics,
    #[allow(dead_code)]
    index_stats: IndexStatistics,
    #[allow(dead_code)]
    advanced_optimizer: AdvancedOptimizer,
    #[allow(dead_code)]
    streaming_executor: Option<StreamingExecutor>,
    plan_cache: Arc<Mutex<PlanCache>>,
    execution_history: Arc<Mutex<ExecutionHistory>>,
    adaptive_thresholds: AdaptiveThresholds,
}

/// Configuration for integrated query planning
#[derive(Debug, Clone)]
pub struct IntegratedPlannerConfig {
    /// Enable adaptive optimization based on execution feedback
    pub adaptive_optimization: bool,
    /// Enable cross-query optimization
    pub cross_query_optimization: bool,
    /// Memory threshold for switching to streaming (bytes)
    pub streaming_threshold: usize,
    /// Enable machine learning-enhanced cost estimation
    pub ml_cost_estimation: bool,
    /// Plan cache size
    pub plan_cache_size: usize,
    /// Enable parallel plan exploration
    pub parallel_planning: bool,
    /// Statistics collection interval
    pub stats_collection_interval: Duration,
    /// Enable advanced index recommendations
    pub advanced_index_recommendations: bool,
}

impl Default for IntegratedPlannerConfig {
    fn default() -> Self {
        Self {
            adaptive_optimization: true,
            cross_query_optimization: true,
            streaming_threshold: 512 * 1024 * 1024, // 512MB
            ml_cost_estimation: true,
            plan_cache_size: 1000,
            parallel_planning: true,
            stats_collection_interval: Duration::from_secs(60),
            advanced_index_recommendations: true,
        }
    }
}

/// Comprehensive execution plan
#[derive(Debug, Clone)]
pub struct IntegratedExecutionPlan {
    /// Optimized algebra expression
    pub optimized_algebra: Algebra,
    /// Estimated execution cost
    pub estimated_cost: CostEstimate,
    /// Index usage plan
    pub index_plan: IndexUsagePlan,
    /// Whether to use streaming execution
    pub use_streaming: bool,
    /// Recommended memory allocation
    pub memory_allocation: usize,
    /// Expected execution time
    pub expected_duration: Duration,
    /// Confidence in the plan (0.0 to 1.0)
    pub confidence: f64,
    /// Adaptive hints for execution
    pub adaptive_hints: AdaptiveHints,
    /// Alternative plans for fallback
    pub alternative_plans: Vec<AlternativePlan>,
}

/// Adaptive hints for execution tuning
#[derive(Debug, Clone, Default)]
pub struct AdaptiveHints {
    /// Suggested batch size for operations
    pub batch_size: Option<usize>,
    /// Suggested parallelism level
    pub parallelism_level: Option<usize>,
    /// Memory allocation suggestions
    pub memory_hints: MemoryHints,
    /// Index access patterns
    pub index_access_patterns: Vec<IndexAccessPattern>,
    /// Join algorithm recommendations
    pub join_algorithms: Vec<JoinAlgorithmHint>,
}

/// Memory allocation hints
#[derive(Debug, Clone, Default)]
pub struct MemoryHints {
    /// Minimum memory requirement
    pub min_memory: usize,
    /// Optimal memory allocation
    pub optimal_memory: usize,
    /// Maximum beneficial memory
    pub max_memory: usize,
    /// Memory allocation strategy
    pub allocation_strategy: MemoryStrategy,
}

/// Memory allocation strategies
#[derive(Debug, Clone, Default)]
pub enum MemoryStrategy {
    Conservative,
    #[default]
    Balanced,
    Aggressive,
    Adaptive,
}

/// Index access pattern hint
#[derive(Debug, Clone)]
pub struct IndexAccessPattern {
    pub index_type: IndexType,
    pub access_pattern: AccessPattern,
    pub expected_selectivity: f64,
    pub prefetch_hint: bool,
}

/// Access patterns for index optimization
#[derive(Debug, Clone)]
pub enum AccessPattern {
    Sequential,
    Random,
    Clustered,
    Sparse,
    Range,
}

/// Join algorithm hint
#[derive(Debug, Clone)]
pub struct JoinAlgorithmHint {
    pub left_pattern_idx: usize,
    pub right_pattern_idx: usize,
    pub recommended_algorithm: JoinAlgorithm,
    pub estimated_cost: f64,
    pub memory_requirement: usize,
}

/// Join algorithm types
#[derive(Debug, Clone)]
pub enum JoinAlgorithm {
    HashJoin,
    SortMergeJoin,
    NestedLoopJoin,
    IndexNestedLoopJoin,
    StreamingHashJoin,
    SymmetricHashJoin,
}

/// Alternative execution plan
#[derive(Debug, Clone)]
pub struct AlternativePlan {
    pub plan: IntegratedExecutionPlan,
    pub trigger_conditions: Vec<TriggerCondition>,
    pub fallback_priority: usize,
}

/// Conditions for switching to alternative plans
#[derive(Debug, Clone)]
pub enum TriggerCondition {
    MemoryPressure(f64),
    ExecutionTimeExceeded(Duration),
    CardinalityMismatch(f64),
    IndexUnavailable(IndexType),
    ConcurrencyLimit,
}

/// Plan cache for optimization reuse
#[derive(Debug)]
pub struct PlanCache {
    plans: HashMap<u64, CachedPlan>,
    access_counts: HashMap<u64, usize>,
    last_access: HashMap<u64, Instant>,
    max_size: usize,
}

/// Cached execution plan with metadata
#[derive(Debug, Clone)]
pub struct CachedPlan {
    pub plan: IntegratedExecutionPlan,
    pub creation_time: Instant,
    pub access_count: usize,
    pub average_accuracy: f64,
    pub invalidation_triggers: Vec<InvalidationTrigger>,
}

/// Triggers for plan cache invalidation
#[derive(Debug, Clone)]
pub enum InvalidationTrigger {
    StatisticsUpdate,
    IndexChange,
    DataSizeChange(f64),
    TimeElapsed(Duration),
}

/// Execution history for adaptive learning
#[derive(Debug)]
pub struct ExecutionHistory {
    executions: VecDeque<ExecutionRecord>,
    #[allow(dead_code)]
    pattern_performance: HashMap<String, PatternPerformance>,
    max_history_size: usize,
}

/// Record of query execution
#[derive(Debug, Clone)]
pub struct ExecutionRecord {
    pub query_hash: u64,
    pub plan_hash: u64,
    pub actual_duration: Duration,
    pub estimated_duration: Duration,
    pub actual_cardinality: usize,
    pub estimated_cardinality: usize,
    pub memory_used: usize,
    pub index_hits: HashMap<IndexType, usize>,
    pub execution_timestamp: Instant,
    pub success: bool,
    pub error_info: Option<String>,
}

/// Performance metrics for query patterns
#[derive(Debug, Clone, Default)]
pub struct PatternPerformance {
    pub total_executions: usize,
    pub successful_executions: usize,
    pub average_accuracy: f64,
    pub average_duration: Duration,
    pub best_plan_hash: Option<u64>,
    pub worst_plan_hash: Option<u64>,
}

/// Adaptive thresholds that adjust based on system performance
#[derive(Debug, Clone)]
pub struct AdaptiveThresholds {
    pub streaming_memory_threshold: usize,
    pub parallel_execution_threshold: f64,
    pub index_recommendation_threshold: f64,
    pub plan_cache_accuracy_threshold: f64,
    pub statistics_staleness_threshold: Duration,
}

impl Default for AdaptiveThresholds {
    fn default() -> Self {
        Self {
            streaming_memory_threshold: 512 * 1024 * 1024, // 512MB
            parallel_execution_threshold: 100.0,           // Cost units
            index_recommendation_threshold: 0.1,           // 10% improvement
            plan_cache_accuracy_threshold: 0.8,            // 80% accuracy
            statistics_staleness_threshold: Duration::from_secs(3600), // 1 hour
        }
    }
}

impl IntegratedQueryPlanner {
    /// Create a new integrated query planner
    pub fn new(config: IntegratedPlannerConfig) -> Result<Self> {
        let cost_config = crate::cost_model::CostModelConfig::default();
        let cost_model = Arc::new(Mutex::new(CostModel::new(cost_config)));
        let statistics_collector = Arc::new(Mutex::new(StatisticsCollector::new()));
        let statistics = Statistics::new();
        let index_stats = IndexStatistics::default();

        let advanced_config = AdvancedOptimizerConfig {
            enable_ml_optimization: config.ml_cost_estimation,
            cross_query_optimization: config.cross_query_optimization,
            parallel_optimization: config.parallel_planning,
            ..Default::default()
        };

        // Create a separate StatisticsCollector for the AdvancedOptimizer to avoid type conflicts
        let advanced_optimizer_stats = Arc::new(StatisticsCollector::new());
        let advanced_optimizer = AdvancedOptimizer::new(
            advanced_config,
            cost_model.clone(),
            advanced_optimizer_stats,
        );

        let streaming_executor = if config.streaming_threshold > 0 {
            let streaming_config = StreamingConfig {
                max_memory_usage: config.streaming_threshold,
                ..Default::default()
            };
            Some(StreamingExecutor::new(streaming_config)?)
        } else {
            None
        };

        let plan_cache = Arc::new(Mutex::new(PlanCache::new(config.plan_cache_size)));
        let execution_history = Arc::new(Mutex::new(ExecutionHistory::new(10000)));

        Ok(Self {
            config,
            cost_model,
            statistics_collector,
            statistics,
            index_stats,
            advanced_optimizer,
            streaming_executor,
            plan_cache,
            execution_history,
            adaptive_thresholds: AdaptiveThresholds::default(),
        })
    }

    /// Create an integrated execution plan for a query
    pub fn create_plan(&mut self, algebra: &Algebra) -> Result<IntegratedExecutionPlan> {
        let _span = span!(Level::INFO, "integrated_planning").entered();
        let start_time = Instant::now();

        // Check plan cache first
        let query_hash = self.compute_algebra_hash(algebra);
        if let Some(cached_plan) = self.get_cached_plan(query_hash) {
            debug!("Using cached execution plan");
            return Ok(cached_plan.plan);
        }

        info!("Creating new integrated execution plan");

        // Step 1: Analyze query complexity and characteristics
        let query_analysis = self.analyze_query(algebra)?;

        // Step 2: Collect and update statistics
        self.update_statistics(&query_analysis)?;

        // Step 3: Optimize BGP patterns with index awareness
        let optimized_bgp = self.optimize_bgp_patterns(algebra)?;

        // Step 4: Apply advanced optimizations
        let advanced_optimized = algebra.clone(); // Use algebra directly for now

        // Step 5: Determine execution strategy (streaming vs. in-memory)
        let execution_strategy =
            self.determine_execution_strategy(&advanced_optimized, &query_analysis)?;

        // Step 6: Generate cost estimates
        let cost_estimate =
            self.estimate_execution_cost(&advanced_optimized, &execution_strategy)?;

        // Step 7: Create adaptive hints
        let adaptive_hints = self.generate_adaptive_hints(&advanced_optimized, &cost_estimate)?;

        // Step 8: Generate alternative plans
        let alternative_plans =
            self.generate_alternative_plans(&advanced_optimized, &cost_estimate)?;

        let plan = IntegratedExecutionPlan {
            optimized_algebra: advanced_optimized,
            estimated_cost: cost_estimate.clone(),
            index_plan: optimized_bgp.index_plan,
            use_streaming: execution_strategy.use_streaming,
            memory_allocation: execution_strategy.memory_allocation,
            expected_duration: Duration::from_millis((cost_estimate.total_cost * 10.0) as u64),
            confidence: self.calculate_plan_confidence(&cost_estimate)?,
            adaptive_hints,
            alternative_plans,
        };

        // Cache the plan
        self.cache_plan(query_hash, plan.clone())?;

        let planning_time = start_time.elapsed();
        info!(
            "Plan created in {:?} with confidence {:.2}",
            planning_time, plan.confidence
        );

        Ok(plan)
    }

    /// Update execution statistics based on actual performance
    pub fn update_execution_feedback(
        &mut self,
        plan_hash: u64,
        actual_duration: Duration,
        actual_cardinality: usize,
        memory_used: usize,
        success: bool,
        error_info: Option<String>,
    ) -> Result<()> {
        let _span = span!(Level::DEBUG, "execution_feedback").entered();

        let execution_record = ExecutionRecord {
            query_hash: 0, // Would need to be provided
            plan_hash,
            actual_duration,
            estimated_duration: Duration::from_secs(0), // Would need to be retrieved from plan
            actual_cardinality,
            estimated_cardinality: 0, // Would need to be retrieved from plan
            memory_used,
            index_hits: HashMap::new(),
            execution_timestamp: Instant::now(),
            success,
            error_info,
        };

        // Update execution history
        {
            let mut history = self.execution_history.lock().expect("lock poisoned");
            history.add_execution(execution_record.clone());
        }

        // Update adaptive thresholds based on performance
        self.update_adaptive_thresholds(&execution_record)?;

        // Update cost model with actual vs. estimated performance
        self.update_cost_model(&execution_record)?;

        debug!("Updated execution feedback for plan {}", plan_hash);
        Ok(())
    }

    /// Get recommendations for index creation
    pub fn get_index_recommendations(&self) -> Result<Vec<IndexRecommendation>> {
        let _span = span!(Level::INFO, "index_recommendations").entered();

        let history = self.execution_history.lock().expect("lock poisoned");
        let recommendations = self.analyze_index_opportunities(&history)?;

        info!("Generated {} index recommendations", recommendations.len());
        Ok(recommendations)
    }

    /// Analyze query characteristics for optimization
    fn analyze_query(&self, algebra: &Algebra) -> Result<QueryAnalysis> {
        let mut analysis = QueryAnalysis::default();

        self.analyze_algebra_recursive(algebra, &mut analysis)?;

        // Calculate complexity score
        analysis.complexity_score = self.calculate_complexity_score(&analysis);

        // Estimate memory requirements
        analysis.estimated_memory = self.estimate_memory_requirements(&analysis)?;

        Ok(analysis)
    }

    /// Recursively analyze algebra expression
    fn analyze_algebra_recursive(
        &self,
        algebra: &Algebra,
        analysis: &mut QueryAnalysis,
    ) -> Result<()> {
        match algebra {
            Algebra::Bgp(patterns) => {
                analysis.triple_pattern_count += patterns.len();
                for pattern in patterns {
                    analysis
                        .variables
                        .extend(self.extract_pattern_variables(pattern));
                }
            }
            Algebra::Join { left, right } => {
                analysis.join_count += 1;
                self.analyze_algebra_recursive(left, analysis)?;
                self.analyze_algebra_recursive(right, analysis)?;
            }
            Algebra::Union { left, right } => {
                analysis.union_count += 1;
                self.analyze_algebra_recursive(left, analysis)?;
                self.analyze_algebra_recursive(right, analysis)?;
            }
            Algebra::Filter { pattern, condition } => {
                analysis.filter_count += 1;
                analysis.has_complex_filters = self.is_complex_filter(condition);
                self.analyze_algebra_recursive(pattern, analysis)?;
            }
            Algebra::Group { pattern, .. } => {
                analysis.has_aggregation = true;
                self.analyze_algebra_recursive(pattern, analysis)?;
            }
            Algebra::OrderBy { pattern, .. } => {
                analysis.has_sorting = true;
                self.analyze_algebra_recursive(pattern, analysis)?;
            }
            _ => {
                // Handle other algebra types
            }
        }
        Ok(())
    }

    /// Calculate complexity score for query
    fn calculate_complexity_score(&self, analysis: &QueryAnalysis) -> f64 {
        let mut score = 0.0;

        score += analysis.triple_pattern_count as f64 * 1.0;
        score += analysis.join_count as f64 * 5.0;
        score += analysis.union_count as f64 * 3.0;
        score += analysis.filter_count as f64 * 2.0;

        if analysis.has_aggregation {
            score += 10.0;
        }
        if analysis.has_sorting {
            score += 8.0;
        }
        if analysis.has_complex_filters {
            score += 5.0;
        }

        score
    }

    /// Extract variables from a triple pattern
    fn extract_pattern_variables(&self, pattern: &TriplePattern) -> HashSet<Variable> {
        let mut variables = HashSet::new();

        if let Term::Variable(var) = &pattern.subject {
            variables.insert(var.clone());
        }
        if let Term::Variable(var) = &pattern.predicate {
            variables.insert(var.clone());
        }
        if let Term::Variable(var) = &pattern.object {
            variables.insert(var.clone());
        }

        variables
    }

    /// Check if filter expression is complex
    #[allow(clippy::only_used_in_recursion)]
    fn is_complex_filter(&self, expression: &Expression) -> bool {
        // Simplified complexity check
        match expression {
            Expression::Function { .. } => true,
            Expression::Exists(_) | Expression::NotExists(_) => true,
            Expression::Binary { left, right, .. } => {
                self.is_complex_filter(left) || self.is_complex_filter(right)
            }
            _ => false,
        }
    }

    /// Generate adaptive hints for execution
    fn generate_adaptive_hints(
        &self,
        _algebra: &Algebra,
        cost_estimate: &CostEstimate,
    ) -> Result<AdaptiveHints> {
        let mut hints = AdaptiveHints::default();

        // Calculate optimal batch size based on memory and cardinality
        if cost_estimate.cardinality > 10000 {
            hints.batch_size = Some((cost_estimate.cardinality / 100).max(1000));
        }

        // Determine parallelism level
        if cost_estimate.total_cost > 100.0 {
            hints.parallelism_level = Some(
                std::thread::available_parallelism()
                    .map(|n| n.get())
                    .unwrap_or(1)
                    .min(4),
            );
        }

        // Memory allocation hints
        hints.memory_hints = self.calculate_memory_hints(cost_estimate)?;

        Ok(hints)
    }

    /// Calculate memory allocation hints
    fn calculate_memory_hints(&self, cost_estimate: &CostEstimate) -> Result<MemoryHints> {
        let base_memory = 64 * 1024 * 1024; // 64MB base
        let cardinality_memory = cost_estimate.cardinality * 100; // ~100 bytes per result

        Ok(MemoryHints {
            min_memory: base_memory,
            optimal_memory: base_memory + cardinality_memory,
            max_memory: (base_memory + cardinality_memory) * 2,
            allocation_strategy: MemoryStrategy::Balanced,
        })
    }

    /// Compute hash for algebra expression
    fn compute_algebra_hash(&self, algebra: &Algebra) -> u64 {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};

        let mut hasher = DefaultHasher::new();
        format!("{algebra:?}").hash(&mut hasher);
        hasher.finish()
    }

    /// Get cached plan if available and valid
    fn get_cached_plan(&self, query_hash: u64) -> Option<CachedPlan> {
        let cache = self.plan_cache.lock().expect("lock poisoned");
        cache.get_plan(query_hash)
    }

    /// Cache execution plan
    fn cache_plan(&self, query_hash: u64, plan: IntegratedExecutionPlan) -> Result<()> {
        let mut cache = self.plan_cache.lock().expect("lock poisoned");
        cache.insert_plan(query_hash, plan);
        Ok(())
    }

    /// Calculate confidence in execution plan
    fn calculate_plan_confidence(&self, _cost_estimate: &CostEstimate) -> Result<f64> {
        // Base confidence on cost model accuracy and statistics quality
        let base_confidence = 0.7;
        let stats_factor = 0.2; // Would be calculated from statistics quality
        let history_factor = 0.1; // Would be calculated from execution history

        Ok(base_confidence + stats_factor + history_factor)
    }

    // Additional implementation methods would continue here...
    // For brevity, I'm including the most important components
}

/// Query analysis results
#[derive(Debug, Default)]
pub struct QueryAnalysis {
    pub triple_pattern_count: usize,
    pub join_count: usize,
    pub union_count: usize,
    pub filter_count: usize,
    pub variables: HashSet<Variable>,
    pub has_aggregation: bool,
    pub has_sorting: bool,
    pub has_complex_filters: bool,
    pub complexity_score: f64,
    pub estimated_memory: usize,
}

/// Execution strategy determination
#[derive(Debug)]
pub struct ExecutionStrategy {
    pub use_streaming: bool,
    pub memory_allocation: usize,
    pub parallel_execution: bool,
    pub index_recommendations: Vec<IndexType>,
}

/// Index recommendation
#[derive(Debug, Clone)]
pub struct IndexRecommendation {
    pub index_type: IndexType,
    pub estimated_benefit: f64,
    pub creation_cost: f64,
    pub maintenance_cost: f64,
    pub confidence: f64,
}

// Implementation of helper structs
impl PlanCache {
    fn new(max_size: usize) -> Self {
        Self {
            plans: HashMap::new(),
            access_counts: HashMap::new(),
            last_access: HashMap::new(),
            max_size,
        }
    }

    fn get_plan(&self, query_hash: u64) -> Option<CachedPlan> {
        self.plans.get(&query_hash).cloned()
    }

    fn insert_plan(&mut self, query_hash: u64, plan: IntegratedExecutionPlan) {
        // Implement LRU eviction if cache is full
        if self.plans.len() >= self.max_size {
            self.evict_lru();
        }

        let cached_plan = CachedPlan {
            plan,
            creation_time: Instant::now(),
            access_count: 0,
            average_accuracy: 0.0,
            invalidation_triggers: vec![
                InvalidationTrigger::TimeElapsed(Duration::from_secs(3600)),
                InvalidationTrigger::StatisticsUpdate,
            ],
        };

        self.plans.insert(query_hash, cached_plan);
        self.access_counts.insert(query_hash, 0);
        self.last_access.insert(query_hash, Instant::now());
    }

    fn evict_lru(&mut self) {
        if let Some(oldest_key) = self
            .last_access
            .iter()
            .min_by_key(|&(_, &instant)| instant)
            .map(|(&key, _)| key)
        {
            self.plans.remove(&oldest_key);
            self.access_counts.remove(&oldest_key);
            self.last_access.remove(&oldest_key);
        }
    }
}

impl ExecutionHistory {
    fn new(max_size: usize) -> Self {
        Self {
            executions: VecDeque::new(),
            pattern_performance: HashMap::new(),
            max_history_size: max_size,
        }
    }

    fn add_execution(&mut self, record: ExecutionRecord) {
        if self.executions.len() >= self.max_history_size {
            self.executions.pop_front();
        }
        self.executions.push_back(record);
    }
}

/// Implementation placeholder methods for the main struct
impl IntegratedQueryPlanner {
    fn update_statistics(&mut self, _analysis: &QueryAnalysis) -> Result<()> {
        // Update statistics collector with query patterns
        Ok(())
    }

    fn optimize_bgp_patterns(&mut self, algebra: &Algebra) -> Result<OptimizedBGP> {
        // Create BGPOptimizer with required statistics
        // let _bgp_optimizer = BGPOptimizer::new(&self.statistics, &self.index_stats);

        // Extract BGP patterns from algebra
        let bgp_patterns = self.extract_bgp_patterns(algebra);

        // Optimize each BGP with the optimizer
        let mut optimized_patterns = Vec::new();
        let mut total_cost = 0.0;
        let mut pattern_selectivity = Vec::new();
        let mut join_selectivity = HashMap::new();
        let mut pattern_indexes = Vec::new();

        for pattern in &bgp_patterns {
            // Calculate pattern selectivity based on statistics
            let selectivity = self.estimate_pattern_selectivity(pattern);
            let cardinality = (1_000_000.0 * selectivity).max(1.0) as usize;

            let pattern_sel = PatternSelectivity {
                pattern: pattern.clone(),
                selectivity,
                cardinality,
                factors: SelectivityFactors {
                    subject_selectivity: 1.0,
                    predicate_selectivity: 1.0,
                    object_selectivity: 1.0,
                    type_selectivity: 1.0,
                    literal_selectivity: 1.0,
                    index_factor: 1.0,
                    distribution_factor: 1.0,
                },
            };
            pattern_selectivity.push(pattern_sel);

            // Determine index usage for this pattern
            let index_hint = self.suggest_index_for_pattern(pattern);
            if let Some((pattern_idx, index_type)) = index_hint {
                pattern_indexes.push(IndexAssignment {
                    pattern_idx,
                    index_type,
                    scan_cost: selectivity * 5.0, // Estimated scan cost
                });
            }

            total_cost += selectivity * 10.0; // Base cost per pattern
            optimized_patterns.push(pattern.clone());
        }

        // Calculate join selectivity between patterns
        for i in 0..bgp_patterns.len() {
            for j in i + 1..bgp_patterns.len() {
                let join_vars = self.find_join_variables(&bgp_patterns[i], &bgp_patterns[j]);
                if !join_vars.is_empty() {
                    let selectivity =
                        self.estimate_join_selectivity(&bgp_patterns[i], &bgp_patterns[j]);
                    join_selectivity.insert((i, j), selectivity);
                }
            }
        }

        // Calculate overall selectivity
        let overall_selectivity = pattern_selectivity
            .iter()
            .map(|p| p.selectivity)
            .product::<f64>()
            * join_selectivity.values().product::<f64>();

        Ok(OptimizedBGP {
            patterns: optimized_patterns,
            estimated_cost: total_cost,
            selectivity_info: SelectivityInfo {
                pattern_selectivity,
                join_selectivity,
                overall_selectivity,
            },
            index_plan: IndexUsagePlan {
                pattern_indexes,
                join_indexes: vec![], // Would be computed based on join analysis
                index_intersections: vec![], // Would be computed for complex patterns
                bloom_filter_candidates: vec![], // Would be suggested for large joins
                recommended_indices: vec![], // Would be suggested based on patterns
                access_patterns: vec![], // Would be analyzed from query structure
                estimated_cost_reduction: 0.0, // Would be computed based on index usage
            },
        })
    }

    fn determine_execution_strategy(
        &self,
        _algebra: &Algebra,
        analysis: &QueryAnalysis,
    ) -> Result<ExecutionStrategy> {
        Ok(ExecutionStrategy {
            use_streaming: analysis.estimated_memory > self.config.streaming_threshold,
            memory_allocation: analysis.estimated_memory,
            parallel_execution: analysis.complexity_score
                > self.adaptive_thresholds.parallel_execution_threshold,
            index_recommendations: vec![],
        })
    }

    #[allow(clippy::only_used_in_recursion)]
    fn estimate_execution_cost(
        &self,
        algebra: &Algebra,
        strategy: &ExecutionStrategy,
    ) -> Result<CostEstimate> {
        let mut cpu_cost = 0.0;
        let mut io_cost = 0.0;
        let mut memory_cost = strategy.memory_allocation as f64 / 1024.0 / 1024.0; // Memory cost in MB
        let network_cost = 0.0;

        // Recursively calculate costs based on algebra structure
        let estimated_cardinality = match algebra {
            Algebra::Bgp(patterns) => {
                // Cost for BGP evaluation
                cpu_cost += patterns.len() as f64 * 2.0; // Base cost per pattern
                io_cost += patterns.len() as f64 * 1.0; // I/O cost for pattern matching
                (patterns.len() * 100).max(1) // Estimate based on pattern count
            }
            Algebra::Join { left, right } => {
                // Recursive cost calculation for joins
                let left_cost = self.estimate_execution_cost(left, strategy)?;
                let right_cost = self.estimate_execution_cost(right, strategy)?;

                cpu_cost += left_cost.cpu_cost + right_cost.cpu_cost;
                io_cost += left_cost.io_cost + right_cost.io_cost;

                // Join cost is proportional to the product of cardinalities
                let join_cost = (left_cost.cardinality * right_cost.cardinality) as f64 * 0.001;
                cpu_cost += join_cost;

                ((left_cost.cardinality as f64 * right_cost.cardinality as f64 * 0.1) as usize)
                    .max(1)
            }
            Algebra::Union { left, right } => {
                let left_cost = self.estimate_execution_cost(left, strategy)?;
                let right_cost = self.estimate_execution_cost(right, strategy)?;

                cpu_cost += left_cost.cpu_cost + right_cost.cpu_cost;
                io_cost += left_cost.io_cost + right_cost.io_cost;
                left_cost.cardinality + right_cost.cardinality
            }
            Algebra::Filter { pattern, .. } => {
                let pattern_cost = self.estimate_execution_cost(pattern, strategy)?;
                cpu_cost += pattern_cost.cpu_cost + 5.0; // Additional cost for filtering
                io_cost += pattern_cost.io_cost;
                (pattern_cost.cardinality as f64 * 0.5) as usize // Filtering reduces cardinality
            }
            Algebra::Group {
                pattern, variables, ..
            } => {
                let pattern_cost = self.estimate_execution_cost(pattern, strategy)?;
                cpu_cost += pattern_cost.cpu_cost + variables.len() as f64 * 3.0; // Grouping cost
                io_cost += pattern_cost.io_cost;
                (pattern_cost.cardinality as f64 * 0.2) as usize // Grouping reduces cardinality
            }
            Algebra::OrderBy {
                pattern,
                conditions,
            } => {
                let pattern_cost = self.estimate_execution_cost(pattern, strategy)?;
                let sort_cost = (pattern_cost.cardinality as f64).log2() * conditions.len() as f64; // O(n log n) sort
                cpu_cost += pattern_cost.cpu_cost + sort_cost;
                io_cost += pattern_cost.io_cost;
                pattern_cost.cardinality
            }
            _ => {
                // Default costs for other algebra types
                cpu_cost += 1.0;
                io_cost += 0.5;
                100
            }
        };

        // Apply strategy-specific adjustments
        if strategy.use_streaming {
            memory_cost *= 0.5; // Streaming reduces memory usage
            io_cost *= 1.2; // But increases I/O
        }

        if strategy.parallel_execution {
            cpu_cost *= 0.7; // Parallel execution improves CPU efficiency
        }

        Ok(CostEstimate::new(
            cpu_cost,
            io_cost,
            memory_cost,
            network_cost,
            estimated_cardinality,
        ))
    }

    fn estimate_memory_requirements(&self, analysis: &QueryAnalysis) -> Result<usize> {
        let base_memory = 64 * 1024 * 1024; // 64MB
        let variable_factor = analysis.variables.len() * 1024 * 1024; // 1MB per variable
        let complexity_factor = (analysis.complexity_score * 1024.0 * 1024.0) as usize;

        Ok(base_memory + variable_factor + complexity_factor)
    }

    fn generate_alternative_plans(
        &self,
        _algebra: &Algebra,
        _cost_estimate: &CostEstimate,
    ) -> Result<Vec<AlternativePlan>> {
        // Generate alternative execution plans for fallback
        Ok(vec![])
    }

    fn update_adaptive_thresholds(&mut self, record: &ExecutionRecord) -> Result<()> {
        // Update adaptive thresholds based on execution performance
        let accuracy_ratio = if record.estimated_duration.as_millis() > 0 {
            record.actual_duration.as_millis() as f64 / record.estimated_duration.as_millis() as f64
        } else {
            1.0
        };

        // If our estimates are consistently off, adjust thresholds
        if accuracy_ratio > 2.0 {
            // We're underestimating, be more conservative
            self.adaptive_thresholds.streaming_memory_threshold =
                (self.adaptive_thresholds.streaming_memory_threshold as f64 * 1.1) as usize;
            self.adaptive_thresholds.parallel_execution_threshold *= 1.1;
        } else if accuracy_ratio < 0.5 {
            // We're overestimating, be more aggressive
            self.adaptive_thresholds.streaming_memory_threshold =
                (self.adaptive_thresholds.streaming_memory_threshold as f64 * 0.9) as usize;
            self.adaptive_thresholds.parallel_execution_threshold *= 0.9;
        }

        // Update plan cache accuracy threshold based on actual success rate
        if record.success {
            self.adaptive_thresholds.plan_cache_accuracy_threshold =
                (self.adaptive_thresholds.plan_cache_accuracy_threshold * 0.95 + 0.05).min(0.95);
        } else {
            self.adaptive_thresholds.plan_cache_accuracy_threshold =
                (self.adaptive_thresholds.plan_cache_accuracy_threshold * 0.95).max(0.5);
        }

        debug!("Updated adaptive thresholds based on execution feedback");
        Ok(())
    }

    fn update_cost_model(&mut self, record: &ExecutionRecord) -> Result<()> {
        // Update cost model with actual vs. estimated performance
        let _cost_model = self.cost_model.lock().expect("lock poisoned");

        // Calculate estimation error
        let duration_error = if record.estimated_duration.as_millis() > 0 {
            (record.actual_duration.as_millis() as f64
                - record.estimated_duration.as_millis() as f64)
                .abs()
                / record.estimated_duration.as_millis() as f64
        } else {
            0.0
        };

        let cardinality_error = if record.estimated_cardinality > 0 {
            (record.actual_cardinality as f64 - record.estimated_cardinality as f64).abs()
                / record.estimated_cardinality as f64
        } else {
            0.0
        };

        // Update cost model parameters based on errors
        // This is a simplified approach - in practice, you'd use more sophisticated ML techniques
        if duration_error > 0.5 {
            info!(
                "Large duration estimation error: {:.2}, updating cost model",
                duration_error
            );
            // Adjust cost factors based on the error
        }

        if cardinality_error > 0.5 {
            info!(
                "Large cardinality estimation error: {:.2}, updating statistics",
                cardinality_error
            );
            // Update cardinality estimation parameters
        }

        // Update statistics collector with actual execution data
        if let Ok(mut stats_collector) = self.statistics_collector.lock() {
            if let Err(e) = stats_collector.update_execution_statistics(
                record.actual_duration,
                record.actual_cardinality,
                record.memory_used,
            ) {
                tracing::warn!("Failed to update execution statistics: {}", e);
            }
        } else {
            tracing::warn!("Failed to acquire lock for statistics collector");
        }

        debug!("Updated cost model with execution feedback");
        Ok(())
    }

    fn analyze_index_opportunities(
        &self,
        _history: &ExecutionHistory,
    ) -> Result<Vec<IndexRecommendation>> {
        // Analyze execution history to recommend new indexes
        // Basic index recommendations based on common query patterns
        // In a full implementation, this would analyze actual execution history

        let recommendations = vec![
            // Recommend B-tree index for frequently filtered properties
            IndexRecommendation {
                index_type: IndexType::BTree,
                estimated_benefit: 0.3, // 30% improvement
                creation_cost: 100.0,
                maintenance_cost: 10.0,
                confidence: 0.8,
            },
            // Recommend hash index for equality lookups
            IndexRecommendation {
                index_type: IndexType::Hash,
                estimated_benefit: 0.5, // 50% improvement for exact matches
                creation_cost: 50.0,
                maintenance_cost: 5.0,
                confidence: 0.9,
            },
        ];

        Ok(recommendations)
    }

    /// Extract BGP patterns from algebra expression
    #[allow(clippy::only_used_in_recursion)]
    fn extract_bgp_patterns(&self, algebra: &Algebra) -> Vec<TriplePattern> {
        match algebra {
            Algebra::Bgp(patterns) => patterns.clone(),
            Algebra::Join { left, right } => {
                let mut patterns = self.extract_bgp_patterns(left);
                patterns.extend(self.extract_bgp_patterns(right));
                patterns
            }
            Algebra::Union { left, right } => {
                let mut patterns = self.extract_bgp_patterns(left);
                patterns.extend(self.extract_bgp_patterns(right));
                patterns
            }
            Algebra::Filter { pattern, .. } => self.extract_bgp_patterns(pattern),
            _ => Vec::new(),
        }
    }

    /// Estimate selectivity of a triple pattern
    fn estimate_pattern_selectivity(&self, pattern: &TriplePattern) -> f64 {
        // Basic selectivity estimation based on pattern structure
        let mut selectivity: f64 = 1.0;

        // Reduce selectivity for each concrete term (non-variable)
        if !matches!(pattern.subject, Term::Variable(_)) {
            selectivity *= 0.1; // Subject specified reduces selectivity to 10%
        }
        if !matches!(pattern.predicate, Term::Variable(_)) {
            selectivity *= 0.2; // Predicate specified reduces selectivity to 20%
        }
        if !matches!(pattern.object, Term::Variable(_)) {
            selectivity *= 0.1; // Object specified reduces selectivity to 10%
        }

        // Ensure minimum selectivity
        selectivity.max(0.001)
    }

    /// Suggest an index for a specific pattern
    fn suggest_index_for_pattern(&self, pattern: &TriplePattern) -> Option<(usize, IndexType)> {
        // Suggest index based on pattern characteristics
        match (&pattern.subject, &pattern.predicate, &pattern.object) {
            // If subject is variable but predicate is concrete, suggest predicate index
            (Term::Variable(_), Term::Iri(_), _) => Some((1, IndexType::BTree)),
            // If object is concrete, suggest object index
            (_, _, Term::Literal(_)) => Some((2, IndexType::Hash)),
            // If subject is concrete, suggest subject index
            (Term::Iri(_), _, _) => Some((0, IndexType::Hash)),
            _ => None,
        }
    }

    /// Find join variables between two patterns
    fn find_join_variables(&self, left: &TriplePattern, right: &TriplePattern) -> Vec<Variable> {
        let left_vars = self.extract_pattern_variables(left);
        let right_vars = self.extract_pattern_variables(right);

        left_vars.intersection(&right_vars).cloned().collect()
    }

    /// Estimate join selectivity between two patterns
    fn estimate_join_selectivity(&self, left: &TriplePattern, right: &TriplePattern) -> f64 {
        let join_vars = self.find_join_variables(left, right);

        if join_vars.is_empty() {
            return 1.0; // Cartesian product
        }

        // Estimate based on number of join variables
        // More join variables typically means higher selectivity
        match join_vars.len() {
            1 => 0.1,  // Single variable join
            2 => 0.05, // Two variable join - more selective
            _ => 0.01, // Multiple variable joins are very selective
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_integrated_planner_creation() {
        let config = IntegratedPlannerConfig::default();
        let planner = IntegratedQueryPlanner::new(config);
        assert!(planner.is_ok());
    }

    #[test]
    fn test_query_analysis() {
        let config = IntegratedPlannerConfig::default();
        let planner = IntegratedQueryPlanner::new(config).unwrap();

        let algebra = Algebra::Bgp(vec![]);
        let analysis = planner.analyze_query(&algebra).unwrap();

        assert_eq!(analysis.triple_pattern_count, 0);
        assert_eq!(analysis.join_count, 0);
    }

    #[test]
    fn test_plan_cache() {
        let mut cache = PlanCache::new(10);

        let plan = IntegratedExecutionPlan {
            optimized_algebra: Algebra::Bgp(vec![]),
            estimated_cost: CostEstimate {
                cpu_cost: 10.0,
                io_cost: 5.0,
                memory_cost: 1.0,
                network_cost: 0.0,
                total_cost: 16.0,
                cardinality: 1000,
                selectivity: 1.0,
                operation_costs: HashMap::new(),
            },
            index_plan: IndexUsagePlan {
                pattern_indexes: vec![],
                join_indexes: vec![],
                index_intersections: vec![],
                bloom_filter_candidates: vec![],
                recommended_indices: vec![],
                access_patterns: vec![],
                estimated_cost_reduction: 0.0,
            },
            use_streaming: false,
            memory_allocation: 1024,
            expected_duration: Duration::from_millis(100),
            confidence: 0.8,
            adaptive_hints: AdaptiveHints::default(),
            alternative_plans: vec![],
        };

        cache.insert_plan(12345, plan);
        assert!(cache.get_plan(12345).is_some());
    }
}