use std::io::Write;
use std::path::Path;
use dataprof::{ChunkSize, EngineType, Profiler, StopCondition, TruncationReason};
use dataprof::{
ColumnStats, CsvParserConfig, DataType, MetricConfidence, QualityDimension, analyze_csv_file,
};
use serde_json::json;
use tempfile::NamedTempFile;
#[test]
fn serialized_text_stats_match_column_analysis_and_csv_engines() {
for values in [
vec!["hello", "world", "test", "hello", "world"],
vec!["東京", "café", "e\u{301}", "東京", "", " ", "NULL"],
vec!["", " ", "NULL"],
] {
let data: Vec<String> = values.iter().map(|value| (*value).to_owned()).collect();
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "text,row").unwrap();
for (row, value) in values.iter().enumerate() {
writeln!(csv, "{value},{row}").unwrap();
}
csv.flush().unwrap();
let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
let expected = serde_json::to_value(&standard.column_profiles[0].stats).unwrap();
if values
.iter()
.all(|value| dataprof::is_null_like_token(value.trim()))
{
assert_eq!(expected, json!("None"), "{values:?}");
} else {
assert!(expected["Text"].is_object(), "{values:?}");
assert!(expected["Text"].get("most_frequent").is_none());
assert!(expected["Text"].get("least_frequent").is_none());
}
for column in [
dataprof::analyze_column("text", &data),
dataprof::analyze_column_fast("text", &data),
dataprof::analyze_column_with_analysis_options(
"text",
&data,
&dataprof::AnalysisOptions::default(),
),
] {
assert_eq!(column.data_type, standard.column_profiles[0].data_type);
assert_eq!(column.null_count, standard.column_profiles[0].null_count);
assert_eq!(column.total_count, standard.column_profiles[0].total_count);
assert_eq!(
serde_json::to_value(&column.stats).unwrap(),
expected,
"column analysis, {values:?}"
);
}
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let report = Profiler::new()
.engine(engine)
.analyze_file(csv.path())
.unwrap();
assert_eq!(
serde_json::to_value(&report.column_profiles[0].stats).unwrap(),
expected,
"{engine:?}, {values:?}"
);
}
}
}
#[test]
fn historical_text_frequencies_survive_rust_round_trip() {
for frequencies in [
json!([{"value": "hello", "count": 2, "percentage": 100.0}]),
json!([]),
] {
let document = json!({"Text": {
"min_length": 5,
"max_length": 5,
"avg_length": 5.0,
"most_frequent": frequencies,
"least_frequent": frequencies,
}});
let restored: ColumnStats = serde_json::from_value(document.clone()).unwrap();
assert_eq!(serde_json::to_value(restored).unwrap(), document);
}
}
fn create_sorted_30k_csv() -> NamedTempFile {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,value").unwrap();
for i in 0..30_000 {
writeln!(f, "{},{:.2}", i, i as f64 / 2.0).unwrap();
}
f.flush().unwrap();
f
}
#[test]
fn null_and_non_finite_numeric_tokens_match_across_csv_engines() {
for (token, null_count) in [
("", 1),
("N/A", 1),
("pending", 0),
("NaN", 1),
("inf", 0),
("-inf", 0),
("1e400", 0),
] {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "amount,label").unwrap();
for i in 0..10 {
writeln!(csv, "{i}.5,row{i}").unwrap();
}
writeln!(csv, "{token},special").unwrap();
csv.flush().unwrap();
let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
let expected = &standard.column_profiles[0];
assert_eq!(expected.data_type, DataType::Float, "{token:?}");
assert_eq!(expected.total_count, 11, "{token:?}");
assert_eq!(expected.null_count, null_count, "{token:?}");
assert_eq!(expected.invalid_count, Some(1 - null_count), "{token:?}");
let ColumnStats::Numeric(stats) = &expected.stats else {
panic!("expected numeric statistics for {token:?}");
};
assert_eq!(stats.min, 0.5, "{token:?}");
assert_eq!(stats.max, 9.5, "{token:?}");
assert_eq!(stats.mean, 5.0, "{token:?}");
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let report = Profiler::new()
.engine(engine)
.analyze_file(csv.path())
.unwrap();
assert_eq!(
serde_json::to_value(&report.column_profiles).unwrap(),
serde_json::to_value(&standard.column_profiles).unwrap(),
"{engine:?}, token {token:?}"
);
}
}
}
#[test]
fn test_header_only_csv_columns_match_across_engines_and_serialization() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,age").unwrap();
csv.flush().unwrap();
let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default())
.expect("standard CSV analysis should succeed");
let reports = [
("standard", standard),
(
"auto",
Profiler::new()
.engine(EngineType::Auto)
.analyze_file(csv.path())
.expect("auto analysis should succeed"),
),
(
"incremental",
Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(csv.path())
.expect("incremental analysis should succeed"),
),
(
"columnar",
Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(csv.path())
.expect("columnar analysis should succeed"),
),
];
for (engine, report) in reports {
assert_eq!(report.execution.rows_processed, 0, "{engine}");
assert_eq!(
report
.column_profiles
.iter()
.map(|column| (column.name.as_str(), column.data_type.clone()))
.collect::<Vec<_>>(),
vec![("name", DataType::String), ("age", DataType::String)],
"{engine}"
);
assert!(report.quality.is_some(), "{engine}");
let serialized = serde_json::to_value(&report).expect("report should serialize");
let serialized_columns = serialized["column_profiles"]
.as_array()
.expect("column_profiles should be an array");
assert_eq!(
serialized_columns
.iter()
.map(|column| column["name"].as_str().expect("column name"))
.collect::<Vec<_>>(),
vec!["name", "age"],
"{engine}"
);
assert!(serialized["quality"].is_object(), "{engine}");
}
}
#[test]
fn test_unassessable_csv_reports_no_overall_score_across_engines() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,age").unwrap();
csv.flush().unwrap();
let reports = [
(
"standard",
analyze_csv_file(csv.path(), &CsvParserConfig::default())
.expect("standard CSV analysis should succeed"),
),
(
"auto",
Profiler::new()
.engine(EngineType::Auto)
.analyze_file(csv.path())
.expect("auto analysis should succeed"),
),
(
"incremental",
Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(csv.path())
.expect("incremental analysis should succeed"),
),
(
"columnar",
Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(csv.path())
.expect("columnar analysis should succeed"),
),
];
for (engine, report) in reports {
assert_eq!(report.quality_score(), None, "{engine}");
let quality = report.quality.as_ref().expect("quality present");
assert!(quality.metrics.assessed_dimensions().is_empty(), "{engine}");
assert_eq!(quality.score(), None, "{engine}");
assert_eq!(quality.metrics.overall_score(), None, "{engine}");
assert!(
matches!(quality.confidence, MetricConfidence::NotAssessed),
"{engine}: confidence must not claim exactness about a score that \
was never computed, got {:?}",
quality.confidence
);
let serialized = serde_json::to_value(&report).expect("report should serialize");
assert_eq!(
serialized["quality"]["confidence"],
json!("NotAssessed"),
"{engine}"
);
}
}
#[test]
fn test_empty_csv_quality_presence_matches_across_engines_and_serialization() {
let csv = NamedTempFile::new().unwrap();
let path = csv.path();
let reports = [
(
"standard",
analyze_csv_file(path, &CsvParserConfig::default())
.expect("standard CSV analysis should succeed"),
),
(
"auto",
Profiler::new()
.engine(EngineType::Auto)
.analyze_file(path)
.expect("auto analysis should succeed"),
),
(
"incremental",
Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(path)
.expect("incremental analysis should succeed"),
),
(
"columnar",
Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(path)
.expect("columnar analysis should succeed"),
),
];
for (engine, report) in reports {
assert!(report.column_profiles.is_empty(), "{engine}");
assert_eq!(report.execution.rows_processed, 0, "{engine}");
assert_eq!(
report.quality_status,
dataprof::QualityAnalysisStatus::Computed,
"{engine}"
);
assert!(report.quality_score().is_none(), "{engine}");
let quality = report.quality.as_ref().expect("empty input was analyzed");
assert!(quality.metrics.assessed_dimensions().is_empty(), "{engine}");
let serialized = serde_json::to_value(&report).expect("report should serialize");
assert!(serialized["quality"].is_object(), "{engine}");
assert_eq!(
serialized["quality_status"],
json!({"state": "computed"}),
"{engine}"
);
assert_eq!(
serialized["quality"]["confidence"],
json!("NotAssessed"),
"{engine}"
);
}
}
#[test]
fn whitespace_padded_numeric_stats_match_across_csv_engines() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,amount").unwrap();
writeln!(csv, "Alice, 1 ").unwrap();
writeln!(csv, "Bob, 2 ").unwrap();
csv.flush().unwrap();
let reports = [
(
"standard",
analyze_csv_file(csv.path(), &CsvParserConfig::default())
.expect("standard CSV analysis should succeed"),
),
(
"auto",
Profiler::new()
.engine(EngineType::Auto)
.analyze_file(csv.path())
.expect("auto analysis should succeed"),
),
(
"incremental",
Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(csv.path())
.expect("incremental analysis should succeed"),
),
(
"columnar",
Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(csv.path())
.expect("columnar analysis should succeed"),
),
];
for (engine, report) in reports {
let amount = report
.column_profiles
.iter()
.find(|column| column.name == "amount")
.expect("amount profile");
assert_eq!(amount.data_type, DataType::Integer, "{engine}");
assert_eq!(amount.invalid_count, Some(0), "{engine}");
let ColumnStats::Numeric(stats) = &amount.stats else {
panic!("{engine} should produce numeric statistics");
};
assert_eq!(stats.min, 1.0, "{engine}");
assert_eq!(stats.max, 2.0, "{engine}");
assert_eq!(stats.mean, 1.5, "{engine}");
}
}
#[test]
fn incremental_chunk_smaller_than_header_preserves_the_file() {
let mut csv = NamedTempFile::new().unwrap();
write!(csv, "alpha,beta\n1,2\n3,4\n").unwrap();
csv.flush().unwrap();
let report = Profiler::new()
.engine(EngineType::Incremental)
.chunk_size(ChunkSize::Fixed(5))
.analyze_file(csv.path())
.expect("incremental analysis should succeed");
assert_eq!(report.execution.rows_processed, 2);
assert_eq!(report.execution.ragged_row_count, 0);
assert!(report.execution.source_exhausted);
assert_eq!(
report
.column_profiles
.iter()
.map(|column| column.name.as_str())
.collect::<Vec<_>>(),
["alpha", "beta"]
);
}
#[test]
fn multiline_csv_crossing_default_chunk_boundary_matches_columnar() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "id,bio").unwrap();
writeln!(csv, "0,{}", "x".repeat(65_490)).unwrap();
writeln!(csv, "1,\"hello\n{}\"", "w".repeat(100)).unwrap();
writeln!(csv, "2,end").unwrap();
csv.flush().unwrap();
let columnar = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(csv.path())
.expect("columnar analysis should succeed");
for engine in [EngineType::Auto, EngineType::Incremental] {
let report = Profiler::new()
.engine(engine)
.analyze_file(csv.path())
.expect("incremental analysis should succeed");
assert_eq!(
report.execution.engine.as_deref(),
Some("incremental"),
"{engine:?}"
);
assert_eq!(report.execution.rows_processed, 3, "{engine:?}");
assert_eq!(report.execution.ragged_row_count, 0, "{engine:?}");
assert_eq!(
report
.column_profiles
.iter()
.map(|column| (&column.name, &column.data_type, column.total_count))
.collect::<Vec<_>>(),
columnar
.column_profiles
.iter()
.map(|column| (&column.name, &column.data_type, column.total_count))
.collect::<Vec<_>>(),
"{engine:?}"
);
}
}
#[test]
fn test_base_numeric_stats_exact_beyond_sample_capacity() {
let csv = create_sorted_30k_csv();
let path = csv.path();
let std_report = analyze_csv_file(path, &CsvParserConfig::default())
.expect("standard CSV analysis should succeed");
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(path)
.expect("Arrow CSV analysis should succeed");
for (engine, report) in [("std", &std_report), ("arrow", &arrow_report)] {
let id = report
.column_profiles
.iter()
.find(|c| c.name == "id")
.expect("id column");
let ColumnStats::Numeric(n) = &id.stats else {
panic!("id should have numeric stats");
};
assert_eq!(n.min, 0.0, "[{engine}] min must be exact");
assert_eq!(n.max, 29_999.0, "[{engine}] max must be exact");
assert!(
(n.mean - 14_999.5).abs() < 1e-6,
"[{engine}] mean must be exact, got {}",
n.mean
);
assert_eq!(
n.is_approximate,
Some(true),
"[{engine}] sampled order statistics must be disclosed as approximate"
);
assert_eq!(id.invalid_count, Some(0), "[{engine}] invalid_count");
}
let standard = serde_json::to_value(&std_report.column_profiles).unwrap();
let arrow = serde_json::to_value(&arrow_report.column_profiles).unwrap();
assert_eq!(
std_report.column_profiles.len(),
arrow_report.column_profiles.len()
);
for index in 0..std_report.column_profiles.len() {
for field in ["min", "max", "mean", "std_dev", "variance"] {
assert!(standard[index]["stats"]["Numeric"][field].is_number());
assert!(arrow[index]["stats"]["Numeric"][field].is_number());
assert_eq!(
standard[index]["stats"]["Numeric"][field], arrow[index]["stats"]["Numeric"][field],
"column {index}, serialized {field}"
);
}
}
}
#[test]
fn test_invalid_count_surfaces_unparseable_numeric_values() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "amount,label").unwrap();
for i in 0..5 {
writeln!(f, "{}.5,row{}", i, i).unwrap();
}
writeln!(f, "\"12,50\",rowx").unwrap();
writeln!(f, ",rownull").unwrap();
f.flush().unwrap();
let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default()).unwrap();
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.unwrap();
for (engine, report) in [("std", &std_report), ("arrow", &arrow_report)] {
let amount = report
.column_profiles
.iter()
.find(|c| c.name == "amount")
.unwrap();
assert_eq!(amount.data_type, DataType::Float, "[{engine}]");
assert_eq!(amount.total_count, 7, "[{engine}]");
assert_eq!(amount.null_count, 1, "[{engine}]");
assert_eq!(
amount.invalid_count,
Some(1),
"[{engine}] the unparseable value must be counted, not silently dropped"
);
let label = report
.column_profiles
.iter()
.find(|c| c.name == "label")
.unwrap();
assert_eq!(
label.invalid_count, None,
"[{engine}] non-numeric columns are not checked: None, not 0"
);
}
}
#[test]
fn test_small_data_not_flagged_approximate() {
let csv = create_test_csv();
let report = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
let age = report
.column_profiles
.iter()
.find(|c| c.name == "age")
.unwrap();
let ColumnStats::Numeric(n) = &age.stats else {
panic!("age should be numeric");
};
assert_eq!(
n.is_approximate,
Some(false),
"full-coverage stats are not approximate"
);
assert_eq!(n.min, 20.0);
assert_eq!(n.max, 69.0);
}
fn create_test_csv() -> NamedTempFile {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "name,age,salary,score").unwrap();
for i in 0..100 {
writeln!(
f,
"Person{},{},{:.2},{:.1}",
i,
20 + i % 50,
30000.0 + i as f64 * 500.0,
50.0 + (i % 50) as f64
)
.unwrap();
}
f.flush().unwrap();
f
}
#[test]
fn test_standard_vs_arrow_csv_numeric_stats() {
let csv = create_test_csv();
let path = csv.path();
let std_report = analyze_csv_file(path, &CsvParserConfig::default())
.expect("standard CSV analysis should succeed");
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(path)
.expect("Arrow CSV analysis should succeed");
assert_eq!(
serde_json::to_value(&std_report.column_profiles).unwrap(),
serde_json::to_value(&arrow_report.column_profiles).unwrap(),
"serialized column profiles must match across CSV engines"
);
assert_eq!(
serde_json::to_value(&std_report.quality).unwrap(),
serde_json::to_value(&arrow_report.quality).unwrap(),
"serialized quality must match across CSV engines"
);
for (standard, arrow) in std_report
.column_profiles
.iter()
.zip(&arrow_report.column_profiles)
{
if let (ColumnStats::Numeric(a), ColumnStats::Numeric(b)) = (&standard.stats, &arrow.stats)
{
for (field, left, right) in [
("min", a.min, b.min),
("max", a.max, b.max),
("mean", a.mean, b.mean),
(
"std_dev",
a.std_dev.expect("finite spread"),
b.std_dev.expect("finite spread"),
),
(
"variance",
a.variance.expect("finite spread"),
b.variance.expect("finite spread"),
),
] {
let tolerance = (1e-9 * left.abs().max(right.abs())).max(1e-12);
assert!(
(left - right).abs() <= tolerance,
"{} raw {field}: {left} vs {right}",
standard.name
);
}
}
}
}
#[test]
fn test_precision_quality_matches_standard_and_columnar_csv() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "amount").unwrap();
for value in ["1.25", "2.35", "3.45", "4.5"] {
writeln!(csv, "{value}").unwrap();
}
csv.flush().unwrap();
let standard = analyze_csv_file(csv.path(), &CsvParserConfig::default()).unwrap();
let columnar = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(csv.path())
.unwrap();
let standard_precision = standard
.quality
.unwrap()
.metrics
.precision
.expect("standard precision");
let columnar_precision = columnar
.quality
.unwrap()
.metrics
.precision
.expect("columnar precision");
assert_eq!(
standard_precision.numeric_values_checked,
columnar_precision.numeric_values_checked
);
assert_eq!(
standard_precision.inconsistent_precision_values,
columnar_precision.inconsistent_precision_values
);
assert!(
(standard_precision.decimal_places_consistency
- columnar_precision.decimal_places_consistency)
.abs()
< 0.01
);
}
#[test]
fn test_mixed_data_column_type_consistency() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,value,date_col").unwrap();
writeln!(f, "1,100,2024-01-01").unwrap();
writeln!(f, "2,200,2024-01-02").unwrap();
writeln!(f, "3,N/A,2024-01-03").unwrap();
writeln!(f, "4,400,2024-01-04").unwrap();
writeln!(f, "5,500,2024-01-05").unwrap();
f.flush().unwrap();
let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default())
.expect("standard CSV should succeed");
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.expect("Arrow CSV should succeed");
for std_col in &std_report.column_profiles {
let arrow_col = arrow_report
.column_profiles
.iter()
.find(|c| c.name == std_col.name)
.unwrap_or_else(|| panic!("Column '{}' missing from Arrow report", std_col.name));
assert_eq!(
std_col.data_type, arrow_col.data_type,
"Type mismatch for column '{}': std={:?}, arrow={:?}",
std_col.name, std_col.data_type, arrow_col.data_type
);
}
}
#[test]
fn test_problematic_date_columns_stay_consistent_across_engines() {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("examples/test_datasets");
for (file_name, column_name) in [
("sales_data_problematic.csv", "order_date"),
("sensor_data_outliers.csv", "timestamp"),
] {
let path = root.join(file_name);
let std_report =
analyze_csv_file(&path, &CsvParserConfig::default()).unwrap_or_else(|err| {
panic!("standard CSV analysis failed for {}: {}", file_name, err)
});
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(&path)
.unwrap_or_else(|err| panic!("Arrow CSV analysis failed for {}: {}", file_name, err));
let std_col = std_report
.column_profiles
.iter()
.find(|c| c.name == column_name)
.unwrap_or_else(|| {
panic!(
"Column '{}' missing from standard report for {}",
column_name, file_name
)
});
let arrow_col = arrow_report
.column_profiles
.iter()
.find(|c| c.name == column_name)
.unwrap_or_else(|| {
panic!(
"Column '{}' missing from Arrow report for {}",
column_name, file_name
)
});
assert_eq!(std_col.data_type, DataType::Date);
assert_eq!(
std_col.data_type, arrow_col.data_type,
"Type mismatch for '{}' in {}: std={:?}, arrow={:?}",
column_name, file_name, std_col.data_type, arrow_col.data_type
);
}
}
#[test]
fn test_streaming_vs_batch_quality_confidence() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,name,value").unwrap();
for i in 0..200 {
let name = if i % 10 == 0 { "" } else { "Alice" }; let value = if i % 20 == 0 {
""
} else {
&format!("{}", i * 100)
}; writeln!(f, "{},{},{}", i, name, value).unwrap();
}
f.flush().unwrap();
let batch_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.expect("Batch analysis should succeed");
let streaming_report = Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(f.path())
.expect("Streaming analysis should succeed");
let batch_quality = batch_report
.quality
.as_ref()
.expect("Batch report should have quality");
let streaming_quality = streaming_report
.quality
.as_ref()
.expect("Streaming report should have quality");
assert!(
matches!(batch_quality.confidence, MetricConfidence::Exact),
"Batch engine should produce Exact confidence, got {:?}",
batch_quality.confidence
);
let streaming_m = &streaming_quality.metrics;
assert!(
streaming_m.missing_values_ratio() > 0.0,
"Streaming engine should detect missing values from empty CSV fields"
);
let batch_m = &batch_quality.metrics;
assert!(
(batch_m.key_uniqueness() - streaming_m.key_uniqueness()).abs() < 5.0,
"key_uniqueness: batch={:.2} vs streaming={:.2}",
batch_m.key_uniqueness(),
streaming_m.key_uniqueness()
);
}
#[test]
fn test_streaming_bifurcation_with_large_dataset() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,category,amount").unwrap();
for i in 0..15_000 {
let cat = if i % 3 == 0 {
"A"
} else if i % 3 == 1 {
"B"
} else {
"C"
};
let amount = if i % 100 == 0 {
""
} else {
&format!("{:.2}", i as f64 * 1.5)
};
writeln!(f, "{},{},{}", i, cat, amount).unwrap();
}
f.flush().unwrap();
let report = Profiler::new()
.engine(EngineType::Incremental)
.identifier_columns(vec!["id".to_string()])
.analyze_file(f.path())
.expect("Large streaming analysis should succeed");
let quality = report
.quality
.as_ref()
.expect("Should have quality assessment");
match &quality.confidence {
MetricConfidence::Mixed {
exact_dimensions,
sampled_dimensions,
sample_size,
} => {
assert!(
exact_dimensions.contains(&"completeness".to_string()),
"completeness should be exact"
);
assert!(
exact_dimensions.contains(&"key_uniqueness".to_string()),
"key_uniqueness should be exact"
);
assert!(
sampled_dimensions.contains(&"consistency".to_string()),
"consistency should be sampled"
);
assert!(
sampled_dimensions.contains(&"accuracy".to_string()),
"accuracy should be sampled"
);
assert!(
*sample_size < 15_000,
"sample_size ({}) should be less than total rows (15000)",
sample_size
);
}
other => panic!(
"Expected Mixed confidence for large streaming dataset, got {:?}",
other
),
}
let m = &quality.metrics;
assert!(
m.missing_values_ratio() > 0.0,
"Should detect some missing values"
);
assert!(
m.missing_values_ratio() < 2.0,
"Missing ratio should be small (~0.33%)"
);
}
#[test]
fn test_distinct_count_marked_exact_on_small_data_across_engines() {
let csv = create_test_csv(); let path = csv.path();
let std_report = analyze_csv_file(path, &CsvParserConfig::default()).unwrap();
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(path)
.unwrap();
for std_col in &std_report.column_profiles {
let arrow_col = arrow_report
.column_profiles
.iter()
.find(|c| c.name == std_col.name)
.unwrap();
assert_eq!(
std_col.unique_count_is_approximate,
Some(false),
"standard engine should mark '{}' exact",
std_col.name
);
assert_eq!(
arrow_col.unique_count_is_approximate,
Some(false),
"columnar engine should mark '{}' exact",
std_col.name
);
assert_eq!(
std_col.unique_count, arrow_col.unique_count,
"exact unique_count should match across engines for '{}'",
std_col.name
);
}
}
#[test]
fn test_table_sized_cardinality_is_exact_across_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,category").unwrap();
let rows = 20_000;
for i in 0..rows {
writeln!(f, "id-{i},{}", if i % 2 == 0 { "A" } else { "B" }).unwrap();
}
f.flush().unwrap();
let std_report = analyze_csv_file(f.path(), &CsvParserConfig::default()).unwrap();
let incremental_report = Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(f.path())
.unwrap();
let arrow_report = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.unwrap();
for report in [&std_report, &incremental_report, &arrow_report] {
for (name, distinct) in [("id", rows), ("category", 2)] {
let column = report
.column_profiles
.iter()
.find(|c| c.name == name)
.expect("column present");
assert_eq!(column.unique_count_is_approximate, Some(false), "{name}");
assert_eq!(column.unique_count, Some(distinct), "{name}");
}
}
}
#[test]
fn test_profiler_selective_dimensions_only_completeness() {
let csv = create_test_csv();
let report = Profiler::new()
.quality_dimensions(vec![QualityDimension::Completeness])
.analyze_file(csv.path())
.unwrap();
let quality = report.quality.expect("quality should be present");
let m = &quality.metrics;
assert!(m.completeness.is_some(), "completeness should be computed");
assert!(m.consistency.is_none(), "consistency should be skipped");
assert!(m.uniqueness.is_none(), "uniqueness should be skipped");
assert!(m.accuracy.is_none(), "accuracy should be skipped");
assert!(m.timeliness.is_none(), "timeliness should be skipped");
assert!(m.validity.is_none(), "validity should be skipped");
assert!(m.precision.is_none(), "precision should be skipped");
let score = m.overall_score().expect("completeness was assessed");
let completeness_score = m.completeness.as_ref().unwrap().complete_records_ratio;
assert!(
(score - completeness_score).abs() < 0.01,
"score {score} should equal completeness {completeness_score}"
);
}
#[test]
fn test_profiler_selective_dimensions_subset() {
let csv = create_test_csv();
let report = Profiler::new()
.quality_dimensions(vec![
QualityDimension::Completeness,
QualityDimension::Uniqueness,
])
.analyze_file(csv.path())
.unwrap();
let quality = report.quality.expect("quality should be present");
let m = &quality.metrics;
assert!(m.completeness.is_some());
assert!(m.consistency.is_none());
assert!(m.uniqueness.is_some());
assert!(m.accuracy.is_none());
assert!(m.timeliness.is_none());
assert!(m.validity.is_none());
assert!(m.precision.is_none());
}
#[test]
fn test_profiler_all_dimensions_default() {
let csv = create_test_csv();
let report = Profiler::new().analyze_file(csv.path()).unwrap();
let quality = report.quality.expect("quality should be present");
let m = &quality.metrics;
assert!(m.completeness.is_some());
assert!(m.consistency.is_some());
assert!(m.uniqueness.is_some());
assert!(m.accuracy.is_some());
assert!(m.timeliness.is_some());
assert!(m.validity.is_some());
assert!(m.precision.is_some());
}
#[test]
fn test_duplicate_rows_with_nulls_match_across_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,name,value,city").unwrap();
writeln!(f, "1,Alice,10.5,Rome").unwrap();
writeln!(f, "2,,20.0,Milan").unwrap();
writeln!(f, "2,,20.0,Milan").unwrap(); writeln!(f, "3,Bob,,Naples").unwrap();
writeln!(f, "4,Alice,10.5,").unwrap();
writeln!(f, "1,Alice,10.5,Rome").unwrap(); f.flush().unwrap();
let incremental = Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(f.path())
.expect("incremental analysis should succeed");
let columnar = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.expect("columnar analysis should succeed");
let inc_uniq = incremental
.quality
.as_ref()
.and_then(|q| q.metrics.uniqueness.as_ref())
.expect("incremental uniqueness should be assessed");
let col_uniq = columnar
.quality
.as_ref()
.and_then(|q| q.metrics.uniqueness.as_ref())
.expect("columnar uniqueness should be assessed");
assert_eq!(inc_uniq.rows_checked, 6, "incremental rows_checked");
assert_eq!(col_uniq.rows_checked, 6, "columnar rows_checked");
assert_eq!(inc_uniq.duplicate_rows, 2, "incremental duplicate_rows");
assert_eq!(col_uniq.duplicate_rows, 2, "columnar duplicate_rows");
assert_eq!(
inc_uniq.key_uniqueness, col_uniq.key_uniqueness,
"key_uniqueness must match across engines"
);
let inc_score = incremental.quality_score().expect("incremental score");
let col_score = columnar.quality_score().expect("columnar score");
assert!(
(inc_score - col_score).abs() < 0.01,
"overall quality must not depend on the engine: incremental={inc_score} columnar={col_score}"
);
}
#[test]
fn test_duplicate_rows_without_nulls_match_across_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "id,name").unwrap();
writeln!(f, "1,Alice").unwrap();
writeln!(f, "2,Bob").unwrap();
writeln!(f, "2,Bob").unwrap();
writeln!(f, "3,Carol").unwrap();
f.flush().unwrap();
for engine in [EngineType::Incremental, EngineType::Columnar] {
let report = Profiler::new()
.engine(engine)
.analyze_file(f.path())
.expect("analysis should succeed");
let uniq = report
.quality
.as_ref()
.and_then(|q| q.metrics.uniqueness.as_ref())
.expect("uniqueness should be assessed");
assert_eq!(uniq.rows_checked, 4, "{engine:?} rows_checked");
assert_eq!(uniq.duplicate_rows, 1, "{engine:?} duplicate_rows");
}
}
#[test]
fn test_duplicate_headers_rejected_across_csv_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "x,x,y").unwrap();
writeln!(f, "1,2,a").unwrap();
writeln!(f, "3,4,b").unwrap();
f.flush().unwrap();
let err = analyze_csv_file(f.path(), &CsvParserConfig::default())
.expect_err("standard engine must reject duplicate headers");
assert_eq!(err.category(), "duplicate_column_name");
let msg = err.to_string();
assert!(msg.contains("'x'"), "names the offender: {msg}");
assert!(!msg.contains('1'), "must not echo cell values: {msg}");
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let err = Profiler::new()
.engine(engine)
.analyze_file(f.path())
.expect_err("engine must reject duplicate headers");
assert_eq!(
err.category(),
"duplicate_column_name",
"{engine:?} should surface the duplicate-column error directly"
);
}
}
#[test]
fn test_unique_headers_preserve_column_count_and_totals() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "a,b,c").unwrap();
writeln!(f, "1,2,3").unwrap();
writeln!(f, "4,5,6").unwrap();
f.flush().unwrap();
for engine in [EngineType::Incremental, EngineType::Columnar] {
let report = Profiler::new()
.engine(engine)
.analyze_file(f.path())
.expect("unique headers should profile");
assert_eq!(report.column_profiles.len(), 3, "{engine:?} column count");
for col in &report.column_profiles {
assert_eq!(col.total_count, 2, "{engine:?} {} total_count", col.name);
}
}
}
#[test]
fn ragged_rows_report_identically_across_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "name,age,city").unwrap();
writeln!(f, "Alice,25,Rome").unwrap();
writeln!(f, "Bob,30").unwrap(); writeln!(f, "Carol,35,LA,EXTRA").unwrap(); writeln!(f, "Dave,40,SF").unwrap();
f.flush().unwrap();
let incremental = Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(f.path())
.expect("incremental recovers ragged rows");
let columnar = Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.expect("columnar recovers ragged rows");
assert_eq!(incremental.execution.rows_processed, 4);
assert_eq!(
columnar.execution.rows_processed, incremental.execution.rows_processed,
"both engines must see every row"
);
assert_eq!(incremental.execution.ragged_row_count, 2);
assert_eq!(
columnar.execution.ragged_row_count, incremental.execution.ragged_row_count,
"the columnar engine must not report a repaired scan as a clean one"
);
assert_eq!(
columnar.column_profiles.len(),
3,
"the surplus field must not become a fourth column"
);
}
#[test]
fn surplus_fields_do_not_leak_into_duplicate_detection() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "name,age,city").unwrap();
writeln!(f, "Alice,25,Rome").unwrap();
writeln!(f, "Alice,25,Rome,X").unwrap();
f.flush().unwrap();
for engine in [EngineType::Incremental, EngineType::Columnar] {
let report = Profiler::new()
.engine(engine)
.analyze_file(f.path())
.expect("ragged rows profile");
let uniqueness = report
.quality
.as_ref()
.and_then(|q| q.metrics.uniqueness.as_ref())
.unwrap_or_else(|| panic!("{engine:?} uniqueness should be assessed"));
assert_eq!(
uniqueness.duplicate_rows, 1,
"{engine:?} must see the truncated row as a duplicate"
);
}
}
#[test]
fn type_homogeneity_agrees_across_engines() {
let mut f = NamedTempFile::new().unwrap();
writeln!(f, "amount,name,when").unwrap();
for i in 0..200 {
let amount = if i % 4 == 0 {
"pending".to_string()
} else {
format!("{}.50", 100 + i)
};
writeln!(f, "{amount},person{i},2024-01-15").unwrap();
}
f.flush().unwrap();
let standard = analyze_csv_file(f.path(), &CsvParserConfig::default())
.expect("standard CSV analysis should succeed");
let reports = [
("standard", standard),
(
"auto",
Profiler::new()
.engine(EngineType::Auto)
.analyze_file(f.path())
.expect("auto analysis should succeed"),
),
(
"incremental",
Profiler::new()
.engine(EngineType::Incremental)
.analyze_file(f.path())
.expect("incremental analysis should succeed"),
),
(
"columnar",
Profiler::new()
.engine(EngineType::Columnar)
.analyze_file(f.path())
.expect("columnar analysis should succeed"),
),
];
for (engine, report) in reports {
let counts = |name: &str| {
let column = report
.column_profiles
.iter()
.find(|column| column.name == name)
.unwrap_or_else(|| panic!("[{engine}] {name} column"));
(
column.data_type.clone(),
column
.type_homogeneity
.unwrap_or_else(|| panic!("[{engine}] {name} must be classified")),
)
};
let (amount_type, amount) = counts("amount");
assert_eq!(
amount_type,
DataType::String,
"[{engine}] the fixture must reproduce the inference fallback"
);
assert_eq!(amount.numeric, 150, "[{engine}] amount numeric");
assert_eq!(amount.text, 50, "[{engine}] amount text");
assert_eq!(amount.dominant_share(), Some(0.75), "[{engine}] amount");
let (name_type, name) = counts("name");
assert_eq!(name_type, DataType::String, "[{engine}] name");
assert_eq!(name.text, 200, "[{engine}] name text");
assert_eq!(name.dominant_share(), Some(1.0), "[{engine}] name");
let (_, when) = counts("when");
assert_eq!(when.date, 200, "[{engine}] when date");
let serialized = serde_json::to_value(&report).expect("report should serialize");
assert_eq!(
serialized["column_profiles"][0]["type_homogeneity"],
json!({"numeric": 150, "date": 0, "boolean": 0, "text": 50}),
"[{engine}] serialized amount counts"
);
}
}
#[test]
fn a_ragged_row_past_the_row_cap_is_ignored_by_every_engine() {
for row_past_cap in ["Bob,30", "Bob,30,Milan,extra"] {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,age,city").unwrap();
writeln!(csv, "Alice,25,Rome").unwrap();
writeln!(csv, "{row_past_cap}").unwrap();
csv.flush().unwrap();
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let report = Profiler::new()
.engine(engine)
.csv_flexible(true)
.stop_when(StopCondition::MaxRows(1))
.analyze_file(csv.path())
.unwrap_or_else(|error| {
panic!("{engine:?} failed over a row past the cap ({row_past_cap:?}): {error}")
});
assert_eq!(report.execution.rows_processed, 1, "{engine:?}");
assert_eq!(report.execution.ragged_row_count, 0, "{engine:?}");
assert!(
matches!(
report.execution.truncation_reason,
Some(TruncationReason::MaxRows(1))
),
"{engine:?}: {:?}",
report.execution.truncation_reason
);
}
}
}
#[test]
fn strict_parsing_rejects_a_ragged_row_past_the_row_cap_on_every_engine() {
for row_past_cap in ["Bob,30", "Bob,30,Milan,extra"] {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,age,city").unwrap();
writeln!(csv, "Alice,25,Rome").unwrap();
writeln!(csv, "{row_past_cap}").unwrap();
csv.flush().unwrap();
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let error = Profiler::new()
.engine(engine)
.csv_flexible(false)
.stop_when(StopCondition::MaxRows(1))
.analyze_file(csv.path())
.expect_err(&format!("{engine:?} must reject {row_past_cap:?}"));
let message = error.to_string();
assert!(message.contains("fields"), "{engine:?}: {message}");
assert!(
!message.contains("incorrect number of fields"),
"{engine:?}: {message}"
);
}
}
}
#[test]
fn a_cap_at_or_beyond_the_file_length_is_not_a_truncation() {
let mut csv = NamedTempFile::new().unwrap();
writeln!(csv, "name,age,city").unwrap();
writeln!(csv, "Alice,25,Rome").unwrap();
writeln!(csv, "Bob,30,Milan").unwrap();
csv.flush().unwrap();
for cap in [2u64, 5] {
for engine in [
EngineType::Auto,
EngineType::Incremental,
EngineType::Columnar,
] {
let report = Profiler::new()
.engine(engine)
.stop_when(StopCondition::MaxRows(cap))
.analyze_file(csv.path())
.expect("a cap the file does not exceed should profile the file");
assert_eq!(report.execution.rows_processed, 2, "{engine:?} cap {cap}");
assert!(
report.execution.truncation_reason.is_none(),
"{engine:?} cap {cap}: {:?}",
report.execution.truncation_reason
);
}
}
}