lake-pulse 0.1.0

// Copyright 2025 Adobe. All rights reserved.
// This file is licensed to you under the Apache License,
// Version 2.0 (http://www.apache.org/licenses/LICENSE-2.0)
// or the MIT license (http://opensource.org/licenses/MIT),
// at your option.
//
// Unless required by applicable law or agreed to in writing,
// this software is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR REPRESENTATIONS OF ANY KIND, either express or
// implied. See the LICENSE-MIT and LICENSE-APACHE files for the
// specific language governing permissions and limitations under
// each license.

use crate::analyze::common::{
    calculate_recommended_retention, calculate_retention_efficiency,
    calculate_schema_stability_score, calculate_storage_cost_impact, is_breaking_change,
    SchemaChange,
};
use crate::analyze::metrics::{
    ClusteringInfo, DeletionVectorMetrics, FileCompactionMetrics, HealthMetrics,
    SchemaEvolutionMetrics, TableConstraintsMetrics, TimeTravelMetrics,
};
use crate::analyze::table_analyzer::TableAnalyzer;
use crate::storage::{FileMetadata, StorageProvider};
use async_trait::async_trait;
use futures::stream::{self, StreamExt};
use serde_json::Value;
use std::error::Error;
use std::sync::Arc;
use std::time::SystemTime;
use tracing::info;

/// Intermediate structure to hold aggregated results from parallel metadata processing.
///
/// This structure accumulates metrics extracted from Iceberg metadata files
/// during parallel processing. It serves as a temporary container before the final
/// metrics are computed and stored in the `HealthMetrics` structure.
///
/// # Fields
///
/// * `partition_columns` - Columns used for table partitioning
/// * `total_snapshots` - Total number of table snapshots/versions
/// * `total_historical_size` - Total size of historical data in bytes
/// * `oldest_timestamp` - Timestamp of the oldest snapshot (milliseconds)
/// * `newest_timestamp` - Timestamp of the newest snapshot (milliseconds)
/// * `total_constraints` - Total number of table constraints
/// * `check_constraints` - Number of CHECK constraints
/// * `not_null_constraints` - Number of NOT NULL constraints
/// * `unique_constraints` - Number of UNIQUE constraints
/// * `foreign_key_constraints` - Number of FOREIGN KEY constraints
/// * `sort_order_columns` - Columns used for sorting/ordering
/// * `schema_changes` - List of all schema changes detected
/// * `delete_files_count` - Number of delete files (Iceberg's merge-on-read feature)
/// * `delete_files_size` - Total size of delete files in bytes
#[derive(Debug, Default)]
pub struct MetadataProcessingResult {
    pub partition_columns: Vec<String>,
    pub total_snapshots: usize,
    pub total_historical_size: u64,
    pub oldest_timestamp: u64,
    pub newest_timestamp: u64,
    pub total_constraints: usize,
    pub check_constraints: usize,
    pub not_null_constraints: usize,
    pub unique_constraints: usize,
    pub foreign_key_constraints: usize,
    pub sort_order_columns: Vec<String>,
    pub schema_changes: Vec<SchemaChange>,
    pub delete_files_count: u64,
    pub delete_files_size: u64,
}

/// Apache Iceberg-specific analyzer for processing Iceberg metadata.
///
/// This analyzer implements the `TableAnalyzer` trait and provides functionality
/// to parse Iceberg metadata files, extract metrics, and analyze table health.
/// It supports parallel processing of metadata files for improved performance.
///
/// # Fields
///
/// * `storage_provider` - The storage backend used to read files (S3, ADLS, local, etc.)
/// * `parallelism` - Number of concurrent tasks for parallel metadata processing
///
/// # Examples
///
/// ```no_run
/// use std::sync::Arc;
/// use lake_pulse::storage::StorageProvider;
/// use lake_pulse::analyze::iceberg::IcebergAnalyzer;
///
/// # async fn example(storage: Arc<dyn StorageProvider>) {
/// let analyzer = IcebergAnalyzer::new(storage, 4);
/// // Use analyzer to process Iceberg tables
/// # }
/// ```
pub struct IcebergAnalyzer {
    storage_provider: Arc<dyn StorageProvider>,
    parallelism: usize,
}

impl IcebergAnalyzer {
    /// Create a new IcebergAnalyzer.
    ///
    /// # Arguments
    ///
    /// * `storage_provider` - The storage provider to use for reading files
    /// * `parallelism` - The number of concurrent tasks to use for metadata processing
    ///
    /// # Returns
    ///
    /// A new `IcebergAnalyzer` instance configured with the specified storage provider and parallelism.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use std::sync::Arc;
    /// use lake_pulse::storage::StorageProvider;
    /// use lake_pulse::analyze::iceberg::IcebergAnalyzer;
    ///
    /// # async fn example(storage: Arc<dyn StorageProvider>) {
    /// let analyzer = IcebergAnalyzer::new(storage, 4);
    /// # }
    /// ```
    pub fn new(storage_provider: Arc<dyn StorageProvider>, parallelism: usize) -> Self {
        Self {
            storage_provider,
            parallelism,
        }
    }

    /// Categorize files into data files and Iceberg metadata files.
    ///
    /// Separates Parquet data files from Iceberg metadata files (metadata.json, manifest files).
    /// Data files are identified by `.parquet` extension, while metadata files include
    /// metadata.json and manifest files.
    ///
    /// # Arguments
    ///
    /// * `objects` - All files discovered in the table location
    ///
    /// # Returns
    ///
    /// A tuple of `(data_files, metadata_files)` where:
    /// * `data_files` - Vector of Parquet data files
    /// * `metadata_files` - Vector of Iceberg metadata and manifest files
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use lake_pulse::analyze::iceberg::IcebergAnalyzer;
    /// # use lake_pulse::storage::FileMetadata;
    /// # use std::sync::Arc;
    /// # fn example(analyzer: &IcebergAnalyzer, files: Vec<FileMetadata>) {
    /// let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);
    /// println!("Found {} data files and {} metadata files",
    ///          data_files.len(), metadata_files.len());
    /// # }
    /// ```
    pub fn categorize_iceberg_files(
        &self,
        objects: Vec<FileMetadata>,
    ) -> (Vec<FileMetadata>, Vec<FileMetadata>) {
        let mut data_files = Vec::new();
        let mut metadata_files = Vec::new();

        for obj in objects {
            if obj.path.ends_with(".parquet") {
                data_files.push(obj);
            } else if obj.path.contains("metadata.json") || obj.path.contains("manifest") {
                metadata_files.push(obj);
            }
        }

        (data_files, metadata_files)
    }

    /// Find referenced files from Iceberg metadata.
    ///
    /// Parses Iceberg metadata.json and manifest files to extract all file paths
    /// referenced in the current table snapshot. This identifies which data files
    /// are currently part of the table's active state.
    ///
    /// # Arguments
    ///
    /// * `metadata_files` - The metadata files to parse
    ///
    /// # Returns
    ///
    /// A `Result` containing:
    /// * `Ok(Vec<String>)` - Vector of file paths referenced in the Iceberg metadata
    /// * `Err` - If file reading or JSON parsing fails
    ///
    /// # Errors
    ///
    /// This function will return an error if:
    /// * Any metadata file cannot be read from storage
    /// * JSON parsing of metadata or manifest files fails
    /// * File content is not valid UTF-8
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use lake_pulse::analyze::iceberg::IcebergAnalyzer;
    /// # use lake_pulse::storage::FileMetadata;
    /// # async fn example(analyzer: &IcebergAnalyzer, metadata: &Vec<FileMetadata>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    /// let referenced_files = analyzer.find_referenced_files(metadata).await?;
    /// println!("Found {} referenced data files", referenced_files.len());
    /// # Ok(())
    /// # }
    /// ```
    pub async fn find_referenced_files(
        &self,
        metadata_files: &[FileMetadata],
    ) -> Result<Vec<String>, Box<dyn Error + Send + Sync>> {
        let storage_provider = Arc::clone(&self.storage_provider);

        // Find the current metadata.json file (usually the one with highest version or named metadata.json)
        let metadata_json_files: Vec<_> = metadata_files
            .iter()
            .filter(|f| {
                f.path.ends_with("metadata.json")
                    || f.path.contains("v") && f.path.ends_with(".json")
            })
            .cloned()
            .collect();

        // Process the most recent metadata file
        let Some(metadata_file) = metadata_json_files.last() else {
            info!("No Iceberg metadata.json files found");
            return Ok(Vec::new());
        };
        info!("Processing metadata file={}", &metadata_file.path);

        let content = storage_provider
            .read_file(&metadata_file.path)
            .await
            .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)?;

        let content_str = String::from_utf8_lossy(&content);
        let json: Value = serde_json::from_str(&content_str)
            .map_err(|e| Box::new(e) as Box<dyn Error + Send + Sync>)?;

        let mut referenced_files = Vec::new();

        // Get manifest list from current snapshot
        let manifest_list_paths = self.get_manifest_list(&json).await?;

