etl-unit 0.1.0

Semantic data model for ETL units — qualities and measurements over subjects and time. Built on Polars.
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119

//! Phase 5: Universe Assembly
//!
//! Assembles the final Universe with metadata.
//!
//! ## Type Signature
//!
//! ```text
//! assemble_universe: (
//!     HashMap<CanonicalColumnName, MeasurementData>,
//!     HashMap<CanonicalColumnName, QualityData>,
//!     EtlSchema,
//!     &EtlUniverseBuildPlan,
//!     Vec<CrushedComponent>,
//!     Duration
//! ) -> EtlResult<Universe>
//! ```

use std::{collections::HashMap, time::Duration};

use super::{
    CrushedComponent, CrushedComponentInfo, MeasurementData, QualityData, Universe,
    UniverseBuildInfo,
};
use crate::{
    CanonicalColumnName, error::EtlResult, schema::EtlSchema, source::EtlUniverseBuildPlan,
};

/// Assemble the final Universe with metadata.
pub fn assemble_universe(
    measurements: HashMap<CanonicalColumnName, MeasurementData>,
    qualities: HashMap<CanonicalColumnName, QualityData>,
    schema: EtlSchema,
    plan: &EtlUniverseBuildPlan,
    crushed_components: Vec<CrushedComponent>,
    duration: Duration,
) -> EtlResult<Universe> {
    let build_info = build_info(
        &measurements,
        &qualities,
        &schema,
        plan,
        crushed_components,
        duration,
    );
    Ok(Universe::new(measurements, qualities, schema, build_info))
}

/// Build the audit/info structure.
fn build_info(
    measurements: &HashMap<CanonicalColumnName, MeasurementData>,
    qualities: &HashMap<CanonicalColumnName, QualityData>,
    schema: &EtlSchema,
    plan: &EtlUniverseBuildPlan,
    crushed_components: Vec<CrushedComponent>,
    duration: Duration,
) -> UniverseBuildInfo {
    // Count total rows and unique subjects across all measurements
    let (total_rows, subject_count) = measurements
        .values()
        .next()
        .map(|m| {
            let rows = m.fragment().height();
            let subjects = m
                .fragment()
                .as_dataframe()
                .ok()
                .and_then(|df| {
                    df.column(schema.subject.as_str())
                        .map(|c| c.n_unique().unwrap_or(0))
                        .ok()
                })
                .unwrap_or(0);
            (rows, subjects)
        })
        .unwrap_or((0, 0));

    // Convert CrushedComponent to CrushedComponentInfo
    let crushed_info: Vec<CrushedComponentInfo> = crushed_components
        .into_iter()
        .map(|c| {
            CrushedComponentInfo::new(
                c.measurement_name.as_str(),
                c.component_name.as_str(),
                c.input_units_without,
                c.input_units_with,
                &format!("{:?}", c.aggregation).to_lowercase(),
            )
        })
        .collect();

    // Collect signal policy stats from all measurements
    let signal_policy_stats = measurements
        .values()
        .filter_map(|m| m.signal_policy_stats().cloned())
        .collect();

    UniverseBuildInfo::builder(&schema.name)
        .sources_used(plan.sources.iter().map(|s| s.name.clone()).collect())
        .base_units_extracted(
            measurements
                .keys()
                .map(|k| k.as_str().to_string())
                .collect(),
        )
        .derivations_computed(
            schema
                .derivations
                .iter()
                .map(|d| d.name.as_str().to_string())
                .collect(),
        )
        .qualities_joined(qualities.keys().map(|k| k.as_str().to_string()).collect())
        .components_crushed(crushed_info)
        .row_count(total_rows)
        .subject_count(subject_count)
        .signal_policy_stats(signal_policy_stats)
        .build_duration(duration)
        .build()
}