hawdb_executor/
load_csv.rs

1// Copyright 2026 Nowledge
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! `LOAD CSV` file access policy and record-to-row conversion.
16//!
17//! Files are read only beneath an explicitly configured import directory.
18//! Records are streamed through a fixed-size buffer; no plan or cache retains
19//! a resolved location or any file contents.
20
21mod reader;
22#[cfg(test)]
23mod tests;
24
25pub use reader::{CsvField, CsvRecordReader, CSV_READ_BUFFER_BYTES};
26
27use hawdb_core::{HawDBError, Result, Value};
28use hawdb_plan_cypher::{CsvLocation, LoadCsvSource};
29use std::collections::BTreeMap;
30use std::fs::File;
31use std::io::Read;
32use std::path::{Path, PathBuf};
33
34/// Resource bounds for one `LOAD CSV` source.
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub struct CsvImportLimits {
37    /// Maximum bytes read from one file, checked before opening against the
38    /// file length and again while streaming.
39    pub max_file_bytes: u64,
40    /// Maximum field bytes in one record, excluding quotes and separators.
41    pub max_record_bytes: usize,
42    /// Maximum fields in one record, including the header record.
43    pub max_fields_per_record: usize,
44}
45
46impl Default for CsvImportLimits {
47    fn default() -> Self {
48        Self {
49            max_file_bytes: 4 * 1024 * 1024 * 1024,
50            max_record_bytes: 4 * 1024 * 1024,
51            max_fields_per_record: 4096,
52        }
53    }
54}
55
56/// The directory `LOAD CSV` may read from, and its per-file limits.
57///
58/// Without a policy, `LOAD CSV` is refused. With one, the canonical path of
59/// every source, after resolving `..` components and symbolic links, must be
60/// a regular file inside the canonical import root.
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub struct CsvImportPolicy {
63    root: PathBuf,
64    limits: CsvImportLimits,
65}
66
67impl CsvImportPolicy {
68    pub fn new(root: impl Into<PathBuf>, limits: CsvImportLimits) -> Self {
69        Self {
70            root: root.into(),
71            limits,
72        }
73    }
74
75    pub fn root(&self) -> &Path {
76        &self.root
77    }
78
79    pub fn limits(&self) -> CsvImportLimits {
80        self.limits
81    }
82
83    /// Resolves `location` to a canonical file path inside the import root.
84    ///
85    /// Errors that depend only on the statement are `Semantic`: a malformed
86    /// or non-local location and a location outside the import directory
87    /// (after resolving symbolic links, or lexically when the target does not
88    /// exist). Filesystem conditions observed at run time are `Execution`: an
89    /// import root or source that cannot be accessed.
90    pub fn resolve(&self, location: &str) -> Result<PathBuf> {
91        let root = std::fs::canonicalize(&self.root).map_err(|error| {
92            HawDBError::Execution(format!(
93                "LOAD CSV import root '{}' cannot be resolved: {error}",
94                self.root.display()
95            ))
96        })?;
97        if !root.is_dir() {
98            return Err(HawDBError::Execution(format!(
99                "LOAD CSV import root '{}' is not a directory",
100                root.display()
101            )));
102        }
103        let path = local_path(location)?;
104        let path = if path.is_absolute() {
105            path
106        } else {
107            root.join(path)
108        };
109        let canonical = std::fs::canonicalize(&path).map_err(|error| {
110            if lexically_outside(&path, &root, &self.root) {
111                outside_root(location)
112            } else {
113                HawDBError::Execution(format!("LOAD CSV cannot access '{location}': {error}"))
114            }
115        })?;
116        if !canonical.starts_with(&root) {
117            return Err(outside_root(location));
118        }
119        Ok(canonical)
120    }
121
122    /// Opens `source` for streaming after resolving and validating its location.
123    ///
124    /// The source must be a regular file before it is opened: opening a FIFO
125    /// or device could block or never end. The opened handle is checked again
126    /// for a file replaced between the two system calls.
127    pub fn open(
128        &self,
129        source: &LoadCsvSource,
130        parameters: &BTreeMap<String, Value>,
131    ) -> Result<CsvRows<File>> {
132        let location = source_location(source, parameters)?;
133        let path = self.resolve(location)?;
134        let inspect = |metadata: std::io::Result<std::fs::Metadata>| {
135            let metadata = metadata.map_err(|error| {
136                HawDBError::Execution(format!("LOAD CSV cannot inspect '{location}': {error}"))
137            })?;
138            if !metadata.is_file() {
139                return Err(HawDBError::Execution(format!(
140                    "LOAD CSV source '{location}' is not a regular file"
141                )));
142            }
143            if metadata.len() > self.limits.max_file_bytes {
144                return Err(HawDBError::Execution(format!(
145                    "LOAD CSV source '{location}' has {} bytes, exceeding max_file_bytes {}",
146                    metadata.len(),
147                    self.limits.max_file_bytes
148                )));
149            }
150            Ok(())
151        };
152        inspect(std::fs::metadata(&path))?;
153        let file = File::open(&path).map_err(|error| {
154            HawDBError::Execution(format!("LOAD CSV cannot open '{location}': {error}"))
155        })?;
156        inspect(file.metadata())?;
157        CsvRows::new(file, source, self.limits)
158    }
159}
160
161fn outside_root(location: &str) -> HawDBError {
162    HawDBError::Semantic(format!(
163        "LOAD CSV source '{location}' is outside the configured import directory"
164    ))
165}
166
167/// Whether `path`, with `.` and `..` applied lexically, lies outside both the
168/// canonical and the configured import root. Used only for a target that
169/// cannot be canonicalized, so a missing file outside the root reports the
170/// policy violation rather than its absence.
171fn lexically_outside(path: &Path, canonical_root: &Path, configured_root: &Path) -> bool {
172    let normalize = |path: &Path| {
173        let mut normalized = PathBuf::new();
174        for component in path.components() {
175            match component {
176                std::path::Component::CurDir => {}
177                std::path::Component::ParentDir => {
178                    normalized.pop();
179                }
180                component => normalized.push(component),
181            }
182        }
183        normalized
184    };
185    let path = normalize(path);
186    let configured = std::path::absolute(configured_root)
187        .map(|root| normalize(&root))
188        .ok();
189    !path.starts_with(canonical_root)
190        && configured.is_none_or(|configured| !path.starts_with(configured))
191}
192
193/// Opens a `LOAD CSV` source, refusing when no import policy is configured.
194pub fn open_load_csv(
195    policy: Option<&CsvImportPolicy>,
196    source: &LoadCsvSource,
197    parameters: &BTreeMap<String, Value>,
198) -> Result<CsvRows<File>> {
199    policy
200        .ok_or_else(|| {
201            HawDBError::Semantic(
202                "LOAD CSV is disabled: set DatabaseConfig::csv_import_root to an import directory"
203                    .to_string(),
204            )
205        })?
206        .open(source, parameters)
207}
208
209fn source_location<'a>(
210    source: &'a LoadCsvSource,
211    parameters: &'a BTreeMap<String, Value>,
212) -> Result<&'a str> {
213    match &source.location {
214        CsvLocation::Literal(location) => Ok(location),
215        CsvLocation::Parameter(name) => match parameters.get(name) {
216            Some(Value::String(location)) => Ok(location),
217            Some(_) => Err(HawDBError::Semantic(format!(
218                "LOAD CSV source parameter '${name}' must be a string"
219            ))),
220            None => Err(HawDBError::Semantic(format!("missing parameter '${name}'"))),
221        },
222    }
223}
224
225/// Converts a `file:` URL or plain path to a filesystem path. `file:` URL
226/// paths are relative to the import directory, as in Neo4j. Every other URL
227/// scheme, including `http` and `https`, is refused.
228fn local_path(location: &str) -> Result<PathBuf> {
229    if location.is_empty() || location.contains('\0') {
230        return Err(HawDBError::Semantic(
231            "LOAD CSV source must be a non-empty file path".to_string(),
232        ));
233    }
234    let Some(scheme) = url_scheme(location) else {
235        return Ok(PathBuf::from(location));
236    };
237    if !scheme.eq_ignore_ascii_case("file") {
238        return Err(HawDBError::Semantic(format!(
239            "LOAD CSV supports only local files; '{scheme}' URLs are not supported"
240        )));
241    }
242    let rest = &location[scheme.len() + 1..];
243    let path = if let Some(authority_path) = rest.strip_prefix("//") {
244        let (host, path) = authority_path
245            .find('/')
246            .map_or((authority_path, ""), |index| authority_path.split_at(index));
247        if !host.is_empty() && !host.eq_ignore_ascii_case("localhost") {
248            return Err(HawDBError::Semantic(format!(
249                "LOAD CSV file URL host '{host}' is not supported; use file:///path relative to the import directory"
250            )));
251        }
252        path
253    } else {
254        rest
255    };
256    if path.contains(['?', '#']) {
257        return Err(HawDBError::Semantic(
258            "LOAD CSV file URLs must not contain a query or fragment".to_string(),
259        ));
260    }
261    if !path.starts_with('/') {
262        return Err(HawDBError::Semantic(
263            "LOAD CSV file URLs must have the form file:///path".to_string(),
264        ));
265    }
266    let decoded = percent_decode(path)?;
267    // Like Neo4j, `file:///x.csv` names `x.csv` inside the import directory;
268    // the URL path is never an absolute filesystem path.
269    Ok(PathBuf::from(decoded.trim_start_matches('/')))
270}
271
272/// Returns the URL scheme when `location` starts with `scheme:`. A single
273/// letter is a Windows drive, not a scheme.
274fn url_scheme(location: &str) -> Option<&str> {
275    let (scheme, _) = location.split_once(':')?;
276    let mut bytes = scheme.bytes();
277    let first = bytes.next()?;
278    (first.is_ascii_alphabetic()
279        && scheme.len() > 1
280        && bytes.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'-' | b'.')))
281    .then_some(scheme)
282}
283
284fn percent_decode(text: &str) -> Result<String> {
285    let bytes = text.as_bytes();
286    let mut decoded = Vec::with_capacity(bytes.len());
287    let mut index = 0;
288    while index < bytes.len() {
289        if bytes[index] == b'%' {
290            let hex = text
291                .get(index + 1..index + 3)
292                .and_then(|hex| u8::from_str_radix(hex, 16).ok())
293                .ok_or_else(|| {
294                    HawDBError::Semantic("LOAD CSV file URL has an invalid percent escape".into())
295                })?;
296            if hex == 0 {
297                return Err(HawDBError::Semantic(
298                    "LOAD CSV file URL must not contain NUL".to_string(),
299                ));
300            }
301            decoded.push(hex);
302            index += 3;
303        } else {
304            decoded.push(bytes[index]);
305            index += 1;
306        }
307    }
308    String::from_utf8(decoded)
309        .map_err(|_| HawDBError::Semantic("LOAD CSV file URL is not valid UTF-8".to_string()))
310}
311
312/// Streams the rows `LOAD CSV` binds to its variable.
313///
314/// Without headers a row is a list of strings. With headers it is a map from
315/// each header to the field in the same position. An empty unquoted field
316/// and a field missing from a short record are `NULL`; a quoted empty field
317/// is the empty string. Fields beyond the header count are ignored.
318pub struct CsvRows<R> {
319    reader: CsvRecordReader<R>,
320    headers: Option<Vec<String>>,
321    with_headers: bool,
322}
323
324impl<R: Read> CsvRows<R> {
325    pub fn new(input: R, source: &LoadCsvSource, limits: CsvImportLimits) -> Result<Self> {
326        Ok(Self {
327            reader: CsvRecordReader::new(input, source.field_terminator, limits)?,
328            headers: None,
329            with_headers: source.with_headers,
330        })
331    }
332
333    pub fn next_row(&mut self) -> Result<Option<Value>> {
334        if self.with_headers && self.headers.is_none() {
335            let Some(header) = self.reader.next_record()? else {
336                return Ok(None);
337            };
338            let mut seen = std::collections::BTreeSet::new();
339            let headers = header
340                .into_iter()
341                .map(|name| {
342                    let name = name.unwrap_or_default();
343                    if !seen.insert(name.clone()) {
344                        return Err(HawDBError::Execution(format!(
345                            "LOAD CSV header repeats column '{name}'"
346                        )));
347                    }
348                    Ok(name)
349                })
350                .collect::<Result<Vec<_>>>()?;
351            self.headers = Some(headers);
352        }
353        let Some(record) = self.reader.next_record()? else {
354            return Ok(None);
355        };
356        Ok(Some(match &self.headers {
357            None => Value::List(record.into_iter().map(field_value).collect()),
358            Some(headers) => {
359                let mut fields = record.into_iter();
360                Value::Map(
361                    headers
362                        .iter()
363                        .map(|name| (name.clone(), fields.next().map_or(Value::Null, field_value)))
364                        .collect(),
365                )
366            }
367        }))
368    }
369
370    /// Data records returned so far, excluding a header record.
371    pub fn rows_read(&self) -> u64 {
372        self.reader
373            .records()
374            .saturating_sub(u64::from(self.headers.is_some()))
375    }
376}
377
378fn field_value(field: CsvField) -> Value {
379    field.map_or(Value::Null, Value::String)
380}
hawdb_executor/load_csv.rs

hawdb_executor/
load_csv.rs
