rust_data_processing/ingestion/
xml.rs

1//! Row-oriented XML interchange (`<rdp_records><record>…</record></rdp_records>`).
2//!
3//! Field names in XML elements match [`Schema`] field names. Values are encoded as UTF-8 text.
4
5use std::collections::HashMap;
6use std::fs::File;
7use std::io::{BufReader, Write};
8use std::path::Path;
9
10use quick_xml::Writer;
11use quick_xml::escape::unescape;
12use quick_xml::events::{BytesDecl, BytesEnd, BytesStart, BytesText, Event};
13
14use crate::error::{IngestionError, IngestionResult};
15use crate::types::{DataSet, DataType, Schema, Value};
16
17const ROOT: &str = "rdp_records";
18const RECORD: &str = "record";
19
20fn xml_err(message: impl Into<String>) -> IngestionError {
21    IngestionError::SchemaMismatch {
22        message: message.into(),
23    }
24}
25
26fn value_to_text(v: &Value) -> String {
27    match v {
28        Value::Null => String::new(),
29        Value::Int64(i) => i.to_string(),
30        Value::Float64(f) => f.to_string(),
31        Value::Bool(b) => b.to_string(),
32        Value::Utf8(s) => s.clone(),
33    }
34}
35
36fn parse_text_to_value(raw: &str, dt: &DataType) -> IngestionResult<Value> {
37    let trimmed = raw.trim();
38    if trimmed.is_empty() {
39        return Ok(Value::Null);
40    }
41    match dt {
42        DataType::Int64 => trimmed
43            .parse::<i64>()
44            .map(Value::Int64)
45            .map_err(|_| xml_err(format!("invalid Int64 in XML: {trimmed:?}"))),
46        DataType::Float64 => trimmed
47            .parse::<f64>()
48            .map(Value::Float64)
49            .map_err(|_| xml_err(format!("invalid Float64 in XML: {trimmed:?}"))),
50        DataType::Bool => match trimmed.to_ascii_lowercase().as_str() {
51            "true" | "1" | "yes" => Ok(Value::Bool(true)),
52            "false" | "0" | "no" => Ok(Value::Bool(false)),
53            _ => Err(xml_err(format!("invalid Bool in XML: {trimmed:?}"))),
54        },
55        DataType::Utf8 => Ok(Value::Utf8(trimmed.to_string())),
56    }
57}
58
59fn row_from_field_map(
60    schema: &Schema,
61    fields: &HashMap<String, String>,
62) -> IngestionResult<Vec<Value>> {
63    let mut row = Vec::with_capacity(schema.fields.len());
64    for field in &schema.fields {
65        let raw = fields.get(&field.name).map(String::as_str).unwrap_or("");
66        row.push(parse_text_to_value(raw, &field.data_type)?);
67    }
68    Ok(row)
69}
70
71/// Write a [`DataSet`] to a single XML file (see module docs for layout).
72pub fn export_dataset_to_xml(path: &Path, ds: &DataSet) -> IngestionResult<()> {
73    let mut file = File::create(path)?;
74    let mut writer = Writer::new_with_indent(&mut file, b' ', 2);
75    writer
76        .write_event(Event::Decl(BytesDecl::new("1.0", Some("UTF-8"), None)))
77        .map_err(|e| xml_err(format!("write xml decl: {e}")))?;
78    writer
79        .write_event(Event::Start(BytesStart::new(ROOT)))
80        .map_err(|e| xml_err(format!("write xml root: {e}")))?;
81
82    for row in &ds.rows {
83        writer
84            .write_event(Event::Start(BytesStart::new(RECORD)))
85            .map_err(|e| xml_err(format!("write record start: {e}")))?;
86        for (field, cell) in ds.schema.fields.iter().zip(row.iter()) {
87            writer
88                .write_event(Event::Start(BytesStart::new(field.name.as_str())))
89                .map_err(|e| xml_err(format!("write field start: {e}")))?;
90            let text = value_to_text(cell);
91            if !text.is_empty() {
92                writer
93                    .write_event(Event::Text(BytesText::new(&text)))
94                    .map_err(|e| xml_err(format!("write field text: {e}")))?;
95            }
96            writer
97                .write_event(Event::End(BytesEnd::new(field.name.as_str())))
98                .map_err(|e| xml_err(format!("write field end: {e}")))?;
99        }
100        writer
101            .write_event(Event::End(BytesEnd::new(RECORD)))
102            .map_err(|e| xml_err(format!("write record end: {e}")))?;
103    }
104
105    writer
106        .write_event(Event::End(BytesEnd::new(ROOT)))
107        .map_err(|e| xml_err(format!("write xml root end: {e}")))?;
108    writer
109        .write_event(Event::Eof)
110        .map_err(|e| xml_err(format!("write xml eof: {e}")))?;
111    file.flush()?;
112    Ok(())
113}
114
115/// Ingest row-oriented XML into a [`DataSet`] using the provided [`Schema`].
116pub fn ingest_xml_from_path(path: impl AsRef<Path>, schema: &Schema) -> IngestionResult<DataSet> {
117    let file = File::open(path.as_ref())?;
118    let reader = BufReader::new(file);
119    let mut xml = quick_xml::Reader::from_reader(reader);
120    xml.config_mut().trim_text(true);
121
122    let mut rows: Vec<Vec<Value>> = Vec::new();
123    let mut in_record = false;
124    let mut field_values: HashMap<String, String> = HashMap::new();
125    let mut active_field: Option<String> = None;
126    let mut buf = Vec::new();
127
128    loop {
129        buf.clear();
130        match xml.read_event_into(&mut buf) {
131            Ok(Event::Start(e)) => {
132                let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
133                if name == RECORD {
134                    in_record = true;
135                    field_values.clear();
136                } else if in_record && name != ROOT {
137                    active_field = Some(name);
138                }
139            }
140            Ok(Event::Text(e)) => {
141                if let Some(ref field) = active_field {
142                    let decoded = e
143                        .xml10_content()
144                        .map_err(|err| xml_err(format!("xml text: {err}")))?;
145                    let text = unescape(&decoded)
146                        .map_err(|err| xml_err(format!("xml text: {err}")))?;
147                    field_values
148                        .entry(field.clone())
149                        .and_modify(|v| {
150                            v.push(' ');
151                            v.push_str(&text);
152                        })
153                        .or_insert_with(|| text.into_owned());
154                }
155            }
156            Ok(Event::End(e)) => {
157                let name = String::from_utf8_lossy(e.name().as_ref()).into_owned();
158                if name == RECORD && in_record {
159                    rows.push(row_from_field_map(schema, &field_values)?);
160                    in_record = false;
161                    active_field = None;
162                } else if active_field.as_deref() == Some(name.as_str()) {
163                    active_field = None;
164                }
165            }
166            Ok(Event::Eof) => break,
167            Err(e) => return Err(xml_err(format!("xml parse: {e}"))),
168            _ => {}
169        }
170    }
171
172    Ok(DataSet::new(schema.clone(), rows))
173}
174
175#[cfg(test)]
176mod tests {
177    use super::*;
178    use crate::types::Field;
179
180    fn sample_schema() -> Schema {
181        Schema::new(vec![
182            Field::new("stationCode", DataType::Utf8),
183            Field::new("lat", DataType::Float64),
184            Field::new("label", DataType::Utf8),
185        ])
186    }
187
188    #[test]
189    fn xml_roundtrip_preserves_rows() {
190        let schema = sample_schema();
191        let ds = DataSet::new(
192            schema.clone(),
193            vec![vec![
194                Value::Utf8("ACW00011604".into()),
195                Value::Float64(17.1167),
196                Value::Utf8("ST JOHNS".into()),
197            ]],
198        );
199        let dir = std::env::temp_dir().join(format!("rdp_xml_test_{}", std::process::id()));
200        std::fs::create_dir_all(&dir).unwrap();
201        let path = dir.join("stations.xml");
202        export_dataset_to_xml(&path, &ds).unwrap();
203        let back = ingest_xml_from_path(&path, &schema).unwrap();
204        assert_eq!(back.row_count(), 1);
205        assert_eq!(back.rows[0][0], Value::Utf8("ACW00011604".into()));
206        std::fs::remove_dir_all(dir).ok();
207    }
208
209    #[test]
210    fn xml_ingest_reads_committed_fixture_shape() {
211        let manifest = Path::new(env!("CARGO_MANIFEST_DIR"));
212        let path = manifest.join("tests/fixtures/ghcn/ghcn_stations_sample.json");
213        if !path.exists() {
214            return;
215        }
216        let schema = Schema::new(vec![
217            Field::new("id", DataType::Utf8),
218            Field::new("latitude", DataType::Float64),
219            Field::new("longitude", DataType::Float64),
220            Field::new("elevation", DataType::Float64),
221            Field::new("name", DataType::Utf8),
222            Field::new("state", DataType::Utf8),
223        ]);
224        let json_ds = crate::ingestion::json::ingest_json_from_path(&path, &schema).unwrap();
225        let xml_path = manifest.join("tests/fixtures/ghcn/_tmp_roundtrip.xml");
226        export_dataset_to_xml(&xml_path, &json_ds).unwrap();
227        let xml_schema = Schema::new(vec![
228            Field::new("stationCode", DataType::Utf8),
229            Field::new("lat", DataType::Float64),
230            Field::new("lon", DataType::Float64),
231            Field::new("elev_m", DataType::Float64),
232            Field::new("label", DataType::Utf8),
233            Field::new("region", DataType::Utf8),
234        ]);
235        let xml_ds = DataSet::new(
236            xml_schema.clone(),
237            json_ds
238                .rows
239                .iter()
240                .map(|row| {
241                    vec![
242                        row[0].clone(),
243                        row[1].clone(),
244                        row[2].clone(),
245                        row[3].clone(),
246                        row[4].clone(),
247                        row[5].clone(),
248                    ]
249                })
250                .collect(),
251        );
252        export_dataset_to_xml(&xml_path, &xml_ds).unwrap();
253        let read = ingest_xml_from_path(&xml_path, &xml_schema).unwrap();
254        assert_eq!(read.row_count(), json_ds.row_count());
255        std::fs::remove_file(xml_path).ok();
256    }
257}
rust_data_processing/ingestion/xml.rs

rust_data_processing/ingestion/
xml.rs
