use std::io::Write;
use dataprof::{ProfileReport, Profiler};
use tempfile::NamedTempFile;
struct Case {
name: &'static str,
records: &'static str,
duplicates: usize,
arrow: bool,
}
const CASES: [Case; 9] = [
Case {
name: "key order",
records: r#"{"id": 1, "a": {"x": 1, "y": 2}}
{"id": 1, "a": {"y": 2, "x": 1}}"#,
duplicates: 1,
arrow: true,
},
Case {
name: "nested key order",
records: r#"{"id": 1, "a": {"b": {"x": 1, "y": [1, 2]}, "c": "k"}}
{"id": 1, "a": {"c": "k", "b": {"y": [1, 2], "x": 1}}}"#,
duplicates: 1,
arrow: true,
},
Case {
name: "comma inside a list element",
records: r#"{"id": 1, "tags": ["p, q"]}
{"id": 1, "tags": ["p", "q"]}"#,
duplicates: 0,
arrow: true,
},
Case {
name: "null element against an empty list",
records: r#"{"id": 1, "tags": ["p"]}
{"id": 2, "tags": [null]}
{"id": 2, "tags": []}"#,
duplicates: 0,
arrow: true,
},
Case {
name: "null child repeated",
records: r#"{"id": 1, "a": {"x": null, "y": "1"}}
{"id": 1, "a": {"x": null, "y": "1"}}
{"id": 1, "a": {"x": "1", "y": null}}"#,
duplicates: 1,
arrow: true,
},
Case {
name: "empty object against an empty list",
records: r#"{"id": 1, "a": {}}
{"id": 1, "a": []}"#,
duplicates: 0,
arrow: false,
},
Case {
name: "a string spelling an object against the object",
records: r#"{"id": 1, "a": "{\"x\":1,\"y\":2}"}
{"id": 1, "a": {"y": 2, "x": 1}}"#,
duplicates: 0,
arrow: false,
},
Case {
name: "a string spelling an array against the array",
records: r#"{"id": 1, "a": "[1,2]"}
{"id": 1, "a": [1, 2]}"#,
duplicates: 0,
arrow: false,
},
Case {
name: "equal strings spelling an object",
records: r#"{"id": 1, "a": "{\"x\":1}"}
{"id": 1, "a": "{\"x\":1}"}"#,
duplicates: 1,
arrow: true,
},
];
fn file(suffix: &str, contents: &[u8]) -> NamedTempFile {
let mut file = NamedTempFile::with_suffix(suffix).unwrap();
file.write_all(contents).unwrap();
file.flush().unwrap();
file
}
fn json_array(records: &str) -> String {
format!("[{}]", records.lines().collect::<Vec<_>>().join(","))
}
#[cfg(feature = "parquet")]
fn parquet(records: &str) -> NamedTempFile {
use std::io::Cursor;
use std::sync::Arc;
use arrow::json::ReaderBuilder;
use arrow::json::reader::infer_json_schema;
use parquet::arrow::ArrowWriter;
let (schema, _) = infer_json_schema(Cursor::new(records), None).unwrap();
let schema = Arc::new(schema);
let reader = ReaderBuilder::new(schema.clone())
.build(Cursor::new(records))
.unwrap();
let file = NamedTempFile::with_suffix(".parquet").unwrap();
let mut writer = ArrowWriter::try_new(file.reopen().unwrap(), schema, None).unwrap();
for batch in reader {
writer.write(&batch.unwrap()).unwrap();
}
writer.close().unwrap();
file
}
fn duplicates(report: &ProfileReport) -> Option<usize> {
report
.quality
.as_ref()
.and_then(|quality| quality.metrics.uniqueness.as_ref())
.map(|uniqueness| uniqueness.duplicate_rows)
}
#[test]
fn equal_nested_rows_are_duplicates_on_every_file_route() {
let mut wrong = Vec::new();
for case in &CASES {
let mut routes = vec![
("jsonl", file(".jsonl", case.records.as_bytes())),
("json", file(".json", json_array(case.records).as_bytes())),
];
#[cfg(feature = "parquet")]
if case.arrow {
routes.push(("parquet", parquet(case.records)));
}
for (route, source) in routes {
let report = Profiler::new()
.analyze_file(source.path())
.unwrap_or_else(|e| panic!("[{} / {route}] {e}", case.name));
let found = duplicates(&report);
if found != Some(case.duplicates) {
wrong.push(format!(
"{} / {route}: {found:?}, expected {}",
case.name, case.duplicates
));
}
}
}
assert!(
wrong.is_empty(),
"wrong duplicate counts:\n{}",
wrong.join("\n")
);
}
#[cfg(feature = "async-streaming")]
#[tokio::test]
async fn equal_nested_rows_are_duplicates_on_the_async_route() {
use dataprof::{AsyncSourceInfo, BytesSource, FileFormat};
let mut wrong = Vec::new();
for case in &CASES {
let source = BytesSource::new(
bytes::Bytes::from(case.records.to_string()),
AsyncSourceInfo::new(case.name, FileFormat::Jsonl)
.size_hint(Some(case.records.len() as u64)),
);
let report = Profiler::new()
.profile_stream(source)
.await
.unwrap_or_else(|e| panic!("[{} / async] {e}", case.name));
let found = duplicates(&report);
if found != Some(case.duplicates) {
wrong.push(format!(
"{} / async: {found:?}, expected {}",
case.name, case.duplicates
));
}
}
assert!(
wrong.is_empty(),
"wrong duplicate counts:\n{}",
wrong.join("\n")
);
}