        // For each manifest in the manifest list, extract data file paths
        for manifest_path in &manifest_list_paths {
            info!("Reading manifest file: {}", manifest_path);

            match storage_provider.read_file(manifest_path).await {
                Ok(manifest_content) => {
                    // Try to parse as JSON (some manifests might be JSON)
                    if let Ok(manifest_str) = String::from_utf8(manifest_content.clone()) {
                        if let Ok(manifest_json) = serde_json::from_str::<Value>(&manifest_str) {
                            // Extract data file paths from manifest entries
                            if let Some(entries) =
                                manifest_json.get("entries").and_then(|e| e.as_array())
                            {
                                for entry in entries {
                                    if let Some(data_file) = entry.get("data-file") {
                                        if let Some(file_path) =
                                            data_file.get("file-path").and_then(|p| p.as_str())
                                        {
                                            referenced_files.push(file_path.to_string());
                                        }
                                    }
                                }
                            }
                        }
                    }
                    // Note: Iceberg manifests are typically Avro files, not JSON
                    // A full implementation would use an Avro parser here
                    // For now, we're doing a best-effort JSON parse
                }
                Err(e) => {
                    info!("Could not read manifest file {}: {}", manifest_path, e);
                }
            }
        }

        info!(
            "Found {} referenced files from manifests",
            referenced_files.len()
        );
        Ok(referenced_files)
    }

    /// Get manifest list paths from metadata JSON
    async fn get_manifest_list(
        &self,
        metadata: &Value,
    ) -> Result<Vec<String>, Box<dyn Error + Send + Sync>> {
        let mut manifest_list = Vec::new();

        // Get current snapshot ID
        if let Some(current_snapshot_id) = metadata
            .get("current-snapshot-id")
            .and_then(|id| id.as_i64())
        {
            // Find the snapshot with this ID
            if let Some(snapshots) = metadata.get("snapshots").and_then(|s| s.as_array()) {
                for snapshot in snapshots {
                    if let Some(snapshot_id) =
                        snapshot.get("snapshot-id").and_then(|id| id.as_i64())
                    {
                        if snapshot_id == current_snapshot_id {
                            // Get manifest list path
                            if let Some(manifest_list_path) =
                                snapshot.get("manifest-list").and_then(|p| p.as_str())
                            {
                                manifest_list.push(manifest_list_path.to_string());
                            }
                            break;
                        }
                    }
                }
            }
        }

        Ok(manifest_list)
    }

    /// Process a single Iceberg metadata file and extract metrics
    pub async fn process_single_metadata_file(
        &self,
        file_path: &str,
        file_index: usize,
    ) -> Result<MetadataProcessingResult, Box<dyn Error + Send + Sync>> {
        let content = self.storage_provider.read_file(file_path).await?;
        let content_str = String::from_utf8_lossy(&content);

        let mut result = MetadataProcessingResult {
            oldest_timestamp: chrono::Utc::now().timestamp() as u64,
            ..Default::default()
        };

        let json: Value = serde_json::from_str(&content_str)?;

        let current_version = file_index as u64;

        // Extract partition spec
        if let Some(partition_specs) = json.get("partition-specs").and_then(|s| s.as_array()) {
            for spec in partition_specs {
                if let Some(fields) = spec.get("fields").and_then(|f| f.as_array()) {
                    let partition_columns: Vec<String> = fields
                        .iter()
                        .filter_map(|f| {
                            f.get("name")
                                .and_then(|n| n.as_str())
                                .map(|s| s.to_string())
                        })
                        .collect();
                    if !partition_columns.is_empty() && result.partition_columns.is_empty() {
                        result.partition_columns = partition_columns;
                    }
                }
            }
        }

        // Extract sort order
        if let Some(sort_orders) = json.get("sort-orders").and_then(|s| s.as_array()) {
            for order in sort_orders {
                if let Some(fields) = order.get("fields").and_then(|f| f.as_array()) {
                    let sort_columns: Vec<String> = fields
                        .iter()
                        .filter_map(|f| {
                            f.get("source-id")
                                .and_then(|_| f.get("transform").and_then(|t| t.as_str()))
                                .map(|s| s.to_string())
                        })
                        .collect();
                    if !sort_columns.is_empty() && result.sort_order_columns.is_empty() {
                        result.sort_order_columns = sort_columns;
                    }
                }
            }
        }

        // Extract snapshot metrics
        self.extract_snapshot_metrics(&json, &mut result);

        // Extract schema and constraints
        self.extract_schema_and_constraints(&json, &mut result, current_version);

        Ok(result)
    }

    /// Extract snapshot/time travel metrics from Iceberg metadata.
    ///
    /// Parses snapshot information from Iceberg metadata to collect timestamp data,
    /// estimate historical data size, and extract delete file metrics (for Iceberg v2+).
    ///
    /// # Arguments
    ///
    /// * `json` - The Iceberg metadata JSON
    /// * `result` - The result object to update (mutated in place)
    fn extract_snapshot_metrics(&self, json: &Value, result: &mut MetadataProcessingResult) {
        if let Some(snapshots) = json.get("snapshots").and_then(|s| s.as_array()) {
            result.total_snapshots = snapshots.len();

            for snapshot in snapshots {
                if let Some(timestamp_ms) = snapshot.get("timestamp-ms").and_then(|t| t.as_u64()) {
                    result.oldest_timestamp = result.oldest_timestamp.min(timestamp_ms);
                    result.newest_timestamp = result.newest_timestamp.max(timestamp_ms);
                }

                // Estimate snapshot size from summary
                if let Some(summary) = snapshot.get("summary").and_then(|s| s.as_object()) {
                    if let Some(total_data_files) =
                        summary.get("total-data-files").and_then(|v| v.as_str())
                    {
                        // Rough estimate based on file count
                        if let Ok(file_count) = total_data_files.parse::<u64>() {
                            result.total_historical_size += file_count * 1024 * 1024;
                            // Estimate 1MB per file
                        }
                    }

                    // Extract delete file metrics (Iceberg v2+)
                    if let Some(total_delete_files) =
                        summary.get("total-delete-files").and_then(|v| v.as_str())
                    {
                        if let Ok(delete_count) = total_delete_files.parse::<u64>() {
                            result.delete_files_count += delete_count;
                        }
                    }

                    if let Some(total_position_deletes) = summary
                        .get("total-position-deletes")
                        .and_then(|v| v.as_str())
                    {
                        if let Ok(position_deletes) = total_position_deletes.parse::<u64>() {
                            result.delete_files_count += position_deletes;
                        }
                    }

                    if let Some(total_equality_deletes) = summary
                        .get("total-equality-deletes")
                        .and_then(|v| v.as_str())
                    {
                        if let Ok(equality_deletes) = total_equality_deletes.parse::<u64>() {
                            result.delete_files_count += equality_deletes;
                        }
                    }
                }
            }
        }
    }

    /// Extract schema and constraints from Iceberg metadata.
    ///
    /// Parses schema definitions to identify and count various types of constraints
    /// and detect schema changes over time.
    ///
    /// # Arguments
    ///
    /// * `json` - The Iceberg metadata JSON
    /// * `result` - The result object to update (mutated in place)
    /// * `current_version` - The current version of the schema
    fn extract_schema_and_constraints(
        &self,
        json: &Value,
        result: &mut MetadataProcessingResult,
        current_version: u64,
    ) {
        if let Some(schemas) = json.get("schemas").and_then(|s| s.as_array()) {
            for (idx, schema) in schemas.iter().enumerate() {
                if let Some(fields) = schema.get("fields").and_then(|f| f.as_array()) {
                    let constraints = self.extract_constraints_from_schema(fields);
                    result.total_constraints += constraints.0;
                    result.check_constraints += constraints.1;
                    result.not_null_constraints += constraints.2;
                    result.unique_constraints += constraints.3;
                    result.foreign_key_constraints += constraints.4;

                    // Track schema changes
                    let breaking = if idx > 0 {
                        is_breaking_change(&result.schema_changes, schema)
                    } else {
                        false
                    };

                    result.schema_changes.push(SchemaChange::new(
                        current_version + idx as u64,
                        0, // Iceberg doesn't track schema change timestamps directly
                        schema.clone(),
                        breaking,
                    ));
                }
            }
        }
    }

    /// Update health metrics from Iceberg metadata files.
    ///
    /// Main entry point for extracting comprehensive health metrics from Iceberg
    /// metadata. Processes all metadata files in parallel, aggregates results, and
    /// populates the HealthMetrics structure with delete file metrics, schema evolution,
    /// time travel, constraint, partition, and compaction metrics.
    ///
    /// # Arguments
    ///
    /// * `metadata_files` - The metadata files to analyze
    /// * `data_files_total_size` - Total size of all data files in bytes
    /// * `data_files_total_files` - Total number of data files
    /// * `metrics` - The metrics object to update (mutated in place)
    ///
    /// # Returns
    ///
    /// A `Result` indicating success or failure of the metrics extraction.
    ///
    /// # Errors
    ///
    /// This function will return an error if:
    /// * Any metadata file cannot be read or parsed
    /// * Schema metric calculation fails
    /// * Parallel processing encounters errors
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use lake_pulse::analyze::iceberg::IcebergAnalyzer;
    /// # use lake_pulse::analyze::metrics::HealthMetrics;
    /// # use lake_pulse::storage::FileMetadata;
    /// # async fn example(analyzer: &IcebergAnalyzer, metadata: &Vec<FileMetadata>, mut metrics: HealthMetrics) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    /// analyzer.update_metrics_from_iceberg_metadata(
    ///     metadata,
    ///     1024 * 1024 * 1024, // 1GB total data size
    ///     100,                 // 100 data files
    ///     &mut metrics
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn update_metrics_from_iceberg_metadata(
        &self,
        metadata_files: &[FileMetadata],
        data_files_total_size: u64,
        data_files_total_files: usize,
        metrics: &mut HealthMetrics,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        let parallelism = self.parallelism.max(1);

        // Filter to only process actual metadata JSON files, not manifest files or CRC files
        let metadata_json_files: Vec<_> = metadata_files
            .iter()
            .filter(|f| {
                (f.path.ends_with("metadata.json")
                    || f.path.contains("/v") && f.path.ends_with(".json"))
                    && !f.path.ends_with(".crc")
            })
            .cloned()
            .collect();

        if metadata_json_files.is_empty() {
            info!("No Iceberg metadata JSON files found to process");
            return Ok(());
        }

        let min_file_name = metadata_json_files
            .iter()
            .min_by_key(|f| f.path.clone())
            .map(|f| f.path.clone());
        let max_file_name = metadata_json_files
            .iter()
            .max_by_key(|f| f.path.clone())
            .map(|f| f.path.clone());
        metrics.metadata_health.first_file_name = min_file_name.clone();
        metrics.metadata_health.last_file_name = max_file_name.clone();

        info!(
            "Updating the metrics from Iceberg metadata files, parallelism={}, metadata_files_count={}, \
            data_files_count={}, min_file={:?}, max_file={:?}",
            parallelism,
            metadata_json_files.len(),
            data_files_total_files,
            min_file_name,
            max_file_name
        );

        let update_metrics_start = SystemTime::now();
        let metadata_files_owned: Vec<_> =
            metadata_json_files.iter().cloned().enumerate().collect();
        let results: Vec<Result<MetadataProcessingResult, Box<dyn Error + Send + Sync>>> =
            stream::iter(metadata_files_owned)
                .map(|(index, metadata_file)| {
                    info!(
                        "Starting processing Iceberg metadata file={}, index={}",
                        &metadata_file.path.clone(),
                        index
                    );
                    let path = metadata_file.path.clone();
                    let analyzer = IcebergAnalyzer {
                        storage_provider: Arc::clone(&self.storage_provider),
                        parallelism,
                    };

                    let result = async move {
                        let process_single_start = SystemTime::now();
                        let r = analyzer.process_single_metadata_file(&path, index).await;
                        info!(
                            "Processed Iceberg metadata file={} in async, took={}",
                            &path,
                            process_single_start
                                .elapsed()
                                .unwrap_or_default()
                                .as_millis(),
                        );
                        r
                    };
                    result
                })
                .buffer_unordered(parallelism)
                .collect()
                .await;

        info!(
            "Finished the metrics from Iceberg metadata files, parallelism={}, metadata_files_count={}, \
            data_files_count={}, min_file={:?}, max_file={:?}, took={}",
            parallelism,
            metadata_files.len(),
            data_files_total_files,
            min_file_name,
            max_file_name,
            update_metrics_start
                .elapsed()
                .unwrap_or_default()
                .as_millis(),
        );

        // Aggregate results from all metadata files
        let mut partition_columns: Vec<String> = vec![];
        let mut total_snapshots = 0;
        let mut total_historical_size = 0u64;
        let mut oldest_timestamp = chrono::Utc::now().timestamp() as u64;
        let mut newest_timestamp = 0u64;
        let mut total_constraints = 0;
        let mut check_constraints = 0;
        let mut not_null_constraints = 0;
        let mut unique_constraints = 0;
        let mut foreign_key_constraints = 0;
        let mut sort_order_columns: Vec<String> = vec![];
        let mut schema_changes = Vec::new();
        let mut current_version = 0u64;
        let mut delete_files_count = 0u64;

        for result in results {
            let r = result?;

            if partition_columns.is_empty() && !r.partition_columns.is_empty() {
                partition_columns = r.partition_columns;
            }

            total_snapshots += r.total_snapshots;
            total_historical_size += r.total_historical_size;
            oldest_timestamp = oldest_timestamp.min(r.oldest_timestamp);
            newest_timestamp = newest_timestamp.max(r.newest_timestamp);

            total_constraints += r.total_constraints;
            check_constraints += r.check_constraints;
            not_null_constraints += r.not_null_constraints;
            unique_constraints += r.unique_constraints;
            foreign_key_constraints += r.foreign_key_constraints;
            delete_files_count += r.delete_files_count;

            if sort_order_columns.is_empty() {
                sort_order_columns = r.sort_order_columns;
            }

            schema_changes.extend(r.schema_changes);
            current_version = current_version.max(schema_changes.len() as u64);
        }

        // Set schema evolution metrics
        if !schema_changes.is_empty() {
            metrics.schema_evolution =
                self.calculate_schema_metrics(schema_changes, current_version)?;
        }

        // Set deletion vector metrics (delete files in Iceberg)
        if delete_files_count > 0 {
            let deletion_vector_metrics = self.calculate_deletion_vector_metrics(
                delete_files_count,
                oldest_timestamp,
                newest_timestamp,
            );
            metrics.deletion_vector_metrics = Some(deletion_vector_metrics);
        }

        // Set time travel metrics
        if total_snapshots != 0 {
            let now = chrono::Utc::now().timestamp() as u64;
            let oldest_age_days = (now - oldest_timestamp / 1000) as f64 / 86400.0;
            let newest_age_days = (now - newest_timestamp / 1000) as f64 / 86400.0;
            let avg_snapshot_size = total_historical_size as f64 / total_snapshots as f64;

            let storage_cost_impact = calculate_storage_cost_impact(
                total_historical_size,
                total_snapshots,
                oldest_age_days,
            );
            let retention_efficiency =
                calculate_retention_efficiency(total_snapshots, oldest_age_days, newest_age_days);
            let recommended_retention =
                calculate_recommended_retention(total_snapshots, oldest_age_days);

            metrics.time_travel_metrics = Some(TimeTravelMetrics {
                total_snapshots,
                oldest_snapshot_age_days: oldest_age_days,
                newest_snapshot_age_days: newest_age_days,
                total_historical_size_bytes: total_historical_size,
                avg_snapshot_size_bytes: avg_snapshot_size,
                storage_cost_impact_score: storage_cost_impact,
                retention_efficiency_score: retention_efficiency,
                recommended_retention_days: recommended_retention,
            });
        }

        // Set table constraints metrics
        if total_constraints != 0 {
            let constraint_violation_risk =
                self.calculate_constraint_violation_risk(total_constraints, check_constraints);
            let data_quality_score =
                self.calculate_data_quality_score(total_constraints, constraint_violation_risk);
            let constraint_coverage_score =
                self.calculate_constraint_coverage_score(total_constraints, check_constraints);

            metrics.table_constraints = Some(TableConstraintsMetrics {
                total_constraints,
                check_constraints,
                not_null_constraints,
                unique_constraints,
                foreign_key_constraints,
                constraint_violation_risk,
                data_quality_score,
                constraint_coverage_score,
            });
        }

        // For Iceberg, use partition columns for clustering
        let partition_count = metrics.partitions.len();
        let cluster_count = partition_count.max(1);
        let avg_files_per_cluster = if cluster_count > 0 {
            data_files_total_files as f64 / cluster_count as f64
        } else {
            0.0
        };

        let avg_cluster_size_bytes = if cluster_count > 0 {
            data_files_total_size as f64 / cluster_count as f64
        } else {
            0.0
        };

        metrics.clustering = Some(ClusteringInfo {
            clustering_columns: partition_columns.to_vec(),
            cluster_count,
            avg_files_per_cluster,
            avg_cluster_size_bytes,
        });

        // Calculate file compaction metrics
        let file_compaction = self.calculate_file_compaction_metrics(
            data_files_total_files,
            data_files_total_size,
            &sort_order_columns,
        );
        metrics.file_compaction = Some(file_compaction);

        Ok(())
    }

    /// Calculate file compaction metrics
    fn calculate_file_compaction_metrics(
        &self,
        total_files: usize,
        total_size: u64,
        sort_order_columns: &[String],
    ) -> FileCompactionMetrics {
        // Estimate small files (< 16MB) based on average file size
        let avg_file_size = if total_files > 0 {
            total_size as f64 / total_files as f64
        } else {
            0.0
        };

        let small_file_threshold = 16 * 1024 * 1024; // 16MB
        let small_files_count = if avg_file_size < small_file_threshold as f64 {
            (total_files as f64 * 0.7).ceil() as usize // Estimate 70% are small
        } else {
            (total_files as f64 * 0.2).ceil() as usize // Estimate 20% are small
        };

        let small_files_size =
            (small_files_count as f64 * avg_file_size.min(small_file_threshold as f64)) as u64;

        // Calculate potential savings
        let target_file_size = 128 * 1024 * 1024; // 128MB target
        let estimated_savings = if small_files_count > 1 {
            let files_per_target =
                (target_file_size as f64 / avg_file_size.max(1.0)).ceil() as usize;
            let target_files = (small_files_count as f64 / files_per_target as f64).ceil() as usize;
            let estimated_target_size = target_files as u64 * target_file_size / 2;
            small_files_size.saturating_sub(estimated_target_size)
        } else {
            0
        };

        let compaction_opportunity =
            self.calculate_compaction_opportunity(small_files_count, total_files);

        let compaction_priority =
            self.calculate_compaction_priority(compaction_opportunity, small_files_count);

        FileCompactionMetrics {
            compaction_opportunity_score: compaction_opportunity,
            small_files_count,
            small_files_size_bytes: small_files_size,
            potential_compaction_files: small_files_count,
            estimated_compaction_savings_bytes: estimated_savings,
            recommended_target_file_size_bytes: target_file_size,
            compaction_priority,
            z_order_opportunity: !sort_order_columns.is_empty(),
            z_order_columns: sort_order_columns.to_vec(),
        }
    }

    /// Calculate compaction opportunity score
    fn calculate_compaction_opportunity(&self, small_files: usize, total_files: usize) -> f64 {
        if total_files == 0 {
            return 0.0;
        }

        let small_file_ratio = small_files as f64 / total_files as f64;

        if small_file_ratio > 0.8 {
            1.0
        } else if small_file_ratio > 0.6 {
            0.8
        } else if small_file_ratio > 0.4 {
            0.6
        } else if small_file_ratio > 0.2 {
            0.4
        } else {
            0.2
        }
    }

    /// Calculate compaction priority
    fn calculate_compaction_priority(&self, opportunity_score: f64, small_files: usize) -> String {
        if opportunity_score > 0.8 || small_files > 100 {
            "critical".to_string()
        } else if opportunity_score > 0.6 || small_files > 50 {
            "high".to_string()
        } else if opportunity_score > 0.4 || small_files > 20 {
            "medium".to_string()
        } else {
            "low".to_string()
        }
    }

    /// Calculate deletion vector metrics from delete files
    fn calculate_deletion_vector_metrics(
        &self,
        delete_files_count: u64,
        oldest_timestamp: u64,
        _newest_timestamp: u64,
    ) -> DeletionVectorMetrics {
        // Estimate deletion vector size (rough estimate: 1KB per delete file)
        let total_dv_size = delete_files_count * 1024;
        let avg_dv_size = if delete_files_count > 0 {
            total_dv_size as f64 / delete_files_count as f64
        } else {
            0.0
        };

        // Calculate age in days
        let now = chrono::Utc::now().timestamp() as u64;
        let oldest_age_days = if oldest_timestamp > 0 {
            (now - oldest_timestamp / 1000) as f64 / 86400.0
        } else {
            0.0
        };

        // Estimate deleted rows (rough estimate: 100 rows per delete file)
        let deleted_rows = delete_files_count * 100;

        // Calculate impact score
        let impact_score =
            self.calculate_deletion_impact_score(delete_files_count, oldest_age_days);

        DeletionVectorMetrics {
            deletion_vector_count: delete_files_count as usize,
            total_deletion_vector_size_bytes: total_dv_size,
            avg_deletion_vector_size_bytes: avg_dv_size,
            deletion_vector_age_days: oldest_age_days,
            deleted_rows_count: deleted_rows,
            deletion_vector_impact_score: impact_score,
        }
    }

    /// Calculate deletion impact score
    fn calculate_deletion_impact_score(&self, delete_count: u64, age_days: f64) -> f64 {
        let mut score: f64 = 0.0;

        // More delete files = higher impact
        if delete_count > 1000 {
            score += 0.5;
        } else if delete_count > 500 {
            score += 0.4;
        } else if delete_count > 100 {
            score += 0.3;
        } else if delete_count > 10 {
            score += 0.2;
        } else {
            score += 0.1;
        }

        // Older delete files = higher impact
        if age_days > 90.0 {
            score += 0.5;
        } else if age_days > 30.0 {
            score += 0.3;
        } else if age_days > 7.0 {
            score += 0.2;
        } else {
            score += 0.1;
        }

        score.min(1.0)
    }

    /// Calculate schema metrics from schema changes
    fn calculate_schema_metrics(
        &self,
        changes: Vec<SchemaChange>,
        current_version: u64,
    ) -> Result<Option<SchemaEvolutionMetrics>, Box<dyn Error + Send + Sync>> {
        let total_changes = changes.len();
        let breaking_changes = changes.iter().filter(|c| c.is_breaking).count();
        let non_breaking_changes = total_changes - breaking_changes;

        // Calculate time-based metrics
        let now = chrono::Utc::now().timestamp() as u64;
        let days_since_last = if let Some(last_change) = changes.last() {
            if last_change.timestamp > 0 {
                (now - last_change.timestamp / 1000) as f64 / 86400.0
            } else {
                365.0 // No timestamp available
            }
        } else {
            365.0 // No changes in a year = very stable
        };

        // Calculate change frequency (changes per day)
        let total_days = match (changes.first(), changes.last()) {
            (Some(first), Some(last)) if changes.len() > 1 => {
                let first_change = first.timestamp / 1000;
                let last_change = last.timestamp / 1000;
                if first_change > 0 && last_change > 0 {
                    ((last_change - first_change) as f64 / 86400.0).max(1.0_f64)
                } else {
                    1.0
                }
            }
            _ => 1.0,
        };

        let change_frequency = total_changes as f64 / total_days;

        // Calculate stability score using shared function
        let stability_score = calculate_schema_stability_score(
            total_changes,
            breaking_changes,
            change_frequency,
            days_since_last,
        );

        Ok(Some(SchemaEvolutionMetrics {
            total_schema_changes: total_changes,
            breaking_changes,
            non_breaking_changes,
            schema_stability_score: stability_score,
            days_since_last_change: days_since_last,
            schema_change_frequency: change_frequency,
            current_schema_version: current_version,
        }))
    }

    /// Extract constraints from Iceberg schema fields
    fn extract_constraints_from_schema(
        &self,
        fields: &Vec<Value>,
    ) -> (usize, usize, usize, usize, usize) {
        let mut total = 0;
        let mut check = 0;
        let mut not_null = 0;
        let mut unique = 0;
        let mut foreign_key = 0;

        for field in fields {
            total += 1;

            // Check for NOT NULL constraint (required field in Iceberg)
            if let Some(required) = field.get("required") {
                if required.as_bool().unwrap_or(false) {
                    not_null += 1;
                }
            }

            // Check for other constraints in field metadata
            if let Some(metadata) = field.get("metadata").and_then(|m| m.as_object()) {
                for (key, _value) in metadata {
                    let key_lower = key.to_lowercase();
                    if key_lower.contains("constraint") || key_lower.contains("check") {
                        check += 1;
                    }
                    if key_lower.contains("unique") {
                        unique += 1;
                    }
                    if key_lower.contains("foreign") || key_lower.contains("reference") {
                        foreign_key += 1;
                    }
                }
            }

            // Also check field doc strings
            if let Some(doc) = field.get("doc").and_then(|d| d.as_str()) {
                let doc_lower = doc.to_lowercase();
                if doc_lower.contains("constraint") || doc_lower.contains("check") {
                    check += 1;
                }
                if doc_lower.contains("unique") {
                    unique += 1;
                }
                if doc_lower.contains("foreign") || doc_lower.contains("reference") {
                    foreign_key += 1;
                }
            }
        }

        (total, check, not_null, unique, foreign_key)
    }

    /// Calculate constraint violation risk
    fn calculate_constraint_violation_risk(
        &self,
        total_constraints: usize,
        check_constraints: usize,
    ) -> f64 {
        if total_constraints == 0 {
            return 0.0;
        }

        // Higher risk with more complex constraints
        let complexity_ratio = check_constraints as f64 / total_constraints as f64;
        if complexity_ratio > 0.5 {
            0.8
        } else if complexity_ratio > 0.3 {
            0.6
        } else if complexity_ratio > 0.1 {
            0.4
        } else {
            0.2
        }
    }

    /// Calculate data quality score
    fn calculate_data_quality_score(&self, total_constraints: usize, violation_risk: f64) -> f64 {
        let mut score = 1.0;

        // Reward having constraints
        if total_constraints > 10 {
            score += 0.2;
        } else if total_constraints > 5 {
            score += 0.1;
        }

        // Penalize violation risk
        score -= violation_risk * 0.5;

        score.clamp(0.0_f64, 1.0_f64)
    }

    /// Calculate constraint coverage score
    fn calculate_constraint_coverage_score(
        &self,
        total_constraints: usize,
        check_constraints: usize,
    ) -> f64 {
        if total_constraints == 0 {
            return 0.0;
        }

        let coverage_ratio = check_constraints as f64 / total_constraints as f64;
        if coverage_ratio > 0.5 {
            1.0
        } else if coverage_ratio > 0.3 {
            0.7
        } else if coverage_ratio > 0.1 {
            0.4
        } else {
            0.1
        }
    }
}

