heliosdb-nano 3.23.2

//! Distributed Query EXPLAIN
//!
//! This module provides distributed execution plan analysis including:
//! - Multi-node execution plans
//! - Network communication costs
//! - Data transfer visualization
//! - Partition pruning display
//! - Cross-partition join analysis
//! - Distributed transaction coordination

#![allow(unused_variables)]

use crate::Result;
use serde::{Deserialize, Serialize};

/// Node in a distributed cluster
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterNode {
    pub node_id: String,
    pub host: String,
    pub port: u16,
    pub role: NodeRole,
    pub data_size_mb: f64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum NodeRole {
    Coordinator,
    Worker,
    Replica,
}

/// Partition information
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PartitionInfo {
    pub partition_id: u32,
    pub table_name: String,
    pub node_id: String,
    pub row_count: usize,
    pub size_mb: f64,
    pub partition_key: String,
    pub key_range: Option<(String, String)>,
}

/// Network operation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetworkOperation {
    pub operation_type: NetworkOpType,
    pub source_node: String,
    pub target_node: String,
    pub data_size_mb: f64,
    pub estimated_time_ms: f64,
    pub compression_enabled: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum NetworkOpType {
    Shuffle,
    Broadcast,
    Gather,
    Scatter,
    Replicate,
}

impl std::fmt::Display for NetworkOpType {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            NetworkOpType::Shuffle => write!(f, "SHUFFLE"),
            NetworkOpType::Broadcast => write!(f, "BROADCAST"),
            NetworkOpType::Gather => write!(f, "GATHER"),
            NetworkOpType::Scatter => write!(f, "SCATTER"),
            NetworkOpType::Replicate => write!(f, "REPLICATE"),
        }
    }
}

/// Partition pruning analysis
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PartitionPruning {
    pub total_partitions: usize,
    pub scanned_partitions: usize,
    pub pruned_partitions: usize,
    pub pruning_efficiency: f64,
    pub pruning_predicates: Vec<String>,
}

/// Cross-partition join analysis
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CrossPartitionJoin {
    pub join_type: String,
    pub left_partitions: usize,
    pub right_partitions: usize,
    pub shuffle_required: bool,
    pub broadcast_candidate: Option<String>,
    pub estimated_network_cost_ms: f64,
}

/// Distributed transaction coordination
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedTransaction {
    pub coordinator_node: String,
    pub participant_nodes: Vec<String>,
    pub two_phase_commit: bool,
    pub coordination_overhead_ms: f64,
    pub network_round_trips: usize,
}

/// Complete distributed EXPLAIN output
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedExplain {
    pub cluster_nodes: Vec<ClusterNode>,
    pub partitions: Vec<PartitionInfo>,
    pub network_operations: Vec<NetworkOperation>,
    pub partition_pruning: Option<PartitionPruning>,
    pub cross_partition_joins: Vec<CrossPartitionJoin>,
    pub distributed_transaction: Option<DistributedTransaction>,
    pub total_network_cost_ms: f64,
    pub total_data_transfer_mb: f64,
    pub parallelism_degree: usize,
}

/// Distributed EXPLAIN analyzer
pub struct DistributedExplainAnalyzer {
    cluster_size: usize,
    enable_compression: bool,
    network_bandwidth_mbps: f64,
}

impl DistributedExplainAnalyzer {
    pub fn new(cluster_size: usize) -> Self {
        Self {
            cluster_size,
            enable_compression: true,
            network_bandwidth_mbps: 1000.0, // 1 Gbps default
        }
    }

    pub fn with_compression(mut self, enable: bool) -> Self {
        self.enable_compression = enable;
        self
    }

    pub fn with_network_bandwidth(mut self, mbps: f64) -> Self {
        self.network_bandwidth_mbps = mbps;
        self
    }

