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::{future::Future, sync::Arc};
19
20use indexmap::{IndexMap, map::Entry};
21use tokio::runtime::Handle;
22
23use crate::{
24    catalog::factory as catalog_factory,
25    common::storage,
26    writer::{factory as writer_factory, feather as feather_writer, traits as writer_traits},
27};
28
29pub mod binary_heap;
30pub mod catalog;
31pub mod compare;
32pub mod feather;
33pub mod kmerge_batch;
34pub mod migration;
35pub mod parquet;
36pub mod session;
37
38/// Returns the persistence-owned catalog-factory registry.
39///
40/// The neutral registry types live in [`catalog_factory`]. Concrete built-in
41/// backend registrations are assembled here so consumers such as backtest, live,
42/// CLI tools, and Python bindings can share the same defaults without making the
43/// shared factory module depend on backend-specific modules.
44#[must_use]
45pub fn default_catalog_factories() -> catalog_factory::CatalogFactoryRegistry {
46    let mut registry = catalog_factory::CatalogFactoryRegistry::new();
47    register_builtin_catalog_factories(&mut registry);
48    registry
49}
50
51/// Merges user-provided factories into the persistence default registry.
52///
53/// # Errors
54///
55/// Returns an error if a user-provided name collides with an existing built-in
56/// or user-provided entry.
57pub fn extend_catalog_factories(
58    extra: impl IntoIterator<Item = (String, catalog_factory::CatalogFactory)>,
59) -> anyhow::Result<catalog_factory::CatalogFactoryRegistry> {
60    extend_factories(default_catalog_factories(), extra, "Catalog")
61}
62
63/// Returns the persistence-owned writer-factory registry.
64///
65/// Mirrors [`default_catalog_factories`]: the neutral registry types live in
66/// [`writer_factory`], and concrete built-in registrations are assembled here.
67#[must_use]
68pub fn default_writer_factories() -> writer_factory::WriterFactoryRegistry {
69    let mut registry = writer_factory::WriterFactoryRegistry::new();
70    register_builtin_writer_factories(&mut registry);
71    registry
72}
73
74/// Merges user-provided writer factories into the persistence default registry.
75///
76/// # Errors
77///
78/// Returns an error if a user-provided name collides with an existing entry.
79pub fn extend_writer_factories(
80    extra: impl IntoIterator<Item = (String, writer_factory::WriterFactory)>,
81) -> anyhow::Result<writer_factory::WriterFactoryRegistry> {
82    extend_factories(default_writer_factories(), extra, "Writer")
83}
84
85fn extend_factories<T>(
86    mut registry: IndexMap<String, T>,
87    extra: impl IntoIterator<Item = (String, T)>,
88    kind: &str,
89) -> anyhow::Result<IndexMap<String, T>> {
90    for (name, factory) in extra {
91        match registry.entry(name) {
92            Entry::Vacant(entry) => {
93                entry.insert(factory);
94            }
95            Entry::Occupied(entry) => {
96                anyhow::bail!("{kind} factory already registered: {}", entry.key());
97            }
98        }
99    }
100    Ok(registry)
101}
102
103fn register_builtin_writer_factories(registry: &mut writer_factory::WriterFactoryRegistry) {
104    parquet::writer::register_factory(registry);
105    registry.insert(
106        writer_factory::FEATHER_WRITER_FACTORY_NAME.to_string(),
107        Arc::new(
108            |config: &writer_factory::WriterConnectConfig, clock: feather_writer::WriterClock| {
109                let storage = storage::create_storage_backend_from_path(
110                    &config.uri,
111                    config.storage_options.clone(),
112                )?;
113                Ok(Box::new(
114                    feather_writer::FeatherWriter::new(
115                        storage.base_path.clone(),
116                        storage.object_store.clone(),
117                        clock,
118                        config.rotation_config.clone(),
119                        None,
120                        None,
121                        config.flush_interval_ms,
122                    )
123                    .with_record_filter(config.record_filter.clone()),
124                ) as writer_traits::StreamingDataSink)
125            },
126        ),
127    );
128}
129
130fn register_builtin_catalog_factories(registry: &mut catalog_factory::CatalogFactoryRegistry) {
131    parquet::register_catalog_factory(registry);
132}
133
134/// Runs an async operation from a synchronous persistence API.
135///
136/// `block_in_place` permits re-entering the shared multi-thread Nautilus runtime and executes the
137/// closure directly when called outside a Tokio runtime.
138///
139/// # Panics
140///
141/// Panics when called from a Tokio `current_thread` runtime. The shared Nautilus runtime is
142/// required to be multi-threaded.
143pub(crate) fn block_on<F>(runtime: &Handle, future: F) -> F::Output
144where
145    F: Future,
146{
147    tokio::task::block_in_place(|| runtime.block_on(future))
148}
nautilus_persistence/backend/mod.rs

nautilus_persistence/backend/
mod.rs
