kore_fileformat 1.3.3

/// Query Optimization Integration for KORE v1.6.0
/// 
/// Full integration layer combining all v1.6.0 components:
/// - Statistics Engine (analysis)
/// - Query Optimizer (planning)
/// - Join Strategies (execution methods)
/// - Predicate Pushdown (filtering)
/// - Adaptive Executor (runtime)

use crate::query_statistics_v1::{CostModel, TableStats, StatisticsCollector};
use crate::query_optimizer_v1::{LogicalExpr, PhysicalExpr, OptimizerContext, optimize};
use crate::join_strategies_v1::{Row, JoinStrategy, JoinStrategyConfig, choose_join_strategy, execute_join};
use crate::predicate_pushdown_v1::{Predicate, ChunkStats, PredicateResult, filter_chunks};
use crate::adaptive_executor_v1::{AdaptiveExecutionContext, AdaptiveQueryPipeline, ExecutionHint};

use std::collections::HashMap;

/// Complete query optimization engine
pub struct QueryOptimizationEngine {
    /// Table statistics
    pub table_stats: HashMap<String, TableStats>,
    /// Cost model for estimation
    pub cost_model: CostModel,
    /// Join strategy configuration
    pub join_config: JoinStrategyConfig,
    /// Adaptive execution enabled
    pub enable_adaptive: bool,
}

impl QueryOptimizationEngine {
    /// Create new optimization engine
    pub fn new() -> Self {
        QueryOptimizationEngine {
            table_stats: HashMap::new(),
            cost_model: CostModel::default(),
            join_config: JoinStrategyConfig::default(),
            enable_adaptive: true,
        }
    }

    /// Register table statistics
    pub fn register_table(&mut self, name: String, stats: TableStats) {
        self.table_stats.insert(name, stats);
    }

    /// Optimize a logical query plan
    pub fn optimize_query(&self, logical: LogicalExpr) -> Result<PhysicalExpr, String> {
        let context = OptimizerContext::new(self.table_stats.clone());
        optimize(logical, &context)
    }

    /// Execute an optimized query (simplified)
    pub fn execute(&self, physical: &PhysicalExpr) -> Result<Vec<Row>, String> {
        self.execute_physical_plan(physical)
    }

    fn execute_physical_plan(&self, plan: &PhysicalExpr) -> Result<Vec<Row>, String> {
        match plan {
            PhysicalExpr::SeqScan { table, .. } => {
                // Return dummy data
                let mut row = HashMap::new();
                row.insert("id".to_string(), "1".to_string());
                Ok(vec![row])
            }

            PhysicalExpr::Filter { input, predicate, selectivity } => {
                let input_rows = self.execute_physical_plan(input)?;
                // Simplified: apply selectivity filter
                let filtered_count = ((input_rows.len() as f64) * selectivity).ceil() as usize;
                Ok(input_rows.into_iter().take(filtered_count).collect())
            }

            PhysicalExpr::Project { input, columns } => {
                let input_rows = self.execute_physical_plan(input)?;
                // Simplified: project columns
                Ok(input_rows)
            }

            PhysicalExpr::HashJoin { left, right, join_keys, .. } => {
                let left_rows = self.execute_physical_plan(left)?;
                let right_rows = self.execute_physical_plan(right)?;
                execute_join(&left_rows, &right_rows, join_keys, JoinStrategy::Hash)
            }

            PhysicalExpr::NestedLoopJoin { left, right, join_keys, .. } => {
                let left_rows = self.execute_physical_plan(left)?;
                let right_rows = self.execute_physical_plan(right)?;
                execute_join(&left_rows, &right_rows, join_keys, JoinStrategy::NestedLoop)
            }

            PhysicalExpr::SortMergeJoin { left, right, join_keys, .. } => {
                let left_rows = self.execute_physical_plan(left)?;
                let right_rows = self.execute_physical_plan(right)?;
                execute_join(&left_rows, &right_rows, join_keys, JoinStrategy::SortMerge)
            }

            PhysicalExpr::GroupBy { input, group_keys, .. } => {
                let input_rows = self.execute_physical_plan(input)?;
                // Simplified: just return input
                Ok(input_rows)
            }

            PhysicalExpr::Sort { input, order_keys, .. } => {
                let mut rows = self.execute_physical_plan(input)?;
                // Simplified: just return in order
                Ok(rows)
            }

            PhysicalExpr::IndexScan { .. } => {
                // Not implemented
                Err("IndexScan not yet implemented".to_string())
            }
        }
    }

