wp-connector-utils 0.1.2

Shared utilities for WP connector implementations — Arrow IPC encode/decode, WireFormat, NDJSON conversion, and batch metadata processing
Documentation 
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
120
121
122
123
124
125
126
127
128
129
130
131

//! Arrow schema inference from `DataRecord` fields.

use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
use wp_model_core::model::DataRecord;

// ---------------------------------------------------------------------------
// Schema inference (migrated from arrow_conv/schema.rs)
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// Schema inference
// ---------------------------------------------------------------------------

/// Infer an Arrow schema from field names.
///
/// All fields default to `Utf8` (conservative).
pub fn infer_arrow_schema(fields: &[String]) -> Schema {
    Schema::new(
        fields
            .iter()
            .map(|f| Field::new(f.as_str(), DataType::Utf8, true))
            .collect::<Vec<_>>(),
    )
}

/// Map wp_model_core DataType to Arrow DataType.
fn wp_type_to_arrow(dt: &wp_model_core::model::DataType) -> DataType {
    use wp_model_core::model::DataType as WpDt;
    match dt {
        WpDt::Bool => DataType::Boolean,
        WpDt::Digit => DataType::Int64,
        WpDt::Float => DataType::Float64,
        WpDt::Port => DataType::Int32,
        WpDt::Time
        | WpDt::TimeISO
        | WpDt::TimeRFC3339
        | WpDt::TimeRFC2822
        | WpDt::TimeTIMESTAMP
        | WpDt::TimeCLF => DataType::Timestamp(TimeUnit::Nanosecond, None),
        WpDt::Hex | WpDt::Base64 => DataType::Binary,
        WpDt::Chars
        | WpDt::Symbol
        | WpDt::PeekSymbol
        | WpDt::IP
        | WpDt::IpNet
        | WpDt::Domain
        | WpDt::Email
        | WpDt::Url
        | WpDt::SN
        | WpDt::IdCard
        | WpDt::MobilePhone
        | WpDt::KV
        | WpDt::KvArr
        | WpDt::Json
        | WpDt::ExactJson
        | WpDt::HttpRequest
        | WpDt::HttpStatus
        | WpDt::HttpAgent
        | WpDt::HttpMethod
        | WpDt::Auto
        | WpDt::ProtoText
        | WpDt::Obj
        | WpDt::Ignore => DataType::Utf8,
        WpDt::Array(_) => DataType::Utf8,
    }
}

/// Infer an Arrow schema from a DataRecord using actual field types (get_meta()).
///
/// Fields with `DataType::Ignore` are excluded from the schema.
pub fn infer_schema_from_record(record: &DataRecord) -> Schema {
    Schema::new(
        record
            .items
            .iter()
            .filter(|f| !matches!(f.get_meta(), wp_model_core::model::DataType::Ignore))
            .map(|f| {
                let arrow_type = wp_type_to_arrow(f.get_meta());
                Field::new(f.get_name(), arrow_type, true)
            })
            .collect::<Vec<_>>(),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use arrow::datatypes::DataType;
    use wp_model_core::model::{DataRecord, Field as ModelField, FieldStorage};

    #[test]
    fn schema_inferred_from_record() {
        let rec = DataRecord::from(vec![
            FieldStorage::from(ModelField::from_chars("name", "a")),
            FieldStorage::from(ModelField::from_digit("count", 1)),
        ]);
        let schema = infer_schema_from_record(&rec);
        assert_eq!(schema.fields().len(), 2);
        assert_eq!(schema.field(0).name(), "name");
        assert_eq!(schema.field(1).name(), "count");
    }

    #[test]
    fn ignore_field_excluded() {
        let rec = DataRecord::from(vec![
            FieldStorage::from(ModelField::from_chars("name", "a")),
            FieldStorage::from(ModelField::from_ignore("junk")),
        ]);
        let schema = infer_schema_from_record(&rec);
        assert_eq!(schema.fields().len(), 1);
        assert_eq!(schema.field(0).name(), "name");
    }

    #[test]
    fn bool_maps_to_boolean() {
        let dt = wp_type_to_arrow(&wp_model_core::model::DataType::Bool);
        assert_eq!(dt, DataType::Boolean);
    }

    #[test]
    fn digit_maps_to_int64() {
        let dt = wp_type_to_arrow(&wp_model_core::model::DataType::Digit);
        assert_eq!(dt, DataType::Int64);
    }

    #[test]
    fn hex_maps_to_binary() {
        let dt = wp_type_to_arrow(&wp_model_core::model::DataType::Hex);
        assert_eq!(dt, DataType::Binary);
    }
}