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;
11mod json_documents;
12pub mod manifest;
13mod output_directory;
14mod schema_resolution;
15pub mod sink;
16pub mod source;
17pub mod text_decoder;
18
19pub use config::{FileFormat, FileSinkConfig, FileSourceConfig};
20pub use manifest::FileIngestionManifest;
21pub use sink::FileSink;
22pub use source::FileSource;
23pub use text_decoder::TextLineDecoder;
24
25/// Registers the file source connector in the registry.
26///
27/// This is called by `LaminarDB::register_builtin_connectors()` when the
28/// `files` feature is enabled, and makes `FROM FILES (...)` available in
29/// `CREATE SOURCE` statements.
30///
31/// # Errors
32///
33/// Returns an error if the connector name is already registered or the registry is frozen.
34pub fn register_file_source(
35    registry: &ConnectorRegistry,
36) -> Result<(), crate::error::ConnectorError> {
37    use crate::config::ConfigKeySpec;
38    let info = ConnectorInfo {
39        schema_capabilities: crate::schema::resolution::SchemaCapabilities {
40            bounded_sampling: true,
41            built_in_formats: vec![
42                "text".into(),
43                "txt".into(),
44                #[cfg(feature = "fpml")]
45                "fpml".into(),
46            ],
47            ..crate::schema::resolution::SchemaCapabilities::metadata(
48                &["parquet", "arrow_ipc", "arrow"],
49                false,
50                crate::schema::resolution::SchemaPreparation::None,
51            )
52        },
53        name: "files".to_string(),
54        display_name: "File Source (AutoLoader)".to_string(),
55        version: env!("CARGO_PKG_VERSION").to_string(),
56        is_source: true,
57        is_sink: false,
58        config_keys: vec![
59            ConfigKeySpec::required("path", "Local directory path or glob pattern"),
60            ConfigKeySpec::optional(
61                "format",
62                "Data format (csv, tsv, json, jsonl, text, txt, parquet, arrow)",
63                "auto-detect",
64            ),
65            ConfigKeySpec::optional(
66                "glob_pattern",
67                "Optional glob pattern to filter files by name",
68                "*",
69            ),
70            ConfigKeySpec::optional(
71                "schema.inference",
72                "Opt-in bounded CSV/JSON inference (4 files, 1 MiB, 1000 rows)",
73                "false",
74            ),
75            #[cfg(feature = "fpml")]
76            ConfigKeySpec::optional(
77                "fpml.profile",
78                "Bundled FpML projection profile (format = 'fpml'); the default has one row per trade without product economics, which the asset-class profiles add",
79                crate::fpml::DEFAULT_PROFILE,
80            ),
81            #[cfg(feature = "fpml")]
82            ConfigKeySpec::optional(
83                "fpml.profile.path",
84                "Operator FpML profile file, instead of a bundled profile",
85                "",
86            ),
87            #[cfg(feature = "fpml")]
88            ConfigKeySpec::optional(
89                "fpml.schema.path",
90                "Directory of a digest-pinned schema closure (closure.toml)",
91                "bundled FpML 5.13 REC-2 confirmation view",
92            ),
93            #[cfg(feature = "fpml")]
94            ConfigKeySpec::optional(
95                "fpml.references",
96                "Reference checks: resolve (ids exist) or schema (also annotated target types)",
97                "resolve",
98            ),
99        ],
100    };
101    registry.register_source(
102        "files",
103        info,
104        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
105            Ok(Box::new(FileSource::with_registry(
106                registry.map(Arc::as_ref),
107            )))
108        }),
109    )
110}
111
112/// Registers the file sink connector in the registry.
113///
114/// Makes `INTO FILES (...)` available in `CREATE SINK` statements.
115///
116/// # Errors
117///
118/// Returns an error if the connector name is already registered or the registry is frozen.
119pub fn register_file_sink(
120    registry: &ConnectorRegistry,
121) -> Result<(), crate::error::ConnectorError> {
122    use crate::config::ConfigKeySpec;
123    let info = ConnectorInfo {
124        schema_capabilities: crate::schema::resolution::SchemaCapabilities::declared(true),
125        name: "files".to_string(),
126        display_name: "File Sink".to_string(),
127        version: env!("CARGO_PKG_VERSION").to_string(),
128        is_source: false,
129        is_sink: true,
130        config_keys: vec![
131            ConfigKeySpec::required("path", "Output directory path"),
132            ConfigKeySpec::required("format", "Output format (csv, json, text, parquet, arrow)"),
133            ConfigKeySpec::optional("prefix", "Immutable output file name prefix", "part"),
134        ],
135    };
136    registry.register_sink(
137        "files",
138        info,
139        Arc::new(|_config, registry: Option<&Arc<prometheus::Registry>>| {
140            Ok(Box::new(FileSink::with_registry(registry.map(Arc::as_ref))))
141        }),
142    )
143}
144#[cfg(test)]
145mod tests;
laminar_connectors/files/mod.rs

laminar_connectors/files/
mod.rs