    /// Get estimated cost for a physical plan
    pub fn estimate_cost(&self, plan: &PhysicalExpr) -> f64 {
        plan.total_cost()
    }

    /// Get estimated output rows
    pub fn estimate_rows(&self, plan: &PhysicalExpr) -> u64 {
        plan.estimated_rows(&self.table_stats)
    }

    /// Full optimization pipeline: analyze -> optimize -> execute
    pub fn full_pipeline(
        &self,
        logical: LogicalExpr,
    ) -> Result<(PhysicalExpr, Vec<Row>, f64), String> {
        // Step 1: Optimize
        let physical = self.optimize_query(logical)?;

        // Step 2: Estimate cost
        let cost = self.estimate_cost(&physical);

        // Step 3: Execute
        let results = self.execute(&physical)?;

        Ok((physical, results, cost))
    }
}

/// Example usage demonstrating the full integration
pub fn example_optimization_pipeline() -> Result<(), String> {
    // Create engine
    let mut engine = QueryOptimizationEngine::new();

    // Register tables
    let mut users_stats = TableStats::new("users".to_string(), 10000);
    let mut orders_stats = TableStats::new("orders".to_string(), 50000);

    engine.register_table("users".to_string(), users_stats);
    engine.register_table("orders".to_string(), orders_stats);

    // Create a logical query: SELECT * FROM users WHERE age > 18
    let logical = LogicalExpr::Filter {
        input: Box::new(LogicalExpr::Scan {
            table: "users".to_string(),
        }),
        predicate: "age > 18".to_string(),
    };

    // Optimize and execute
    let (physical, results, cost) = engine.full_pipeline(logical)?;

    println!("Optimized plan: {:?}", physical);
    println!("Estimated cost: {}", cost);
    println!("Results: {} rows", results.len());

    Ok(())
}

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

    #[test]
    fn test_engine_creation() {
        let engine = QueryOptimizationEngine::new();
        assert_eq!(engine.table_stats.len(), 0);
    }

    #[test]
    fn test_register_table() {
        let mut engine = QueryOptimizationEngine::new();
        let stats = TableStats::new("users".to_string(), 1000);
        engine.register_table("users".to_string(), stats);
        assert_eq!(engine.table_stats.len(), 1);
    }

    #[test]
    fn test_optimize_scan() {
        let mut engine = QueryOptimizationEngine::new();
        let stats = TableStats::new("users".to_string(), 1000);
        engine.register_table("users".to_string(), stats);

        let logical = LogicalExpr::Scan {
            table: "users".to_string(),
        };

        let result = engine.optimize_query(logical);
        assert!(result.is_ok());
    }

    #[test]
    fn test_optimize_filter() {
        let mut engine = QueryOptimizationEngine::new();
        let stats = TableStats::new("users".to_string(), 1000);
        engine.register_table("users".to_string(), stats);

        let logical = LogicalExpr::Filter {
            input: Box::new(LogicalExpr::Scan {
                table: "users".to_string(),
            }),
            predicate: "age > 18".to_string(),
        };

        let result = engine.optimize_query(logical);
        assert!(result.is_ok());
        let physical = result.unwrap();
        assert!(matches!(physical, PhysicalExpr::Filter { .. }));
    }

    #[test]
    fn test_full_pipeline() {
        let mut engine = QueryOptimizationEngine::new();
        let stats = TableStats::new("users".to_string(), 1000);
        engine.register_table("users".to_string(), stats);

        let logical = LogicalExpr::Scan {
            table: "users".to_string(),
        };

        let result = engine.full_pipeline(logical);
        assert!(result.is_ok());

        let (physical, results, cost) = result.unwrap();
        assert!(matches!(physical, PhysicalExpr::SeqScan { .. }));
        assert!(cost > 0.0);
    }

    #[test]
    fn test_estimate_cost() {
        let engine = QueryOptimizationEngine::new();
        let plan = PhysicalExpr::SeqScan {
            table: "users".to_string(),
            cost: 100.0,
        };

        let cost = engine.estimate_cost(&plan);
        assert_eq!(cost, 100.0);
    }

    #[test]
    fn test_example_pipeline() {
        let result = example_optimization_pipeline();
        assert!(result.is_ok());
    }
}