lake-pulse 0.1.3

// 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 chrono::DateTime;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::{Display, Formatter, Result as FmtResult};

/// Iceberg table specific metrics extracted from table metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IcebergMetrics {
    /// Current snapshot ID
    pub current_snapshot_id: Option<i64>,

    /// Table format version
    pub format_version: i32,

    /// Table UUID
    pub table_uuid: String,

    /// Table metadata
    pub metadata: TableMetadata,

    /// Table properties
    pub table_properties: HashMap<String, String>,

    /// Snapshot information
    pub snapshot_info: SnapshotMetrics,

    /// Schema information
    pub schema_info: SchemaMetrics,

    /// Partition spec information
    pub partition_spec: PartitionSpecMetrics,

    /// Sort order information
    pub sort_order: SortOrderMetrics,

    /// File statistics
    pub file_stats: FileStatistics,

    /// Manifest statistics
    pub manifest_stats: ManifestStatistics,
}

/// Iceberg table metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableMetadata {
    /// Table location (base path)
    pub location: String,

    /// Last updated timestamp (milliseconds since epoch)
    pub last_updated_ms: Option<i64>,

    /// Last column ID assigned
    pub last_column_id: i32,

    /// Current schema ID
    pub current_schema_id: i32,

    /// Number of schemas
    pub schema_count: usize,

    /// Default spec ID
    pub default_spec_id: i32,

    /// Number of partition specs
    pub partition_spec_count: usize,

    /// Default sort order ID
    pub default_sort_order_id: i32,

    /// Number of sort orders
    pub sort_order_count: usize,

    /// Last sequence number
    pub last_sequence_number: Option<i64>,
}

/// Snapshot-level metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SnapshotMetrics {
    /// Total number of snapshots
    pub total_snapshots: usize,

    /// Current snapshot ID
    pub current_snapshot_id: Option<i64>,

    /// Current snapshot timestamp (milliseconds since epoch)
    pub current_snapshot_timestamp_ms: Option<i64>,

    /// Parent snapshot ID (if available)
    pub parent_snapshot_id: Option<i64>,

    /// Snapshot operation (append, replace, overwrite, delete)
    pub operation: Option<String>,

    /// Snapshot summary statistics
    pub summary: HashMap<String, String>,

    /// Manifest list location
    pub manifest_list: Option<String>,

    /// Schema ID at snapshot time
    pub schema_id: Option<i32>,

    /// Sequence number
    pub sequence_number: Option<i64>,
}

/// Schema-level metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaMetrics {
    /// Current schema ID
    pub schema_id: i32,

    /// Number of fields in current schema
    pub field_count: usize,

    /// Schema as JSON string
    pub schema_string: String,

    /// List of field names
    pub field_names: Vec<String>,

    /// Number of nested fields (structs, lists, maps)
    pub nested_field_count: usize,

    /// Number of required fields
    pub required_field_count: usize,

    /// Number of optional fields
    pub optional_field_count: usize,
}

/// Partition spec metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PartitionSpecMetrics {
    /// Partition spec ID
    pub spec_id: i32,

    /// Number of partition fields
    pub partition_field_count: usize,

    /// Partition field names
    pub partition_fields: Vec<String>,

    /// Partition transforms (identity, bucket, truncate, year, month, day, hour)
    pub partition_transforms: Vec<String>,

    /// Whether the table is partitioned
    pub is_partitioned: bool,

    /// Estimated number of unique partitions (if available)
    pub estimated_partition_count: Option<usize>,
}

/// Sort order metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SortOrderMetrics {
    /// Sort order ID
    pub order_id: i32,

    /// Number of sort fields
    pub sort_field_count: usize,

    /// Sort field names
    pub sort_fields: Vec<String>,

    /// Sort directions (asc, desc)
    pub sort_directions: Vec<String>,

    /// Null orders (nulls-first, nulls-last)
    pub null_orders: Vec<String>,

    /// Whether the table has a sort order defined
    pub is_sorted: bool,
}

