laminar_connectors/files/
config.rs

1//! Configuration types for the file source and sink connectors.
2
3use std::collections::HashMap;
4use std::time::Duration;
5
6use crate::config::ConnectorConfig;
7use crate::error::ConnectorError;
8use laminar_core::time::parse_duration_str;
9
10/// Parsed configuration for [`super::source::FileSource`].
11#[derive(Debug, Clone)]
12pub struct FileSourceConfig {
13    /// Directory path, glob pattern, or cloud URL.
14    pub path: String,
15
16    /// Data format (`csv`, `json`, `text`, `parquet`). `None` = auto-detect.
17    pub format: Option<FileFormat>,
18
19    /// Discovery polling interval.
20    pub poll_interval: Duration,
21
22    /// Wait after last modify event before considering a file complete.
23    pub stabilisation_delay: Duration,
24
25    /// Maximum files to process per `poll_batch` call.
26    pub max_files_per_poll: usize,
27
28    /// Whether to append a `_metadata` struct column.
29    pub include_metadata: bool,
30
31    /// Safety limit for reading a single file (bytes). Primarily for Parquet.
32    pub max_file_bytes: usize,
33
34    /// Additional glob pattern to filter discovered file names.
35    pub glob_pattern: Option<String>,
36
37    /// CSV-specific: field delimiter.
38    pub csv_delimiter: u8,
39
40    /// CSV-specific: whether the first row is a header.
41    pub csv_has_header: bool,
42}
43
44impl FileSourceConfig {
45    /// Parse from connector properties.
46    ///
47    /// # Errors
48    ///
49    /// Returns `ConnectorError::ConfigurationError` if required options are
50    /// missing or values cannot be parsed.
51    pub fn from_connector_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
52        let props = config.properties();
53
54        let path = props
55            .get("path")
56            .cloned()
57            .ok_or_else(|| ConnectorError::ConfigurationError("'path' is required".into()))?;
58
59        let format = props
60            .get("format")
61            .map(|s| FileFormat::parse(s))
62            .transpose()?;
63
64        let poll_interval = parse_duration(props, "poll_interval", Duration::from_secs(10))?;
65        let stabilisation_delay =
66            parse_duration(props, "stabilisation_delay", Duration::from_secs(1))?;
67        let max_files_per_poll = parse_usize(props, "max_files_per_poll", 100)?;
68        if max_files_per_poll == 0 {
69            return Err(ConnectorError::ConfigurationError(
70                "'max_files_per_poll' must be greater than zero".into(),
71            ));
72        }
73        let include_metadata = parse_bool(props, "include_metadata", false)?;
74        for removed in [
75            "allow_overwrites",
76            "manifest_retention_count",
77            "manifest_retention_age_days",
78        ] {
79            if props.contains_key(removed) {
80                return Err(ConnectorError::ConfigurationError(format!(
81                    "file source option '{removed}' was removed; processed paths are immutable and retained in one exact checkpoint inventory"
82                )));
83            }
84        }
85        let max_file_bytes = parse_usize(props, "max_file_bytes", 256 * 1024 * 1024)?;
86        if max_file_bytes == 0 {
87            return Err(ConnectorError::ConfigurationError(
88                "'max_file_bytes' must be greater than zero".into(),
89            ));
90        }
91        let glob_pattern = props.get("glob_pattern").cloned();
92
93        // Default delimiter: tab for tsv, comma otherwise.
94        let is_tsv = props
95            .get("format")
96            .is_some_and(|f| f.eq_ignore_ascii_case("tsv"));
97        let csv_delimiter = props
98            .get("csv.delimiter")
99            .and_then(|s| s.as_bytes().first().copied())
100            .unwrap_or(if is_tsv { b'\t' } else { b',' });
101        let csv_has_header = parse_bool(props, "csv.has_header", true)?;
102
103        Ok(Self {
104            path,
105            format,
106            poll_interval,
107            stabilisation_delay,
108            max_files_per_poll,
109            include_metadata,
110            max_file_bytes,
111            glob_pattern,
112            csv_delimiter,
113            csv_has_header,
114        })
115    }
116}
117
118/// Parsed configuration for [`super::sink::FileSink`].
119#[derive(Debug, Clone)]
120pub struct FileSinkConfig {
121    /// Output directory path.
122    pub path: String,
123
124    /// Output format.
125    pub format: FileFormat,
126
127    /// File name prefix for immutable output files.
128    pub prefix: String,
129
130    /// Maximum file size before rotation (row formats only).
131    pub max_file_size: Option<usize>,
132
133    /// Parquet compression codec.
134    pub compression: String,
135
136    /// Maximum batches to buffer between bulk-format flushes (default: 10,000).
137    pub max_buffered_batches: usize,
138}
139
140impl FileSinkConfig {
141    /// Parse from connector properties.
142    ///
143    /// # Errors
144    ///
145    /// Returns `ConnectorError::ConfigurationError` if required options are
146    /// missing or values cannot be parsed.
147    pub fn from_connector_config(config: &ConnectorConfig) -> Result<Self, ConnectorError> {
148        let props = config.properties();
149
150        let path = props
151            .get("path")
152            .cloned()
153            .ok_or_else(|| ConnectorError::ConfigurationError("'path' is required".into()))?;
154
155        let format = props
156            .get("format")
157            .ok_or_else(|| {
158                ConnectorError::ConfigurationError("'format' is required for file sink".into())
159            })
160            .and_then(|s| FileFormat::parse(s))?;
161
162        if props.contains_key("mode") {
163            return Err(ConnectorError::ConfigurationError(
164                "file sink option 'mode' was removed; files are always published as immutable rolling files"
165                    .into(),
166            ));
167        }
168
169        let prefix = props
170            .get("prefix")
171            .cloned()
172            .unwrap_or_else(|| "part".to_string());
173        if prefix.is_empty()
174            || prefix == "."
175            || prefix == ".."
176            || prefix.contains('/')
177            || prefix.contains('\\')
178        {
179            return Err(ConnectorError::ConfigurationError(
180                "file sink 'prefix' must be a non-empty file-name component".into(),
181            ));
182        }
183
184        let max_file_size = props
185            .get("max_file_size")
186            .map(|s| {
187                s.parse::<usize>().map_err(|e| {
188                    ConnectorError::ConfigurationError(format!("invalid max_file_size: {e}"))
189                })
190            })
191            .transpose()?;
192
193        let compression = props
194            .get("compression")
195            .cloned()
196            .unwrap_or_else(|| "snappy".to_string());
197
198        if props.contains_key("max_epoch_batches") {
199            return Err(ConnectorError::ConfigurationError(
200                "file sink option 'max_epoch_batches' was replaced by 'max_buffered_batches'"
201                    .into(),
202            ));
203        }
204        let max_buffered_batches = props
205            .get("max_buffered_batches")
206            .map(|s| {
207                s.parse::<usize>().map_err(|e| {
208                    ConnectorError::ConfigurationError(format!("invalid max_buffered_batches: {e}"))
209                })
210            })
211            .transpose()?
212            .unwrap_or(10_000);
213        if max_buffered_batches == 0 {
214            return Err(ConnectorError::ConfigurationError(
215                "max_buffered_batches must be > 0".into(),
216            ));
217        }
218
219        Ok(Self {
220            path,
221            format,
222            prefix,
223            max_file_size,
224            compression,
225            max_buffered_batches,
226        })
227    }
228}
229
230/// Supported file formats.
231#[derive(Debug, Clone, Copy, PartialEq, Eq)]
232pub enum FileFormat {
233    /// Comma-separated values.
234    Csv,
235    /// Newline-delimited JSON.
236    Json,
237    /// Plain text (one line per record).
238    Text,
239    /// Apache Parquet.
240    Parquet,
241    /// Arrow IPC file format (random-access `.arrow` files).
242    ArrowIpc,
243}
244
245impl FileFormat {
246    /// Parse a format string.
247    ///
248    /// # Errors
249    ///
250    /// Returns `ConnectorError::ConfigurationError` for unknown formats.
251    pub fn parse(s: &str) -> Result<Self, ConnectorError> {
252        match s.to_lowercase().as_str() {
253            "csv" | "tsv" => Ok(Self::Csv),
254            "json" | "jsonl" | "ndjson" | "json_lines" => Ok(Self::Json),
255            "text" | "txt" | "plain" => Ok(Self::Text),
256            "parquet" | "parq" => Ok(Self::Parquet),
257            "arrow" | "ipc" | "arrow_ipc" => Ok(Self::ArrowIpc),
258            other => Err(ConnectorError::ConfigurationError(format!(
259                "unknown file format: '{other}' (expected csv, json, text, parquet, or arrow)"
260            ))),
261        }
262    }
263
264    /// Detect format from a file path extension.
265    pub fn from_extension(path: &str) -> Option<Self> {
266        let ext = path.rsplit('.').next()?.to_lowercase();
267        match ext.as_str() {
268            "csv" | "tsv" => Some(Self::Csv),
269            "json" | "jsonl" | "ndjson" => Some(Self::Json),
270            "txt" | "log" => Some(Self::Text),
271            "parquet" | "parq" => Some(Self::Parquet),
272            "arrow" | "ipc" => Some(Self::ArrowIpc),
273            _ => None,
274        }
275    }
276
277    /// Returns the canonical file extension for this format.
278    #[must_use]
279    pub fn extension(&self) -> &'static str {
280        match self {
281            Self::Csv => "csv",
282            Self::Json => "jsonl",
283            Self::Text => "txt",
284            Self::Parquet => "parquet",
285            Self::ArrowIpc => "arrow",
286        }
287    }
288
289    /// Whether this is a columnar/bulk format (cannot be truncated mid-file).
290    #[must_use]
291    pub fn is_bulk_format(&self) -> bool {
292        matches!(self, Self::Parquet | Self::ArrowIpc)
293    }
294}
295
296// ── Helpers ──────────────────────────────────────────────────────────
297
298fn parse_duration(
299    props: &HashMap<String, String>,
300    key: &str,
301    default: Duration,
302) -> Result<Duration, ConnectorError> {
303    match props.get(key) {
304        Some(s) => parse_duration_str(s).ok_or_else(|| {
305            ConnectorError::ConfigurationError(format!("invalid duration for {key}: '{s}'"))
306        }),
307        None => Ok(default),
308    }
309}
310
311fn parse_usize(
312    props: &HashMap<String, String>,
313    key: &str,
314    default: usize,
315) -> Result<usize, ConnectorError> {
316    match props.get(key) {
317        Some(s) => s
318            .parse()
319            .map_err(|e| ConnectorError::ConfigurationError(format!("invalid {key}: {e}"))),
320        None => Ok(default),
321    }
322}
323
324fn parse_bool(
325    props: &HashMap<String, String>,
326    key: &str,
327    default: bool,
328) -> Result<bool, ConnectorError> {
329    match props.get(key) {
330        Some(s) => match s.to_lowercase().as_str() {
331            "true" | "1" | "yes" => Ok(true),
332            "false" | "0" | "no" => Ok(false),
333            other => Err(ConnectorError::ConfigurationError(format!(
334                "invalid boolean for {key}: '{other}'"
335            ))),
336        },
337        None => Ok(default),
338    }
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    #[test]
346    fn test_file_format_parse() {
347        assert_eq!(FileFormat::parse("csv").unwrap(), FileFormat::Csv);
348        assert_eq!(FileFormat::parse("JSON").unwrap(), FileFormat::Json);
349        assert_eq!(FileFormat::parse("jsonl").unwrap(), FileFormat::Json);
350        assert_eq!(FileFormat::parse("ndjson").unwrap(), FileFormat::Json);
351        assert_eq!(FileFormat::parse("text").unwrap(), FileFormat::Text);
352        assert_eq!(FileFormat::parse("parquet").unwrap(), FileFormat::Parquet);
353        assert_eq!(FileFormat::parse("parq").unwrap(), FileFormat::Parquet);
354        assert_eq!(FileFormat::parse("arrow").unwrap(), FileFormat::ArrowIpc);
355        assert_eq!(FileFormat::parse("ipc").unwrap(), FileFormat::ArrowIpc);
356        assert_eq!(
357            FileFormat::parse("arrow_ipc").unwrap(),
358            FileFormat::ArrowIpc
359        );
360        assert!(FileFormat::parse("xml").is_err());
361    }
362
363    #[test]
364    fn test_file_format_from_extension() {
365        assert_eq!(
366            FileFormat::from_extension("/data/logs/app.csv"),
367            Some(FileFormat::Csv)
368        );
369        assert_eq!(
370            FileFormat::from_extension("events.jsonl"),
371            Some(FileFormat::Json)
372        );
373        assert_eq!(
374            FileFormat::from_extension("data.parquet"),
375            Some(FileFormat::Parquet)
376        );
377        assert_eq!(
378            FileFormat::from_extension("log.txt"),
379            Some(FileFormat::Text)
380        );
381        assert_eq!(
382            FileFormat::from_extension("data.arrow"),
383            Some(FileFormat::ArrowIpc)
384        );
385        assert_eq!(
386            FileFormat::from_extension("stream.ipc"),
387            Some(FileFormat::ArrowIpc)
388        );
389        assert_eq!(FileFormat::from_extension("file.bin"), None);
390    }
391
392    #[test]
393    fn test_file_format_extension() {
394        assert_eq!(FileFormat::Csv.extension(), "csv");
395        assert_eq!(FileFormat::Json.extension(), "jsonl");
396        assert_eq!(FileFormat::Text.extension(), "txt");
397        assert_eq!(FileFormat::Parquet.extension(), "parquet");
398        assert_eq!(FileFormat::ArrowIpc.extension(), "arrow");
399    }
400
401    #[test]
402    fn test_file_format_is_bulk() {
403        assert!(!FileFormat::Csv.is_bulk_format());
404        assert!(!FileFormat::Json.is_bulk_format());
405        assert!(!FileFormat::Text.is_bulk_format());
406        assert!(FileFormat::ArrowIpc.is_bulk_format());
407        assert!(FileFormat::Parquet.is_bulk_format());
408    }
409
410    #[test]
411    fn test_source_config_from_connector() {
412        let mut config = ConnectorConfig::new("files");
413        config.set("path", "/data/logs/*.csv");
414        config.set("format", "csv");
415        config.set("max_files_per_poll", "50");
416        config.set("include_metadata", "true");
417
418        let src = FileSourceConfig::from_connector_config(&config).unwrap();
419        assert_eq!(src.path, "/data/logs/*.csv");
420        assert_eq!(src.format, Some(FileFormat::Csv));
421        assert_eq!(src.max_files_per_poll, 50);
422        assert!(src.include_metadata);
423    }
424
425    #[test]
426    fn removed_probabilistic_manifest_options_are_rejected() {
427        for option in [
428            "allow_overwrites",
429            "manifest_retention_count",
430            "manifest_retention_age_days",
431        ] {
432            let mut config = ConnectorConfig::new("files");
433            config.set("path", "/data/logs");
434            config.set("format", "text");
435            config.set(option, "1");
436            let error = FileSourceConfig::from_connector_config(&config)
437                .expect_err("removed lossy manifest option must fail closed");
438            assert!(error.to_string().contains(option));
439        }
440    }
441
442    #[test]
443    fn source_poll_bounds_must_be_nonzero() {
444        for option in ["max_files_per_poll", "max_file_bytes"] {
445            let mut config = ConnectorConfig::new("files");
446            config.set("path", "/data/logs");
447            config.set("format", "text");
448            config.set(option, "0");
449            let error = FileSourceConfig::from_connector_config(&config)
450                .expect_err("zero source bound must fail closed");
451            assert!(error.to_string().contains(option));
452        }
453    }
454
455    #[test]
456    fn test_source_config_missing_path() {
457        let config = ConnectorConfig::new("files");
458        assert!(FileSourceConfig::from_connector_config(&config).is_err());
459    }
460
461    #[test]
462    fn test_sink_config_from_connector() {
463        let mut config = ConnectorConfig::new("files");
464        config.set("path", "/output");
465        config.set("format", "parquet");
466        config.set("compression", "zstd");
467
468        let sink = FileSinkConfig::from_connector_config(&config).unwrap();
469        assert_eq!(sink.path, "/output");
470        assert_eq!(sink.format, FileFormat::Parquet);
471        assert_eq!(sink.compression, "zstd");
472    }
473
474    #[test]
475    fn test_sink_config_missing_format() {
476        let mut config = ConnectorConfig::new("files");
477        config.set("path", "/output");
478        assert!(FileSinkConfig::from_connector_config(&config).is_err());
479    }
480
481    #[test]
482    fn test_parse_duration_str() {
483        assert_eq!(parse_duration_str("10").unwrap(), Duration::from_secs(10));
484        assert_eq!(parse_duration_str("10s").unwrap(), Duration::from_secs(10));
485        assert_eq!(
486            parse_duration_str("500ms").unwrap(),
487            Duration::from_millis(500)
488        );
489    }
490
491    #[test]
492    fn test_removed_sink_mode_option_is_rejected() {
493        for old_mode in ["append", "rolling"] {
494            let mut config = ConnectorConfig::new("files");
495            config.set("path", "/output");
496            config.set("format", "csv");
497            config.set("mode", old_mode);
498
499            let error = FileSinkConfig::from_connector_config(&config).unwrap_err();
500            assert!(error.to_string().contains("option 'mode' was removed"));
501        }
502    }
503}
laminar_connectors/files/config.rs

laminar_connectors/files/
config.rs
