nautilus_persistence/backend/
mod.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//! Provides an Apache Parquet backend powered by [DataFusion](https://arrow.apache.org/datafusion).
17
18use std::sync::Arc;
19
20use indexmap::{IndexMap, map::Entry};
21
22use crate::{
23    catalog::factory as catalog_factory,
24    common::paths,
25    writer::{factory as writer_factory, feather as feather_writer, traits as writer_traits},
26};
27
28pub mod migration;
29pub mod parquet;
30
31/// Returns the persistence-owned catalog-factory registry.
32///
33/// The neutral registry types live in [`catalog_factory`]. Concrete built-in
34/// backend registrations are assembled here so consumers such as backtest, live,
35/// CLI tools, and Python bindings can share the same defaults without making the
36/// shared factory module depend on backend-specific modules.
37#[must_use]
38pub fn default_catalog_factories() -> catalog_factory::CatalogFactoryRegistry {
39    let mut registry = catalog_factory::CatalogFactoryRegistry::new();
40    register_builtin_catalog_factories(&mut registry);
41    registry
42}
43
44/// Merges user-provided factories into the persistence default registry.
45///
46/// # Errors
47///
48/// Returns an error if a user-provided name collides with an existing built-in
49/// or user-provided entry.
50pub fn extend_catalog_factories(
51    extra: impl IntoIterator<Item = (String, catalog_factory::CatalogFactory)>,
52) -> anyhow::Result<catalog_factory::CatalogFactoryRegistry> {
53    extend_factories(default_catalog_factories(), extra, "Catalog")
54}
55
56/// Returns the persistence-owned writer-factory registry.
57///
58/// Mirrors [`default_catalog_factories`]: the neutral registry types live in
59/// [`writer_factory`], and concrete built-in registrations are assembled here.
60#[must_use]
61pub fn default_writer_factories() -> writer_factory::WriterFactoryRegistry {
62    let mut registry = writer_factory::WriterFactoryRegistry::new();
63    register_builtin_writer_factories(&mut registry);
64    registry
65}
66
67/// Merges user-provided writer factories into the persistence default registry.
68///
69/// # Errors
70///
71/// Returns an error if a user-provided name collides with an existing entry.
72pub fn extend_writer_factories(
73    extra: impl IntoIterator<Item = (String, writer_factory::WriterFactory)>,
74) -> anyhow::Result<writer_factory::WriterFactoryRegistry> {
75    extend_factories(default_writer_factories(), extra, "Writer")
76}
77
78fn extend_factories<T>(
79    mut registry: IndexMap<String, T>,
80    extra: impl IntoIterator<Item = (String, T)>,
81    kind: &str,
82) -> anyhow::Result<IndexMap<String, T>> {
83    for (name, factory) in extra {
84        match registry.entry(name) {
85            Entry::Vacant(entry) => {
86                entry.insert(factory);
87            }
88            Entry::Occupied(entry) => {
89                anyhow::bail!("{kind} factory already registered: {}", entry.key());
90            }
91        }
92    }
93
94    Ok(registry)
95}
96
97fn register_builtin_catalog_factories(registry: &mut catalog_factory::CatalogFactoryRegistry) {
98    parquet::register_catalog_factory(registry);
99}
100
101fn register_builtin_writer_factories(registry: &mut writer_factory::WriterFactoryRegistry) {
102    parquet::writer::register_factory(registry);
103    registry.insert(
104        writer_factory::FEATHER_WRITER_FACTORY_NAME.to_string(),
105        Arc::new(
106            |config: &writer_factory::WriterConnectConfig, clock: feather_writer::WriterClock| {
107                let directory = paths::local_writer_directory(&config.uri)?;
108                feather_writer::recover_partial_feather_files(&directory);
109
110                Ok(Box::new(
111                    feather_writer::FeatherWriter::new(
112                        directory,
113                        clock,
114                        config.rotation_config.clone(),
115                        None,
116                        config.flush_interval_ms,
117                    )
118                    .with_record_filter(config.record_filter.clone()),
119                ) as writer_traits::StreamingDataSink)
120            },
121        ),
122    );
123}
124
125#[cfg(test)]
126mod tests {
127    use rstest::rstest;
128
129    use super::*;
130
131    fn unused_catalog_factory() -> catalog_factory::CatalogFactory {
132        Arc::new(|_| anyhow::bail!("not invoked"))
133    }
134
135    #[rstest]
136    fn extend_catalog_factories_accepts_external_factory() {
137        let factories =
138            extend_catalog_factories(vec![("ExternalTest".to_string(), unused_catalog_factory())])
139                .unwrap();
140
141        assert_eq!(
142            factories.keys().map(String::as_str).collect::<Vec<_>>(),
143            vec![
144                catalog_factory::PARQUET_CATALOG_FACTORY_NAME,
145                "ExternalTest"
146            ],
147        );
148    }
149
150    #[rstest]
151    #[case::builtin(vec!["Parquet"], "Parquet")]
152    #[case::user(vec!["Duplicate", "Duplicate"], "Duplicate")]
153    fn extend_catalog_factories_rejects_duplicate_names(
154        #[case] names: Vec<&str>,
155        #[case] duplicate: &str,
156    ) {
157        let extra = names
158            .into_iter()
159            .map(|name| (name.to_string(), unused_catalog_factory()));
160
161        let error = extend_catalog_factories(extra)
162            .err()
163            .expect("duplicate registration should fail");
164
165        assert_eq!(
166            error.to_string(),
167            format!("Catalog factory already registered: {duplicate}"),
168        );
169    }
170}
nautilus_persistence/backend/mod.rs

nautilus_persistence/backend/
mod.rs