/// File-level statistics from Iceberg table
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileStatistics {
    /// Total number of data files
    pub num_data_files: usize,

    /// Total size of data files in bytes
    pub total_data_size_bytes: u64,

    /// Average data file size in bytes
    pub avg_data_file_size_bytes: f64,

    /// Minimum data file size in bytes
    pub min_data_file_size_bytes: u64,

    /// Maximum data file size in bytes
    pub max_data_file_size_bytes: u64,

    /// Total number of records across all data files
    pub total_records: Option<i64>,

    /// Number of delete files (position deletes and equality deletes)
    pub num_delete_files: usize,

    /// Total size of delete files in bytes
    pub total_delete_size_bytes: u64,

    /// Number of position delete files
    pub num_position_delete_files: usize,

    /// Number of equality delete files
    pub num_equality_delete_files: usize,
}

/// Manifest file statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ManifestStatistics {
    /// Total number of manifest files
    pub num_manifest_files: usize,

    /// Total size of manifest files in bytes
    pub total_manifest_size_bytes: u64,

    /// Average manifest file size in bytes
    pub avg_manifest_file_size_bytes: f64,

    /// Number of data manifests
    pub num_data_manifests: usize,

    /// Number of delete manifests
    pub num_delete_manifests: usize,

    /// Total number of manifest lists
    pub num_manifest_lists: usize,

    /// Total size of manifest lists in bytes
    pub total_manifest_list_size_bytes: u64,
}

impl IcebergMetrics {
    /// Create a new IcebergMetrics instance with default values
    pub fn new() -> Self {
        Self {
            current_snapshot_id: None,
            format_version: 1,
            table_uuid: String::new(),
            metadata: TableMetadata {
                location: String::new(),
                last_updated_ms: None,
                last_column_id: 0,
                current_schema_id: 0,
                schema_count: 0,
                default_spec_id: 0,
                partition_spec_count: 0,
                default_sort_order_id: 0,
                sort_order_count: 0,
                last_sequence_number: None,
            },
            table_properties: HashMap::new(),
            snapshot_info: SnapshotMetrics {
                total_snapshots: 0,
                current_snapshot_id: None,
                current_snapshot_timestamp_ms: None,
                parent_snapshot_id: None,
                operation: None,
                summary: HashMap::new(),
                manifest_list: None,
                schema_id: None,
                sequence_number: None,
            },
            schema_info: SchemaMetrics {
                schema_id: 0,
                field_count: 0,
                schema_string: String::new(),
                field_names: Vec::new(),
                nested_field_count: 0,
                required_field_count: 0,
                optional_field_count: 0,
            },
            partition_spec: PartitionSpecMetrics {
                spec_id: 0,
                partition_field_count: 0,
                partition_fields: Vec::new(),
                partition_transforms: Vec::new(),
                is_partitioned: false,
                estimated_partition_count: None,
            },
            sort_order: SortOrderMetrics {
                order_id: 0,
                sort_field_count: 0,
                sort_fields: Vec::new(),
                sort_directions: Vec::new(),
                null_orders: Vec::new(),
                is_sorted: false,
            },
            file_stats: FileStatistics {
                num_data_files: 0,
                total_data_size_bytes: 0,
                avg_data_file_size_bytes: 0.0,
                min_data_file_size_bytes: 0,
                max_data_file_size_bytes: 0,
                total_records: None,
                num_delete_files: 0,
                total_delete_size_bytes: 0,
                num_position_delete_files: 0,
                num_equality_delete_files: 0,
            },
            manifest_stats: ManifestStatistics {
                num_manifest_files: 0,
                total_manifest_size_bytes: 0,
                avg_manifest_file_size_bytes: 0.0,
                num_data_manifests: 0,
                num_delete_manifests: 0,
                num_manifest_lists: 0,
                total_manifest_list_size_bytes: 0,
            },
        }
    }
}

impl Default for IcebergMetrics {
    fn default() -> Self {
        Self::new()
    }
}

impl Display for IcebergMetrics {
    fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
        writeln!(f)?;
        writeln!(f, " Iceberg Specific Metrics")?;
        writeln!(f, "{}", "━".repeat(80))?;

