laminar_connectors/files/
mod.rs

1//! File source and sink connectors.
2
3use std::sync::Arc;
4
5use crate::config::ConnectorInfo;
6use crate::registry::ConnectorRegistry;
7
8pub mod arrow_ipc_codec;
9pub mod config;
10pub mod discovery;
11pub mod manifest;
12pub mod sink;
13pub mod source;
14pub mod text_decoder;
15
16pub use config::{FileFormat, FileSinkConfig, FileSourceConfig};
17pub use manifest::FileIngestionManifest;
18pub use sink::FileSink;
19pub use source::FileSource;
20pub use text_decoder::TextLineDecoder;
21
22/// Registers the file source connector in the registry.
23///
24/// This is called by `LaminarDB::register_builtin_connectors()` when the
25/// `files` feature is enabled, and makes `FROM FILES (...)` available in
26/// `CREATE SOURCE` statements.
27///
28/// # Errors
29///
30/// Returns an error if the connector name is already registered or the registry is frozen.
31pub fn register_file_source(
32    registry: &ConnectorRegistry,
33) -> Result<(), crate::error::ConnectorError> {
34    use crate::config::ConfigKeySpec;
35    let info = ConnectorInfo {
36        name: "files".to_string(),
37        display_name: "File Source (AutoLoader)".to_string(),
38        version: env!("CARGO_PKG_VERSION").to_string(),
39        is_source: true,
40        is_sink: false,
41        config_keys: vec![
42            ConfigKeySpec::required("path", "Local directory path or glob pattern"),
43            ConfigKeySpec::optional(
44                "format",
45                "Data format (csv, tsv, json, jsonl, text, txt, parquet, arrow)",
46                "auto-detect",
47            ),
48            ConfigKeySpec::optional(
49                "glob_pattern",
50                "Optional glob pattern to filter files by name",
51                "*",
52            ),
53        ],
54    };
55    registry.register_source(
56        "files",
57        info,
58        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
59            Ok(Box::new(FileSource::with_registry(
60                registry.map(Arc::as_ref),
61            )))
62        }),
63    )
64}
65
66/// Registers the file sink connector in the registry.
67///
68/// Makes `INTO FILES (...)` available in `CREATE SINK` statements.
69///
70/// # Errors
71///
72/// Returns an error if the connector name is already registered or the registry is frozen.
73pub fn register_file_sink(
74    registry: &ConnectorRegistry,
75) -> Result<(), crate::error::ConnectorError> {
76    use crate::config::ConfigKeySpec;
77    let info = ConnectorInfo {
78        name: "files".to_string(),
79        display_name: "File Sink".to_string(),
80        version: env!("CARGO_PKG_VERSION").to_string(),
81        is_source: false,
82        is_sink: true,
83        config_keys: vec![
84            ConfigKeySpec::required("path", "Output directory path"),
85            ConfigKeySpec::required("format", "Output format (csv, json, text, parquet, arrow)"),
86            ConfigKeySpec::optional("prefix", "Immutable output file name prefix", "part"),
87        ],
88    };
89    registry.register_sink(
90        "files",
91        info,
92        Arc::new(|_config, registry: Option<&Arc<prometheus::Registry>>| {
93            Ok(Box::new(FileSink::with_registry(registry.map(Arc::as_ref))))
94        }),
95    )
96}
97#[cfg(test)]
98mod tests;
laminar_connectors/files/mod.rs

laminar_connectors/files/
mod.rs