    /// Analyze distributed execution for a query
    pub fn analyze(
        &self,
        query_type: &str,
        tables: &[String],
        has_join: bool,
        has_aggregation: bool,
    ) -> Result<DistributedExplain> {
        let cluster_nodes = self.generate_cluster_topology();
        let partitions = self.generate_partition_info(tables);
        let partition_pruning = self.analyze_partition_pruning(query_type, &partitions);

        let (network_operations, cross_partition_joins) = if has_join {
            self.analyze_distributed_join(&partitions, &cluster_nodes)
        } else if has_aggregation {
            (self.analyze_distributed_aggregation(&partitions, &cluster_nodes), vec![])
        } else {
            (self.analyze_distributed_scan(&partitions, &cluster_nodes), vec![])
        };

        let distributed_transaction = if query_type != "SELECT" {
            Some(self.analyze_distributed_transaction(&cluster_nodes))
        } else {
            None
        };

        let total_network_cost_ms: f64 = network_operations.iter()
            .map(|op| op.estimated_time_ms)
            .sum();

        let total_data_transfer_mb: f64 = network_operations.iter()
            .map(|op| op.data_size_mb)
            .sum();

        let parallelism_degree = cluster_nodes.iter()
            .filter(|n| matches!(n.role, NodeRole::Worker))
            .count();

        Ok(DistributedExplain {
            cluster_nodes,
            partitions,
            network_operations,
            partition_pruning,
            cross_partition_joins,
            distributed_transaction,
            total_network_cost_ms,
            total_data_transfer_mb,
            parallelism_degree,
        })
    }

    fn generate_cluster_topology(&self) -> Vec<ClusterNode> {
        let mut nodes = vec![
            ClusterNode {
                node_id: "coordinator".to_string(),
                host: "10.0.0.1".to_string(),
                port: 5432,
                role: NodeRole::Coordinator,
                data_size_mb: 0.0,
            }
        ];

        for i in 0..self.cluster_size {
            nodes.push(ClusterNode {
                node_id: format!("worker-{}", i + 1),
                host: format!("10.0.0.{}", i + 10),
                port: 5432,
                role: NodeRole::Worker,
                data_size_mb: 10000.0,
            });
        }

        nodes
    }

    fn generate_partition_info(&self, tables: &[String]) -> Vec<PartitionInfo> {
        let mut partitions = Vec::new();
        let partitions_per_table = 16;

        for table in tables {
            for i in 0..partitions_per_table {
                let node_idx = i % self.cluster_size;
                partitions.push(PartitionInfo {
                    partition_id: i as u32,
                    table_name: table.clone(),
                    node_id: format!("worker-{}", node_idx + 1),
                    row_count: 100000,
                    size_mb: 500.0,
                    partition_key: "id".to_string(),
                    key_range: Some((
                        format!("{}", i * 1000),
                        format!("{}", (i + 1) * 1000 - 1),
                    )),
                });
            }
        }

        partitions
    }

    fn analyze_partition_pruning(
        &self,
        query_type: &str,
        partitions: &[PartitionInfo],
    ) -> Option<PartitionPruning> {
        if query_type == "SELECT" {
            let total = partitions.len();
            let scanned = (total as f64 * 0.25) as usize; // 25% scanned
            let pruned = total - scanned;

            Some(PartitionPruning {
                total_partitions: total,
                scanned_partitions: scanned,
                pruned_partitions: pruned,
                pruning_efficiency: (pruned as f64 / total as f64) * 100.0,
                pruning_predicates: vec![
                    "id BETWEEN 1000 AND 5000".to_string(),
                    "partition_key IN (1, 2, 3, 4)".to_string(),
                ],
            })
        } else {
            None
        }
    }

    fn analyze_distributed_scan(
        &self,
        partitions: &[PartitionInfo],
        nodes: &[ClusterNode],
    ) -> Vec<NetworkOperation> {
        let mut operations = Vec::new();
        let partitions_per_worker = partitions.len() / self.cluster_size.max(1);

        // Gather results from workers to coordinator
        for i in 0..self.cluster_size {
            let data_size = partitions_per_worker as f64 * 50.0; // 50 MB per partition result
            let compressed_size = if self.enable_compression {
                data_size * 0.3 // 70% compression
            } else {
                data_size
            };

            operations.push(NetworkOperation {
                operation_type: NetworkOpType::Gather,
                source_node: format!("worker-{}", i + 1),
                target_node: "coordinator".to_string(),
                data_size_mb: compressed_size,
                estimated_time_ms: self.calculate_transfer_time(compressed_size),
                compression_enabled: self.enable_compression,
            });
        }

        operations
    }