        // Basic info
        writeln!(f, " {:<40} {:>32}", "Format Version", self.format_version)?;
        if !self.table_uuid.is_empty() {
            writeln!(f, " {:<31}  {:>40}", "Table UUID", &self.table_uuid)?;
        }
        if let Some(snapshot_id) = self.current_snapshot_id {
            writeln!(f, " {:<40} {:>32}", "Current Snapshot ID", snapshot_id)?;
        }

        // Metadata
        if let Some(last_updated) = self.metadata.last_updated_ms {
            if let Some(dt) = DateTime::from_timestamp_millis(last_updated) {
                writeln!(f, " {:<40} {:>32}", "Last Updated", dt.to_rfc3339())?;
            }
        }
        writeln!(
            f,
            " {:<40} {:>32}",
            "Schema Count", self.metadata.schema_count
        )?;
        writeln!(
            f,
            " {:<40} {:>32}",
            "Partition Spec Count", self.metadata.partition_spec_count
        )?;
        writeln!(
            f,
            " {:<40} {:>32}",
            "Sort Order Count", self.metadata.sort_order_count
        )?;
        if let Some(seq) = self.metadata.last_sequence_number {
            writeln!(f, " {:<40} {:>32}", "Last Sequence Number", seq)?;
        }

        // Table Properties
        if !self.table_properties.is_empty() {
            self.table_properties
                .iter()
                .enumerate()
                .try_for_each(|(i, (k, v))| {
                    if i == 0 {
                        writeln!(
                            f,
                            " {:<20} {:>52}",
                            "Table Properties",
                            format!("{}: {}", k, v)
                        )
                    } else {
                        writeln!(f, " {:<20} {:>52}", "", format!("{}: {}", k, v))
                    }
                })?;
        }

