qualia_core_db/sparql_library/parsers/
csv_parser.rs

1//! CSV Parser for QualiaDB
2//!
3//! Zero-allocation CSV parser that streams data directly into NQuin format.
4//! Supports common data types with proper inline type tagging.
5
6use atoi::atoi;
7use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
8use csv::ReaderBuilder;
9use std::io::Read;
10
11use crate::mini_parser::hash_token;
12use crate::NQuin;
13
14/// Data type for CSV field mapping
15#[derive(Debug, Clone, Copy)]
16pub enum CsvDatatype {
17    Integer,
18    Float,
19    DateTime,
20    StringRef,
21}
22
23/// Column mapping configuration
24#[derive(Debug, Clone)]
25pub struct CsvColumnMapping {
26    pub source_key: String,
27    pub column_index: Option<usize>,
28    pub predicate_hash: u64,
29    pub datatype: CsvDatatype,
30}
31
32/// CSV parsing configuration
33#[derive(Debug, Clone)]
34pub struct CsvMappingProfile {
35    pub base_class_hash: u64,
36    pub fields: Vec<CsvColumnMapping>,
37}
38
39/// Parse CSV data from a reader and stream Quins
40pub fn parse_csv_to_quins<R: Read>(
41    reader: R,
42    profile: &mut CsvMappingProfile,
43    mut on_quin: impl FnMut(NQuin),
44) -> Result<(), String> {
45    let mut rdr = ReaderBuilder::new().has_headers(true).from_reader(reader);
46
47    // Resolve header indices
48    let headers = rdr
49        .byte_headers()
50        .map_err(|e| format!("Failed to read headers: {}", e))?
51        .clone();
52    for field in profile.fields.iter_mut() {
53        field.column_index = headers
54            .iter()
55            .position(|h| h == field.source_key.as_bytes());
56    }
57
58    let mut record = csv::ByteRecord::new();
59
60    // Zero-allocation stream loop
61    while rdr
62        .read_byte_record(&mut record)
63        .map_err(|e| format!("CSV read error: {}", e))?
64    {
65        let subject_hash: u64 = rand::random(); // Ephemeral Subject ID for the row
66
67        for field in &profile.fields {
68            if let Some(idx) = field.column_index {
69                if let Some(raw_bytes) = record.get(idx) {
70                    // Pack into Quin without String allocation
71                    match field.datatype {
72                        CsvDatatype::Integer => {
73                            let val: u64 = atoi::<u64>(raw_bytes).unwrap_or(0);
74                            let quin = NQuin {
75                                subject: subject_hash,
76                                predicate: field.predicate_hash,
77                                object: val | (0b001 << 60), // INLINE_TAG_INTEGER
78                                context: 0,
79                                metadata: 0,
80                                parity: NQuin::calculate_parity(
81                                    subject_hash,
82                                    field.predicate_hash,
83                                    val | (0b001 << 60),
84                                    0,
85                                    0,
86                                ),
87                            };
88                            on_quin(quin);
89                        }
90                        CsvDatatype::Float => {
91                            let str_slice = std::str::from_utf8(raw_bytes).unwrap_or("0.0");
92                            let float_val: f32 = str_slice.parse::<f32>().unwrap_or(0.0);
93                            let float_bits: u32 = float_val.to_bits();
94                            let inline_tag: u64 = 0b010 << 60;
95                            let packed_object: u64 = inline_tag | (float_bits as u64);
96
97                            let quin = NQuin {
98                                subject: subject_hash,
99                                predicate: field.predicate_hash,
100                                object: packed_object,
101                                context: 0,
102                                metadata: 0,
103                                parity: NQuin::calculate_parity(
104                                    subject_hash,
105                                    field.predicate_hash,
106                                    packed_object,
107                                    0,
108                                    0,
109                                ),
110                            };
111                            on_quin(quin);
112                        }
113                        CsvDatatype::StringRef => {
114                            let s = std::str::from_utf8(raw_bytes).unwrap_or("");
115                            let quin = NQuin {
116                                subject: subject_hash,
117                                predicate: field.predicate_hash,
118                                object: hash_token(s),
119                                context: 0,
120                                metadata: 0,
121                                parity: NQuin::calculate_parity(
122                                    subject_hash,
123                                    field.predicate_hash,
124                                    hash_token(s),
125                                    0,
126                                    0,
127                                ),
128                            };
129                            on_quin(quin);
130                        }
131                        CsvDatatype::DateTime => {
132                            let s = std::str::from_utf8(raw_bytes).unwrap_or("");
133                            let millis: u64 = parse_datetime_millis(s).unwrap_or(0);
134                            let quin = NQuin {
135                                subject: subject_hash,
136                                predicate: field.predicate_hash,
137                                object: (0b011u64 << 60) | millis,
138                                context: 0,
139                                metadata: 0,
140                                parity: NQuin::calculate_parity(
141                                    subject_hash,
142                                    field.predicate_hash,
143                                    (0b011u64 << 60) | millis,
144                                    0,
145                                    0,
146                                ),
147                            };
148                            on_quin(quin);
149                        }
150                    }
151                }
152            }
153        }
154    }
155
156    Ok(())
157}
158
159/// Try parsing common datetime string formats into Unix milliseconds.
160fn parse_datetime_millis(s: &str) -> Option<u64> {
161    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
162        return Some(dt.timestamp_millis() as u64);
163    }
164    for fmt in &[
165        "%Y-%m-%dT%H:%M:%S",
166        "%Y-%m-%d %H:%M:%S",
167        "%Y-%m-%dT%H:%M:%SZ",
168    ] {
169        if let Ok(nd) = NaiveDateTime::parse_from_str(s, fmt) {
170            return Some(Utc.from_utc_datetime(&nd).timestamp_millis() as u64);
171        }
172    }
173    if let Ok(d) = NaiveDate::parse_from_str(s, "%Y-%m-%d") {
174        let nd = d.and_hms_opt(0, 0, 0)?;
175        return Some(Utc.from_utc_datetime(&nd).timestamp_millis() as u64);
176    }
177    None
178}
qualia_core_db/sparql_library/parsers/csv_parser.rs

qualia_core_db/sparql_library/parsers/
csv_parser.rs