    fn analyze_distributed_aggregation(
        &self,
        partitions: &[PartitionInfo],
        nodes: &[ClusterNode],
    ) -> Vec<NetworkOperation> {
        let mut operations = Vec::new();

        // Phase 1: Local aggregation on each worker (no network transfer)

        // Phase 2: Shuffle partial aggregates for final aggregation
        for i in 0..self.cluster_size {
            let data_size = 10.0; // Aggregated results are small
            operations.push(NetworkOperation {
                operation_type: NetworkOpType::Shuffle,
                source_node: format!("worker-{}", i + 1),
                target_node: "coordinator".to_string(),
                data_size_mb: data_size,
                estimated_time_ms: self.calculate_transfer_time(data_size),
                compression_enabled: self.enable_compression,
            });
        }

        operations
    }

    // SAFETY: tables[0] and tables[1] are guarded by tables.len() >= 2 check.
    #[allow(clippy::indexing_slicing)]
    fn analyze_distributed_join(
        &self,
        partitions: &[PartitionInfo],
        nodes: &[ClusterNode],
    ) -> (Vec<NetworkOperation>, Vec<CrossPartitionJoin>) {
        let mut operations = Vec::new();
        let mut joins = Vec::new();

        // Determine if broadcast or shuffle join
        let tables: Vec<_> = partitions.iter()
            .map(|p| p.table_name.clone())
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();

        if tables.len() >= 2 {
            let left_size: f64 = partitions.iter()
                .filter(|p| p.table_name == tables[0])
                .map(|p| p.size_mb)
                .sum();

            let right_size: f64 = partitions.iter()
                .filter(|p| p.table_name == tables[1])
                .map(|p| p.size_mb)
                .sum();

            let (shuffle_required, broadcast_candidate) = if right_size < 100.0 {
                // Small table - broadcast join
                (false, Some(tables[1].clone()))
            } else {
                // Large tables - shuffle join
                (true, None)
            };

            if let Some(ref broadcast_table) = broadcast_candidate {
                // Broadcast small table to all workers
                for i in 0..self.cluster_size {
                    operations.push(NetworkOperation {
                        operation_type: NetworkOpType::Broadcast,
                        source_node: "coordinator".to_string(),
                        target_node: format!("worker-{}", i + 1),
                        data_size_mb: right_size,
                        estimated_time_ms: self.calculate_transfer_time(right_size),
                        compression_enabled: self.enable_compression,
                    });
                }
            } else {
                // Shuffle both tables
                for i in 0..self.cluster_size {
                    let shuffle_size = (left_size + right_size) / self.cluster_size as f64;
                    operations.push(NetworkOperation {
                        operation_type: NetworkOpType::Shuffle,
                        source_node: format!("worker-{}", i + 1),
                        target_node: format!("worker-{}", (i + 1) % self.cluster_size + 1),
                        data_size_mb: shuffle_size,
                        estimated_time_ms: self.calculate_transfer_time(shuffle_size),
                        compression_enabled: self.enable_compression,
                    });
                }
            }

            let network_cost: f64 = operations.iter().map(|op| op.estimated_time_ms).sum();

            joins.push(CrossPartitionJoin {
                join_type: "Hash Join".to_string(),
                left_partitions: partitions.iter().filter(|p| p.table_name == tables[0]).count(),
                right_partitions: partitions.iter().filter(|p| p.table_name == tables[1]).count(),
                shuffle_required,
                broadcast_candidate,
                estimated_network_cost_ms: network_cost,
            });
        }

        (operations, joins)
    }

    fn analyze_distributed_transaction(&self, nodes: &[ClusterNode]) -> DistributedTransaction {
        let worker_nodes: Vec<_> = nodes.iter()
            .filter(|n| matches!(n.role, NodeRole::Worker))
            .map(|n| n.node_id.clone())
            .collect();

        DistributedTransaction {
            coordinator_node: "coordinator".to_string(),
            participant_nodes: worker_nodes.clone(),
            two_phase_commit: true,
            coordination_overhead_ms: worker_nodes.len() as f64 * 5.0, // 5ms per participant
            network_round_trips: 4, // 2PC requires 4 round trips
        }
    }

    fn calculate_transfer_time(&self, size_mb: f64) -> f64 {
        // Transfer time = data size / bandwidth + network latency
        let transfer_time_ms = (size_mb / self.network_bandwidth_mbps * 1000.0) * 8.0; // Convert to ms
        let network_latency_ms = 2.0; // 2ms baseline latency
        transfer_time_ms + network_latency_ms
    }

