apif_source_row/
definition.rs

1use crate::detect::SourceFormat;
2use crate::filter::FilterCondition;
3use serde::Deserialize;
4use std::str::FromStr;
5
6#[derive(Debug, Clone, Deserialize)]
7pub struct SourceDefinition {
8    pub file: String,
9    #[serde(default)]
10    pub name: Option<String>,
11    #[serde(default, deserialize_with = "deserialize_format_opt")]
12    pub format: Option<SourceFormat>,
13    #[serde(default)]
14    pub delimiter: Option<u8>,
15    #[serde(default)]
16    pub header: Option<bool>,
17    #[serde(default)]
18    pub indexed_by: Option<Vec<String>>,
19    #[serde(default)]
20    pub index_mode: Option<IndexMode>,
21    #[serde(default)]
22    pub memory_budget: Option<String>,
23    #[serde(default)]
24    pub filter: Option<Vec<FilterCondition>>,
25    #[serde(default)]
26    pub join_type: Option<JoinType>,
27}
28
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
30#[serde(rename_all = "snake_case")]
31pub enum JoinType {
32    Inner,
33    Left,
34    Cross,
35}
36
37#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
38#[serde(rename_all = "snake_case")]
39pub enum IndexMode {
40    #[default]
41    OnDemand,
42    BuildOnce,
43    Memory,
44}
45
46fn deserialize_format_opt<'de, D>(de: D) -> Result<Option<SourceFormat>, D::Error>
47where
48    D: serde::Deserializer<'de>,
49{
50    let s: Option<String> = Option::deserialize(de)?;
51    match s {
52        None => Ok(None),
53        Some(ref val) => SourceFormat::from_str(val)
54            .map(Some)
55            .map_err(serde::de::Error::custom),
56    }
57}
58
59impl SourceDefinition {
60    pub fn from_file_raw(file: &str, key_column: &str, format: Option<&SourceFormat>) -> Self {
61        Self {
62            file: file.to_string(),
63            name: None,
64            format: format.cloned(),
65            delimiter: None,
66            header: None,
67            indexed_by: Some(vec![key_column.to_string()]),
68            index_mode: Some(IndexMode::BuildOnce),
69            memory_budget: None,
70            filter: None,
71            join_type: None,
72        }
73    }
74
75    pub fn join_type_or_default(&self) -> JoinType {
76        self.join_type.unwrap_or(JoinType::Left)
77    }
78
79    pub fn effective_index_mode(&self) -> IndexMode {
80        self.index_mode.unwrap_or_default()
81    }
82
83    pub fn indexed_columns(&self) -> Vec<&str> {
84        match &self.indexed_by {
85            None => Vec::new(),
86            Some(v) => v.iter().map(|s| s.as_str()).collect(),
87        }
88    }
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94
95    #[test]
96    fn deserialize_minimal() {
97        let def: SourceDefinition = serde_yaml_ng::from_str("file: data/users.csv").unwrap();
98        assert_eq!(def.file, "data/users.csv");
99        assert!(def.name.is_none());
100        assert!(def.format.is_none());
101        assert!(def.indexed_by.is_none());
102    }
103
104    #[test]
105    fn deserialize_full() {
106        let yaml = "\
107file: data/pvz.csv
108name: pvz
109format: csv
110indexed_by: [pvz_id]
111index_mode: build_once
112memory_budget: 256mb
113";
114        let def: SourceDefinition = serde_yaml_ng::from_str(yaml).unwrap();
115        assert_eq!(def.file, "data/pvz.csv");
116        assert_eq!(def.name.as_deref(), Some("pvz"));
117        assert_eq!(def.format, Some(SourceFormat::Csv));
118        assert_eq!(def.effective_index_mode(), IndexMode::BuildOnce);
119        assert_eq!(def.indexed_columns(), vec!["pvz_id"]);
120        assert_eq!(def.memory_budget.as_deref(), Some("256mb"));
121    }
122
123    #[test]
124    fn deserialize_multi_indexed_by() {
125        let yaml = "\
126file: data/pvz.csv
127indexed_by:
128  - pvz_id
129  - region_id
130";
131        let def: SourceDefinition = serde_yaml_ng::from_str(yaml).unwrap();
132        assert_eq!(def.indexed_columns(), vec!["pvz_id", "region_id"]);
133    }
134
135    #[test]
136    fn index_mode_default_is_on_demand() {
137        let def: SourceDefinition = serde_yaml_ng::from_str("file: x.csv").unwrap();
138        assert_eq!(def.effective_index_mode(), IndexMode::OnDemand);
139    }
140
141    #[test]
142    fn index_mode_all_variants() {
143        assert_eq!(IndexMode::default(), IndexMode::OnDemand);
144        assert_ne!(IndexMode::OnDemand, IndexMode::BuildOnce);
145        assert_ne!(IndexMode::BuildOnce, IndexMode::Memory);
146    }
147
148    #[test]
149    fn deserialize_format_aliases() {
150        for fmt in &["csv", "tsv", "ndjson", "json", "jsonl"] {
151            let yaml = format!("file: x\nformat: {fmt}");
152            let def: SourceDefinition = serde_yaml_ng::from_str(&yaml).unwrap();
153            assert!(def.format.is_some());
154        }
155    }
156
157    #[test]
158    fn deserialize_invalid_format_errors() {
159        let yaml = "file: x\nformat: excel";
160        let result: Result<SourceDefinition, _> = serde_yaml_ng::from_str(yaml);
161        assert!(result.is_err());
162    }
163
164    #[test]
165    fn deserialize_filter_conditions() {
166        let yaml = "\
167file: data/pvz.csv
168filter:
169  - field: status
170    in: [active, suspended]
171  - field: created_at
172    gte: \"2024-01-01\"
173";
174        let def: SourceDefinition = serde_yaml_ng::from_str(yaml).unwrap();
175        let filter = def.filter.expect("filter should exist");
176        assert_eq!(filter.len(), 2);
177        assert_eq!(filter[0].field, "status");
178        assert_eq!(filter[0].in_values.as_ref().map(Vec::len), Some(2));
179        assert_eq!(filter[1].field, "created_at");
180        assert_eq!(filter[1].gte.as_deref(), Some("2024-01-01"));
181    }
182}
apif_source_row/definition.rs

apif_source_row/
definition.rs
