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;
23
24use crate::catalog::traits::DataCatalog;
25
26/// Conventional name of the Parquet catalog factory registration.
27pub const PARQUET_CATALOG_FACTORY_NAME: &str = "Parquet";
28
29/// Minimal connection-config supplied to [`CatalogFactory`].
30#[derive(Debug, Clone)]
31pub struct CatalogConnectConfig {
32    /// Resolved URI for the catalog backend (e.g. `file:///tmp/cat`, `s3://bucket/cat`).
33    pub uri: String,
34    /// Optional storage-backend options (credentials, region, ...) passed to `object_store`.
35    pub storage_options: Option<AHashMap<String, String>>,
36    /// Backend-specific catalog parameters.
37    pub params: Option<Params>,
38}
39
40impl CatalogConnectConfig {
41    /// Creates a new [`CatalogConnectConfig`].
42    #[must_use]
43    pub fn new(uri: impl Into<String>, storage_options: Option<AHashMap<String, String>>) -> Self {
44        Self {
45            uri: uri.into(),
46            storage_options,
47            params: None,
48        }
49    }
50
51    /// Builds a [`CatalogConnectConfig`] from `path` + optional `fs_protocol`.
52    #[must_use]
53    pub fn from_path_and_protocol(
54        path: &str,
55        fs_protocol: Option<&str>,
56        storage_options: Option<AHashMap<String, String>>,
57    ) -> Self {
58        let uri = match fs_protocol {
59            _ if path.contains("://") => path.to_string(),
60            Some(protocol) => format!("{protocol}://{path}"),
61            None => path.to_string(),
62        };
63        Self::new(uri, storage_options)
64    }
65}
66
67/// Factory for a named catalog backend.
68pub type CatalogFactory =
69    Arc<dyn Fn(&CatalogConnectConfig) -> anyhow::Result<DataCatalog> + Send + Sync>;
70
71/// Ordered registry of catalog factories keyed by name.
72pub type CatalogFactoryRegistry = IndexMap<String, CatalogFactory>;
73
74#[cfg(test)]
75mod tests {
76    use rstest::rstest;
77
78    use super::*;
79
80    #[rstest]
81    fn from_path_and_protocol_joins_scheme() {
82        let cfg = CatalogConnectConfig::from_path_and_protocol("bucket/cat", Some("s3"), None);
83        assert_eq!(cfg.uri, "s3://bucket/cat");
84
85        let cfg = CatalogConnectConfig::from_path_and_protocol("/tmp/cat", None, None);
86        assert_eq!(cfg.uri, "/tmp/cat");
87
88        let cfg = CatalogConnectConfig::from_path_and_protocol(
89            "postgres://user:pass@localhost/catalog",
90            Some("file"),
91            None,
92        );
93        assert_eq!(cfg.uri, "postgres://user:pass@localhost/catalog");
94    }
95}
nautilus_persistence/catalog/factory.rs

nautilus_persistence/catalog/
factory.rs