        // Snapshot Info
        writeln!(f, " Snapshot Info")?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Total Snapshots", self.snapshot_info.total_snapshots
        )?;
        if let Some(op) = &self.snapshot_info.operation {
            writeln!(f, "   {:<38} {:>32}", "Last Operation", op)?;
        }
        if let Some(ts) = self.snapshot_info.current_snapshot_timestamp_ms {
            if let Some(dt) = DateTime::from_timestamp_millis(ts) {
                writeln!(f, "   {:<38} {:>32}", "Snapshot Time", dt.to_rfc3339())?;
            }
        }

        // Schema Info
        writeln!(f, " Schema Info")?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Schema ID", self.schema_info.schema_id
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Field Count", self.schema_info.field_count
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Required Fields", self.schema_info.required_field_count
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Optional Fields", self.schema_info.optional_field_count
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Nested Fields", self.schema_info.nested_field_count
        )?;

        // Partition Spec
        writeln!(f, " Partition Spec")?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Is Partitioned",
            if self.partition_spec.is_partitioned {
                "Yes"
            } else {
                "No"
            }
        )?;
        if self.partition_spec.is_partitioned {
            writeln!(
                f,
                "   {:<38} {:>32}",
                "Partition Fields", self.partition_spec.partition_field_count
            )?;
            if !self.partition_spec.partition_fields.is_empty() {
                writeln!(
                    f,
                    "   {:<38} {:>32}",
                    "Field Names",
                    self.partition_spec.partition_fields.join(", ")
                )?;
            }
            if !self.partition_spec.partition_transforms.is_empty() {
                writeln!(
                    f,
                    "   {:<38} {:>32}",
                    "Transforms",
                    self.partition_spec.partition_transforms.join(", ")
                )?;
            }
        }

        // Sort Order
        if self.sort_order.is_sorted {
            writeln!(f, " Sort Order")?;
            writeln!(
                f,
                "   {:<38} {:>32}",
                "Sort Fields", self.sort_order.sort_field_count
            )?;
            if !self.sort_order.sort_fields.is_empty() {
                writeln!(
                    f,
                    "   {:<38} {:>32}",
                    "Field Names",
                    self.sort_order.sort_fields.join(", ")
                )?;
            }
        }

        // File Statistics
        let file_stats = &self.file_stats;
        let total_size_mb = file_stats.total_data_size_bytes as f64 / (1024.0 * 1024.0);
        let avg_size_mb = file_stats.avg_data_file_size_bytes / (1024.0 * 1024.0);

        writeln!(f, " File Statistics")?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Data Files", file_stats.num_data_files
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Total Data Size",
            format!("{:.2} MB", total_size_mb)
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Avg Data File Size",
            format!("{:.2} MB", avg_size_mb)
        )?;
        if let Some(records) = file_stats.total_records {
            writeln!(f, "   {:<38} {:>32}", "Total Records", records)?;
        }
        if file_stats.num_delete_files > 0 {
            writeln!(
                f,
                "   {:<38} {:>32}",
                "Delete Files", file_stats.num_delete_files
            )?;
            writeln!(
                f,
                "   {:<38} {:>32}",
                "Position Delete Files", file_stats.num_position_delete_files
            )?;
            writeln!(
                f,
                "   {:<38} {:>32}",
                "Equality Delete Files", file_stats.num_equality_delete_files
            )?;
        }

        // Manifest Statistics
        writeln!(f, " Manifest Statistics")?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Manifest Files", self.manifest_stats.num_manifest_files
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Data Manifests", self.manifest_stats.num_data_manifests
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Delete Manifests", self.manifest_stats.num_delete_manifests
        )?;
        writeln!(
            f,
            "   {:<38} {:>32}",
            "Manifest Lists", self.manifest_stats.num_manifest_lists
        )?;

        Ok(())
    }
}

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

    #[test]
    fn test_iceberg_metrics_new() {
        let metrics = IcebergMetrics::new();

        assert_eq!(metrics.current_snapshot_id, None);
        assert_eq!(metrics.format_version, 1);
        assert_eq!(metrics.table_uuid, "");
        assert_eq!(metrics.metadata.location, "");
        assert_eq!(metrics.table_properties.len(), 0);
        assert_eq!(metrics.snapshot_info.total_snapshots, 0);
        assert_eq!(metrics.schema_info.field_count, 0);
        assert_eq!(metrics.partition_spec.partition_field_count, 0);
        assert_eq!(metrics.sort_order.sort_field_count, 0);
        assert_eq!(metrics.file_stats.num_data_files, 0);
        assert_eq!(metrics.manifest_stats.num_manifest_files, 0);
    }

    #[test]
    fn test_iceberg_metrics_default() {
        let metrics = IcebergMetrics::default();

        assert_eq!(metrics.current_snapshot_id, None);
        assert_eq!(metrics.format_version, 1);
        assert_eq!(metrics.table_uuid, "");
    }

    #[test]
    fn test_table_metadata_structure() {
        let metadata = TableMetadata {
            location: "/path/to/table".to_string(),
            last_updated_ms: Some(1234567890),
            last_column_id: 10,
            current_schema_id: 1,
            schema_count: 2,
            default_spec_id: 0,
            partition_spec_count: 1,
            default_sort_order_id: 0,
            sort_order_count: 1,
            last_sequence_number: Some(5),
        };

        assert_eq!(metadata.location, "/path/to/table");
        assert_eq!(metadata.last_updated_ms, Some(1234567890));
        assert_eq!(metadata.last_column_id, 10);
        assert_eq!(metadata.current_schema_id, 1);
        assert_eq!(metadata.schema_count, 2);
    }

    #[test]
    fn test_snapshot_metrics_structure() {
        let mut summary = HashMap::new();
        summary.insert("total-records".to_string(), "1000".to_string());

        let snapshot = SnapshotMetrics {
            total_snapshots: 5,
            current_snapshot_id: Some(123),
            current_snapshot_timestamp_ms: Some(1234567890),
            parent_snapshot_id: Some(122),
            operation: Some("append".to_string()),
            summary,
            manifest_list: Some("/path/to/manifest".to_string()),
            schema_id: Some(1),
            sequence_number: Some(10),
        };

        assert_eq!(snapshot.total_snapshots, 5);
        assert_eq!(snapshot.current_snapshot_id, Some(123));
        assert_eq!(snapshot.parent_snapshot_id, Some(122));
        assert_eq!(snapshot.operation, Some("append".to_string()));
        assert_eq!(
            snapshot.summary.get("total-records"),
            Some(&"1000".to_string())
        );
    }

    #[test]
    fn test_schema_metrics_structure() {
        let schema = SchemaMetrics {
            schema_id: 1,
            field_count: 5,
            schema_string: "{}".to_string(),
            field_names: vec!["id".to_string(), "name".to_string()],
            nested_field_count: 1,
            required_field_count: 3,
            optional_field_count: 2,
        };

        assert_eq!(schema.schema_id, 1);
        assert_eq!(schema.field_count, 5);
        assert_eq!(schema.field_names.len(), 2);
        assert_eq!(schema.required_field_count + schema.optional_field_count, 5);
    }

    #[test]
    fn test_partition_spec_metrics_structure() {
        let partition_spec = PartitionSpecMetrics {
            spec_id: 0,
            partition_field_count: 2,
            partition_fields: vec!["year".to_string(), "month".to_string()],
            partition_transforms: vec!["year".to_string(), "month".to_string()],
            is_partitioned: true,
            estimated_partition_count: Some(100),
        };

        assert_eq!(partition_spec.spec_id, 0);
        assert_eq!(partition_spec.partition_field_count, 2);
        assert_eq!(partition_spec.partition_fields.len(), 2);
        assert_eq!(partition_spec.partition_transforms.len(), 2);
        assert!(partition_spec.is_partitioned);
    }

    #[test]
    fn test_partition_spec_unpartitioned() {
        let partition_spec = PartitionSpecMetrics {
            spec_id: 0,
            partition_field_count: 0,
            partition_fields: vec![],
            partition_transforms: vec![],
            is_partitioned: false,
            estimated_partition_count: None,
        };

        assert_eq!(partition_spec.partition_field_count, 0);
        assert!(!partition_spec.is_partitioned);
        assert_eq!(partition_spec.partition_fields.len(), 0);
    }

    #[test]
    fn test_sort_order_metrics_structure() {
        let sort_order = SortOrderMetrics {
            order_id: 1,
            sort_field_count: 2,
            sort_fields: vec!["1".to_string(), "2".to_string()],
            sort_directions: vec!["asc".to_string(), "desc".to_string()],
            null_orders: vec!["nulls-first".to_string(), "nulls-last".to_string()],
            is_sorted: true,
        };

        assert_eq!(sort_order.order_id, 1);
        assert_eq!(sort_order.sort_field_count, 2);
        assert_eq!(sort_order.sort_fields.len(), 2);
        assert_eq!(sort_order.sort_directions.len(), 2);
        assert_eq!(sort_order.null_orders.len(), 2);
        assert!(sort_order.is_sorted);
    }

    #[test]
    fn test_sort_order_unsorted() {
        let sort_order = SortOrderMetrics {
            order_id: 0,
            sort_field_count: 0,
            sort_fields: vec![],
            sort_directions: vec![],
            null_orders: vec![],
            is_sorted: false,
        };

        assert_eq!(sort_order.sort_field_count, 0);
        assert!(!sort_order.is_sorted);
    }

    #[test]
    fn test_file_statistics_structure() {
        let file_stats = FileStatistics {
            num_data_files: 10,
            total_data_size_bytes: 1000000,
            avg_data_file_size_bytes: 100000.0,
            min_data_file_size_bytes: 50000,
            max_data_file_size_bytes: 150000,
            total_records: Some(10000),
            num_delete_files: 2,
            total_delete_size_bytes: 20000,
            num_position_delete_files: 1,
            num_equality_delete_files: 1,
        };

        assert_eq!(file_stats.num_data_files, 10);
        assert_eq!(file_stats.total_data_size_bytes, 1000000);
        assert_eq!(file_stats.avg_data_file_size_bytes, 100000.0);
        assert_eq!(file_stats.total_records, Some(10000));
        assert_eq!(
            file_stats.num_delete_files,
            file_stats.num_position_delete_files + file_stats.num_equality_delete_files
        );
    }

    #[test]
    fn test_file_statistics_empty() {
        let file_stats = FileStatistics {
            num_data_files: 0,
            total_data_size_bytes: 0,
            avg_data_file_size_bytes: 0.0,
            min_data_file_size_bytes: 0,
            max_data_file_size_bytes: 0,
            total_records: None,
            num_delete_files: 0,
            total_delete_size_bytes: 0,
            num_position_delete_files: 0,
            num_equality_delete_files: 0,
        };

        assert_eq!(file_stats.num_data_files, 0);
        assert_eq!(file_stats.total_data_size_bytes, 0);
        assert_eq!(file_stats.avg_data_file_size_bytes, 0.0);
        assert_eq!(file_stats.total_records, None);
    }

    #[test]
    fn test_manifest_statistics_structure() {
        let manifest_stats = ManifestStatistics {
            num_manifest_files: 5,
            total_manifest_size_bytes: 500000,
            avg_manifest_file_size_bytes: 100000.0,
            num_data_manifests: 4,
            num_delete_manifests: 1,
            num_manifest_lists: 1,
            total_manifest_list_size_bytes: 500000,
        };

        assert_eq!(manifest_stats.num_manifest_files, 5);
        assert_eq!(
            manifest_stats.num_manifest_files,
            manifest_stats.num_data_manifests + manifest_stats.num_delete_manifests
        );
        assert_eq!(manifest_stats.total_manifest_size_bytes, 500000);
        assert_eq!(manifest_stats.avg_manifest_file_size_bytes, 100000.0);
    }

    #[test]
    fn test_manifest_statistics_empty() {
        let manifest_stats = ManifestStatistics {
            num_manifest_files: 0,
            total_manifest_size_bytes: 0,
            avg_manifest_file_size_bytes: 0.0,
            num_data_manifests: 0,
            num_delete_manifests: 0,
            num_manifest_lists: 0,
            total_manifest_list_size_bytes: 0,
        };

        assert_eq!(manifest_stats.num_manifest_files, 0);
        assert_eq!(manifest_stats.total_manifest_size_bytes, 0);
        assert_eq!(manifest_stats.avg_manifest_file_size_bytes, 0.0);
    }

    #[test]
    fn test_iceberg_metrics_serialization() {
        let metrics = IcebergMetrics::new();

        // Test that metrics can be serialized to JSON
        let json = serde_json::to_string(&metrics);
        assert!(
            json.is_ok(),
            "Should be able to serialize IcebergMetrics to JSON"
        );

        // Test that metrics can be deserialized from JSON
        let json_str = json.unwrap();
        let deserialized: Result<IcebergMetrics, _> = serde_json::from_str(&json_str);
        assert!(
            deserialized.is_ok(),
            "Should be able to deserialize IcebergMetrics from JSON"
        );
    }

    #[test]
    fn test_table_metadata_serialization() {
        let metadata = TableMetadata {
            location: "/path/to/table".to_string(),
            last_updated_ms: Some(1234567890),
            last_column_id: 10,
            current_schema_id: 1,
            schema_count: 2,
            default_spec_id: 0,
            partition_spec_count: 1,
            default_sort_order_id: 0,
            sort_order_count: 1,
            last_sequence_number: Some(5),
        };

        let json = serde_json::to_string(&metadata);
        assert!(json.is_ok());

        let deserialized: Result<TableMetadata, _> = serde_json::from_str(&json.unwrap());
        assert!(deserialized.is_ok());
    }

    #[test]
    fn test_snapshot_metrics_clone() {
        let snapshot = SnapshotMetrics {
            total_snapshots: 5,
            current_snapshot_id: Some(123),
            current_snapshot_timestamp_ms: Some(1234567890),
            parent_snapshot_id: Some(122),
            operation: Some("append".to_string()),
            summary: HashMap::new(),
            manifest_list: Some("/path/to/manifest".to_string()),
            schema_id: Some(1),
            sequence_number: Some(10),
        };

        let cloned = snapshot.clone();
        assert_eq!(snapshot.total_snapshots, cloned.total_snapshots);
        assert_eq!(snapshot.current_snapshot_id, cloned.current_snapshot_id);
    }

    #[test]
    fn test_iceberg_metrics_debug() {
        let metrics = IcebergMetrics::new();
        let debug_str = format!("{:?}", metrics);
        assert!(!debug_str.is_empty());
        assert!(debug_str.contains("IcebergMetrics"));
    }

    #[test]
    fn test_iceberg_metrics_display() {
        let mut metrics = IcebergMetrics::new();
        metrics.format_version = 2;
        metrics.table_uuid = "test-uuid-12345".to_string();
        metrics.current_snapshot_id = Some(123456789);
        metrics.snapshot_info = SnapshotMetrics {
            total_snapshots: 5,
            current_snapshot_id: Some(123456789),
            current_snapshot_timestamp_ms: Some(1700000000000),
            parent_snapshot_id: Some(123456788),
            operation: Some("append".to_string()),
            summary: HashMap::new(),
            manifest_list: Some("/path/to/manifest".to_string()),
            schema_id: Some(1),
            sequence_number: Some(10),
        };
        metrics.schema_info = SchemaMetrics {
            schema_id: 1,
            field_count: 10,
            schema_string: String::new(),
            field_names: vec!["id".to_string(), "name".to_string()],
            required_field_count: 5,
            optional_field_count: 5,
            nested_field_count: 2,
        };
        metrics.partition_spec = PartitionSpecMetrics {
            spec_id: 0,
            partition_field_count: 1,
            partition_fields: vec!["date".to_string()],
            partition_transforms: vec!["day".to_string()],
            is_partitioned: true,
            estimated_partition_count: Some(30),
        };
        metrics.sort_order = SortOrderMetrics {
            order_id: 0,
            sort_field_count: 1,
            sort_fields: vec!["timestamp".to_string()],
            sort_directions: vec!["asc".to_string()],
            null_orders: vec!["nulls-first".to_string()],
            is_sorted: true,
        };
        metrics.file_stats = FileStatistics {
            num_data_files: 100,
            total_data_size_bytes: 1024 * 1024 * 500,
            avg_data_file_size_bytes: 1024.0 * 1024.0 * 5.0,
            min_data_file_size_bytes: 1024,
            max_data_file_size_bytes: 1024 * 1024 * 10,
            total_records: Some(1000000),
            num_delete_files: 5,
            total_delete_size_bytes: 1024 * 100,
            num_position_delete_files: 3,
            num_equality_delete_files: 2,
        };
        metrics.manifest_stats = ManifestStatistics {
            num_manifest_files: 10,
            total_manifest_size_bytes: 1024 * 1024,
            avg_manifest_file_size_bytes: 1024.0 * 100.0,
            num_data_manifests: 8,
            num_delete_manifests: 2,
            num_manifest_lists: 5,
            total_manifest_list_size_bytes: 1024 * 50,
        };
        metrics
            .table_properties
            .insert("write.format.default".to_string(), "parquet".to_string());

        let display = format!("{}", metrics);

        assert!(display.contains("Iceberg Specific Metrics"));
        assert!(display.contains("Format Version"));
        assert!(display.contains("2"));
        assert!(display.contains("test-uuid-12345"));
        assert!(display.contains("Snapshot Info"));
        assert!(display.contains("Schema Info"));
        assert!(display.contains("Partition Spec"));
        assert!(display.contains("Sort Order"));
        assert!(display.contains("File Statistics"));
        assert!(display.contains("Manifest Statistics"));
        assert!(display.contains("Table Properties"));
    }

    #[test]
    fn test_iceberg_metrics_display_minimal() {
        let metrics = IcebergMetrics::new();
        let display = format!("{}", metrics);

        assert!(display.contains("Iceberg Specific Metrics"));
        assert!(display.contains("Format Version"));
        assert!(display.contains("File Statistics"));
    }

    #[test]
    fn test_iceberg_metrics_display_not_partitioned() {
        let mut metrics = IcebergMetrics::new();
        metrics.partition_spec.is_partitioned = false;

        let display = format!("{}", metrics);

        assert!(display.contains("Is Partitioned"));
        assert!(display.contains("No"));
    }

    #[test]
    fn test_iceberg_metrics_display_with_deletes() {
        let mut metrics = IcebergMetrics::new();
        metrics.file_stats.num_delete_files = 10;
        metrics.file_stats.num_position_delete_files = 6;
        metrics.file_stats.num_equality_delete_files = 4;

        let display = format!("{}", metrics);

        assert!(display.contains("Delete Files"));
        assert!(display.contains("Position Delete Files"));
        assert!(display.contains("Equality Delete Files"));
    }
}