// Implement the TableAnalyzer trait for IcebergAnalyzer
#[async_trait]
impl TableAnalyzer for IcebergAnalyzer {
    fn categorize_files(
        &self,
        objects: Vec<FileMetadata>,
    ) -> (Vec<FileMetadata>, Vec<FileMetadata>) {
        self.categorize_iceberg_files(objects)
    }

    async fn find_referenced_files(
        &self,
        metadata_files: &[FileMetadata],
    ) -> Result<Vec<String>, Box<dyn Error + Send + Sync>> {
        self.find_referenced_files(metadata_files).await
    }

    async fn update_metrics_from_metadata(
        &self,
        metadata_files: &[FileMetadata],
        data_files_total_size: u64,
        data_files_total_files: usize,
        metrics: &mut HealthMetrics,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        self.update_metrics_from_iceberg_metadata(
            metadata_files,
            data_files_total_size,
            data_files_total_files,
            metrics,
        )
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::analyze::common::detect_breaking_schema_changes;
    use crate::storage::error::StorageResult;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::OnceLock;

    // Mock storage provider for testing
    struct MockStorageProvider {
        options: OnceLock<HashMap<String, String>>,
        test_data: Option<Vec<u8>>,
    }

    impl MockStorageProvider {
        fn new() -> Self {
            Self {
                options: OnceLock::new(),
                test_data: None,
            }
        }

        fn with_data(data: Vec<u8>) -> Self {
            Self {
                options: OnceLock::new(),
                test_data: Some(data),
            }
        }
    }

    #[async_trait]
    impl StorageProvider for MockStorageProvider {
        fn base_path(&self) -> &str {
            "/mock/base/path"
        }

        async fn validate_connection(&self, _path: &str) -> StorageResult<()> {
            Ok(())
        }

        async fn list_files(
            &self,
            _path: &str,
            _recursive: bool,
        ) -> StorageResult<Vec<FileMetadata>> {
            Ok(vec![])
        }

        async fn discover_partitions(
            &self,
            _path: &str,
            _exclude_prefixes: Vec<&str>,
        ) -> StorageResult<Vec<String>> {
            Ok(vec![])
        }

        async fn list_files_parallel(
            &self,
            _path: &str,
            _partitions: Vec<String>,
            _parallelism: usize,
        ) -> StorageResult<Vec<FileMetadata>> {
            Ok(vec![])
        }

        async fn read_file(&self, _path: &str) -> StorageResult<Vec<u8>> {
            if let Some(ref data) = self.test_data {
                Ok(data.clone())
            } else {
                Ok(vec![])
            }
        }

        async fn exists(&self, _path: &str) -> StorageResult<bool> {
            Ok(true)
        }

        async fn get_metadata(&self, _path: &str) -> StorageResult<FileMetadata> {
            Ok(FileMetadata {
                path: "test".to_string(),
                size: 0,
                last_modified: None,
            })
        }

        fn options(&self) -> &HashMap<String, String> {
            self.options.get_or_init(HashMap::new)
        }

        fn clean_options(&self) -> HashMap<String, String> {
            HashMap::new()
        }

        fn uri_from_path(&self, path: &str) -> String {
            path.to_string()
        }
    }

    // Helper function to create a test IcebergAnalyzer
    fn create_test_analyzer() -> IcebergAnalyzer {
        let mock_storage = Arc::new(MockStorageProvider::new());
        IcebergAnalyzer::new(mock_storage, 4)
    }

    #[test]
    fn test_iceberg_analyzer_new() {
        let mock_storage = Arc::new(MockStorageProvider::new());
        let analyzer = IcebergAnalyzer::new(mock_storage, 8);

        assert_eq!(analyzer.parallelism, 8);
    }

    #[test]
    fn test_categorize_iceberg_files_data_only() {
        let analyzer = create_test_analyzer();
        let files = vec![
            FileMetadata {
                path: "table/data/part-00000.parquet".to_string(),
                size: 1024,
                last_modified: None,
            },
            FileMetadata {
                path: "table/data/part-00001.parquet".to_string(),
                size: 2048,
                last_modified: None,
            },
        ];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        assert_eq!(data_files.len(), 2);
        assert_eq!(metadata_files.len(), 0);
    }

    #[test]
    fn test_categorize_iceberg_files_metadata_only() {
        let analyzer = create_test_analyzer();
        let files = vec![FileMetadata {
            path: "table/metadata/v1.metadata.json".to_string(),
            size: 512,
            last_modified: None,
        }];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        assert_eq!(data_files.len(), 0);
        assert_eq!(metadata_files.len(), 1);
    }

    #[test]
    fn test_categorize_iceberg_files_mixed() {
        let analyzer = create_test_analyzer();
        let files = vec![
            FileMetadata {
                path: "table/data/part-00000.parquet".to_string(),
                size: 1024,
                last_modified: None,
            },
            FileMetadata {
                path: "table/metadata/v1.metadata.json".to_string(),
                size: 512,
                last_modified: None,
            },
            FileMetadata {
                path: "table/data/part-00001.parquet".to_string(),
                size: 2048,
                last_modified: None,
            },
            FileMetadata {
                path: "table/metadata/manifest-list.avro".to_string(),
                size: 256,
                last_modified: None,
            },
        ];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        assert_eq!(data_files.len(), 2);
        assert_eq!(metadata_files.len(), 2);
    }

    #[test]
    fn test_categorize_iceberg_files_empty() {
        let analyzer = create_test_analyzer();
        let files = vec![];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        assert_eq!(data_files.len(), 0);
        assert_eq!(metadata_files.len(), 0);
    }

    #[test]
    fn test_categorize_iceberg_files_non_parquet() {
        let analyzer = create_test_analyzer();
        let files = vec![
            FileMetadata {
                path: "table/data.csv".to_string(),
                size: 1024,
                last_modified: None,
            },
            FileMetadata {
                path: "table/README.md".to_string(),
                size: 512,
                last_modified: None,
            },
        ];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        // Non-parquet, non-metadata files should be ignored
        assert_eq!(data_files.len(), 0);
        assert_eq!(metadata_files.len(), 0);
    }

    #[test]
    fn test_categorize_iceberg_files_manifest_files() {
        let analyzer = create_test_analyzer();
        let files = vec![FileMetadata {
            path: "table/metadata/manifest-list.avro".to_string(),
            size: 256,
            last_modified: None,
        }];

        let (data_files, metadata_files) = analyzer.categorize_iceberg_files(files);

        assert_eq!(data_files.len(), 0);
        assert_eq!(metadata_files.len(), 1);
        assert!(metadata_files.iter().all(|f| f.path.contains("manifest")));
    }

    #[test]
    fn test_schema_change_creation() {
        let schema = json!({
            "type": "struct",
            "fields": [
                {"name": "id", "type": "long", "required": true}
            ]
        });

        let change = SchemaChange {
            version: 1,
            timestamp: 1234567890,
            schema: schema.clone(),
            is_breaking: false,
        };

        assert_eq!(change.version, 1);
        assert_eq!(change.timestamp, 1234567890);
        assert!(!change.is_breaking);
        assert_eq!(change.schema, schema);
    }

    #[test]
    fn test_schema_change_clone() {
        let schema = json!({"type": "struct", "fields": []});
        let change = SchemaChange {
            version: 1,
            timestamp: 1234567890,
            schema: schema.clone(),
            is_breaking: true,
        };

        let cloned = change.clone();
        assert_eq!(change.version, cloned.version);
        assert_eq!(change.timestamp, cloned.timestamp);
        assert_eq!(change.is_breaking, cloned.is_breaking);
    }

    #[test]
    fn test_schema_change_debug() {
        let schema = json!({"type": "struct"});
        let change = SchemaChange {
            version: 1,
            timestamp: 1234567890,
            schema,
            is_breaking: false,
        };

        let debug_str = format!("{:?}", change);
        assert!(!debug_str.is_empty());
        assert!(debug_str.contains("SchemaChange"));
    }

    #[test]
    fn test_metadata_processing_result_default() {
        let result = MetadataProcessingResult::default();

        assert_eq!(result.partition_columns.len(), 0);
        assert_eq!(result.total_snapshots, 0);
        assert_eq!(result.total_historical_size, 0);
        assert_eq!(result.oldest_timestamp, 0);
        assert_eq!(result.newest_timestamp, 0);
        assert_eq!(result.total_constraints, 0);
        assert_eq!(result.check_constraints, 0);
        assert_eq!(result.not_null_constraints, 0);
        assert_eq!(result.unique_constraints, 0);
        assert_eq!(result.foreign_key_constraints, 0);
        assert_eq!(result.sort_order_columns.len(), 0);
        assert_eq!(result.schema_changes.len(), 0);
        assert_eq!(result.delete_files_count, 0);
        assert_eq!(result.delete_files_size, 0);
    }

    #[test]
    fn test_detect_breaking_schema_changes_field_removed() {
        let old_schema = json!({
            "fields": [
                {"name": "id", "type": "long"},
                {"name": "name", "type": "string"}
            ]
        });

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let breaking = detect_breaking_schema_changes(&old_schema, &new_schema);
        assert!(breaking, "Removing a field should be a breaking change");
    }

    #[test]
    fn test_detect_breaking_schema_changes_type_changed() {
        let old_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "string"}
            ]
        });

        let breaking = detect_breaking_schema_changes(&old_schema, &new_schema);
        assert!(breaking, "Changing field type should be a breaking change");
    }

    #[test]
    fn test_detect_breaking_schema_changes_nullable_changed() {
        let old_schema = json!({
            "fields": [
                {"name": "id", "type": "long", "nullable": false}
            ]
        });

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "long", "nullable": true}
            ]
        });

        let breaking = detect_breaking_schema_changes(&old_schema, &new_schema);
        assert!(
            breaking,
            "Making a field nullable should be a breaking change"
        );
    }

    #[test]
    fn test_detect_breaking_schema_changes_field_added() {
        let old_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "long"},
                {"name": "name", "type": "string"}
            ]
        });

        let breaking = detect_breaking_schema_changes(&old_schema, &new_schema);
        assert!(!breaking, "Adding a field should not be a breaking change");
    }

    #[test]
    fn test_detect_breaking_schema_changes_no_change() {
        let old_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let breaking = detect_breaking_schema_changes(&old_schema, &new_schema);
        assert!(!breaking, "No change should not be breaking");
    }

    #[test]
    fn test_is_breaking_change_no_previous() {
        let new_schema = json!({"fields": []});

        let breaking = is_breaking_change(&[], &new_schema);
        assert!(!breaking, "First schema should not be breaking");
    }

    #[test]
    fn test_is_breaking_change_with_previous() {
        let previous_changes = vec![SchemaChange::new(
            1,
            1234567890,
            json!({
                "fields": [
                    {"name": "id", "type": "long"},
                    {"name": "name", "type": "string"}
                ]
            }),
            false,
        )];

        let new_schema = json!({
            "fields": [
                {"name": "id", "type": "long"}
            ]
        });

        let breaking = is_breaking_change(&previous_changes, &new_schema);
        assert!(breaking, "Removing a field should be breaking");
    }

    #[test]
    fn test_calculate_schema_stability_score_perfect() {
        let score = calculate_schema_stability_score(0, 0, 0.0, 100.0);

        assert!(score >= 0.9, "Perfect stability should have high score");
        assert!(score <= 1.0);
    }

    #[test]
    fn test_calculate_schema_stability_score_moderate() {
        let score = calculate_schema_stability_score(15, 2, 0.2, 10.0);

        assert!(score >= 0.0);
        assert!(score <= 1.0);
    }

    #[test]
    fn test_calculate_schema_stability_score_unstable() {
        let score = calculate_schema_stability_score(100, 20, 2.0, 1.0);

        assert!(score >= 0.0);
        assert!(score < 0.5, "Unstable schema should have low score");
    }

    #[test]
    fn test_calculate_schema_stability_score_clamped() {
        // Extreme values that would result in negative score
        let score = calculate_schema_stability_score(1000, 100, 10.0, 0.1);

        assert!(score >= 0.0, "Score should be clamped to 0.0");
        assert!(score <= 1.0, "Score should be clamped to 1.0");
    }

    #[test]
    fn test_calculate_storage_cost_impact_zero() {
        let impact = calculate_storage_cost_impact(0, 0, 0.0);

        assert_eq!(impact, 0.0);
    }

    #[test]
    fn test_calculate_storage_cost_impact_low() {
        // 2GB, 50 snapshots, 10 days old
        let impact = calculate_storage_cost_impact(2 * 1024 * 1024 * 1024, 50, 10.0);

        assert!(impact >= 0.1, "Should have some impact from size");
        assert!(impact < 0.5);
    }

    #[test]
    fn test_calculate_storage_cost_impact_high() {
        // 200GB, 2000 snapshots, 400 days old
        let impact = calculate_storage_cost_impact(200 * 1024 * 1024 * 1024, 2000, 400.0);

        assert!(impact > 0.5);
        assert!(impact <= 1.0);
    }

    #[test]
    fn test_calculate_retention_efficiency_perfect() {
        // Low snapshot count, reasonable retention
        let efficiency = calculate_retention_efficiency(30, 30.0, 0.0);

        assert!(efficiency >= 0.8);
        assert!(efficiency <= 1.0);
    }

    #[test]
    fn test_calculate_retention_efficiency_poor() {
        // Too many snapshots, too long retention
        let efficiency = calculate_retention_efficiency(2000, 500.0, 0.0);

        assert!(efficiency < 0.5);
    }

    #[test]
    fn test_calculate_recommended_retention_few_snapshots() {
        let retention = calculate_recommended_retention(50, 20.0);

        assert_eq!(
            retention, 180,
            "Few snapshots should recommend longer retention"
        );
    }

    #[test]
    fn test_calculate_recommended_retention_many_snapshots() {
        let retention = calculate_recommended_retention(1500, 400.0);

        assert_eq!(
            retention, 30,
            "Many snapshots should recommend shorter retention"
        );
    }

    // Tests for extract_constraints_from_schema
    #[test]
    fn test_extract_constraints_from_schema_empty() {
        let analyzer = create_test_analyzer();
        let fields = vec![];

        let (total, check, not_null, unique, foreign_key) =
            analyzer.extract_constraints_from_schema(&fields);

        assert_eq!(total, 0);
        assert_eq!(check, 0);
        assert_eq!(not_null, 0);
        assert_eq!(unique, 0);
        assert_eq!(foreign_key, 0);
    }

    #[test]
    fn test_extract_constraints_from_schema_not_null() {
        let analyzer = create_test_analyzer();
        let fields = vec![
            json!({"name": "id", "type": "long", "required": true}),
            json!({"name": "name", "type": "string", "required": false}),
        ];

        let (total, _check, not_null, _unique, _foreign_key) =
            analyzer.extract_constraints_from_schema(&fields);

        assert_eq!(total, 2);
        assert_eq!(not_null, 1);
    }

    #[test]
    fn test_extract_constraints_from_schema_with_metadata() {
        let analyzer = create_test_analyzer();
        let fields = vec![json!({
            "name": "id",
            "type": "long",
            "required": true,
            "metadata": {
                "constraint": "id > 0",
                "unique": true
            }
        })];

        let (total, check, not_null, unique, _foreign_key) =
            analyzer.extract_constraints_from_schema(&fields);

        assert_eq!(total, 1);
        assert_eq!(not_null, 1);
        assert_eq!(check, 1);
        assert_eq!(unique, 1);
    }

    #[test]
    fn test_extract_constraints_from_schema_with_doc() {
        let analyzer = create_test_analyzer();
        let fields = vec![json!({
            "name": "user_id",
            "type": "long",
            "doc": "Foreign key reference to users table"
        })];

        let (total, _check, _not_null, _unique, foreign_key) =
            analyzer.extract_constraints_from_schema(&fields);

        assert_eq!(total, 1);
        assert_eq!(foreign_key, 1);
    }

    // Tests for extract_snapshot_metrics
    #[test]
    fn test_extract_snapshot_metrics_empty() {
        let analyzer = create_test_analyzer();
        let json = json!({});
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_snapshot_metrics(&json, &mut result);

        assert_eq!(result.total_snapshots, 0);
        assert_eq!(result.oldest_timestamp, 0);
        assert_eq!(result.newest_timestamp, 0);
    }

    #[test]
    fn test_extract_snapshot_metrics_single_snapshot() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "snapshots": [
                {
                    "timestamp-ms": 1234567890000_u64,
                    "summary": {
                        "total-data-files": "10"
                    }
                }
            ]
        });
        let mut result = MetadataProcessingResult {
            oldest_timestamp: u64::MAX, // Initialize to max so min() works correctly
            ..Default::default()
        };

        analyzer.extract_snapshot_metrics(&json, &mut result);

        assert_eq!(result.total_snapshots, 1);
        assert_eq!(result.oldest_timestamp, 1234567890000_u64);
        assert_eq!(result.newest_timestamp, 1234567890000_u64);
        assert!(result.total_historical_size > 0);
    }

    #[test]
    fn test_extract_snapshot_metrics_multiple_snapshots() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "snapshots": [
                {
                    "timestamp-ms": 1000000000000_u64,
                    "summary": {
                        "total-data-files": "5"
                    }
                },
                {
                    "timestamp-ms": 2000000000000_u64,
                    "summary": {
                        "total-data-files": "10"
                    }
                }
            ]
        });
        let mut result = MetadataProcessingResult {
            oldest_timestamp: u64::MAX, // Initialize to max so min() works correctly
            ..Default::default()
        };

        analyzer.extract_snapshot_metrics(&json, &mut result);

        assert_eq!(result.total_snapshots, 2);
        assert_eq!(result.oldest_timestamp, 1000000000000_u64);
        assert_eq!(result.newest_timestamp, 2000000000000_u64);
    }

    #[test]
    fn test_extract_snapshot_metrics_with_delete_files() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "snapshots": [
                {
                    "timestamp-ms": 1234567890000_u64,
                    "summary": {
                        "total-data-files": "10",
                        "total-delete-files": "5",
                        "total-position-deletes": "3",
                        "total-equality-deletes": "2"
                    }
                }
            ]
        });
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_snapshot_metrics(&json, &mut result);

        assert_eq!(result.total_snapshots, 1);
        assert_eq!(result.delete_files_count, 10); // 5 + 3 + 2
    }

    // Tests for extract_schema_and_constraints
    #[test]
    fn test_extract_schema_and_constraints_empty() {
        let analyzer = create_test_analyzer();
        let json = json!({});
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_schema_and_constraints(&json, &mut result, 1);

        assert_eq!(result.total_constraints, 0);
        assert_eq!(result.schema_changes.len(), 0);
    }

    #[test]
    fn test_extract_schema_and_constraints_single_schema() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "schemas": [
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true},
                        {"name": "name", "type": "string", "required": false}
                    ]
                }
            ]
        });
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_schema_and_constraints(&json, &mut result, 1);

        assert_eq!(result.total_constraints, 2);
        assert_eq!(result.not_null_constraints, 1);
        assert_eq!(result.schema_changes.len(), 1);
        assert!(!result.schema_changes[0].is_breaking);
    }

    #[test]
    fn test_extract_schema_and_constraints_multiple_schemas() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "schemas": [
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true}
                    ]
                },
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true},
                        {"name": "name", "type": "string", "required": false}
                    ]
                }
            ]
        });
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_schema_and_constraints(&json, &mut result, 1);

        assert_eq!(result.schema_changes.len(), 2);
        assert!(!result.schema_changes[0].is_breaking);
        assert!(!result.schema_changes[1].is_breaking); // Adding field is not breaking
    }

    #[test]
    fn test_extract_schema_and_constraints_with_breaking_change() {
        let analyzer = create_test_analyzer();
        let json = json!({
            "schemas": [
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true},
                        {"name": "name", "type": "string", "required": false}
                    ]
                },
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true}
                    ]
                }
            ]
        });
        let mut result = MetadataProcessingResult::default();

        analyzer.extract_schema_and_constraints(&json, &mut result, 1);

        assert_eq!(result.schema_changes.len(), 2);
        assert!(!result.schema_changes[0].is_breaking);
        assert!(result.schema_changes[1].is_breaking); // Removing field is breaking
    }

    // Tests for calculate_compaction_opportunity
    #[test]
    fn test_calculate_compaction_opportunity_zero_files() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_compaction_opportunity(0, 0);

        assert_eq!(score, 0.0);
    }

    #[test]
    fn test_calculate_compaction_opportunity_low() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_compaction_opportunity(10, 100);

        assert_eq!(score, 0.2); // 10% ratio
    }

    #[test]
    fn test_calculate_compaction_opportunity_medium() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_compaction_opportunity(50, 100);

        assert_eq!(score, 0.6); // 50% ratio
    }

    #[test]
    fn test_calculate_compaction_opportunity_high() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_compaction_opportunity(85, 100);

        assert_eq!(score, 1.0); // 85% ratio
    }

    // Tests for calculate_compaction_priority
    #[test]
    fn test_calculate_compaction_priority_low() {
        let analyzer = create_test_analyzer();

        let priority = analyzer.calculate_compaction_priority(0.3, 10);

        assert_eq!(priority, "low");
    }

    #[test]
    fn test_calculate_compaction_priority_medium() {
        let analyzer = create_test_analyzer();

        let priority = analyzer.calculate_compaction_priority(0.5, 25);

        assert_eq!(priority, "medium");
    }

    #[test]
    fn test_calculate_compaction_priority_high() {
        let analyzer = create_test_analyzer();

        let priority = analyzer.calculate_compaction_priority(0.7, 60);

        assert_eq!(priority, "high");
    }

    #[test]
    fn test_calculate_compaction_priority_critical() {
        let analyzer = create_test_analyzer();

        let priority = analyzer.calculate_compaction_priority(0.9, 150);

        assert_eq!(priority, "critical");
    }

    #[test]
    fn test_calculate_compaction_priority_critical_by_count() {
        let analyzer = create_test_analyzer();

        let priority = analyzer.calculate_compaction_priority(0.5, 150);

        assert_eq!(priority, "critical"); // >100 small files
    }

    // Tests for calculate_file_compaction_metrics
    #[test]
    fn test_calculate_file_compaction_metrics_zero_files() {
        let analyzer = create_test_analyzer();

        let metrics = analyzer.calculate_file_compaction_metrics(0, 0, &[]);

        assert_eq!(metrics.small_files_count, 0);
        assert_eq!(metrics.compaction_opportunity_score, 0.0);
        assert_eq!(metrics.compaction_priority, "low");
    }

    #[test]
    fn test_calculate_file_compaction_metrics_small_files() {
        let analyzer = create_test_analyzer();
        let total_files = 100;
        let avg_size = 8 * 1024 * 1024; // 8MB average
        let total_size = total_files as u64 * avg_size;

        let metrics = analyzer.calculate_file_compaction_metrics(total_files, total_size, &[]);

        assert!(metrics.small_files_count > 0);
        assert!(metrics.compaction_opportunity_score > 0.0);
        assert!(metrics.small_files_size_bytes > 0);
    }

    #[test]
    fn test_calculate_file_compaction_metrics_large_files() {
        let analyzer = create_test_analyzer();
        let total_files = 100;
        let avg_size = 128 * 1024 * 1024; // 128MB average
        let total_size = total_files as u64 * avg_size;

        let metrics = analyzer.calculate_file_compaction_metrics(total_files, total_size, &[]);

        // Should estimate fewer small files
        assert!(metrics.small_files_count < total_files);
        assert!(metrics.compaction_opportunity_score < 1.0);
    }

    #[test]
    fn test_calculate_file_compaction_metrics_with_sort_order() {
        let analyzer = create_test_analyzer();
        let sort_columns = vec!["date".to_string(), "user_id".to_string()];

        let metrics =
            analyzer.calculate_file_compaction_metrics(100, 1024 * 1024 * 1024, &sort_columns);

        assert!(metrics.z_order_opportunity);
        assert_eq!(metrics.z_order_columns, sort_columns);
    }

    // Tests for calculate_deletion_impact_score
    #[test]
    fn test_calculate_deletion_impact_score_zero() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_deletion_impact_score(0, 0.0);

        assert!(score >= 0.0);
        assert!(score <= 1.0);
    }

    #[test]
    fn test_calculate_deletion_impact_score_low() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_deletion_impact_score(5, 3.0);

        assert!(score >= 0.0);
        assert!(score < 0.5);
    }

    #[test]
    fn test_calculate_deletion_impact_score_medium() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_deletion_impact_score(150, 40.0);

        assert!(score >= 0.4);
        assert!(score <= 1.0);
    }

    #[test]
    fn test_calculate_deletion_impact_score_high() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_deletion_impact_score(1500, 120.0);

        assert!(score >= 0.8);
        assert!(score <= 1.0);
    }

    // Tests for calculate_deletion_vector_metrics
    #[test]
    fn test_calculate_deletion_vector_metrics_zero() {
        let analyzer = create_test_analyzer();

        let metrics = analyzer.calculate_deletion_vector_metrics(0, 0, 0);

        assert_eq!(metrics.deletion_vector_count, 0);
        assert_eq!(metrics.total_deletion_vector_size_bytes, 0);
        assert_eq!(metrics.deleted_rows_count, 0);
    }

    #[test]
    fn test_calculate_deletion_vector_metrics_with_deletes() {
        let analyzer = create_test_analyzer();
        let delete_count = 100;
        let now = chrono::Utc::now().timestamp() as u64;
        let oldest_timestamp = (now - 86400 * 30) * 1000; // 30 days ago in ms

        let metrics =
            analyzer.calculate_deletion_vector_metrics(delete_count, oldest_timestamp, now * 1000);

        assert_eq!(metrics.deletion_vector_count, 100);
        assert_eq!(metrics.total_deletion_vector_size_bytes, 100 * 1024); // 1KB per delete file
        assert_eq!(metrics.deleted_rows_count, 100 * 100); // 100 rows per delete file
        assert!(metrics.deletion_vector_age_days > 0.0);
        assert!(metrics.deletion_vector_impact_score > 0.0);
    }

    #[test]
    fn test_calculate_deletion_vector_metrics_old_deletes() {
        let analyzer = create_test_analyzer();
        let delete_count = 500;
        let now = chrono::Utc::now().timestamp() as u64;
        let oldest_timestamp = (now - 86400 * 120) * 1000; // 120 days ago in ms

        let metrics =
            analyzer.calculate_deletion_vector_metrics(delete_count, oldest_timestamp, now * 1000);

        assert_eq!(metrics.deletion_vector_count, 500);
        assert!(metrics.deletion_vector_age_days > 100.0);
        assert!(metrics.deletion_vector_impact_score > 0.5); // Should have high impact
    }

    // Tests for calculate_constraint_violation_risk
    #[test]
    fn test_calculate_constraint_violation_risk_zero() {
        let analyzer = create_test_analyzer();

        let risk = analyzer.calculate_constraint_violation_risk(0, 0);

        assert_eq!(risk, 0.0);
    }

    #[test]
    fn test_calculate_constraint_violation_risk_low() {
        let analyzer = create_test_analyzer();

        let risk = analyzer.calculate_constraint_violation_risk(100, 5);

        assert_eq!(risk, 0.2); // 5% check constraint ratio
    }

    #[test]
    fn test_calculate_constraint_violation_risk_medium() {
        let analyzer = create_test_analyzer();

        let risk = analyzer.calculate_constraint_violation_risk(100, 25);

        assert_eq!(risk, 0.4); // 25% check constraint ratio
    }

    #[test]
    fn test_calculate_constraint_violation_risk_high() {
        let analyzer = create_test_analyzer();

        let risk = analyzer.calculate_constraint_violation_risk(100, 60);

        assert_eq!(risk, 0.8); // 60% check constraint ratio
    }

    // Tests for calculate_data_quality_score
    #[test]
    fn test_calculate_data_quality_score_no_constraints() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_data_quality_score(0, 0.0);

        assert_eq!(score, 1.0);
    }

    #[test]
    fn test_calculate_data_quality_score_few_constraints() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_data_quality_score(3, 0.2);

        assert!(score >= 0.8);
        assert!(score <= 1.0);
    }

    #[test]
    fn test_calculate_data_quality_score_many_constraints() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_data_quality_score(15, 0.3);

        assert!(score >= 1.0); // Bonus for many constraints
    }

    #[test]
    fn test_calculate_data_quality_score_high_risk() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_data_quality_score(10, 0.9);

        assert!(score < 1.0); // Penalized for high risk
    }

    // Tests for calculate_constraint_coverage_score
    #[test]
    fn test_calculate_constraint_coverage_score_zero() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_constraint_coverage_score(0, 0);

        assert_eq!(score, 0.0);
    }

    #[test]
    fn test_calculate_constraint_coverage_score_low() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_constraint_coverage_score(100, 5);

        assert_eq!(score, 0.1); // 5% coverage
    }

    #[test]
    fn test_calculate_constraint_coverage_score_medium() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_constraint_coverage_score(100, 25);

        assert_eq!(score, 0.4); // 25% coverage
    }

    #[test]
    fn test_calculate_constraint_coverage_score_high() {
        let analyzer = create_test_analyzer();

        let score = analyzer.calculate_constraint_coverage_score(100, 60);

        assert_eq!(score, 1.0); // 60% coverage
    }

    // Tests for calculate_schema_metrics
    #[test]
    fn test_calculate_schema_metrics_no_changes() {
        let analyzer = create_test_analyzer();
        let changes = vec![];

        let result = analyzer.calculate_schema_metrics(changes, 1);

        assert!(result.is_ok());
        let metrics = result.unwrap();
        assert!(metrics.is_some());
        let metrics = metrics.unwrap();
        assert_eq!(metrics.total_schema_changes, 0);
        assert_eq!(metrics.breaking_changes, 0);
        assert_eq!(metrics.non_breaking_changes, 0);
        assert!(metrics.schema_stability_score > 0.0);
    }

    #[test]
    fn test_calculate_schema_metrics_single_change() {
        let analyzer = create_test_analyzer();
        let now = chrono::Utc::now().timestamp() as u64;
        let changes = vec![SchemaChange {
            version: 1,
            timestamp: (now - 86400 * 10) * 1000, // 10 days ago in ms
            schema: json!({"fields": [{"name": "id", "type": "long"}]}),
            is_breaking: false,
        }];

        let result = analyzer.calculate_schema_metrics(changes, 1);

        assert!(result.is_ok());
        let metrics = result.unwrap().unwrap();
        assert_eq!(metrics.total_schema_changes, 1);
        assert_eq!(metrics.breaking_changes, 0);
        assert_eq!(metrics.non_breaking_changes, 1);
        assert!(metrics.days_since_last_change >= 9.0);
        assert!(metrics.days_since_last_change <= 11.0);
    }

    #[test]
    fn test_calculate_schema_metrics_multiple_changes() {
        let analyzer = create_test_analyzer();
        let now = chrono::Utc::now().timestamp() as u64;
        let changes = vec![
            SchemaChange {
                version: 1,
                timestamp: (now - 86400 * 100) * 1000, // 100 days ago
                schema: json!({"fields": [{"name": "id", "type": "long"}]}),
                is_breaking: false,
            },
            SchemaChange {
                version: 2,
                timestamp: (now - 86400 * 50) * 1000, // 50 days ago
                schema: json!({"fields": [{"name": "id", "type": "long"}, {"name": "name", "type": "string"}]}),
                is_breaking: false,
            },
            SchemaChange {
                version: 3,
                timestamp: (now - 86400 * 10) * 1000, // 10 days ago
                schema: json!({"fields": [{"name": "id", "type": "long"}]}),
                is_breaking: true,
            },
        ];

        let result = analyzer.calculate_schema_metrics(changes, 3);

        assert!(result.is_ok());
        let metrics = result.unwrap().unwrap();
        assert_eq!(metrics.total_schema_changes, 3);
        assert_eq!(metrics.breaking_changes, 1);
        assert_eq!(metrics.non_breaking_changes, 2);
        assert!(metrics.schema_change_frequency > 0.0);
        assert_eq!(metrics.current_schema_version, 3);
    }

    #[test]
    fn test_calculate_schema_metrics_all_breaking() {
        let analyzer = create_test_analyzer();
        let now = chrono::Utc::now().timestamp() as u64;
        let changes = vec![
            SchemaChange {
                version: 1,
                timestamp: (now - 86400 * 20) * 1000,
                schema: json!({"fields": [{"name": "id", "type": "long"}]}),
                is_breaking: true,
            },
            SchemaChange {
                version: 2,
                timestamp: (now - 86400 * 10) * 1000,
                schema: json!({"fields": [{"name": "id", "type": "string"}]}),
                is_breaking: true,
            },
        ];

        let result = analyzer.calculate_schema_metrics(changes, 2);

        assert!(result.is_ok());
        let metrics = result.unwrap().unwrap();
        assert_eq!(metrics.total_schema_changes, 2);
        assert_eq!(metrics.breaking_changes, 2);
        assert_eq!(metrics.non_breaking_changes, 0);
        // With 2 breaking changes, score is 1.0 - 0.2 = 0.8
        assert!(metrics.schema_stability_score < 1.0);
        assert!(metrics.schema_stability_score > 0.5);
    }

    #[test]
    fn test_calculate_schema_metrics_no_timestamps() {
        let analyzer = create_test_analyzer();
        let changes = vec![SchemaChange {
            version: 1,
            timestamp: 0, // No timestamp
            schema: json!({"fields": [{"name": "id", "type": "long"}]}),
            is_breaking: false,
        }];

        let result = analyzer.calculate_schema_metrics(changes, 1);

        assert!(result.is_ok());
        let metrics = result.unwrap().unwrap();
        assert_eq!(metrics.total_schema_changes, 1);
        assert_eq!(metrics.days_since_last_change, 365.0); // Default when no timestamp
    }

    // Async tests for get_manifest_list
    #[tokio::test]
    async fn test_get_manifest_list_empty() {
        let analyzer = create_test_analyzer();
        let metadata = json!({});

        let result = analyzer.get_manifest_list(&metadata).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap().len(), 0);
    }

    #[tokio::test]
    async fn test_get_manifest_list_with_snapshot() {
        let analyzer = create_test_analyzer();
        let metadata = json!({
            "current-snapshot-id": 12345,
            "snapshots": [
                {
                    "snapshot-id": 12345,
                    "manifest-list": "s3://bucket/table/metadata/snap-12345-1-abc.avro"
                }
            ]
        });

        let result = analyzer.get_manifest_list(&metadata).await;

        assert!(result.is_ok());
        let manifests = result.unwrap();
        assert_eq!(manifests.len(), 1);
        assert_eq!(
            manifests[0],
            "s3://bucket/table/metadata/snap-12345-1-abc.avro"
        );
    }

    #[tokio::test]
    async fn test_get_manifest_list_no_current_snapshot() {
        let analyzer = create_test_analyzer();
        let metadata = json!({
            "snapshots": [
                {
                    "snapshot-id": 12345,
                    "manifest-list": "s3://bucket/table/metadata/snap-12345-1-abc.avro"
                }
            ]
        });

        let result = analyzer.get_manifest_list(&metadata).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap().len(), 0);
    }

    #[tokio::test]
    async fn test_get_manifest_list_snapshot_not_found() {
        let analyzer = create_test_analyzer();
        let metadata = json!({
            "current-snapshot-id": 99999,
            "snapshots": [
                {
                    "snapshot-id": 12345,
                    "manifest-list": "s3://bucket/table/metadata/snap-12345-1-abc.avro"
                }
            ]
        });

        let result = analyzer.get_manifest_list(&metadata).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap().len(), 0);
    }

    // Async tests for process_single_metadata_file
    #[tokio::test]
    async fn test_process_single_metadata_file_basic() {
        let metadata_json = json!({
            "format-version": 2,
            "table-uuid": "test-uuid",
            "schemas": [
                {
                    "schema-id": 0,
                    "fields": [
                        {"id": 1, "name": "id", "type": "long", "required": true},
                        {"id": 2, "name": "name", "type": "string", "required": false}
                    ]
                }
            ],
            "partition-specs": [
                {
                    "spec-id": 0,
                    "fields": [
                        {"name": "date", "transform": "day", "source-id": 3}
                    ]
                }
            ],
            "snapshots": [
                {
                    "snapshot-id": 12345,
                    "timestamp-ms": 1234567890000_u64,
                    "summary": {
                        "total-data-files": "10"
                    }
                }
            ]
        });

        let json_bytes = serde_json::to_vec(&metadata_json).unwrap();
        let mock_storage = Arc::new(MockStorageProvider::with_data(json_bytes));
        let analyzer = IcebergAnalyzer::new(mock_storage, 4);

        let result = analyzer.process_single_metadata_file("test.json", 0).await;

        assert!(result.is_ok());
        let metrics = result.unwrap();
        assert_eq!(metrics.total_snapshots, 1);
        assert_eq!(metrics.partition_columns.len(), 1);
        assert_eq!(metrics.partition_columns[0], "date");
        assert_eq!(metrics.schema_changes.len(), 1);
        assert_eq!(metrics.total_constraints, 2);
        assert_eq!(metrics.not_null_constraints, 1);
    }

    #[tokio::test]
    async fn test_process_single_metadata_file_with_delete_files() {
        let metadata_json = json!({
            "format-version": 2,
            "schemas": [
                {
                    "fields": [
                        {"name": "id", "type": "long", "required": true}
                    ]
                }
            ],
            "snapshots": [
                {
                    "snapshot-id": 12345,
                    "timestamp-ms": 1234567890000_u64,
                    "summary": {
                        "total-data-files": "10",
                        "total-delete-files": "5"
                    }
                }
            ]
        });

        let json_bytes = serde_json::to_vec(&metadata_json).unwrap();
        let mock_storage = Arc::new(MockStorageProvider::with_data(json_bytes));
        let analyzer = IcebergAnalyzer::new(mock_storage, 4);

        let result = analyzer.process_single_metadata_file("test.json", 0).await;

        assert!(result.is_ok());
        let metrics = result.unwrap();
        assert_eq!(metrics.delete_files_count, 5);
    }

    #[tokio::test]
    async fn test_process_single_metadata_file_invalid_json() {
        let invalid_json = b"not valid json".to_vec();
        let mock_storage = Arc::new(MockStorageProvider::with_data(invalid_json));
        let analyzer = IcebergAnalyzer::new(mock_storage, 4);

        let result = analyzer.process_single_metadata_file("test.json", 0).await;

        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_process_single_metadata_file_empty() {
        let metadata_json = json!({});
        let json_bytes = serde_json::to_vec(&metadata_json).unwrap();
        let mock_storage = Arc::new(MockStorageProvider::with_data(json_bytes));
        let analyzer = IcebergAnalyzer::new(mock_storage, 4);

        let result = analyzer.process_single_metadata_file("test.json", 0).await;

        assert!(result.is_ok());
        let metrics = result.unwrap();
        assert_eq!(metrics.total_snapshots, 0);
        assert_eq!(metrics.schema_changes.len(), 0);
    }
}