// 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::reader::delta::metrics::DeltaMetrics;
#[cfg(feature = "hudi")]
use crate::reader::hudi::metrics::HudiMetrics;
use crate::reader::iceberg::metrics::IcebergMetrics;
#[cfg(feature = "lance")]
use crate::reader::lance::metrics::LanceMetrics;
use crate::util::ascii_gantt::to_ascii_gantt;
use crate::util::ascii_gantt::GanttConfig;
use serde::{Deserialize, Serialize};
use serde_json::{json, Error as JsonError};
use std::collections::{HashMap, LinkedList};
use std::error::Error;
use std::fmt::{Display, Formatter, Result as FmtResult};
/// Information about a single file in the data lake table.
///
/// Contains metadata about a file including its path, size, modification time,
/// and whether it's referenced by the table metadata.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileInfo {
/// The full path to the file
pub path: String,
/// Size of the file in bytes
pub size_bytes: u64,
/// Last modification timestamp (ISO 8601 format)
pub last_modified: Option<String>,
/// Whether this file is referenced in the table metadata
pub is_referenced: bool,
}
/// Information about a table partition.
///
/// Contains aggregated metrics for a single partition including file counts,
/// sizes, and the list of files within the partition.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PartitionInfo {
/// Partition key-value pairs (e.g., {"year": "2024", "month": "01"})
pub partition_values: HashMap<String, String>,
/// Number of files in this partition
pub file_count: usize,
/// Total size of all files in this partition (bytes)
pub total_size_bytes: u64,
/// Average file size in this partition (bytes)
pub avg_file_size_bytes: f64,
/// List of files in this partition
pub files: Vec<FileInfo>,
}
/// Clustering information for Iceberg tables.
///
/// Iceberg supports clustering data by specific columns to improve
/// query performance through data locality.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusteringInfo {
/// Columns used for clustering
pub clustering_columns: Vec<String>,
/// Number of clusters in the table
pub cluster_count: usize,
/// Average number of files per cluster
pub avg_files_per_cluster: f64,
/// Average cluster size (bytes)
pub avg_cluster_size_bytes: f64,
}
/// Metrics for Delta Lake deletion vectors.
///
/// Deletion vectors are a Delta Lake feature that allows marking rows as deleted
/// without rewriting data files. High deletion vector usage may indicate a need
/// for compaction.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DeletionVectorMetrics {
/// Number of deletion vectors in the table
pub deletion_vector_count: usize,
/// Total size of all deletion vectors (bytes)
pub total_deletion_vector_size_bytes: u64,
/// Average deletion vector size (bytes)
pub avg_deletion_vector_size_bytes: f64,
/// Age of the oldest deletion vector (days)
pub deletion_vector_age_days: f64,
/// Total number of deleted rows tracked by deletion vectors
pub deleted_rows_count: u64,
/// Impact score: 0.0 (no impact) to 1.0 (high impact, compaction recommended)
pub deletion_vector_impact_score: f64,
}
/// Metrics tracking schema evolution over time.
///
/// Monitors how the table schema has changed, helping identify schema stability
/// and potential compatibility issues.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaEvolutionMetrics {
/// Total number of schema changes
pub total_schema_changes: usize,
/// Number of breaking schema changes
pub breaking_changes: usize,
/// Number of non-breaking schema changes
pub non_breaking_changes: usize,
/// Schema stability score: 0.0 (unstable) to 1.0 (very stable)
pub schema_stability_score: f64,
/// Days since the last schema change
pub days_since_last_change: f64,
/// Schema change frequency (changes per day)
pub schema_change_frequency: f64,
/// Current schema version number
pub current_schema_version: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeTravelMetrics {
pub total_snapshots: usize,
pub oldest_snapshot_age_days: f64,
pub newest_snapshot_age_days: f64,
pub total_historical_size_bytes: u64,
pub avg_snapshot_size_bytes: f64,
pub storage_cost_impact_score: f64, // 0.0 = low cost, 1.0 = high cost
pub retention_efficiency_score: f64, // 0.0 = inefficient, 1.0 = very efficient
pub recommended_retention_days: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableConstraintsMetrics {
pub total_constraints: usize,
pub check_constraints: usize,
pub not_null_constraints: usize,
pub unique_constraints: usize,
pub foreign_key_constraints: usize,
pub constraint_violation_risk: f64, // 0.0 = low risk, 1.0 = high risk
pub data_quality_score: f64, // 0.0 = poor quality, 1.0 = excellent quality
pub constraint_coverage_score: f64, // 0.0 = no coverage, 1.0 = full coverage
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileCompactionMetrics {
pub compaction_opportunity_score: f64, // 0.0 = no opportunity, 1.0 = high opportunity
pub small_files_count: usize,
pub small_files_size_bytes: u64,
pub potential_compaction_files: usize,
pub estimated_compaction_savings_bytes: u64,
pub recommended_target_file_size_bytes: u64,
pub compaction_priority: String, // "low", "medium", "high", "critical"
pub z_order_opportunity: bool,
pub z_order_columns: Vec<String>,
}
/// Comprehensive health metrics for a data lake table.
///
/// This struct contains all the metrics collected during table analysis,
/// including file statistics, partition information, data quality metrics,
/// and format-specific metrics.
///
/// # Examples
///
/// ```no_run
/// use lake_pulse::analyze::metrics::HealthMetrics;
///
/// let mut metrics = HealthMetrics::new();
/// metrics.calculate_data_skew();
/// let score = metrics.calculate_health_score();
/// println!("Health score: {:.2}", score);
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthMetrics {
/// Total number of data files in the table
pub total_files: usize,
/// Total size of all data files (bytes)
pub total_size_bytes: u64,
/// List of files not referenced in table metadata
pub unreferenced_files: Vec<FileInfo>,
/// Total size of unreferenced files (bytes)
pub unreferenced_size_bytes: u64,
/// Number of partitions in the table
pub partition_count: usize,
/// Detailed information about each partition
pub partitions: Vec<PartitionInfo>,
/// Clustering information (Iceberg-specific)
pub clustering: Option<ClusteringInfo>,
/// Average file size across all files (bytes)
pub avg_file_size_bytes: f64,
/// Distribution of files by size category
pub file_size_distribution: FileSizeDistribution,
/// List of actionable recommendations
pub recommendations: Vec<String>,
/// Overall health score (0.0 to 1.0, higher is better)
pub health_score: f64,
/// Data skew metrics
pub data_skew: DataSkewMetrics,
/// Metadata health metrics
pub metadata_health: MetadataHealth,
/// Snapshot health metrics
pub snapshot_health: SnapshotHealth,
/// Deletion vector metrics (Delta-specific)
pub deletion_vector_metrics: Option<DeletionVectorMetrics>,
/// Schema evolution metrics
pub schema_evolution: Option<SchemaEvolutionMetrics>,
/// Time travel metrics
pub time_travel_metrics: Option<TimeTravelMetrics>,
/// Table constraints metrics
pub table_constraints: Option<TableConstraintsMetrics>,
/// File compaction opportunity metrics
pub file_compaction: Option<FileCompactionMetrics>,
/// Delta Lake specific metrics
pub delta_table_specific_metrics: Option<DeltaMetrics>,
/// Apache Hudi specific metrics (requires `hudi` feature)
#[cfg(feature = "hudi")]
pub hudi_table_specific_metrics: Option<HudiMetrics>,
/// Apache Iceberg specific metrics
pub iceberg_table_specific_metrics: Option<IcebergMetrics>,
/// Lance specific metrics (requires `lance` feature)
#[cfg(feature = "lance")]
pub lance_table_specific_metrics: Option<LanceMetrics>,
}
/// Distribution of files by size category.
///
/// Categorizes files into size buckets to help identify potential
/// performance issues from too many small files or very large files.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FileSizeDistribution {
/// Number of small files (< 16MB)
pub small_files: usize,
/// Number of medium files (16MB - 128MB)
pub medium_files: usize,
/// Number of large files (128MB - 1GB)
pub large_files: usize,
/// Number of very large files (> 1GB)
pub very_large_files: usize,
}
/// Metrics for detecting data skew in partitions and file sizes.
///
/// Data skew can lead to performance issues where some tasks process
/// significantly more data than others.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataSkewMetrics {
/// Partition skew score: 0.0 (perfectly balanced) to 1.0 (highly skewed)
pub partition_skew_score: f64,
/// File size skew score: 0.0 (uniform sizes) to 1.0 (highly varied)
pub file_size_skew_score: f64,
/// Size of the largest partition (bytes)
pub largest_partition_size: u64,
/// Size of the smallest partition (bytes)
pub smallest_partition_size: u64,
/// Average partition size (bytes)
pub avg_partition_size: u64,
/// Standard deviation of partition sizes
pub partition_size_std_dev: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MetadataHealth {
pub metadata_file_count: usize,
pub metadata_total_size_bytes: u64,
pub avg_metadata_file_size: f64,
pub metadata_growth_rate: f64, // bytes per day (estimated)
pub manifest_file_count: usize, // For Iceberg
pub first_file_name: Option<String>,
pub last_file_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SnapshotHealth {
pub snapshot_count: usize,
pub oldest_snapshot_age_days: f64,
pub newest_snapshot_age_days: f64,
pub avg_snapshot_age_days: f64,
pub snapshot_retention_risk: f64, // 0.0 (good) to 1.0 (high risk)
}
/// Complete health report for a data lake table.
///
/// This is the main output of the analysis process, containing all metrics,
/// recommendations, and timing information.
///
/// # Examples
///
/// ```no_run
/// use lake_pulse::{Analyzer, StorageConfig};
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
/// let config = StorageConfig::local()
/// .with_option("path", "./examples/data");
///
/// let analyzer = Analyzer::builder(config)
/// .build()
/// .await?;
///
/// let report = analyzer.analyze("delta_dataset").await?;
///
/// // Print the report
/// println!("{}", report);
///
/// // Export to JSON
/// let json = report.to_json(false)?;
/// println!("{}", json);
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthReport {
/// Path to the analyzed table
pub table_path: String,
/// Table format type ("delta", "iceberg", "hudi", or "lance")
pub table_type: String,
/// Timestamp when the analysis was performed (ISO 8601 format)
pub analysis_timestamp: String,
/// Comprehensive health metrics
pub metrics: HealthMetrics,
/// Overall health score (0.0 to 1.0, higher is better)
pub health_score: f64,
/// Performance timing metrics
pub timed_metrics: TimedLikeMetrics,
}
/// Performance timing metrics for analysis operations.
///
/// Tracks the duration of various operations during table analysis,
/// useful for performance profiling and optimization.
///
/// # Examples
///
/// ```no_run
/// use lake_pulse::analyze::metrics::TimedLikeMetrics;
///
/// # fn example(metrics: &TimedLikeMetrics) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
/// // Export to Chrome tracing format
/// let tracing_json = metrics.to_chrome_tracing()?;
///
/// // Generate ASCII Gantt chart
/// let gantt = metrics.duration_collection_as_gantt(None)?;
/// println!("{}", gantt);
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimedLikeMetrics {
/// Collection of (operation_name, start_time_micros, duration_micros) tuples
pub duration_collection: LinkedList<(String, u128, u128)>,
}
impl TimedLikeMetrics {
pub fn to_chrome_tracing(&self) -> Result<String, Box<dyn Error + Send + Sync>> {
let mut events = Vec::new();
for (name, start, duration) in &self.duration_collection {
events.push(json!({
"name": name,
"cat": "PERF",
"pid": "1",
"ph": "B",
"ts": start * 1000,
}));
events.push(json!({
"name": name,
"cat": "PERF",
"pid": "1",
"ph": "E",
"ts": start * 1000 + duration * 1000,
}));
}
Ok(serde_json::to_string(&events)?)
}
/// Generate an ASCII Gantt chart representation of the timing data
///
/// This creates a visual timeline showing when each operation started and how long it took.
///
/// # Arguments
///
/// * `config` - Optional configuration for the chart appearance. If None, uses default settings.
///
/// # Example
///
/// ```ignore
/// let gantt = metrics.to_ascii_gantt(None);
/// println!("{}", gantt);
/// ```
///
/// Output example:
/// ```text
/// Timeline (ms):
/// 1000 1550 2100 2650
/// |-----------|-----------|-----------|
/// storage_config_new_dur [] 50ms
/// analyzer_new_dur [====] 300ms
/// analyze_total_dur [========================] 2400ms
/// ```
pub fn duration_collection_as_gantt(
&self,
config: Option<GanttConfig>,
) -> Result<String, Box<dyn Error + Send + Sync>> {
to_ascii_gantt(&self.duration_collection, config)
}
}
impl Display for HealthReport {
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
let report = self;
// Header (no vertical borders for path to allow longer paths)
writeln!(f, "\n{}", "━".repeat(80))?;
writeln!(
f,
" {:<60} Score: {:>5.1}% ",
"Table Health Report",
report.health_score * 100.0
)?;
writeln!(f, "{}", "━".repeat(80))?;
writeln!(f, " {}", report.table_path)?;
writeln!(f, " {} ({})", report.analysis_timestamp, report.table_type)?;
writeln!(f, "{}", "━".repeat(80))?;
// Key Metrics and File Size Distribution (side by side)
writeln!(f)?;
writeln!(f, " {:<41} File Size Distribution", "Key Metrics")?;
writeln!(f, "{}", "━".repeat(80))?;
let dist = &report.metrics.file_size_distribution;
let total_files =
(dist.small_files + dist.medium_files + dist.large_files + dist.very_large_files)
as f64;
let size_gb = report.metrics.total_size_bytes as f64 / (1024.0 * 1024.0 * 1024.0);
let size_str = if size_gb >= 1.0 {
format!("{:.2} GB", size_gb)
} else {
let size_mb = report.metrics.total_size_bytes as f64 / (1024.0 * 1024.0);
format!("{:.2} MB", size_mb)
};
let avg_mb = report.metrics.avg_file_size_bytes / (1024.0 * 1024.0);
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8} {:>5.1}%",
"Total Data Files",
format!("{}", report.metrics.total_files),
"Small (<16MB)",
dist.small_files,
if total_files > 0.0 {
dist.small_files as f64 / total_files * 100.0
} else {
0.0
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8} {:>5.1}%",
"Total Data Size",
size_str,
"Medium (16-128MB)",
dist.medium_files,
if total_files > 0.0 {
dist.medium_files as f64 / total_files * 100.0
} else {
0.0
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8} {:>5.1}%",
"Avg File Size",
format!("{:.2} MB", avg_mb),
"Large (128MB-1GB)",
dist.large_files,
if total_files > 0.0 {
dist.large_files as f64 / total_files * 100.0
} else {
0.0
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8} {:>5.1}%",
"Partitions",
format!("{}", report.metrics.partition_count),
"Very Large (>1GB)",
dist.very_large_files,
if total_files > 0.0 {
dist.very_large_files as f64 / total_files * 100.0
} else {
0.0
}
)?;
// Data Skew Analysis and Metadata Health (side by side)
writeln!(f)?;
writeln!(f, " {:<41} Metadata Health", "Data Skew Analysis")?;
writeln!(f, "{}", "━".repeat(80))?;
let skew = &report.metrics.data_skew;
let meta = &report.metrics.metadata_health;
let largest_mb = skew.largest_partition_size as f64 / (1024.0 * 1024.0);
let smallest_mb = skew.smallest_partition_size as f64 / (1024.0 * 1024.0);
let avg_partition_mb = skew.avg_partition_size as f64 / (1024.0 * 1024.0);
let meta_size_mb = meta.metadata_total_size_bytes as f64 / (1024.0 * 1024.0);
let avg_meta_mb = meta.avg_metadata_file_size / (1024.0 * 1024.0);
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"Partition Skew",
format!("{:.2}", skew.partition_skew_score),
"Count",
format!("{}", meta.metadata_file_count)
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"File Size Skew",
format!("{:.2}", skew.file_size_skew_score),
"Size",
format!("{:.2} MB", meta_size_mb)
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"Largest Partition",
if skew.avg_partition_size > 0 {
format!("{:.2} MB", largest_mb)
} else {
"N/A".to_string()
},
"Avg Size",
if meta.metadata_file_count > 0 {
format!("{:.2} MB", avg_meta_mb)
} else {
"N/A".to_string()
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"Smallest Partition",
if skew.avg_partition_size > 0 {
format!("{:.2} MB", smallest_mb)
} else {
"N/A".to_string()
},
"Manifests",
format!("{}", meta.manifest_file_count)
)?;
// Truncate file names if too long
let first_file = meta.first_file_name.as_deref().unwrap_or("N/A");
let first_file_display = if first_file.len() > 30 {
format!("...{}", &first_file[first_file.len() - 27..])
} else {
first_file.to_string()
};
let last_file = meta.last_file_name.as_deref().unwrap_or("N/A");
let last_file_display = if last_file.len() > 30 {
format!("...{}", &last_file[last_file.len() - 27..])
} else {
last_file.to_string()
};
writeln!(
f,
" {:<19} {:>8} {:<12} {:>15}",
"Avg Partition Size",
if skew.avg_partition_size > 0 {
format!("{:.2} MB", avg_partition_mb)
} else {
"N/A".to_string()
},
"First File",
first_file_display
)?;
writeln!(
f,
" {:<19} {:>8} {:<12} {:>15}",
"", "", "Last File", last_file_display
)?;
// Snapshot Health and Unreferenced Files (side by side)
writeln!(f)?;
writeln!(f, " {:<41} Unreferenced Files", "Snapshot Health")?;
writeln!(f, "{}", "━".repeat(80))?;
let snap = &report.metrics.snapshot_health;
let has_unreferenced = !report.metrics.unreferenced_files.is_empty();
let wasted_gb = report.metrics.unreferenced_size_bytes as f64 / (1024.0 * 1024.0 * 1024.0);
let wasted_str = if wasted_gb >= 1.0 {
format!("{:.2} GB", wasted_gb)
} else {
let wasted_mb = report.metrics.unreferenced_size_bytes as f64 / (1024.0 * 1024.0);
format!("{:.2} MB", wasted_mb)
};
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"Snapshot Count",
format!("{}", snap.snapshot_count),
"Count",
if has_unreferenced {
format!("{}", report.metrics.unreferenced_files.len())
} else {
"0".to_string()
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<19} {:>8}",
"Retention Risk",
format!("{:.1}%", snap.snapshot_retention_risk * 100.0),
"Wasted Space",
if has_unreferenced {
wasted_str
} else {
"0 MB".to_string()
}
)?;
writeln!(
f,
" {:<19} {:>8}",
"Oldest Snapshot",
if snap.oldest_snapshot_age_days > 0.0 {
format!("{:.1} days", snap.oldest_snapshot_age_days)
} else {
"N/A".to_string()
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<39}",
"Newest Snapshot",
if snap.newest_snapshot_age_days > 0.0 {
format!("{:.1} days", snap.newest_snapshot_age_days)
} else {
"N/A".to_string()
},
if has_unreferenced {
"WARNING: Files exist in storage"
} else {
""
}
)?;
writeln!(
f,
" {:<19} {:>8} {:<39}",
"Avg Snapshot Age",
if snap.avg_snapshot_age_days > 0.0 {
format!("{:.1} days", snap.avg_snapshot_age_days)
} else {
"N/A".to_string()
},
if has_unreferenced {
"but not referenced in metadata"
} else {
""
}
)?;
// Clustering (Iceberg) and Deletion Vectors (Delta) - side by side
// Only show clustering if there are actual clustering columns defined
let has_clustering = report
.metrics
.clustering
.as_ref()
.is_some_and(|c| !c.clustering_columns.is_empty());
let has_deletion_vectors = report.metrics.deletion_vector_metrics.is_some();
if has_clustering || has_deletion_vectors {
writeln!(f)?;
writeln!(
f,
" {:<41} {}",
if has_clustering { "Clustering" } else { "" },
if has_deletion_vectors {
"Deletion Vectors"
} else {
""
}
)?;
writeln!(f, "{}", "━".repeat(80))?;
let max_rows = if has_clustering { 4 } else { 5 };
for i in 0..max_rows {
let left = if has_clustering {
if let Some(ref clustering) = report.metrics.clustering {
match i {
0 => format!(
" {:<19} {:>8}",
"Avg Cluster Size",
format!(
"{:.2} MB",
clustering.avg_cluster_size_bytes / (1024.0 * 1024.0)
)
),
1 => format!(
" {:<19} {:>8}",
"Clusters",
format!("{}", clustering.cluster_count)
),
2 => format!(
" {:<19} {:>8}",
"Avg Files/Cluster",
format!("{:.2}", clustering.avg_files_per_cluster)
),
3 => clustering
.clustering_columns
.iter()
.enumerate()
.map(|(i, v)| {
if i == 0 {
format!(" {:<14} {:>13}", "Columns", v)
} else {
format!(" {:<14} {:>13}", "", v)
}
})
.collect::<Vec<_>>()
.join("\n"),
_ => format!("{:<40}", ""),
}
} else {
format!("{:<40}", "")
}
} else {
format!("{:<40}", "")
};
let right = if let Some(ref dv_metrics) = report.metrics.deletion_vector_metrics {
let dv_size_mb =
dv_metrics.total_deletion_vector_size_bytes as f64 / (1024.0 * 1024.0);
let dv_size_str = if dv_size_mb >= 1.0 {
format!("{:.2} MB", dv_size_mb)
} else {
let dv_size_kb =
dv_metrics.total_deletion_vector_size_bytes as f64 / 1024.0;
format!("{:.2} KB", dv_size_kb)
};
match i {
0 => format!(
" {:<19} {:>8}",
"Vectors",
format!("{}", dv_metrics.deletion_vector_count)
),
1 => format!(" {:<19} {:>8}", "DV Size", dv_size_str),
2 => format!(
" {:<19} {:>8}",
"Deleted Rows",
format!("{}", dv_metrics.deleted_rows_count)
),
3 => format!(
" {:<19} {:>8}",
"Oldest Age (days)",
format!("{:.1}", dv_metrics.deletion_vector_age_days)
),
4 => format!(
" {:<19} {:>8}",
"Impact (0-1)",
format!("{:.2}", dv_metrics.deletion_vector_impact_score)
),
_ => format!("{:<40}", ""),
}
} else {
format!("{:<40}", "")
};
if has_clustering && has_deletion_vectors {
writeln!(f, "{} {}", left, right)?;
} else if has_clustering {
writeln!(f, "{}", left)?;
} else {
writeln!(f, "{:<40} {}", "", right)?;
}
}
}
// Schema Evolution and Time Travel - side by side
let has_schema = report.metrics.schema_evolution.is_some();
let has_time_travel = report.metrics.time_travel_metrics.is_some();
if has_schema || has_time_travel {
writeln!(f)?;
writeln!(
f,
" {:<41} {}",
if has_schema { "Schema Evolution" } else { "" },
if has_time_travel {
"Time Travel Analysis"
} else {
""
}
)?;
writeln!(f, "{}", "━".repeat(80))?;
for i in 0..7 {
let left = if let Some(ref schema_metrics) = report.metrics.schema_evolution {
match i {
0 => format!(
" {:<19} {:>8}",
"Total Changes",
format!("{}", schema_metrics.total_schema_changes)
),
1 => format!(
" {:<19} {:>8}",
"Breaking",
format!("{}", schema_metrics.breaking_changes)
),
2 => format!(
" {:<19} {:>8}",
"Non-Breaking",
format!("{}", schema_metrics.non_breaking_changes)
),
3 => format!(
" {:<19} {:>8}",
"Stability (0-1)",
format!("{:.2}", schema_metrics.schema_stability_score)
),
4 => format!(
" {:<19} {:>8}",
"Days Since Last",
format!("{:.1}d", schema_metrics.days_since_last_change)
),
5 => format!(
" {:<19} {:>8}",
"Change Freq",
format!("{:.1}/d", schema_metrics.schema_change_frequency)
),
6 => format!(
" {:<19} {:>8}",
"Version",
format!("{}", schema_metrics.current_schema_version)
),
_ => format!("{:<40}", ""),
}
} else {
format!("{:<40}", "")
};
let right = if let Some(ref tt_metrics) = report.metrics.time_travel_metrics {
let historical_gb =
tt_metrics.total_historical_size_bytes as f64 / (1024.0 * 1024.0 * 1024.0);
let historical_str = if historical_gb >= 1.0 {
format!("{:.2} GB", historical_gb)
} else {
let historical_mb =
tt_metrics.total_historical_size_bytes as f64 / (1024.0 * 1024.0);
format!("{:.2} MB", historical_mb)
};
match i {
0 => format!(
" {:<19} {:>8}",
"Snapshots",
format!("{}", tt_metrics.total_snapshots)
),
1 => format!(
" {:<19} {:>8}",
"Oldest (days)",
format!("{:.1}", tt_metrics.oldest_snapshot_age_days)
),
2 => format!(
" {:<19} {:>8}",
"Newest (days)",
format!("{:.1}", tt_metrics.newest_snapshot_age_days)
),
3 => format!(" {:<19} {:>8}", "Historical Size", historical_str),
4 => format!(
" {:<19} {:>8}",
"Cost Impact (0-1)",
format!("{:.2}", tt_metrics.storage_cost_impact_score)
),
5 => format!(
" {:<19} {:>8}",
"Reten Eff (0-1)",
format!("{:.2}", tt_metrics.retention_efficiency_score)
),
6 => format!(
" {:<19} {:>8}",
"Recommended (days)",
format!("{}", tt_metrics.recommended_retention_days)
),
_ => format!("{:<40}", ""),
}
} else {
format!("{:<40}", "")
};
if has_schema && has_time_travel {
writeln!(f, "{} {}", left, right)?;
} else if has_schema {
writeln!(f, "{}", left)?;
} else {
writeln!(f, "{:<40} {}", "", right)?;
}
}
}
// Table Constraints and File Compaction - side by side
let has_constraints = report.metrics.table_constraints.is_some();
let has_compaction = report.metrics.file_compaction.is_some();
if has_constraints || has_compaction {
writeln!(f)?;
writeln!(
f,
" {:<41} {}",
if has_constraints {
"Table Constraints"
} else {
""
},
if has_compaction {
"File Compaction"
} else {
""
}
)?;
writeln!(f, "{}", "━".repeat(80))?;
let max_rows = 9;
for i in 0..max_rows {
let left = if let Some(ref constraint_metrics) = report.metrics.table_constraints {
match i {
0 => format!(
" {:<19} {:>8}",
"Total",
format!("{}", constraint_metrics.total_constraints)
),
1 => format!(
" {:<19} {:>8}",
"Check",
format!("{}", constraint_metrics.check_constraints)
),
2 => format!(
" {:<19} {:>8}",
"NOT NULL",
format!("{}", constraint_metrics.not_null_constraints)
),
3 => format!(
" {:<19} {:>8}",
"Unique",
format!("{}", constraint_metrics.unique_constraints)
),
4 => format!(
" {:<19} {:>8}",
"Foreign Key",
format!("{}", constraint_metrics.foreign_key_constraints)
),
5 => format!(
" {:<20} {:>7}",
"Violation Risk (0-1)",
format!("{:.2}", constraint_metrics.constraint_violation_risk)
),
6 => format!(
" {:<19} {:>8}",
"Quality Score (0-1)",
format!("{:.2}", constraint_metrics.data_quality_score)
),
7 => format!(
" {:<19} {:>8}",
"Coverage (0-1)",
format!("{:.2}", constraint_metrics.constraint_coverage_score)
),
_ => format!("{:<29}", ""),
}
} else {
format!("{:<40}", "")
};
let right = if let Some(ref compaction_metrics) = report.metrics.file_compaction {
let small_files_mb =
compaction_metrics.small_files_size_bytes as f64 / (1024.0 * 1024.0);
let savings_mb = compaction_metrics.estimated_compaction_savings_bytes as f64
/ (1024.0 * 1024.0);
let savings_str = if savings_mb >= 1.0 {
format!("{:.2} MB", savings_mb)
} else {
let savings_kb =
compaction_metrics.estimated_compaction_savings_bytes as f64 / 1024.0;
format!("{:.2} KB", savings_kb)
};
let target_mb = compaction_metrics.recommended_target_file_size_bytes as f64
/ (1024.0 * 1024.0);
match i {
0 => format!(
" {:<19} {:>8}",
"Opportunity (0-1)",
format!("{:.2}", compaction_metrics.compaction_opportunity_score)
),
1 => format!(
" {:<19} {:>8}",
"Small Files",
format!("{}", compaction_metrics.small_files_count)
),
2 => format!(
" {:<19} {:>8}",
"Small Size",
format!("{:.2} MB", small_files_mb)
),
3 => format!(
" {:<19} {:>8}",
"Potential Files",
format!("{}", compaction_metrics.potential_compaction_files)
),
4 => format!(" {:<19} {:>8}", "Savings", savings_str),
5 => format!(
" {:<19} {:>8}",
"Target Size",
format!("{:.0} MB", target_mb)
),
6 => format!(
" {:<19} {:>8}",
"Priority",
compaction_metrics.compaction_priority.to_uppercase()
),
7 => format!(
" {:<19} {:>8}",
"Z-Order",
if compaction_metrics.z_order_opportunity {
"Yes"
} else {
"No"
}
),
8 => {
if !compaction_metrics.z_order_columns.is_empty() {
let print_rows = compaction_metrics
.z_order_columns
.iter()
.enumerate()
.map(|(i, v)| {
if i == 0 {
format!(" {:<14} {:>13}", "Z-Columns", v)
} else {
format!(" {:<14} {:>13}", "", v)
}
})
.collect::<Vec<_>>();
print_rows.join("\n")
// format!(
// " {:<19} {:>8}",
// "Z-Columns",
// compaction_metrics.z_order_columns.join(", ")
// )
} else {
format!("{:<40}", "")
}
}
_ => format!("{:<40}", ""),
}
} else {
format!("{:<40}", "")
};
// Skip empty rows (e.g., row 8 when no Z-Order columns)
let left_empty = left.trim().is_empty();
let right_empty = right.trim().is_empty();
if left_empty && right_empty {
continue;
}
if has_constraints && has_compaction {
writeln!(f, "{} {}", left, right)?;
} else if has_constraints {
writeln!(f, "{}", left)?;
} else {
writeln!(f, "{:<40} {}", "", right)?;
}
}
}
// Format-Specific Metrics
if let Some(ref delta_metrics) = report.metrics.delta_table_specific_metrics {
write!(f, "{}", delta_metrics)?;
}
if let Some(ref iceberg_metrics) = report.metrics.iceberg_table_specific_metrics {
write!(f, "{}", iceberg_metrics)?;
}
#[cfg(feature = "hudi")]
if let Some(ref hudi_metrics) = report.metrics.hudi_table_specific_metrics {
write!(f, "{}", hudi_metrics)?;
}
#[cfg(feature = "lance")]
if let Some(ref lance_metrics) = report.metrics.lance_table_specific_metrics {
write!(f, "{}", lance_metrics)?;
}
// Recommendations (full width)
writeln!(f)?;
writeln!(f, " Recommendations")?;
writeln!(f, "{}", "━".repeat(80))?;
if !report.metrics.recommendations.is_empty() {
for (i, rec) in report.metrics.recommendations.iter().enumerate() {
writeln!(f, " {}. {}", i + 1, rec)?;
}
} else {
writeln!(f, " No recommendations - table is in excellent health!")?;
}
Ok(())
}
}
impl HealthReport {
pub fn to_json(&self, exclude_files: bool) -> Result<String, JsonError> {
if exclude_files {
let mut report = self.clone();
report.metrics.unreferenced_files = Vec::new();
report
.metrics
.partitions
.iter_mut()
.for_each(|p| p.files = Vec::new());
serde_json::to_string_pretty(&report)
} else {
serde_json::to_string_pretty(self)
}
}
}
impl Default for HealthMetrics {
fn default() -> Self {
Self::new()
}
}
impl HealthMetrics {
pub fn new() -> Self {
Self {
total_files: 0,
total_size_bytes: 0,
unreferenced_files: Vec::new(),
unreferenced_size_bytes: 0,
partition_count: 0,
partitions: Vec::new(),
clustering: None,
avg_file_size_bytes: 0.0,
file_size_distribution: FileSizeDistribution {
small_files: 0,
medium_files: 0,
large_files: 0,
very_large_files: 0,
},
recommendations: Vec::new(),
health_score: 0.0,
data_skew: DataSkewMetrics {
partition_skew_score: 0.0,
file_size_skew_score: 0.0,
largest_partition_size: 0,
smallest_partition_size: 0,
avg_partition_size: 0,
partition_size_std_dev: 0.0,
},
metadata_health: MetadataHealth {
metadata_file_count: 0,
metadata_total_size_bytes: 0,
avg_metadata_file_size: 0.0,
metadata_growth_rate: 0.0,
manifest_file_count: 0,
first_file_name: None,
last_file_name: None,
},
snapshot_health: SnapshotHealth {
snapshot_count: 0,
oldest_snapshot_age_days: 0.0,
newest_snapshot_age_days: 0.0,
avg_snapshot_age_days: 0.0,
snapshot_retention_risk: 0.0,
},
deletion_vector_metrics: None,
schema_evolution: None,
time_travel_metrics: None,
table_constraints: None,
file_compaction: None,
delta_table_specific_metrics: None,
#[cfg(feature = "hudi")]
hudi_table_specific_metrics: None,
iceberg_table_specific_metrics: None,
#[cfg(feature = "lance")]
lance_table_specific_metrics: None,
}
}
pub fn calculate_health_score(&self) -> f64 {
let mut score = 1.0;
// Penalize unreferenced files
if self.total_files > 0 {
let unreferenced_ratio = self.unreferenced_files.len() as f64 / self.total_files as f64;
score -= unreferenced_ratio * 0.3;
}
// Penalize small files (inefficient)
if self.total_files > 0 {
let small_file_ratio =
self.file_size_distribution.small_files as f64 / self.total_files as f64;
score -= small_file_ratio * 0.2;
}
// Penalize very large files (potential performance issues)
if self.total_files > 0 {
let very_large_ratio =
self.file_size_distribution.very_large_files as f64 / self.total_files as f64;
score -= very_large_ratio * 0.1;
}
// Reward good partitioning
if self.partition_count > 0 && self.total_files > 0 {
let avg_files_per_partition = self.total_files as f64 / self.partition_count as f64;
if avg_files_per_partition > 100.0 {
score -= 0.1; // Too many files per partition
} else if avg_files_per_partition < 5.0 {
score -= 0.05; // Too few files per partition
}
}
// Penalize data skew
score -= self.data_skew.partition_skew_score * 0.15;
score -= self.data_skew.file_size_skew_score * 0.1;
// Penalize metadata bloat
if self.metadata_health.metadata_total_size_bytes > 100 * 1024 * 1024 {
// > 100MB
score -= 0.05;
}
// Penalize snapshot retention issues
score -= self.snapshot_health.snapshot_retention_risk * 0.1;
// Penalize deletion vector impact
if let Some(ref dv_metrics) = self.deletion_vector_metrics {
score -= dv_metrics.deletion_vector_impact_score * 0.15;
}
// Factor in schema stability
if let Some(ref schema_metrics) = self.schema_evolution {
score -= (1.0 - schema_metrics.schema_stability_score) * 0.2;
}
// Factor in time travel storage costs
if let Some(ref tt_metrics) = self.time_travel_metrics {
score -= tt_metrics.storage_cost_impact_score * 0.1;
score -= (1.0 - tt_metrics.retention_efficiency_score) * 0.05;
}
// Factor in data quality from constraints
if let Some(ref constraint_metrics) = self.table_constraints {
score -= (1.0 - constraint_metrics.data_quality_score) * 0.15;
score -= constraint_metrics.constraint_violation_risk * 0.1;
}
// Factor in file compaction opportunities
if let Some(ref compaction_metrics) = self.file_compaction {
score -= (1.0 - compaction_metrics.compaction_opportunity_score) * 0.1;
}
score.clamp(0.0, 1.0)
}
pub fn calculate_data_skew(&mut self) {
if self.partitions.is_empty() {
return;
}
let partition_sizes: Vec<u64> =
self.partitions.iter().map(|p| p.total_size_bytes).collect();
let file_counts: Vec<usize> = self.partitions.iter().map(|p| p.file_count).collect();
// Calculate partition size skew
if !partition_sizes.is_empty() {
let total_size: u64 = partition_sizes.iter().sum();
let avg_size = total_size as f64 / partition_sizes.len() as f64;
let variance = partition_sizes
.iter()
.map(|&size| (size as f64 - avg_size).powi(2))
.sum::<f64>()
/ partition_sizes.len() as f64;
let std_dev = variance.sqrt();
let coefficient_of_variation = if avg_size > 0.0 {
std_dev / avg_size
} else {
0.0
};
self.data_skew.partition_skew_score = coefficient_of_variation.min(1.0);
self.data_skew.largest_partition_size = *partition_sizes.iter().max().unwrap_or(&0);
self.data_skew.smallest_partition_size = *partition_sizes.iter().min().unwrap_or(&0);
self.data_skew.avg_partition_size = avg_size as u64;
self.data_skew.partition_size_std_dev = std_dev;
}
// Calculate file count skew
if !file_counts.is_empty() {
let total_files: usize = file_counts.iter().sum();
let avg_files = total_files as f64 / file_counts.len() as f64;
let variance = file_counts
.iter()
.map(|&count| (count as f64 - avg_files).powi(2))
.sum::<f64>()
/ file_counts.len() as f64;
let std_dev = variance.sqrt();
let coefficient_of_variation = if avg_files > 0.0 {
std_dev / avg_files
} else {
0.0
};
self.data_skew.file_size_skew_score = coefficient_of_variation.min(1.0);
}
}
pub fn calculate_snapshot_health(&mut self, snapshot_count: usize) {
self.snapshot_health.snapshot_count = snapshot_count;
// Simplified snapshot age calculation (would need actual timestamps)
self.snapshot_health.oldest_snapshot_age_days = 0.0;
self.snapshot_health.newest_snapshot_age_days = 0.0;
self.snapshot_health.avg_snapshot_age_days = 0.0;
// Calculate retention risk based on snapshot count
if snapshot_count > 100 {
self.snapshot_health.snapshot_retention_risk = 0.8;
} else if snapshot_count > 50 {
self.snapshot_health.snapshot_retention_risk = 0.5;
} else if snapshot_count > 20 {
self.snapshot_health.snapshot_retention_risk = 0.2;
} else {
self.snapshot_health.snapshot_retention_risk = 0.0;
}
}
pub fn generate_recommendations(&mut self) {
// Check for unreferenced files
if !self.unreferenced_files.is_empty() {
self.recommendations.push(format!(
"Found {} unreferenced files ({} bytes). Consider cleaning up orphaned data files.",
self.unreferenced_files.len(),
self.unreferenced_size_bytes
));
}
// Check file size distribution
let total_files = self.total_files as f64;
if total_files > 0.0 {
let small_file_ratio = self.file_size_distribution.small_files as f64 / total_files;
if small_file_ratio > 0.5 {
self.recommendations.push(
"High percentage of small files detected. Consider compacting to improve query performance.".to_string()
);
}
let very_large_ratio =
self.file_size_distribution.very_large_files as f64 / total_files;
if very_large_ratio > 0.1 {
self.recommendations.push(
"Some very large files detected. Consider splitting large files for better parallelism.".to_string()
);
}
}
// Check partitioning
if self.partition_count > 0 {
let avg_files_per_partition = total_files / self.partition_count as f64;
if avg_files_per_partition > 100.0 {
self.recommendations.push(
"High number of files per partition. Consider repartitioning to reduce file count.".to_string()
);
} else if avg_files_per_partition < 5.0 {
self.recommendations.push(
"Low number of files per partition. Consider consolidating partitions."
.to_string(),
);
}
}
// Check for empty partitions
let empty_partitions = self.partitions.iter().filter(|p| p.file_count == 0).count();
if empty_partitions > 0 {
self.recommendations.push(format!(
"Found {} empty partitions. Consider removing empty partition directories.",
empty_partitions
));
}
// Check data skew
if self.data_skew.partition_skew_score > 0.5 {
self.recommendations.push(
"High partition skew detected. Consider repartitioning to balance data distribution.".to_string()
);
}
if self.data_skew.file_size_skew_score > 0.5 {
self.recommendations.push(
"High file size skew detected. Consider running OPTIMIZE to balance file sizes."
.to_string(),
);
}
// Check metadata health
if self.metadata_health.metadata_total_size_bytes > 50 * 1024 * 1024 {
// > 50MB
self.recommendations.push(
"Large metadata size detected. Consider running VACUUM to clean up old transaction logs.".to_string()
);
}
// Check snapshot health
if self.snapshot_health.snapshot_retention_risk > 0.7 {
self.recommendations.push(
"High snapshot retention risk. Consider running VACUUM to remove old snapshots."
.to_string(),
);
}
// Check clustering
if let Some(ref clustering) = self.clustering {
if clustering.avg_files_per_cluster > 50.0 {
self.recommendations.push(
"High number of files per cluster. Consider optimizing clustering strategy."
.to_string(),
);
}
if clustering.clustering_columns.len() > 4 {
self.recommendations.push(
"Too many clustering columns detected. Consider reducing to 4 or fewer columns for optimal performance.".to_string()
);
}
if clustering.clustering_columns.is_empty() {
self.recommendations.push(
"No clustering detected. Consider enabling liquid clustering for better query performance.".to_string()
);
}
}
// Check deletion vectors
if let Some(ref dv_metrics) = self.deletion_vector_metrics {
if dv_metrics.deletion_vector_impact_score > 0.7 {
self.recommendations.push(
"High deletion vector impact detected. Consider running VACUUM to clean up old deletion vectors.".to_string()
);
}
if dv_metrics.deletion_vector_count > 50 {
self.recommendations.push(
"Many deletion vectors detected. Consider optimizing delete operations to reduce fragmentation.".to_string()
);
}
if dv_metrics.deletion_vector_age_days > 30.0 {
self.recommendations.push(
"Old deletion vectors detected. Consider running VACUUM to clean up deletion vectors older than 30 days.".to_string()
);
}
}
// Check schema evolution
if let Some(ref schema_metrics) = self.schema_evolution {
if schema_metrics.schema_stability_score < 0.5 {
self.recommendations.push(
"Unstable schema detected. Consider planning schema changes more carefully to improve performance.".to_string()
);
}
if schema_metrics.breaking_changes > 5 {
self.recommendations.push(
"Many breaking schema changes detected. Consider using schema evolution features to avoid breaking changes.".to_string()
);
}
if schema_metrics.schema_change_frequency > 1.0 {
self.recommendations.push(
"High schema change frequency detected. Consider batching schema changes to reduce performance impact.".to_string()
);
}
if schema_metrics.days_since_last_change < 1.0 {
self.recommendations.push(
"Recent schema changes detected. Monitor query performance for potential issues.".to_string()
);
}
}
// Check time travel storage costs
if let Some(ref tt_metrics) = self.time_travel_metrics {
if tt_metrics.storage_cost_impact_score > 0.7 {
self.recommendations.push(
"High time travel storage costs detected. Consider running VACUUM to clean up old snapshots.".to_string()
);
}
if tt_metrics.retention_efficiency_score < 0.5 {
self.recommendations.push(
"Inefficient snapshot retention detected. Consider optimizing retention policy.".to_string()
);
}
if tt_metrics.total_snapshots > 1000 {
self.recommendations.push(
"High snapshot count detected. Consider reducing retention period to improve performance.".to_string()
);
}
}
// Check table constraints
if let Some(ref constraint_metrics) = self.table_constraints {
if constraint_metrics.data_quality_score < 0.5 {
self.recommendations.push(
"Low data quality score detected. Consider adding more table constraints."
.to_string(),
);
}
if constraint_metrics.constraint_violation_risk > 0.7 {
self.recommendations.push(
"High constraint violation risk detected. Monitor data quality and consider data validation.".to_string()
);
}
if constraint_metrics.constraint_coverage_score < 0.3 {
self.recommendations.push(
"Low constraint coverage detected. Consider adding check constraints for better data quality.".to_string()
);
}
}
// Check file compaction opportunities
if let Some(ref compaction_metrics) = self.file_compaction {
if compaction_metrics.compaction_opportunity_score > 0.7 {
self.recommendations.push(
"High file compaction opportunity detected. Consider running OPTIMIZE to improve performance.".to_string()
);
}
if compaction_metrics.compaction_priority == "critical" {
self.recommendations.push(
"Critical compaction priority detected. Run OPTIMIZE immediately to improve query performance.".to_string()
);
}
if compaction_metrics.z_order_opportunity {
self.recommendations.push(
format!("Z-ordering opportunity detected. Consider running OPTIMIZE ZORDER BY ({}) to improve query performance.",
compaction_metrics.z_order_columns.join(", ")).to_string()
);
}
if compaction_metrics.estimated_compaction_savings_bytes > 100 * 1024 * 1024 {
// > 100MB
let savings_mb = compaction_metrics.estimated_compaction_savings_bytes as f64
/ (1024.0 * 1024.0);
self.recommendations.push(
format!("Significant compaction savings available: {:.1} MB. Consider running OPTIMIZE.", savings_mb).to_string()
);
}
}
// Check Lance-specific index recommendations
#[cfg(feature = "lance")]
if let Some(ref lance_metrics) = self.lance_table_specific_metrics {
let index_info = &lance_metrics.index_info;
// Recommend creating indices for large tables without any indices
if index_info.num_indices == 0 {
if let Some(num_rows) = lance_metrics.metadata.num_rows {
if num_rows > 10_000 {
self.recommendations.push(
"No indices found on Lance table. Consider creating vector or scalar indices for frequently queried columns.".to_string()
);
}
}
}
// Check for tables with many fragments that could benefit from indexing
if lance_metrics.fragment_info.num_fragments > 100 && index_info.num_indices == 0 {
self.recommendations.push(
"Large number of fragments detected without indices. Consider creating indices to improve query performance.".to_string()
);
}
// Check for high deletion ratio that might affect index performance
if let Some(num_deleted) = lance_metrics.metadata.num_deleted_rows {
if let Some(num_rows) = lance_metrics.metadata.num_rows {
if num_rows > 0 {
let deletion_ratio = num_deleted as f64 / (num_rows + num_deleted) as f64;
if deletion_ratio > 0.2 && index_info.num_indices > 0 {
self.recommendations.push(
"High deletion ratio detected with existing indices. Consider rebuilding indices after compaction.".to_string()
);
}
}
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_chrome_tracing_with_real_data_reproduces_overflow() {
// This test reproduces the overflow bug with real Chrome tracing data
// The input has timestamps in microseconds (very large numbers)
let chrome_trace_input = r#"[{"cat":"PERF","name":"storage_config_new_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"storage_config_new_dur","ph":"E","pid":"1","ts":1762455078690000}]"#;
let events: Vec<serde_json::Value> =
serde_json::from_str(chrome_trace_input).expect("Failed to parse Chrome tracing JSON");
// Extract timing data - timestamps are in microseconds
let begin_ts = events[0]["ts"].as_u64().unwrap() as u128;
let end_ts = events[1]["ts"].as_u64().unwrap() as u128;
println!("Begin timestamp (microseconds): {}", begin_ts);
println!("End timestamp (microseconds): {}", end_ts);
println!("Duration (microseconds): {}", end_ts - begin_ts);
// The current code expects milliseconds and multiplies by 1000
// If we naively use these microsecond values as milliseconds, we get overflow
let mut collection = LinkedList::new();
// This is what would cause overflow - treating microseconds as milliseconds
// collection.push_back(("storage_config_new_dur".to_string(), begin_ts, end_ts - begin_ts));
// The correct approach: timestamps in the input are already in microseconds
// So we should NOT multiply by 1000 in to_chrome_tracing(), OR
// we need to store timestamps in milliseconds in our internal format
// For now, let's test with the corrected approach (divide by 1000 to get milliseconds)
collection.push_back((
"storage_config_new_dur".to_string(),
begin_ts / 1000, // Convert to milliseconds
(end_ts - begin_ts) / 1000, // Duration in milliseconds
));
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
// This should work without overflow now
let output = metrics
.to_chrome_tracing()
.expect("Should generate Chrome tracing output");
let output_events: Vec<serde_json::Value> =
serde_json::from_str(&output).expect("Should parse output");
// Verify the output
assert_eq!(output_events.len(), 2);
// The output timestamps should be in microseconds (original format)
let output_begin_ts = output_events[0]["ts"].as_u64().unwrap();
let output_end_ts = output_events[1]["ts"].as_u64().unwrap();
println!("Output begin timestamp: {}", output_begin_ts);
println!("Output end timestamp: {}", output_end_ts);
// Verify the timestamps are approximately correct (within rounding error)
assert!(
(output_begin_ts as i128 - begin_ts as i128).abs() < 1000,
"Begin timestamp should be approximately preserved"
);
assert!(
(output_end_ts as i128 - end_ts as i128).abs() < 1000,
"End timestamp should be approximately preserved"
);
}
#[test]
#[should_panic(expected = "attempt to multiply with overflow")]
fn test_chrome_tracing_overflow_bug() {
// This test demonstrates the overflow bug when using very large timestamps
let mut collection = LinkedList::new();
// Use a timestamp that will overflow when multiplied by 1000
// u128::MAX / 1000 = 340282366920938463463374607431768211
// Anything larger than this will overflow
let large_timestamp: u128 = u128::MAX / 500; // This will overflow when multiplied by 1000
collection.push_back(("test_metric".to_string(), large_timestamp, 1000));
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
// This should panic with overflow in debug mode
let _ = metrics.to_chrome_tracing();
}
#[test]
fn test_parse_full_chrome_trace_and_regenerate() {
// This test parses the FULL Chrome tracing JSON provided by the user
// and attempts to regenerate it, which should expose any overflow issues
let chrome_trace_input = r#"[{"cat":"PERF","name":"storage_config_new_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"storage_config_new_dur","ph":"E","pid":"1","ts":1762455078690000},{"cat":"PERF","name":"analyzer_new_dur","ph":"B","pid":"1","ts":1762455078691000},{"cat":"PERF","name":"analyzer_new_dur","ph":"E","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"validate_connection_dur","ph":"B","pid":"1","ts":1762455078973000},{"cat":"PERF","name":"validate_connection_dur","ph":"E","pid":"1","ts":1762455080930000},{"cat":"PERF","name":"discover_partitions","ph":"B","pid":"1","ts":1762455080932000},{"cat":"PERF","name":"discover_partitions","ph":"E","pid":"1","ts":1762455081493000},{"cat":"PERF","name":"list_files_parallel","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"list_files_parallel","ph":"E","pid":"1","ts":1762455082170000},{"cat":"PERF","name":"detect_table_type","ph":"B","pid":"1","ts":1762455082171000},{"cat":"PERF","name":"detect_table_type","ph":"E","pid":"1","ts":1762455082172000},{"cat":"PERF","name":"categorize_files","ph":"B","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"categorize_files","ph":"E","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"find_referenced_files","ph":"B","pid":"1","ts":1762455082174000},{"cat":"PERF","name":"find_referenced_files","ph":"E","pid":"1","ts":1762455082683000},{"cat":"PERF","name":"find_unreferenced_files","ph":"B","pid":"1","ts":1762455082684000},{"cat":"PERF","name":"find_unreferenced_files","ph":"E","pid":"1","ts":1762455082685000},{"cat":"PERF","name":"analyze_partitioning","ph":"B","pid":"1","ts":1762455082687000},{"cat":"PERF","name":"analyze_partitioning","ph":"E","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"B","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"E","pid":"1","ts":1762455083173000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"B","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"E","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_metadata_health","ph":"B","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_metadata_health","ph":"E","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_data_skew","ph":"B","pid":"1","ts":1762455083177000},{"cat":"PERF","name":"calculate_data_skew","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"B","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"analyze_file_compaction","ph":"B","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"analyze_file_compaction","ph":"E","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"generate_recommendations","ph":"B","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"generate_recommendations","ph":"E","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"calculate_health_score","ph":"B","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"calculate_health_score","ph":"E","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"E","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"B","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"E","pid":"1","ts":1762455085985000},{"cat":"PERF","name":"analyze_total_dur","ph":"B","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"analyze_total_dur","ph":"E","pid":"1","ts":1762455085987000},{"cat":"PERF","name":"total_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"total_dur","ph":"E","pid":"1","ts":1762455085987000}]"#;
// Parse the Chrome tracing JSON
let events: Vec<serde_json::Value> =
serde_json::from_str(chrome_trace_input).expect("Failed to parse Chrome tracing JSON");
// Extract timing data from the parsed events
let mut timing_map: HashMap<String, (u128, u128)> = HashMap::new();
for event in &events {
let name = event["name"].as_str().expect("Event should have a name");
let ts = event["ts"].as_u64().expect("Event should have a timestamp") as u128;
let phase = event["ph"].as_str().expect("Event should have a phase");
if phase == "B" {
// Begin event - store start time (in microseconds)
timing_map.entry(name.to_string()).or_insert((ts, 0)).0 = ts;
} else if phase == "E" {
// End event - calculate duration (in microseconds)
if let Some(entry) = timing_map.get_mut(name) {
entry.1 = ts - entry.0;
}
}
}
// Create TimedLikeMetrics from the parsed data
// Convert from microseconds to milliseconds for internal storage
let mut collection = LinkedList::new();
for (name, (start_ts_us, duration_us)) in timing_map.iter() {
collection.push_back((
name.clone(),
start_ts_us / 1000, // Convert microseconds to milliseconds
duration_us / 1000, // Convert microseconds to milliseconds
));
}
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
// Generate Chrome tracing output - this should work without overflow
let output = metrics
.to_chrome_tracing()
.expect("Should generate Chrome tracing output without overflow");
// Parse the generated output
let output_events: Vec<serde_json::Value> =
serde_json::from_str(&output).expect("Should parse generated output");
// Verify we have the correct number of events
assert_eq!(
output_events.len(),
44,
"Should have 44 events (22 metrics * 2)"
);
// Verify all expected metrics are present
let metric_names: std::collections::HashSet<String> = output_events
.iter()
.filter_map(|e| e["name"].as_str().map(String::from))
.collect();
assert!(metric_names.contains("storage_config_new_dur"));
assert!(metric_names.contains("analyzer_new_dur"));
assert!(metric_names.contains("validate_connection_dur"));
assert!(metric_names.contains("total_dur"));
assert!(metric_names.contains("delta_reader"));
}
#[test]
fn test_chrome_tracing_empty_metrics() {
let metrics = TimedLikeMetrics {
duration_collection: LinkedList::new(),
};
let output = metrics
.to_chrome_tracing()
.expect("Should handle empty metrics");
let events: Vec<serde_json::Value> =
serde_json::from_str(&output).expect("Should parse empty array");
assert_eq!(
events.len(),
0,
"Empty metrics should produce empty events array"
);
}
#[test]
fn test_chrome_tracing_single_metric() {
let mut collection = LinkedList::new();
collection.push_back(("test_metric".to_string(), 1000, 500));
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
let output = metrics
.to_chrome_tracing()
.expect("Should generate output for single metric");
let events: Vec<serde_json::Value> =
serde_json::from_str(&output).expect("Should parse output");
assert_eq!(
events.len(),
2,
"Single metric should produce 2 events (B and E)"
);
// Verify Begin event
assert_eq!(events[0]["name"].as_str(), Some("test_metric"));
assert_eq!(events[0]["ph"].as_str(), Some("B"));
assert_eq!(events[0]["ts"].as_u64(), Some(1000000)); // 1000ms * 1000
// Verify End event
assert_eq!(events[1]["name"].as_str(), Some("test_metric"));
assert_eq!(events[1]["ph"].as_str(), Some("E"));
assert_eq!(events[1]["ts"].as_u64(), Some(1500000)); // (1000 + 500)ms * 1000
}
#[test]
fn test_chrome_tracing_duration_calculation() {
let mut collection = LinkedList::new();
// Add metrics with known durations
collection.push_back(("metric1".to_string(), 0, 100));
collection.push_back(("metric2".to_string(), 50, 200));
collection.push_back(("metric3".to_string(), 100, 50));
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
let output = metrics.to_chrome_tracing().expect("Should generate output");
let events: Vec<serde_json::Value> =
serde_json::from_str(&output).expect("Should parse output");
// Should have 6 events (3 metrics * 2 events each)
assert_eq!(events.len(), 6);
// Verify metric1 duration
let metric1_begin = events
.iter()
.find(|e| e["name"] == "metric1" && e["ph"] == "B")
.expect("Should find metric1 begin");
let metric1_end = events
.iter()
.find(|e| e["name"] == "metric1" && e["ph"] == "E")
.expect("Should find metric1 end");
let duration = metric1_end["ts"].as_u64().unwrap() - metric1_begin["ts"].as_u64().unwrap();
assert_eq!(
duration, 100000,
"metric1 duration should be 100ms (100000 microseconds)"
);
}
/// This test documents the proper way to use Chrome tracing data with TimedLikeMetrics
///
/// Chrome tracing format uses microseconds for timestamps, but TimedLikeMetrics
/// expects milliseconds internally and multiplies by 1000 when generating output.
///
/// IMPORTANT: When parsing Chrome tracing JSON, you MUST divide timestamps by 1000
/// to convert from microseconds to milliseconds before storing in TimedLikeMetrics.
#[test]
fn test_chrome_tracing_format_documentation() {
// Chrome tracing uses microseconds
let chrome_ts_microseconds: u128 = 1762455078689000;
// TimedLikeMetrics expects milliseconds
let internal_ts_milliseconds: u128 = chrome_ts_microseconds / 1000;
// When to_chrome_tracing() is called, it multiplies by 1000 to get back to microseconds
let output_ts_microseconds: u128 = internal_ts_milliseconds * 1000;
// Verify the round-trip works correctly
assert_eq!(
chrome_ts_microseconds, output_ts_microseconds,
"Chrome tracing timestamps should round-trip correctly"
);
println!("Chrome tracing format:");
println!(" Input (microseconds): {}", chrome_ts_microseconds);
println!(" Internal (milliseconds): {}", internal_ts_milliseconds);
println!(" Output (microseconds): {}", output_ts_microseconds);
}
/// This test verifies that the ASCII Gantt chart bug is FIXED
/// It uses the ACTUAL Chrome tracing data with absolute timestamps
/// and should NOT panic after the fix to ascii_gantt.rs
#[test]
fn test_chrome_trace_with_absolute_timestamps_fixed() {
let chrome_trace_input = r#"[{"cat":"PERF","name":"storage_config_new_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"storage_config_new_dur","ph":"E","pid":"1","ts":1762455078690000},{"cat":"PERF","name":"analyzer_new_dur","ph":"B","pid":"1","ts":1762455078691000},{"cat":"PERF","name":"analyzer_new_dur","ph":"E","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"validate_connection_dur","ph":"B","pid":"1","ts":1762455078973000},{"cat":"PERF","name":"validate_connection_dur","ph":"E","pid":"1","ts":1762455080930000},{"cat":"PERF","name":"discover_partitions","ph":"B","pid":"1","ts":1762455080932000},{"cat":"PERF","name":"discover_partitions","ph":"E","pid":"1","ts":1762455081493000},{"cat":"PERF","name":"list_files_parallel","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"list_files_parallel","ph":"E","pid":"1","ts":1762455082170000},{"cat":"PERF","name":"detect_table_type","ph":"B","pid":"1","ts":1762455082171000},{"cat":"PERF","name":"detect_table_type","ph":"E","pid":"1","ts":1762455082172000},{"cat":"PERF","name":"categorize_files","ph":"B","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"categorize_files","ph":"E","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"find_referenced_files","ph":"B","pid":"1","ts":1762455082174000},{"cat":"PERF","name":"find_referenced_files","ph":"E","pid":"1","ts":1762455082683000},{"cat":"PERF","name":"find_unreferenced_files","ph":"B","pid":"1","ts":1762455082684000},{"cat":"PERF","name":"find_unreferenced_files","ph":"E","pid":"1","ts":1762455082685000},{"cat":"PERF","name":"analyze_partitioning","ph":"B","pid":"1","ts":1762455082687000},{"cat":"PERF","name":"analyze_partitioning","ph":"E","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"B","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"E","pid":"1","ts":1762455083173000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"B","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"E","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_metadata_health","ph":"B","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_metadata_health","ph":"E","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_data_skew","ph":"B","pid":"1","ts":1762455083177000},{"cat":"PERF","name":"calculate_data_skew","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"B","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"analyze_file_compaction","ph":"B","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"analyze_file_compaction","ph":"E","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"generate_recommendations","ph":"B","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"generate_recommendations","ph":"E","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"calculate_health_score","ph":"B","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"calculate_health_score","ph":"E","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"E","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"B","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"E","pid":"1","ts":1762455085985000},{"cat":"PERF","name":"analyze_total_dur","ph":"B","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"analyze_total_dur","ph":"E","pid":"1","ts":1762455085987000},{"cat":"PERF","name":"total_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"total_dur","ph":"E","pid":"1","ts":1762455085987000}]"#;
// Parse the Chrome tracing JSON
let events: Vec<serde_json::Value> =
serde_json::from_str(chrome_trace_input).expect("Failed to parse Chrome tracing JSON");
// Extract timing data from the parsed events
let mut timing_map: HashMap<String, (u128, u128)> = HashMap::new();
for event in &events {
let name = event["name"].as_str().expect("Event should have a name");
let ts = event["ts"].as_u64().expect("Event should have a timestamp") as u128;
let phase = event["ph"].as_str().expect("Event should have a phase");
if phase == "B" {
timing_map.entry(name.to_string()).or_insert((ts, 0)).0 = ts;
} else if phase == "E" {
if let Some(entry) = timing_map.get_mut(name) {
entry.1 = ts - entry.0;
}
}
}
// Create TimedLikeMetrics with ABSOLUTE timestamps (in milliseconds)
// This is the scenario that was causing the overflow bug
let mut collection = LinkedList::new();
for (name, (start_ts_us, duration_us)) in timing_map.iter() {
// Convert from microseconds to milliseconds
// These will be LARGE numbers like 1762455078689
collection.push_back((name.clone(), start_ts_us / 1000, duration_us / 1000));
}
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
// This should NOW work without panic after the fix to ascii_gantt.rs
let gantt_output = metrics
.duration_collection_as_gantt(None)
.expect("Should generate ASCII Gantt chart with absolute timestamps after fix");
println!("\nASCII Gantt Chart with absolute timestamps:");
println!("{}", gantt_output);
// Verify the output contains expected metrics
assert!(gantt_output.contains("storage_config_new_dur"));
assert!(gantt_output.contains("analyzer_new_dur"));
assert!(gantt_output.contains("total_dur"));
assert!(gantt_output.contains("Timeline"));
}
/// This test parses the FULL Chrome tracing data and works correctly
/// by using RELATIVE timestamps (offset from the minimum) instead of absolute timestamps
#[test]
fn test_chrome_trace_with_relative_timestamps() {
let chrome_trace_input = r#"[{"cat":"PERF","name":"storage_config_new_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"storage_config_new_dur","ph":"E","pid":"1","ts":1762455078690000},{"cat":"PERF","name":"analyzer_new_dur","ph":"B","pid":"1","ts":1762455078691000},{"cat":"PERF","name":"analyzer_new_dur","ph":"E","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"validate_connection_dur","ph":"B","pid":"1","ts":1762455078973000},{"cat":"PERF","name":"validate_connection_dur","ph":"E","pid":"1","ts":1762455080930000},{"cat":"PERF","name":"discover_partitions","ph":"B","pid":"1","ts":1762455080932000},{"cat":"PERF","name":"discover_partitions","ph":"E","pid":"1","ts":1762455081493000},{"cat":"PERF","name":"list_files_parallel","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"list_files_parallel","ph":"E","pid":"1","ts":1762455082170000},{"cat":"PERF","name":"detect_table_type","ph":"B","pid":"1","ts":1762455082171000},{"cat":"PERF","name":"detect_table_type","ph":"E","pid":"1","ts":1762455082172000},{"cat":"PERF","name":"categorize_files","ph":"B","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"categorize_files","ph":"E","pid":"1","ts":1762455082173000},{"cat":"PERF","name":"find_referenced_files","ph":"B","pid":"1","ts":1762455082174000},{"cat":"PERF","name":"find_referenced_files","ph":"E","pid":"1","ts":1762455082683000},{"cat":"PERF","name":"find_unreferenced_files","ph":"B","pid":"1","ts":1762455082684000},{"cat":"PERF","name":"find_unreferenced_files","ph":"E","pid":"1","ts":1762455082685000},{"cat":"PERF","name":"analyze_partitioning","ph":"B","pid":"1","ts":1762455082687000},{"cat":"PERF","name":"analyze_partitioning","ph":"E","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"B","pid":"1","ts":1762455082689000},{"cat":"PERF","name":"update_metrics_from_metadata","ph":"E","pid":"1","ts":1762455083173000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"B","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_file_size_distribution","ph":"E","pid":"1","ts":1762455083175000},{"cat":"PERF","name":"calculate_metadata_health","ph":"B","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_metadata_health","ph":"E","pid":"1","ts":1762455083176000},{"cat":"PERF","name":"calculate_data_skew","ph":"B","pid":"1","ts":1762455083177000},{"cat":"PERF","name":"calculate_data_skew","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"B","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"calculate_snapshot_health","ph":"E","pid":"1","ts":1762455083179000},{"cat":"PERF","name":"analyze_file_compaction","ph":"B","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"analyze_file_compaction","ph":"E","pid":"1","ts":1762455083180000},{"cat":"PERF","name":"generate_recommendations","ph":"B","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"generate_recommendations","ph":"E","pid":"1","ts":1762455083181000},{"cat":"PERF","name":"calculate_health_score","ph":"B","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"calculate_health_score","ph":"E","pid":"1","ts":1762455083182000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"B","pid":"1","ts":1762455081495000},{"cat":"PERF","name":"analyze_after_validation_dur","ph":"E","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"B","pid":"1","ts":1762455083183000},{"cat":"PERF","name":"delta_reader","ph":"E","pid":"1","ts":1762455085985000},{"cat":"PERF","name":"analyze_total_dur","ph":"B","pid":"1","ts":1762455078972000},{"cat":"PERF","name":"analyze_total_dur","ph":"E","pid":"1","ts":1762455085987000},{"cat":"PERF","name":"total_dur","ph":"B","pid":"1","ts":1762455078689000},{"cat":"PERF","name":"total_dur","ph":"E","pid":"1","ts":1762455085987000}]"#;
// Parse the Chrome tracing JSON
let events: Vec<serde_json::Value> =
serde_json::from_str(chrome_trace_input).expect("Failed to parse Chrome tracing JSON");
// Extract timing data from the parsed events
let mut timing_map: HashMap<String, (u128, u128)> = HashMap::new();
for event in &events {
let name = event["name"].as_str().expect("Event should have a name");
let ts = event["ts"].as_u64().expect("Event should have a timestamp") as u128;
let phase = event["ph"].as_str().expect("Event should have a phase");
if phase == "B" {
timing_map.entry(name.to_string()).or_insert((ts, 0)).0 = ts;
} else if phase == "E" {
if let Some(entry) = timing_map.get_mut(name) {
entry.1 = ts - entry.0;
}
}
}
// Find the minimum timestamp to use as offset
let min_timestamp = timing_map
.values()
.map(|(start, _)| *start)
.min()
.unwrap_or(0);
println!("Min timestamp (microseconds): {}", min_timestamp);
println!("Min timestamp (milliseconds): {}", min_timestamp / 1000);
// Create TimedLikeMetrics with RELATIVE timestamps (offset from minimum)
let mut collection = LinkedList::new();
for (name, (start_ts_us, duration_us)) in timing_map.iter() {
// Use relative timestamps: subtract the minimum and convert to milliseconds
let relative_start_ms = (start_ts_us - min_timestamp) / 1000;
let duration_ms = duration_us / 1000;
collection.push_back((name.clone(), relative_start_ms, duration_ms));
}
let metrics = TimedLikeMetrics {
duration_collection: collection,
};
// Test Chrome tracing generation - should work fine
let chrome_output = metrics
.to_chrome_tracing()
.expect("Should generate Chrome tracing output");
let output_events: Vec<serde_json::Value> =
serde_json::from_str(&chrome_output).expect("Should parse output");
assert_eq!(output_events.len(), 44, "Should have 44 events");
// Test ASCII Gantt generation - should work now with relative timestamps
let gantt_output = metrics
.duration_collection_as_gantt(None)
.expect("Should generate ASCII Gantt chart with relative timestamps");
println!("\nASCII Gantt Chart:");
println!("{}", gantt_output);
// Verify the output contains expected metrics
assert!(gantt_output.contains("storage_config_new_dur"));
assert!(gantt_output.contains("analyzer_new_dur"));
assert!(gantt_output.contains("total_dur"));
}
// HealthMetrics tests
#[test]
fn test_health_metrics_new() {
let metrics = HealthMetrics::new();
assert_eq!(metrics.total_files, 0);
assert_eq!(metrics.total_size_bytes, 0);
assert_eq!(metrics.partition_count, 0);
assert_eq!(metrics.avg_file_size_bytes, 0.0);
assert_eq!(metrics.health_score, 0.0);
assert_eq!(metrics.file_size_distribution.small_files, 0);
assert_eq!(metrics.file_size_distribution.medium_files, 0);
assert_eq!(metrics.file_size_distribution.large_files, 0);
assert_eq!(metrics.file_size_distribution.very_large_files, 0);
assert!(metrics.unreferenced_files.is_empty());
assert!(metrics.partitions.is_empty());
assert!(metrics.recommendations.is_empty());
assert!(metrics.clustering.is_none());
assert!(metrics.deletion_vector_metrics.is_none());
assert!(metrics.schema_evolution.is_none());
assert!(metrics.time_travel_metrics.is_none());
assert!(metrics.table_constraints.is_none());
assert!(metrics.file_compaction.is_none());
}
#[test]
fn test_health_metrics_default() {
let metrics = HealthMetrics::default();
assert_eq!(metrics.total_files, 0);
assert_eq!(metrics.health_score, 0.0);
}
#[test]
fn test_calculate_health_score_perfect() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.file_size_distribution.medium_files = 100; // All medium files
let score = metrics.calculate_health_score();
assert!(
(0.9..=1.0).contains(&score),
"Perfect health should score high"
);
}
#[test]
fn test_calculate_health_score_with_unreferenced_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.unreferenced_files = vec![
FileInfo {
path: "file1.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: false,
};
30
]; // 30% unreferenced
let score = metrics.calculate_health_score();
// Should be penalized by 30% * 0.3 = 0.09
assert!(score < 1.0);
assert!((0.85..=0.95).contains(&score));
}
#[test]
fn test_calculate_health_score_with_small_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.file_size_distribution.small_files = 60; // 60% small files
let score = metrics.calculate_health_score();
// Should be penalized by 60% * 0.2 = 0.12
assert!(score < 1.0);
assert!((0.8..=0.9).contains(&score));
}
#[test]
fn test_calculate_health_score_with_very_large_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.file_size_distribution.very_large_files = 20; // 20% very large
let score = metrics.calculate_health_score();
// Should be penalized by 20% * 0.1 = 0.02
assert!((0.95..=1.0).contains(&score));
}
#[test]
fn test_calculate_health_score_with_data_skew() {
let mut metrics = HealthMetrics::new();
metrics.data_skew.partition_skew_score = 0.8; // High partition skew
metrics.data_skew.file_size_skew_score = 0.6; // Moderate file size skew
let score = metrics.calculate_health_score();
// Should be penalized by 0.8 * 0.15 + 0.6 * 0.1 = 0.12 + 0.06 = 0.18
assert!((0.75..=0.85).contains(&score));
}
#[test]
fn test_calculate_health_score_clamped() {
let mut metrics = HealthMetrics::new();
// Set extreme values that would result in negative score
metrics.total_files = 100;
metrics.unreferenced_files = vec![
FileInfo {
path: "file.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: false,
};
100
];
metrics.file_size_distribution.small_files = 100;
metrics.data_skew.partition_skew_score = 1.0;
metrics.data_skew.file_size_skew_score = 1.0;
let score = metrics.calculate_health_score();
// Score should be clamped to [0.0, 1.0]
assert!((0.0..=1.0).contains(&score));
}
#[test]
fn test_calculate_data_skew_empty_partitions() {
let mut metrics = HealthMetrics::new();
metrics.calculate_data_skew();
assert_eq!(metrics.data_skew.partition_skew_score, 0.0);
assert_eq!(metrics.data_skew.file_size_skew_score, 0.0);
}
#[test]
fn test_calculate_data_skew_balanced() {
let mut metrics = HealthMetrics::new();
// Create balanced partitions
metrics.partitions = vec![
PartitionInfo {
partition_values: HashMap::new(),
file_count: 10,
total_size_bytes: 1000,
avg_file_size_bytes: 100.0,
files: Vec::new(),
};
5
];
metrics.calculate_data_skew();
// Perfectly balanced should have low skew
assert!(metrics.data_skew.partition_skew_score < 0.1);
assert_eq!(metrics.data_skew.largest_partition_size, 1000);
assert_eq!(metrics.data_skew.smallest_partition_size, 1000);
assert_eq!(metrics.data_skew.avg_partition_size, 1000);
}
#[test]
fn test_calculate_data_skew_unbalanced() {
let mut metrics = HealthMetrics::new();
// Create unbalanced partitions
metrics.partitions = vec![
PartitionInfo {
partition_values: HashMap::new(),
file_count: 100,
total_size_bytes: 10000,
avg_file_size_bytes: 100.0,
files: Vec::new(),
},
PartitionInfo {
partition_values: HashMap::new(),
file_count: 10,
total_size_bytes: 1000,
avg_file_size_bytes: 100.0,
files: Vec::new(),
},
PartitionInfo {
partition_values: HashMap::new(),
file_count: 5,
total_size_bytes: 500,
avg_file_size_bytes: 100.0,
files: Vec::new(),
},
];
metrics.calculate_data_skew();
// Unbalanced should have higher skew
assert!(metrics.data_skew.partition_skew_score > 0.3);
assert_eq!(metrics.data_skew.largest_partition_size, 10000);
assert_eq!(metrics.data_skew.smallest_partition_size, 500);
}
#[test]
fn test_calculate_snapshot_health_low_count() {
let mut metrics = HealthMetrics::new();
metrics.calculate_snapshot_health(10);
assert_eq!(metrics.snapshot_health.snapshot_count, 10);
assert_eq!(metrics.snapshot_health.snapshot_retention_risk, 0.0);
}
#[test]
fn test_calculate_snapshot_health_medium_count() {
let mut metrics = HealthMetrics::new();
metrics.calculate_snapshot_health(30);
assert_eq!(metrics.snapshot_health.snapshot_count, 30);
assert_eq!(metrics.snapshot_health.snapshot_retention_risk, 0.2);
}
#[test]
fn test_calculate_snapshot_health_high_count() {
let mut metrics = HealthMetrics::new();
metrics.calculate_snapshot_health(75);
assert_eq!(metrics.snapshot_health.snapshot_count, 75);
assert_eq!(metrics.snapshot_health.snapshot_retention_risk, 0.5);
}
#[test]
fn test_calculate_snapshot_health_very_high_count() {
let mut metrics = HealthMetrics::new();
metrics.calculate_snapshot_health(150);
assert_eq!(metrics.snapshot_health.snapshot_count, 150);
assert_eq!(metrics.snapshot_health.snapshot_retention_risk, 0.8);
}
#[test]
fn test_generate_recommendations_unreferenced_files() {
let mut metrics = HealthMetrics::new();
metrics.unreferenced_files = vec![
FileInfo {
path: "file1.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: false,
},
FileInfo {
path: "file2.parquet".to_string(),
size_bytes: 2048,
last_modified: None,
is_referenced: false,
},
];
metrics.unreferenced_size_bytes = 3072;
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics.recommendations[0].contains("unreferenced files"));
assert!(metrics.recommendations[0].contains("3072"));
}
#[test]
fn test_generate_recommendations_small_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.file_size_distribution.small_files = 60; // 60% small files
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("small files") && r.contains("compacting")));
}
#[test]
fn test_generate_recommendations_very_large_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.file_size_distribution.very_large_files = 15; // 15% very large
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("very large files") && r.contains("splitting")));
}
#[test]
fn test_generate_recommendations_too_many_files_per_partition() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 1000;
metrics.partition_count = 5; // 200 files per partition
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("files per partition") && r.contains("repartitioning")));
}
#[test]
fn test_generate_recommendations_too_few_files_per_partition() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 10;
metrics.partition_count = 5; // 2 files per partition
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Low number of files") && r.contains("consolidating")));
}
#[test]
fn test_generate_recommendations_empty_partitions() {
let mut metrics = HealthMetrics::new();
metrics.partitions = vec![
PartitionInfo {
partition_values: HashMap::new(),
file_count: 0,
total_size_bytes: 0,
avg_file_size_bytes: 0.0,
files: Vec::new(),
},
PartitionInfo {
partition_values: HashMap::new(),
file_count: 10,
total_size_bytes: 1000,
avg_file_size_bytes: 100.0,
files: Vec::new(),
},
];
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("empty partitions")));
}
#[test]
fn test_generate_recommendations_high_partition_skew() {
let mut metrics = HealthMetrics::new();
metrics.data_skew.partition_skew_score = 0.8;
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("partition skew") && r.contains("repartitioning")));
}
#[test]
fn test_generate_recommendations_high_file_size_skew() {
let mut metrics = HealthMetrics::new();
metrics.data_skew.file_size_skew_score = 0.7;
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("file size skew") && r.contains("OPTIMIZE")));
}
#[test]
fn test_generate_recommendations_large_metadata() {
let mut metrics = HealthMetrics::new();
metrics.metadata_health.metadata_total_size_bytes = 60 * 1024 * 1024; // 60MB
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("metadata size") && r.contains("VACUUM")));
}
#[test]
fn test_generate_recommendations_high_snapshot_retention_risk() {
let mut metrics = HealthMetrics::new();
metrics.snapshot_health.snapshot_retention_risk = 0.9;
metrics.generate_recommendations();
assert!(!metrics.recommendations.is_empty());
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("snapshot retention risk") && r.contains("VACUUM")));
}
#[test]
fn test_generate_recommendations_clustering_high_files_per_cluster() {
let mut metrics = HealthMetrics::new();
metrics.clustering = Some(ClusteringInfo {
clustering_columns: vec!["col1".to_string()],
cluster_count: 5,
avg_files_per_cluster: 60.0, // > 50
avg_cluster_size_bytes: 1024.0 * 1024.0,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("files per cluster") && r.contains("optimizing")));
}
#[test]
fn test_generate_recommendations_clustering_too_many_columns() {
let mut metrics = HealthMetrics::new();
metrics.clustering = Some(ClusteringInfo {
clustering_columns: vec![
"col1".to_string(),
"col2".to_string(),
"col3".to_string(),
"col4".to_string(),
"col5".to_string(),
],
cluster_count: 10,
avg_files_per_cluster: 5.0,
avg_cluster_size_bytes: 1024.0 * 1024.0,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Too many clustering columns")));
}
#[test]
fn test_generate_recommendations_clustering_no_columns() {
let mut metrics = HealthMetrics::new();
metrics.clustering = Some(ClusteringInfo {
clustering_columns: vec![],
cluster_count: 0,
avg_files_per_cluster: 0.0,
avg_cluster_size_bytes: 0.0,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("No clustering detected")));
}
#[test]
fn test_generate_recommendations_deletion_vectors_high_impact() {
let mut metrics = HealthMetrics::new();
metrics.deletion_vector_metrics = Some(DeletionVectorMetrics {
deletion_vector_count: 10,
total_deletion_vector_size_bytes: 1024,
avg_deletion_vector_size_bytes: 102.4,
deletion_vector_age_days: 5.0,
deleted_rows_count: 100,
deletion_vector_impact_score: 0.8, // > 0.7
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("deletion vector impact") && r.contains("VACUUM")));
}
#[test]
fn test_generate_recommendations_deletion_vectors_many_vectors() {
let mut metrics = HealthMetrics::new();
metrics.deletion_vector_metrics = Some(DeletionVectorMetrics {
deletion_vector_count: 60, // > 50
total_deletion_vector_size_bytes: 1024,
avg_deletion_vector_size_bytes: 17.0,
deletion_vector_age_days: 5.0,
deleted_rows_count: 100,
deletion_vector_impact_score: 0.3,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Many deletion vectors")));
}
#[test]
fn test_generate_recommendations_deletion_vectors_old() {
let mut metrics = HealthMetrics::new();
metrics.deletion_vector_metrics = Some(DeletionVectorMetrics {
deletion_vector_count: 5,
total_deletion_vector_size_bytes: 1024,
avg_deletion_vector_size_bytes: 204.8,
deletion_vector_age_days: 45.0, // > 30
deleted_rows_count: 100,
deletion_vector_impact_score: 0.3,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Old deletion vectors")));
}
#[test]
fn test_generate_recommendations_schema_unstable() {
let mut metrics = HealthMetrics::new();
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 10,
breaking_changes: 2,
non_breaking_changes: 8,
schema_stability_score: 0.3, // < 0.5
days_since_last_change: 30.0,
schema_change_frequency: 0.5,
current_schema_version: 10,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Unstable schema")));
}
#[test]
fn test_generate_recommendations_schema_many_breaking_changes() {
let mut metrics = HealthMetrics::new();
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 10,
breaking_changes: 7, // > 5
non_breaking_changes: 3,
schema_stability_score: 0.6,
days_since_last_change: 30.0,
schema_change_frequency: 0.5,
current_schema_version: 10,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("breaking schema changes")));
}
#[test]
fn test_generate_recommendations_schema_high_frequency() {
let mut metrics = HealthMetrics::new();
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 10,
breaking_changes: 1,
non_breaking_changes: 9,
schema_stability_score: 0.7,
days_since_last_change: 30.0,
schema_change_frequency: 1.5, // > 1.0
current_schema_version: 10,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("schema change frequency")));
}
#[test]
fn test_generate_recommendations_schema_recent_change() {
let mut metrics = HealthMetrics::new();
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 5,
breaking_changes: 0,
non_breaking_changes: 5,
schema_stability_score: 0.9,
days_since_last_change: 0.5, // < 1.0
schema_change_frequency: 0.1,
current_schema_version: 5,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Recent schema changes")));
}
#[test]
fn test_generate_recommendations_time_travel_high_cost() {
let mut metrics = HealthMetrics::new();
metrics.time_travel_metrics = Some(TimeTravelMetrics {
total_snapshots: 50,
oldest_snapshot_age_days: 90.0,
newest_snapshot_age_days: 0.5,
total_historical_size_bytes: 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 20.0 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.8, // > 0.7
retention_efficiency_score: 0.6,
recommended_retention_days: 30,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("time travel storage costs")));
}
#[test]
fn test_generate_recommendations_time_travel_inefficient_retention() {
let mut metrics = HealthMetrics::new();
metrics.time_travel_metrics = Some(TimeTravelMetrics {
total_snapshots: 50,
oldest_snapshot_age_days: 90.0,
newest_snapshot_age_days: 0.5,
total_historical_size_bytes: 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 20.0 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.3,
retention_efficiency_score: 0.3, // < 0.5
recommended_retention_days: 30,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Inefficient snapshot retention")));
}
#[test]
fn test_generate_recommendations_time_travel_high_snapshot_count() {
let mut metrics = HealthMetrics::new();
metrics.time_travel_metrics = Some(TimeTravelMetrics {
total_snapshots: 1500, // > 1000
oldest_snapshot_age_days: 365.0,
newest_snapshot_age_days: 0.1,
total_historical_size_bytes: 10 * 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 6.7 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.5,
retention_efficiency_score: 0.6,
recommended_retention_days: 30,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("High snapshot count")));
}
#[test]
fn test_generate_recommendations_constraints_low_quality() {
let mut metrics = HealthMetrics::new();
metrics.table_constraints = Some(TableConstraintsMetrics {
total_constraints: 5,
check_constraints: 1,
not_null_constraints: 3,
unique_constraints: 1,
foreign_key_constraints: 0,
constraint_violation_risk: 0.2,
data_quality_score: 0.3, // < 0.5
constraint_coverage_score: 0.5,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Low data quality score")));
}
#[test]
fn test_generate_recommendations_constraints_high_violation_risk() {
let mut metrics = HealthMetrics::new();
metrics.table_constraints = Some(TableConstraintsMetrics {
total_constraints: 5,
check_constraints: 1,
not_null_constraints: 3,
unique_constraints: 1,
foreign_key_constraints: 0,
constraint_violation_risk: 0.8, // > 0.7
data_quality_score: 0.7,
constraint_coverage_score: 0.5,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("constraint violation risk")));
}
#[test]
fn test_generate_recommendations_constraints_low_coverage() {
let mut metrics = HealthMetrics::new();
metrics.table_constraints = Some(TableConstraintsMetrics {
total_constraints: 2,
check_constraints: 0,
not_null_constraints: 2,
unique_constraints: 0,
foreign_key_constraints: 0,
constraint_violation_risk: 0.2,
data_quality_score: 0.7,
constraint_coverage_score: 0.2, // < 0.3
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Low constraint coverage")));
}
#[test]
fn test_generate_recommendations_compaction_high_opportunity() {
let mut metrics = HealthMetrics::new();
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.8, // > 0.7
small_files_count: 100,
small_files_size_bytes: 50 * 1024 * 1024,
potential_compaction_files: 80,
estimated_compaction_savings_bytes: 20 * 1024 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "high".to_string(),
z_order_opportunity: false,
z_order_columns: vec![],
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("file compaction opportunity")));
}
#[test]
fn test_generate_recommendations_compaction_critical_priority() {
let mut metrics = HealthMetrics::new();
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.5,
small_files_count: 200,
small_files_size_bytes: 100 * 1024 * 1024,
potential_compaction_files: 180,
estimated_compaction_savings_bytes: 50 * 1024 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "critical".to_string(),
z_order_opportunity: false,
z_order_columns: vec![],
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Critical compaction priority")));
}
#[test]
fn test_generate_recommendations_compaction_z_order_opportunity() {
let mut metrics = HealthMetrics::new();
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.5,
small_files_count: 50,
small_files_size_bytes: 25 * 1024 * 1024,
potential_compaction_files: 40,
estimated_compaction_savings_bytes: 10 * 1024 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "medium".to_string(),
z_order_opportunity: true,
z_order_columns: vec!["date".to_string(), "region".to_string()],
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Z-ordering opportunity") && r.contains("date, region")));
}
#[test]
fn test_generate_recommendations_compaction_significant_savings() {
let mut metrics = HealthMetrics::new();
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.5,
small_files_count: 500,
small_files_size_bytes: 500 * 1024 * 1024,
potential_compaction_files: 400,
estimated_compaction_savings_bytes: 200 * 1024 * 1024, // > 100MB
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "medium".to_string(),
z_order_opportunity: false,
z_order_columns: vec![],
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Significant compaction savings")));
}
// HealthReport tests
#[test]
fn test_health_report_to_json() {
let report = HealthReport {
table_path: "/path/to/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics: HealthMetrics::new(),
health_score: 0.85,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let json = report.to_json(false).expect("Should serialize to JSON");
assert!(json.contains("table_path"));
assert!(json.contains("/path/to/table"));
assert!(json.contains("delta"));
assert!(json.contains("0.85"));
}
#[test]
fn test_health_report_to_json_exclude_files() {
let mut metrics = HealthMetrics::new();
metrics.unreferenced_files = vec![FileInfo {
path: "file1.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: false,
}];
metrics.partitions = vec![PartitionInfo {
partition_values: HashMap::new(),
file_count: 1,
total_size_bytes: 1024,
avg_file_size_bytes: 1024.0,
files: vec![FileInfo {
path: "file2.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: true,
}],
}];
let report = HealthReport {
table_path: "/path/to/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.85,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let json = report.to_json(true).expect("Should serialize to JSON");
// Should not contain file paths when exclude_files is true
assert!(!json.contains("file1.parquet"));
assert!(!json.contains("file2.parquet"));
}
#[test]
fn test_health_report_display() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.total_size_bytes = 1024 * 1024 * 1024; // 1GB
metrics.avg_file_size_bytes = 10.0 * 1024.0 * 1024.0; // 10MB
metrics.partition_count = 5;
metrics.file_size_distribution.small_files = 20;
metrics.file_size_distribution.medium_files = 60;
metrics.file_size_distribution.large_files = 15;
metrics.file_size_distribution.very_large_files = 5;
let report = HealthReport {
table_path: "/path/to/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.85,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Table Health Report"));
assert!(display.contains("85.0%"));
assert!(display.contains("/path/to/table"));
assert!(display.contains("delta"));
assert!(display.contains("Total Data Files"));
assert!(display.contains("100"));
}
// FileInfo tests
#[test]
fn test_file_info_creation() {
let file_info = FileInfo {
path: "data/file.parquet".to_string(),
size_bytes: 1024 * 1024,
last_modified: Some("2024-01-01T00:00:00Z".to_string()),
is_referenced: true,
};
assert_eq!(file_info.path, "data/file.parquet");
assert_eq!(file_info.size_bytes, 1024 * 1024);
assert_eq!(
file_info.last_modified,
Some("2024-01-01T00:00:00Z".to_string())
);
assert!(file_info.is_referenced);
}
#[test]
fn test_file_info_serialization() {
let file_info = FileInfo {
path: "data/file.parquet".to_string(),
size_bytes: 1024,
last_modified: None,
is_referenced: false,
};
let json = serde_json::to_string(&file_info).expect("Should serialize");
assert!(json.contains("data/file.parquet"));
assert!(json.contains("1024"));
assert!(json.contains("false"));
}
// PartitionInfo tests
#[test]
fn test_partition_info_creation() {
let mut partition_values = HashMap::new();
partition_values.insert("year".to_string(), "2024".to_string());
partition_values.insert("month".to_string(), "01".to_string());
let partition_info = PartitionInfo {
partition_values,
file_count: 10,
total_size_bytes: 10240,
avg_file_size_bytes: 1024.0,
files: Vec::new(),
};
assert_eq!(partition_info.file_count, 10);
assert_eq!(partition_info.total_size_bytes, 10240);
assert_eq!(partition_info.avg_file_size_bytes, 1024.0);
assert_eq!(partition_info.partition_values.len(), 2);
}
// ClusteringInfo tests
#[test]
fn test_clustering_info_creation() {
let clustering_info = ClusteringInfo {
clustering_columns: vec!["col1".to_string(), "col2".to_string()],
cluster_count: 10,
avg_files_per_cluster: 5.5,
avg_cluster_size_bytes: 1024.0 * 1024.0,
};
assert_eq!(clustering_info.clustering_columns.len(), 2);
assert_eq!(clustering_info.cluster_count, 10);
assert_eq!(clustering_info.avg_files_per_cluster, 5.5);
}
// DeletionVectorMetrics tests
#[test]
fn test_deletion_vector_metrics_creation() {
let dv_metrics = DeletionVectorMetrics {
deletion_vector_count: 5,
total_deletion_vector_size_bytes: 5120,
avg_deletion_vector_size_bytes: 1024.0,
deletion_vector_age_days: 15.5,
deleted_rows_count: 1000,
deletion_vector_impact_score: 0.6,
};
assert_eq!(dv_metrics.deletion_vector_count, 5);
assert_eq!(dv_metrics.total_deletion_vector_size_bytes, 5120);
assert_eq!(dv_metrics.deleted_rows_count, 1000);
assert_eq!(dv_metrics.deletion_vector_impact_score, 0.6);
}
// SchemaEvolutionMetrics tests
#[test]
fn test_schema_evolution_metrics_creation() {
let schema_metrics = SchemaEvolutionMetrics {
total_schema_changes: 10,
breaking_changes: 2,
non_breaking_changes: 8,
schema_stability_score: 0.8,
days_since_last_change: 5.0,
schema_change_frequency: 0.5,
current_schema_version: 11,
};
assert_eq!(schema_metrics.total_schema_changes, 10);
assert_eq!(schema_metrics.breaking_changes, 2);
assert_eq!(schema_metrics.non_breaking_changes, 8);
assert_eq!(schema_metrics.schema_stability_score, 0.8);
}
// TimeTravelMetrics tests
#[test]
fn test_time_travel_metrics_creation() {
let tt_metrics = TimeTravelMetrics {
total_snapshots: 50,
oldest_snapshot_age_days: 30.0,
newest_snapshot_age_days: 1.0,
total_historical_size_bytes: 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 20.0 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.5,
retention_efficiency_score: 0.7,
recommended_retention_days: 30,
};
assert_eq!(tt_metrics.total_snapshots, 50);
assert_eq!(tt_metrics.oldest_snapshot_age_days, 30.0);
assert_eq!(tt_metrics.recommended_retention_days, 30);
}
// TableConstraintsMetrics tests
#[test]
fn test_table_constraints_metrics_creation() {
let constraint_metrics = TableConstraintsMetrics {
total_constraints: 10,
check_constraints: 3,
not_null_constraints: 5,
unique_constraints: 1,
foreign_key_constraints: 1,
constraint_violation_risk: 0.2,
data_quality_score: 0.9,
constraint_coverage_score: 0.8,
};
assert_eq!(constraint_metrics.total_constraints, 10);
assert_eq!(constraint_metrics.check_constraints, 3);
assert_eq!(constraint_metrics.not_null_constraints, 5);
assert_eq!(constraint_metrics.data_quality_score, 0.9);
}
// FileCompactionMetrics tests
#[test]
fn test_file_compaction_metrics_creation() {
let compaction_metrics = FileCompactionMetrics {
compaction_opportunity_score: 0.8,
small_files_count: 100,
small_files_size_bytes: 1024 * 1024,
potential_compaction_files: 80,
estimated_compaction_savings_bytes: 512 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "high".to_string(),
z_order_opportunity: true,
z_order_columns: vec!["col1".to_string(), "col2".to_string()],
};
assert_eq!(compaction_metrics.compaction_opportunity_score, 0.8);
assert_eq!(compaction_metrics.small_files_count, 100);
assert_eq!(compaction_metrics.compaction_priority, "high");
assert!(compaction_metrics.z_order_opportunity);
assert_eq!(compaction_metrics.z_order_columns.len(), 2);
}
// ========== Display implementation coverage tests ==========
#[test]
fn test_health_report_display_with_clustering() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 50;
metrics.total_size_bytes = 500 * 1024 * 1024;
metrics.clustering = Some(ClusteringInfo {
clustering_columns: vec!["date".to_string(), "region".to_string()],
cluster_count: 10,
avg_files_per_cluster: 5.0,
avg_cluster_size_bytes: 50.0 * 1024.0 * 1024.0,
});
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "iceberg".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.9,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Clustering"));
assert!(display.contains("Avg Cluster Size"));
assert!(display.contains("Clusters"));
assert!(display.contains("10"));
}
#[test]
fn test_health_report_display_with_deletion_vectors() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.deletion_vector_metrics = Some(DeletionVectorMetrics {
deletion_vector_count: 25,
total_deletion_vector_size_bytes: 2 * 1024 * 1024,
avg_deletion_vector_size_bytes: 80.0 * 1024.0,
deletion_vector_age_days: 15.5,
deleted_rows_count: 50000,
deletion_vector_impact_score: 0.65,
});
let report = HealthReport {
table_path: "/test/delta_table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.75,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Deletion Vectors"));
assert!(display.contains("Vectors"));
assert!(display.contains("25"));
assert!(display.contains("Deleted Rows"));
assert!(display.contains("50000"));
}
#[test]
fn test_health_report_display_with_schema_evolution() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 5,
breaking_changes: 1,
non_breaking_changes: 4,
schema_stability_score: 0.8,
days_since_last_change: 30.0,
schema_change_frequency: 0.1,
current_schema_version: 5,
});
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.85,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Schema Evolution"));
assert!(display.contains("Total Changes"));
assert!(display.contains("Breaking"));
assert!(display.contains("Non-Breaking"));
assert!(display.contains("Stability"));
}
#[test]
fn test_health_report_display_with_time_travel() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.time_travel_metrics = Some(TimeTravelMetrics {
total_snapshots: 50,
oldest_snapshot_age_days: 90.0,
newest_snapshot_age_days: 0.5,
total_historical_size_bytes: 2 * 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 40.0 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.3,
retention_efficiency_score: 0.85,
recommended_retention_days: 30,
});
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.8,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Time Travel"));
assert!(display.contains("Snapshots"));
assert!(display.contains("50"));
assert!(display.contains("Historical Size"));
}
#[test]
fn test_health_report_display_with_constraints_and_compaction() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.table_constraints = Some(TableConstraintsMetrics {
total_constraints: 10,
check_constraints: 3,
not_null_constraints: 5,
unique_constraints: 1,
foreign_key_constraints: 1,
constraint_violation_risk: 0.1,
data_quality_score: 0.95,
constraint_coverage_score: 0.7,
});
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.6,
small_files_count: 40,
small_files_size_bytes: 100 * 1024 * 1024,
potential_compaction_files: 35,
estimated_compaction_savings_bytes: 50 * 1024 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "MEDIUM".to_string(),
z_order_opportunity: false,
z_order_columns: vec![],
});
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.7,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Table Constraints"));
assert!(display.contains("File Compaction"));
assert!(display.contains("NOT NULL"));
assert!(display.contains("Small Files"));
assert!(display.contains("40"));
}
#[test]
fn test_health_report_display_with_unreferenced_files() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.unreferenced_files = vec![
FileInfo {
path: "orphan1.parquet".to_string(),
size_bytes: 1024 * 1024,
last_modified: None,
is_referenced: false,
},
FileInfo {
path: "orphan2.parquet".to_string(),
size_bytes: 2 * 1024 * 1024,
last_modified: None,
is_referenced: false,
},
];
metrics.unreferenced_size_bytes = 3 * 1024 * 1024;
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.6,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Unreferenced Files"));
assert!(display.contains("Count"));
assert!(display.contains("2"));
assert!(display.contains("Wasted Space"));
}
#[test]
fn test_health_report_display_with_all_optional_fields() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 200;
metrics.total_size_bytes = 5 * 1024 * 1024 * 1024;
metrics.avg_file_size_bytes = 25.0 * 1024.0 * 1024.0;
metrics.partition_count = 20;
// Add all optional fields
metrics.clustering = Some(ClusteringInfo {
clustering_columns: vec!["col1".to_string()],
cluster_count: 5,
avg_files_per_cluster: 40.0,
avg_cluster_size_bytes: 1024.0 * 1024.0 * 1024.0,
});
metrics.deletion_vector_metrics = Some(DeletionVectorMetrics {
deletion_vector_count: 10,
total_deletion_vector_size_bytes: 500 * 1024,
avg_deletion_vector_size_bytes: 50.0 * 1024.0,
deletion_vector_age_days: 5.0,
deleted_rows_count: 1000,
deletion_vector_impact_score: 0.2,
});
metrics.schema_evolution = Some(SchemaEvolutionMetrics {
total_schema_changes: 3,
breaking_changes: 0,
non_breaking_changes: 3,
schema_stability_score: 0.95,
days_since_last_change: 60.0,
schema_change_frequency: 0.05,
current_schema_version: 3,
});
metrics.time_travel_metrics = Some(TimeTravelMetrics {
total_snapshots: 100,
oldest_snapshot_age_days: 180.0,
newest_snapshot_age_days: 0.1,
total_historical_size_bytes: 10 * 1024 * 1024 * 1024,
avg_snapshot_size_bytes: 100.0 * 1024.0 * 1024.0,
storage_cost_impact_score: 0.5,
retention_efficiency_score: 0.7,
recommended_retention_days: 90,
});
metrics.table_constraints = Some(TableConstraintsMetrics {
total_constraints: 15,
check_constraints: 5,
not_null_constraints: 8,
unique_constraints: 1,
foreign_key_constraints: 1,
constraint_violation_risk: 0.05,
data_quality_score: 0.98,
constraint_coverage_score: 0.85,
});
metrics.file_compaction = Some(FileCompactionMetrics {
compaction_opportunity_score: 0.3,
small_files_count: 20,
small_files_size_bytes: 50 * 1024 * 1024,
potential_compaction_files: 15,
estimated_compaction_savings_bytes: 20 * 1024 * 1024,
recommended_target_file_size_bytes: 128 * 1024 * 1024,
compaction_priority: "LOW".to_string(),
z_order_opportunity: true,
z_order_columns: vec!["date".to_string()],
});
let report = HealthReport {
table_path: "/production/data/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-06-15T12:00:00Z".to_string(),
metrics,
health_score: 0.92,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
// Verify all sections are present
assert!(display.contains("Table Health Report"));
assert!(display.contains("92.0%"));
assert!(display.contains("Key Metrics"));
assert!(display.contains("File Size Distribution"));
assert!(display.contains("Data Skew Analysis"));
assert!(display.contains("Metadata Health"));
assert!(display.contains("Snapshot Health"));
assert!(display.contains("Clustering"));
assert!(display.contains("Deletion Vectors"));
assert!(display.contains("Schema Evolution"));
assert!(display.contains("Time Travel"));
assert!(display.contains("Table Constraints"));
assert!(display.contains("File Compaction"));
}
#[test]
fn test_health_report_display_size_formatting_gb() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 1000;
metrics.total_size_bytes = 50 * 1024 * 1024 * 1024; // 50 GB
let report = HealthReport {
table_path: "/test/large_table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.8,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("GB"));
}
#[test]
fn test_health_report_display_metadata_file_names_truncation() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.metadata_health.first_file_name =
Some("very_long_metadata_file_name_that_exceeds_thirty_characters.json".to_string());
metrics.metadata_health.last_file_name =
Some("another_very_long_file_name_for_testing_truncation.json".to_string());
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.8,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
// Truncated names should contain "..."
assert!(display.contains("..."));
}
#[test]
fn test_health_report_display_zero_partitions() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 10;
metrics.partition_count = 0;
metrics.data_skew.avg_partition_size = 0;
let report = HealthReport {
table_path: "/test/unpartitioned".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.7,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("N/A"));
}
#[test]
fn test_health_report_display_with_recommendations() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.recommendations = vec![
"Consider running OPTIMIZE".to_string(),
"Too many small files detected".to_string(),
];
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.5,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Recommendations"));
assert!(display.contains("Consider running OPTIMIZE"));
assert!(display.contains("Too many small files detected"));
}
#[test]
fn test_health_report_display_snapshot_ages() {
let mut metrics = HealthMetrics::new();
metrics.total_files = 50;
metrics.snapshot_health.oldest_snapshot_age_days = 45.5;
metrics.snapshot_health.newest_snapshot_age_days = 0.5;
metrics.snapshot_health.avg_snapshot_age_days = 20.0;
let report = HealthReport {
table_path: "/test/table".to_string(),
table_type: "delta".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.8,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
assert!(display.contains("Oldest Snapshot"));
assert!(display.contains("45.5 days"));
assert!(display.contains("Newest Snapshot"));
assert!(display.contains("0.5 days"));
}
#[cfg(feature = "hudi")]
#[test]
fn test_health_report_display_with_hudi_metrics() {
use crate::reader::hudi::metrics::{
FileStatistics, HudiMetrics, PartitionMetrics, TableMetadata, TimelineMetrics,
};
let mut metrics = HealthMetrics::new();
metrics.total_files = 100;
metrics.hudi_table_specific_metrics = Some(HudiMetrics {
table_type: "COPY_ON_WRITE".to_string(),
table_name: "test_hudi_table".to_string(),
metadata: TableMetadata {
name: "test_hudi_table".to_string(),
base_path: "/data/hudi/test_table".to_string(),
schema_string: "{}".to_string(),
field_count: 10,
partition_columns: vec!["date".to_string()],
created_time: Some(1700000000000),
format_provider: "parquet".to_string(),
format_options: std::collections::HashMap::new(),
},
table_properties: {
let mut props = std::collections::HashMap::new();
props.insert("hoodie.table.name".to_string(), "test".to_string());
props
},
file_stats: FileStatistics {
num_files: 50,
total_size_bytes: 1024 * 1024 * 100,
avg_file_size_bytes: 1024.0 * 1024.0 * 2.0,
min_file_size_bytes: 1024,
max_file_size_bytes: 1024 * 1024 * 10,
num_log_files: 10,
total_log_size_bytes: 1024 * 1024 * 20,
},
partition_info: PartitionMetrics {
num_partition_columns: 1,
num_partitions: 30,
partition_paths: vec!["date=2024-01-01".to_string()],
largest_partition_size_bytes: 1024 * 1024 * 50,
smallest_partition_size_bytes: 1024 * 1024,
avg_partition_size_bytes: 1024.0 * 1024.0 * 25.0,
},
timeline_info: TimelineMetrics {
total_commits: 100,
total_delta_commits: 0,
total_compactions: 5,
total_cleans: 10,
total_rollbacks: 2,
total_savepoints: 1,
latest_commit_timestamp: Some("20240101120000000".to_string()),
earliest_commit_timestamp: Some("20231201000000000".to_string()),
pending_compactions: 0,
},
});
let report = HealthReport {
table_path: "/data/hudi/test_table".to_string(),
table_type: "hudi".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.85,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
// Verify Hudi-specific metrics are displayed
assert!(display.contains("Hudi Specific Metrics"));
assert!(display.contains("COPY_ON_WRITE"));
assert!(display.contains("test_hudi_table"));
assert!(display.contains("File Statistics"));
assert!(display.contains("Timeline Info"));
assert!(display.contains("Total Commits"));
}
#[cfg(feature = "lance")]
#[test]
fn test_generate_recommendations_lance_no_indices_large_table() {
use crate::reader::lance::metrics::{
FileStatistics, FragmentMetrics, IndexMetrics, LanceMetrics, TableMetadata,
};
let mut metrics = HealthMetrics::new();
metrics.lance_table_specific_metrics = Some(LanceMetrics {
version: 1,
metadata: TableMetadata {
uuid: "test-uuid".to_string(),
schema_string: String::new(),
field_count: 5,
created_time: None,
last_modified_time: None,
num_rows: Some(50_000), // > 10,000 rows
num_deleted_rows: None,
},
table_properties: std::collections::HashMap::new(),
file_stats: FileStatistics {
num_data_files: 10,
num_deletion_files: 0,
total_data_size_bytes: 1024 * 1024 * 100,
total_deletion_size_bytes: 0,
avg_data_file_size_bytes: 1024.0 * 1024.0 * 10.0,
min_data_file_size_bytes: 1024 * 1024,
max_data_file_size_bytes: 1024 * 1024 * 20,
},
fragment_info: FragmentMetrics {
num_fragments: 10,
num_fragments_with_deletions: 0,
avg_rows_per_fragment: 5000.0,
min_rows_per_fragment: 1000,
max_rows_per_fragment: 10000,
total_physical_rows: 50000,
},
index_info: IndexMetrics {
num_indices: 0, // No indices
indexed_columns: vec![],
index_types: vec![],
total_index_size_bytes: 0,
},
operation_metrics: None,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("No indices found on Lance table")));
}
#[cfg(feature = "lance")]
#[test]
fn test_generate_recommendations_lance_many_fragments_no_indices() {
use crate::reader::lance::metrics::{
FileStatistics, FragmentMetrics, IndexMetrics, LanceMetrics, TableMetadata,
};
let mut metrics = HealthMetrics::new();
metrics.lance_table_specific_metrics = Some(LanceMetrics {
version: 1,
metadata: TableMetadata {
uuid: "test-uuid".to_string(),
schema_string: String::new(),
field_count: 5,
created_time: None,
last_modified_time: None,
num_rows: Some(5_000), // Small table
num_deleted_rows: None,
},
table_properties: std::collections::HashMap::new(),
file_stats: FileStatistics {
num_data_files: 150,
num_deletion_files: 0,
total_data_size_bytes: 1024 * 1024 * 100,
total_deletion_size_bytes: 0,
avg_data_file_size_bytes: 1024.0 * 1024.0,
min_data_file_size_bytes: 1024,
max_data_file_size_bytes: 1024 * 1024 * 5,
},
fragment_info: FragmentMetrics {
num_fragments: 150, // > 100 fragments
num_fragments_with_deletions: 0,
avg_rows_per_fragment: 33.0,
min_rows_per_fragment: 10,
max_rows_per_fragment: 100,
total_physical_rows: 5000,
},
index_info: IndexMetrics {
num_indices: 0, // No indices
indexed_columns: vec![],
index_types: vec![],
total_index_size_bytes: 0,
},
operation_metrics: None,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("Large number of fragments")));
}
#[cfg(feature = "lance")]
#[test]
fn test_generate_recommendations_lance_high_deletion_ratio_with_indices() {
use crate::reader::lance::metrics::{
FileStatistics, FragmentMetrics, IndexMetrics, LanceMetrics, TableMetadata,
};
let mut metrics = HealthMetrics::new();
metrics.lance_table_specific_metrics = Some(LanceMetrics {
version: 1,
metadata: TableMetadata {
uuid: "test-uuid".to_string(),
schema_string: String::new(),
field_count: 5,
created_time: None,
last_modified_time: None,
num_rows: Some(8_000),
num_deleted_rows: Some(3_000), // 3000 / (8000 + 3000) = 27% > 20%
},
table_properties: std::collections::HashMap::new(),
file_stats: FileStatistics {
num_data_files: 10,
num_deletion_files: 5,
total_data_size_bytes: 1024 * 1024 * 100,
total_deletion_size_bytes: 1024 * 100,
avg_data_file_size_bytes: 1024.0 * 1024.0 * 10.0,
min_data_file_size_bytes: 1024 * 1024,
max_data_file_size_bytes: 1024 * 1024 * 20,
},
fragment_info: FragmentMetrics {
num_fragments: 10,
num_fragments_with_deletions: 5,
avg_rows_per_fragment: 800.0,
min_rows_per_fragment: 500,
max_rows_per_fragment: 1500,
total_physical_rows: 11000,
},
index_info: IndexMetrics {
num_indices: 2, // Has indices
indexed_columns: vec!["id".to_string(), "embedding".to_string()],
index_types: vec!["IVF_PQ".to_string()],
total_index_size_bytes: 1024 * 1024 * 10,
},
operation_metrics: None,
});
metrics.generate_recommendations();
assert!(metrics
.recommendations
.iter()
.any(|r| r.contains("High deletion ratio") && r.contains("rebuilding indices")));
}
#[cfg(feature = "lance")]
#[test]
fn test_health_report_display_with_lance_metrics() {
use crate::reader::lance::metrics::{
FileStatistics, FragmentMetrics, IndexMetrics, LanceMetrics, TableMetadata,
};
let mut metrics = HealthMetrics::new();
metrics.lance_table_specific_metrics = Some(LanceMetrics {
version: 5,
metadata: TableMetadata {
uuid: "test-lance-uuid".to_string(),
schema_string: String::new(),
field_count: 10,
created_time: None,
last_modified_time: None,
num_rows: Some(100_000),
num_deleted_rows: Some(500),
},
table_properties: std::collections::HashMap::new(),
file_stats: FileStatistics {
num_data_files: 50,
num_deletion_files: 5,
total_data_size_bytes: 1024 * 1024 * 500,
total_deletion_size_bytes: 1024 * 100,
avg_data_file_size_bytes: 1024.0 * 1024.0 * 10.0,
min_data_file_size_bytes: 1024 * 1024,
max_data_file_size_bytes: 1024 * 1024 * 50,
},
fragment_info: FragmentMetrics {
num_fragments: 50,
num_fragments_with_deletions: 5,
avg_rows_per_fragment: 2000.0,
min_rows_per_fragment: 500,
max_rows_per_fragment: 5000,
total_physical_rows: 100500,
},
index_info: IndexMetrics {
num_indices: 2,
indexed_columns: vec!["id".to_string(), "embedding".to_string()],
index_types: vec!["IVF_PQ".to_string()],
total_index_size_bytes: 1024 * 1024 * 50,
},
operation_metrics: None,
});
let report = HealthReport {
table_path: "/data/lance/test_table.lance".to_string(),
table_type: "lance".to_string(),
analysis_timestamp: "2024-01-01T00:00:00Z".to_string(),
metrics,
health_score: 0.90,
timed_metrics: TimedLikeMetrics {
duration_collection: LinkedList::new(),
},
};
let display = format!("{}", report);
// Verify Lance-specific metrics are displayed
assert!(display.contains("Lance Specific Metrics"));
assert!(display.contains("test-lance-uuid"));
assert!(display.contains("File Statistics"));
assert!(display.contains("Fragment Info"));
assert!(display.contains("Index Info"));
}
}