nautilus_persistence/catalog/
factory.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
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
16//! Catalog factory registry primitives.
17
18use std::sync::Arc;
19
20use ahash::AHashMap;
21use indexmap::IndexMap;
22use nautilus_core::Params;
23use serde_json::Value;
24
25use crate::{
26    catalog::traits::DataCatalog,
27    common::paths::file_protocol_uri,
28    config::{CatalogBackendType, CatalogCompression},
29};
30
31/// Conventional name of the Parquet catalog factory registration.
32pub const PARQUET_CATALOG_FACTORY_NAME: &str = "Parquet";
33
34/// Minimal connection-config supplied to [`CatalogFactory`].
35#[derive(Debug, Clone)]
36pub struct CatalogConnectConfig {
37    /// Resolved URI for the catalog backend (e.g. `file:///tmp/cat`, `s3://bucket/cat`).
38    pub uri: String,
39    /// Optional storage-backend options (credentials, region, ...) passed to `object_store`.
40    pub storage_options: Option<AHashMap<String, String>>,
41    /// The number of rows per batch the catalog reads and writes.
42    pub batch_size: Option<usize>,
43    /// The compression codec of written data files.
44    pub compression: Option<CatalogCompression>,
45    /// The maximum number of rows per written row group.
46    pub max_row_group_size: Option<usize>,
47    /// Backend-specific catalog parameters, which a factory validates with
48    /// [`validate_catalog_params`].
49    pub params: Option<Params>,
50}
51
52impl CatalogConnectConfig {
53    /// Creates a new [`CatalogConnectConfig`].
54    #[must_use]
55    pub fn new(uri: impl Into<String>, storage_options: Option<AHashMap<String, String>>) -> Self {
56        Self {
57            uri: uri.into(),
58            storage_options,
59            batch_size: None,
60            compression: None,
61            max_row_group_size: None,
62            params: None,
63        }
64    }
65
66    /// Builds a [`CatalogConnectConfig`] from `path` + optional `fs_protocol`.
67    ///
68    /// A `file` protocol with a Windows drive path becomes `file:///C:/...`.
69    /// A path that already contains `://` is left unchanged.
70    #[must_use]
71    pub fn from_path_and_protocol(
72        path: &str,
73        fs_protocol: Option<&str>,
74        storage_options: Option<AHashMap<String, String>>,
75    ) -> Self {
76        let uri = match fs_protocol {
77            _ if path.contains("://") => path.to_string(),
78            Some("file") => file_protocol_uri(path),
79            Some(protocol) => format!("{protocol}://{path}"),
80            None => path.to_string(),
81        };
82
83        Self::new(uri, storage_options)
84    }
85}
86
87/// Factory for a named catalog backend.
88pub type CatalogFactory =
89    Arc<dyn Fn(&CatalogConnectConfig) -> anyhow::Result<DataCatalog> + Send + Sync>;
90
91/// Ordered registry of catalog factories keyed by name.
92pub type CatalogFactoryRegistry = IndexMap<String, CatalogFactory>;
93
94/// Resolves a catalog backend through the factory registry and opens the catalog.
95///
96/// All variants resolve through the registry so built-ins and custom names share one code path.
97///
98/// # Errors
99///
100/// Returns an error if the backend's factory is not registered or opening the catalog fails.
101pub fn create_catalog(
102    backend: &CatalogBackendType,
103    config: &CatalogConnectConfig,
104    factories: &CatalogFactoryRegistry,
105) -> anyhow::Result<DataCatalog> {
106    let name = backend.to_string();
107    factories
108        .get(&name)
109        .ok_or_else(|| anyhow::anyhow!("No catalog factory registered for '{name}'"))?(config)
110}
111
112/// The JSON type a catalog param holds.
113#[derive(Debug, Clone, Copy, PartialEq, Eq)]
114pub enum CatalogParamKind {
115    /// An object of named values.
116    Object,
117    /// A non-negative integer.
118    Count,
119    /// A string.
120    Text,
121    /// An array.
122    List,
123}
124
125impl CatalogParamKind {
126    fn accepts(self, value: &Value) -> bool {
127        match self {
128            Self::Object => value.is_object(),
129            Self::Count => value.is_u64(),
130            Self::Text => value.is_string(),
131            Self::List => value.is_array(),
132        }
133    }
134
135    const fn expected(self) -> &'static str {
136        match self {
137            Self::Object => "an object",
138            Self::Count => "a non-negative integer",
139            Self::Text => "a string",
140            Self::List => "a list",
141        }
142    }
143}
144
145/// Checks that `params` holds only the `accepted` keys, each with its declared kind.
146///
147/// A catalog factory calls this with the keys it reads. A `null` value satisfies any declared
148/// kind, but an undeclared key fails whatever its value.
149///
150/// # Errors
151///
152/// Returns an error if a key is not accepted or its value has the wrong kind. The error names the
153/// catalog and the key.
154pub fn validate_catalog_params(
155    catalog: &str,
156    params: Option<&Params>,
157    accepted: &[(&str, CatalogParamKind)],
158) -> anyhow::Result<()> {
159    let Some(params) = params else {
160        return Ok(());
161    };
162
163    for (key, value) in params {
164        let Some((_, kind)) = accepted.iter().find(|(name, _)| name == key) else {
165            if accepted.is_empty() {
166                anyhow::bail!(
167                    "Unknown {catalog} catalog param '{key}': this catalog takes no params"
168                );
169            }
170
171            let names = accepted
172                .iter()
173                .map(|(name, _)| *name)
174                .collect::<Vec<_>>()
175                .join(", ");
176            anyhow::bail!("Unknown {catalog} catalog param '{key}', expected one of {names}");
177        };
178
179        anyhow::ensure!(
180            value.is_null() || kind.accepts(value),
181            "Invalid {catalog} catalog param '{key}': expected {}",
182            kind.expected()
183        );
184    }
185
186    Ok(())
187}
188
189#[cfg(test)]
190mod tests {
191    use rstest::rstest;
192    use serde_json::json;
193
194    use super::*;
195
196    #[rstest]
197    fn from_path_and_protocol_joins_scheme() {
198        let cfg = CatalogConnectConfig::from_path_and_protocol("bucket/cat", Some("s3"), None);
199        assert_eq!(cfg.uri, "s3://bucket/cat");
200
201        let cfg = CatalogConnectConfig::from_path_and_protocol("/tmp/cat", None, None);
202        assert_eq!(cfg.uri, "/tmp/cat");
203
204        let cfg = CatalogConnectConfig::from_path_and_protocol(
205            "postgres://user:pass@localhost/catalog",
206            Some("file"),
207            None,
208        );
209        assert_eq!(cfg.uri, "postgres://user:pass@localhost/catalog");
210    }
211
212    #[rstest]
213    #[case(r"C:\data\catalog", "file:///C:/data/catalog")]
214    #[case("C:/data/catalog", "file:///C:/data/catalog")]
215    #[case(r"D:\", "file:///D:/")]
216    fn from_path_and_protocol_normalizes_windows_drive_paths(
217        #[case] path: &str,
218        #[case] expected: &str,
219    ) {
220        let cfg = CatalogConnectConfig::from_path_and_protocol(path, Some("file"), None);
221        assert_eq!(cfg.uri, expected);
222    }
223
224    #[rstest]
225    fn from_path_and_protocol_keeps_non_drive_file_joins() {
226        let cfg = CatalogConnectConfig::from_path_and_protocol("/tmp/cat", Some("file"), None);
227        assert_eq!(cfg.uri, "file:///tmp/cat");
228
229        let cfg = CatalogConnectConfig::from_path_and_protocol("data/cat", Some("file"), None);
230        assert_eq!(cfg.uri, "file://data/cat");
231
232        let cfg = CatalogConnectConfig::from_path_and_protocol(
233            r"\\server\share\catalog",
234            Some("file"),
235            None,
236        );
237        assert_eq!(cfg.uri, r"file://\\server\share\catalog");
238    }
239
240    #[rstest]
241    fn create_catalog_rejects_unregistered_backend() {
242        let config = CatalogConnectConfig::new("/tmp/catalog", None);
243
244        let error = create_catalog(
245            &CatalogBackendType::External("Missing".to_string()),
246            &config,
247            &CatalogFactoryRegistry::new(),
248        )
249        .expect_err("an unregistered backend should fail");
250
251        assert_eq!(
252            error.to_string(),
253            "No catalog factory registered for 'Missing'"
254        );
255    }
256
257    const ACCEPTED: &[(&str, CatalogParamKind)] = &[
258        ("options", CatalogParamKind::Object),
259        ("rows", CatalogParamKind::Count),
260        ("codec", CatalogParamKind::Text),
261        ("rules", CatalogParamKind::List),
262    ];
263
264    fn params(entries: &[(&str, Value)]) -> Params {
265        let mut params = Params::new();
266
267        for (key, value) in entries {
268            params.insert((*key).to_string(), value.clone());
269        }
270
271        params
272    }
273
274    #[rstest]
275    #[case::absent(None)]
276    #[case::every_kind(Some(params(&[
277        ("options", json!({"region": "eu-west-1"})),
278        ("rows", json!(1024)),
279        ("codec", json!("zstd")),
280        ("rules", json!([])),
281    ])))]
282    #[case::null_counts_as_absent(Some(params(&[("rows", Value::Null)])))]
283    fn validate_catalog_params_accepts_declared_keys(#[case] params: Option<Params>) {
284        assert!(validate_catalog_params("Test", params.as_ref(), ACCEPTED).is_ok());
285    }
286
287    #[rstest]
288    #[case::unknown(
289        "no_such_param",
290        json!(1),
291        "Unknown Test catalog param 'no_such_param', expected one of options, rows, codec, rules"
292    )]
293    #[case::unknown_null(
294        "no_such_param",
295        Value::Null,
296        "Unknown Test catalog param 'no_such_param', expected one of options, rows, codec, rules"
297    )]
298    #[case::text_for_count(
299        "rows",
300        json!("big"),
301        "Invalid Test catalog param 'rows': expected a non-negative integer"
302    )]
303    #[case::negative_count(
304        "rows",
305        json!(-1),
306        "Invalid Test catalog param 'rows': expected a non-negative integer"
307    )]
308    #[case::text_for_object(
309        "options",
310        json!("region=eu-west-1"),
311        "Invalid Test catalog param 'options': expected an object"
312    )]
313    #[case::count_for_text(
314        "codec",
315        json!(6),
316        "Invalid Test catalog param 'codec': expected a string"
317    )]
318    #[case::object_for_list(
319        "rules",
320        json!({}),
321        "Invalid Test catalog param 'rules': expected a list"
322    )]
323    fn validate_catalog_params_rejects_unknown_or_mistyped_key(
324        #[case] key: &str,
325        #[case] value: Value,
326        #[case] expected: &str,
327    ) {
328        let error =
329            validate_catalog_params("Test", Some(&params(&[(key, value)])), ACCEPTED).unwrap_err();
330
331        assert_eq!(error.to_string(), expected);
332    }
333
334    #[rstest]
335    fn validate_catalog_params_rejects_every_key_when_none_accepted() {
336        let error =
337            validate_catalog_params("Test", Some(&params(&[("rows", json!(1))])), &[]).unwrap_err();
338
339        assert_eq!(
340            error.to_string(),
341            "Unknown Test catalog param 'rows': this catalog takes no params"
342        );
343    }
344}
nautilus_persistence/catalog/factory.rs

nautilus_persistence/catalog/
factory.rs
