rust-data-processing 0.2.2

Schema-first ingestion (CSV, JSON, Parquet, Excel) into an in-memory DataSet, plus Polars-backed pipelines, SQL, profiling, validation, and map/reduce-style processing.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

//! Incremental / high-water filtering on file ingest (`P2-E1-S2a`, `P2-E1-S2c`) and option plumbing for DB (`P2-E1-S2b`).

use std::path::PathBuf;

use rust_data_processing::ingestion::{IngestionOptions, ingest_from_path};
use rust_data_processing::types::{DataType, Field, Schema, Value};

fn fixture(name: &str) -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures")
        .join(name)
}

fn events_schema() -> Schema {
    Schema::new(vec![
        Field::new("id", DataType::Int64),
        Field::new("ts", DataType::Int64),
    ])
}

#[test]
fn watermark_file_happy_path_keeps_rows_strictly_above() {
    let path = fixture("watermark_events.csv");
    let schema = events_schema();
    let opts = IngestionOptions {
        watermark_column: Some("ts".to_string()),
        watermark_exclusive_above: Some(Value::Int64(100)),
        ..Default::default()
    };
    let ds = ingest_from_path(&path, &schema, &opts).unwrap();
    assert_eq!(ds.row_count(), 2);
    let ids: Vec<i64> = ds
        .rows
        .iter()
        .map(|r| match r[0] {
            Value::Int64(i) => i,
            _ => panic!("expected id int"),
        })
        .collect();
    assert_eq!(ids, vec![2, 4]);
}

#[test]
fn watermark_file_empty_when_all_at_or_below_high_water() {
    let path = fixture("watermark_events.csv");
    let schema = events_schema();
    let opts = IngestionOptions {
        watermark_column: Some("ts".to_string()),
        watermark_exclusive_above: Some(Value::Int64(200)),
        ..Default::default()
    };
    let ds = ingest_from_path(&path, &schema, &opts).unwrap();
    assert_eq!(ds.row_count(), 0);
}

#[test]
fn watermark_json_matches_csv_semantics() {
    let path = fixture("watermark_events.json");
    let schema = events_schema();
    let opts = IngestionOptions {
        watermark_column: Some("ts".to_string()),
        watermark_exclusive_above: Some(Value::Int64(100)),
        ..Default::default()
    };
    let ds = ingest_from_path(&path, &schema, &opts).unwrap();
    assert_eq!(ds.row_count(), 2);
    let ids: Vec<i64> = ds
        .rows
        .iter()
        .map(|r| match r[0] {
            Value::Int64(i) => i,
            _ => panic!("expected id int"),
        })
        .collect();
    assert_eq!(ids, vec![2, 4]);
}

#[test]
fn watermark_rejects_column_without_value() {
    let path = fixture("watermark_events.csv");
    let schema = events_schema();
    let opts = IngestionOptions {
        watermark_column: Some("ts".to_string()),
        watermark_exclusive_above: None,
        ..Default::default()
    };
    let err = ingest_from_path(&path, &schema, &opts).unwrap_err();
    assert!(err.to_string().contains("watermark_column"));
}

#[test]
fn db_stub_accepts_options_without_panic() {
    use rust_data_processing::ingestion::{ingest_from_db, ingest_from_db_infer};
    let schema = events_schema();
    let opts = IngestionOptions::default();
    let e = ingest_from_db("not a url", "SELECT 1", &schema, &opts).unwrap_err();
    assert!(
        e.to_string().contains("db ingestion is disabled") || e.to_string().contains("invalid")
    );

    let e2 = ingest_from_db_infer("not a url", "SELECT 1", &opts).unwrap_err();
    assert!(
        e2.to_string().contains("db ingestion is disabled") || e2.to_string().contains("invalid")
    );
}