    /// Format distributed EXPLAIN output
    // SAFETY: table_partitions[0] access is within a non-empty filtered collection.
    #[allow(clippy::indexing_slicing)]
    pub fn format_output(&self, explain: &DistributedExplain) -> String {
        let mut output = String::new();

        output.push_str("═══════════════════════════════════════════════════════════════\n");
        output.push_str("         DISTRIBUTED QUERY EXPLAIN ANALYSIS                    \n");
        output.push_str("═══════════════════════════════════════════════════════════════\n\n");

        // Cluster topology
        output.push_str("───────────────────────────────────────────────────────────────\n");
        output.push_str(&format!("  CLUSTER TOPOLOGY ({} nodes)\n", explain.cluster_nodes.len()));
        output.push_str("───────────────────────────────────────────────────────────────\n\n");

        for node in &explain.cluster_nodes {
            output.push_str(&format!("• {} ({:?})\n", node.node_id, node.role));
            output.push_str(&format!("  Host: {}:{}\n", node.host, node.port));
            if node.data_size_mb > 0.0 {
                output.push_str(&format!("  Data Size: {:.2} MB\n", node.data_size_mb));
            }
            output.push_str("\n");
        }

        // Partition information
        if !explain.partitions.is_empty() {
            output.push_str("───────────────────────────────────────────────────────────────\n");
            output.push_str(&format!("  PARTITIONS ({} total)\n", explain.partitions.len()));
            output.push_str("───────────────────────────────────────────────────────────────\n\n");

            let tables: std::collections::HashSet<_> = explain.partitions.iter()
                .map(|p| p.table_name.clone())
                .collect();

            for table in tables {
                let table_partitions: Vec<_> = explain.partitions.iter()
                    .filter(|p| p.table_name == table)
                    .collect();

                output.push_str(&format!("Table: {} ({} partitions)\n", table, table_partitions.len()));
                output.push_str(&format!("  Partition Key: {}\n", table_partitions[0].partition_key));
                output.push_str(&format!("  Total Size: {:.2} MB\n",
                    table_partitions.iter().map(|p| p.size_mb).sum::<f64>()));
                output.push_str("\n");
            }
        }

        // Partition pruning
        if let Some(ref pruning) = explain.partition_pruning {
            output.push_str("───────────────────────────────────────────────────────────────\n");
            output.push_str("  PARTITION PRUNING\n");
            output.push_str("───────────────────────────────────────────────────────────────\n\n");
            output.push_str(&format!("Total Partitions: {}\n", pruning.total_partitions));
            output.push_str(&format!("Scanned: {} ({:.1}%)\n",
                pruning.scanned_partitions,
                (pruning.scanned_partitions as f64 / pruning.total_partitions as f64) * 100.0));
            output.push_str(&format!("Pruned: {} ({:.1}% efficiency)\n",
                pruning.pruned_partitions,
                pruning.pruning_efficiency));

            if !pruning.pruning_predicates.is_empty() {
                output.push_str("\nPruning Predicates:\n");
                for pred in &pruning.pruning_predicates {
                    output.push_str(&format!("  • {}\n", pred));
                }
            }
            output.push_str("\n");
        }

        // Network operations
        if !explain.network_operations.is_empty() {
            output.push_str("───────────────────────────────────────────────────────────────\n");
            output.push_str(&format!("  NETWORK OPERATIONS ({} ops)\n", explain.network_operations.len()));
            output.push_str("───────────────────────────────────────────────────────────────\n\n");

            for op in &explain.network_operations {
                output.push_str(&format!("• {} {} → {}\n",
                    op.operation_type, op.source_node, op.target_node));
                output.push_str(&format!("  Data Transfer: {:.2} MB\n", op.data_size_mb));
                output.push_str(&format!("  Estimated Time: {:.2} ms\n", op.estimated_time_ms));
                output.push_str(&format!("  Compression: {}\n",
                    if op.compression_enabled { "enabled" } else { "disabled" }));
                output.push_str("\n");
            }
        }

        // Cross-partition joins
        if !explain.cross_partition_joins.is_empty() {
            output.push_str("───────────────────────────────────────────────────────────────\n");
            output.push_str("  CROSS-PARTITION JOINS\n");
            output.push_str("───────────────────────────────────────────────────────────────\n\n");

            for join in &explain.cross_partition_joins {
                output.push_str(&format!("{}\n", join.join_type));
                output.push_str(&format!("  Left Partitions: {}\n", join.left_partitions));
                output.push_str(&format!("  Right Partitions: {}\n", join.right_partitions));
                output.push_str(&format!("  Shuffle Required: {}\n", join.shuffle_required));
                if let Some(ref broadcast) = join.broadcast_candidate {
                    output.push_str(&format!("  Broadcast Table: {}\n", broadcast));
                }
                output.push_str(&format!("  Network Cost: {:.2} ms\n", join.estimated_network_cost_ms));
                output.push_str("\n");
            }
        }

        // Distributed transaction
        if let Some(ref txn) = explain.distributed_transaction {
            output.push_str("───────────────────────────────────────────────────────────────\n");
            output.push_str("  DISTRIBUTED TRANSACTION\n");
            output.push_str("───────────────────────────────────────────────────────────────\n\n");
            output.push_str(&format!("Coordinator: {}\n", txn.coordinator_node));
            output.push_str(&format!("Participants: {}\n", txn.participant_nodes.join(", ")));
            output.push_str(&format!("Two-Phase Commit: {}\n", txn.two_phase_commit));
            output.push_str(&format!("Coordination Overhead: {:.2} ms\n", txn.coordination_overhead_ms));
            output.push_str(&format!("Network Round Trips: {}\n", txn.network_round_trips));
            output.push_str("\n");
        }

        // Summary
        output.push_str("───────────────────────────────────────────────────────────────\n");
        output.push_str("  SUMMARY\n");
        output.push_str("───────────────────────────────────────────────────────────────\n\n");
        output.push_str(&format!("Parallelism Degree: {}\n", explain.parallelism_degree));
        output.push_str(&format!("Total Network Cost: {:.2} ms\n", explain.total_network_cost_ms));
        output.push_str(&format!("Total Data Transfer: {:.2} MB\n", explain.total_data_transfer_mb));

        output.push_str("\n═══════════════════════════════════════════════════════════════\n");

        output
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    #[test]
    fn test_cluster_topology() {
        let analyzer = DistributedExplainAnalyzer::new(4);
        let result = analyzer.analyze("SELECT", &["users".to_string()], false, false).unwrap();

        assert_eq!(result.cluster_nodes.len(), 5); // 1 coordinator + 4 workers
        assert_eq!(result.parallelism_degree, 4);
    }

    #[test]
    fn test_partition_pruning() {
        let analyzer = DistributedExplainAnalyzer::new(4);
        let result = analyzer.analyze("SELECT", &["users".to_string()], false, false).unwrap();

        assert!(result.partition_pruning.is_some());
        let pruning = result.partition_pruning.unwrap();
        assert!(pruning.pruned_partitions > 0);
        assert!(pruning.pruning_efficiency > 0.0);
    }

    #[test]
    fn test_distributed_scan() {
        let analyzer = DistributedExplainAnalyzer::new(4);
        let result = analyzer.analyze("SELECT", &["users".to_string()], false, false).unwrap();

        assert!(!result.network_operations.is_empty());
        assert!(result.total_network_cost_ms > 0.0);
    }

    #[test]
    fn test_distributed_join() {
        let analyzer = DistributedExplainAnalyzer::new(4);
        let result = analyzer.analyze(
            "SELECT",
            &["users".to_string(), "orders".to_string()],
            true,
            false,
        ).unwrap();

        assert!(!result.cross_partition_joins.is_empty());
        assert!(!result.network_operations.is_empty());
    }

    #[test]
    fn test_distributed_transaction() {
        let analyzer = DistributedExplainAnalyzer::new(4);
        let result = analyzer.analyze("UPDATE", &["users".to_string()], false, false).unwrap();

        assert!(result.distributed_transaction.is_some());
        let txn = result.distributed_transaction.unwrap();
        assert!(txn.two_phase_commit);
        assert_eq!(txn.participant_nodes.len(), 4);
    }

    #[test]
    fn test_network_compression() {
        let analyzer = DistributedExplainAnalyzer::new(4).with_compression(true);
        let result = analyzer.analyze("SELECT", &["users".to_string()], false, false).unwrap();

        assert!(result.network_operations.iter().all(|op| op.compression_enabled));
    }

    #[test]
    fn test_format_output() {
        let analyzer = DistributedExplainAnalyzer::new(2);
        let result = analyzer.analyze("SELECT", &["users".to_string()], false, false).unwrap();
        let output = analyzer.format_output(&result);

        assert!(output.contains("DISTRIBUTED QUERY"));
        assert!(output.contains("CLUSTER TOPOLOGY"));
        assert!(output.contains("NETWORK OPERATIONS"));
    }
}