nautilus_persistence/backend/parquet/catalog/
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//! Parquet data catalog for efficient storage and retrieval of financial market data.
17//!
18//! This module provides a data catalog implementation that uses Apache Parquet
19//! format for storing financial market data with object store backends. The catalog supports
20//! various data types including quotes, trades, bars, order book data, and other market events.
21//!
22//! # Key Features
23//!
24//! - **Object Store Integration**: Works with local filesystems, S3, and other object stores.
25//! - **Data Type Support**: Handles all major financial data types (quotes, trades, bars, etc.).
26//! - **Time-based Organization**: Organizes data by timestamp ranges for efficient querying.
27//! - **Consolidation**: Merges multiple files to optimize storage and query performance.
28//! - **Validation**: Ensures data integrity with timestamp ordering and interval validation.
29//!
30//! # Architecture
31//!
32//! The catalog organizes data in a hierarchical structure:
33//! ```text
34//! data/
35//! ├── quotes/
36//! │   └── INSTRUMENT_ID/
37//! │       └── start_ts-end_ts.parquet
38//! ├── trades/
39//! │   └── INSTRUMENT_ID/
40//! │       └── start_ts-end_ts.parquet
41//! └── bars/
42//!     └── INSTRUMENT_ID/
43//!         └── start_ts-end_ts.parquet
44//! ```
45//!
46//! # Usage
47//!
48//! ```rust,no_run
49//! use std::path::Path;
50//!
51//! use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
52//!
53//! // Create a new catalog
54//! let catalog = ParquetDataCatalog::new(
55//!     Path::new("/path/to/data"),
56//!     None,       // storage_options
57//!     Some(5000), // batch_size
58//!     None,       // compression (defaults to SNAPPY)
59//!     None,       // max_row_group_size (defaults to 131,072)
60//! );
61//!
62//! // Write data to the catalog
63//! // catalog.write_to_parquet(&data, None, None, None)?;
64//! ```
65
66#![expect(
67    clippy::missing_fields_in_debug,
68    reason = "catalog Debug redacts internal caches"
69)]
70
71use std::{
72    borrow::Cow,
73    collections::{BTreeMap, HashMap, HashSet},
74    fmt::Debug,
75    path::{Path, PathBuf},
76    sync::Arc,
77};
78
79use ahash::AHashMap;
80use arrow::record_batch::RecordBatch;
81use futures::StreamExt;
82use nautilus_core::{
83    Params, UnixNanos,
84    string::{conversions::to_snake_case, urlencoding},
85};
86use nautilus_model::{
87    data::{
88        Bar, CustomData, Data, DataBatch, FundingRateUpdate, HasTsInit, IndexPriceUpdate,
89        InstrumentStatus, MarkPriceUpdate, NautilusDataType, NautilusRecordType, OptionGreeks,
90        OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick, close::InstrumentClose,
91        is_monotonically_increasing_by_init,
92    },
93    instruments::{Instrument, InstrumentAny},
94};
95use nautilus_serialization::arrow::{
96    ArrowSchemaProvider, DecodeDataFromRecordBatch, DecodeTypedFromRecordBatch,
97    EncodeToRecordBatch, catalog_display::catalog_record_batch_to_display,
98    custom::CustomDataDecoder, display::instrument::encode_instruments,
99    record_batch_without_identifier_column,
100};
101use object_store::{ObjectStore, ObjectStoreExt, path::Path as ObjectPath};
102use serde::Serialize;
103
104use crate::{
105    backend::parquet::{
106        intervals::query_interval_diff,
107        io::{
108            append_path_to_file_uri, decode_object_store_segment, is_remote_uri_scheme,
109            read_parquet_from_object_store, read_parquet_schema_from_object_store, remote_full_uri,
110            remote_store_root_url, write_batches_to_object_store,
111        },
112        paths::{extract_bar_type_instrument_id, query_intersects_filename},
113    },
114    catalog::{
115        session::{DEFAULT_DATA_BATCH_CHUNK_SIZE, DataBatchQueryResult, TypedDataBatchSession},
116        traits::{
117            CatalogInstrumentQuery, CatalogMetadata, CatalogQuery, CatalogReader,
118            CatalogRecordQuery, CatalogWriter, DataCatalog, filter_instrument_query_result,
119            filter_instruments_for_request_range,
120        },
121        types::{
122            CatalogAsOf, CatalogDataType, HasCatalogDataType, INSTRUMENT_PATH_PREFIXES,
123            instrument_any_type, instrument_path_prefix, parquet_data_path_prefix,
124            record_path_prefix,
125        },
126    },
127    common::{
128        custom::prepare_custom_data_batch,
129        datafusion::{self as datafusion, DataBackendSession, build_query},
130    },
131};
132
133/// Optional Parquet query file-loading hint parameter key.
134pub const QUERY_OPTIMIZE_FILE_LOADING: &str = "optimize_file_loading";
135
136/// Optional Parquet write overlap bypass parameter key.
137pub const WRITE_SKIP_DISJOINT_CHECK: &str = "skip_disjoint_check";
138
139macro_rules! define_builtin_data_dispatch {
140    (
141        (Instrument, InstrumentAny, Instrument, Instrument, $instrument_prefix:literal),
142        $(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?
143    ) => {
144
145
146        fn query_builtin_batch(
147            catalog: &mut ParquetDataCatalog,
148            data_type: &NautilusDataType,
149            identifiers: Option<Vec<String>>,
150            start: Option<UnixNanos>,
151            end: Option<UnixNanos>,
152            where_clause: Option<&str>,
153            optimize_file_loading: bool,
154        ) -> Option<anyhow::Result<DataBatch>> {
155            match data_type {
156                $(
157                    NautilusDataType::$variant => Some(
158                        catalog
159                            .query_typed_data::<$type>(
160                                identifiers,
161                                start,
162                                end,
163                                where_clause,
164                                None,
165                                optimize_file_loading,
166                            )
167                            .map(|data| DataBatch::$batch(data.into())),
168                    ),
169                )+
170                _ => None,
171            }
172        }
173
174        fn write_catalog_batch(
175            catalog: &ParquetDataCatalog,
176            batch: &DataBatch,
177            start: Option<UnixNanos>,
178            end: Option<UnixNanos>,
179            skip_disjoint_check: Option<bool>,
180        ) -> anyhow::Result<()> {
181            #[allow(unreachable_patterns, reason = "reject unsupported variants introduced by feature unification")]
182            match batch {
183                DataBatch::BookDeltas(data) => {
184                    let deltas = data.iter().flat_map(|batch| batch.deltas.iter().copied()).collect::<Vec<_>>();
185                    catalog.write_grouped_to_parquet(&deltas, start, end, skip_disjoint_check)
186                }
187                DataBatch::Custom(data) => catalog.write_custom_data_batch(data.as_ref(), start, end, skip_disjoint_check).map(|_| ()),
188                DataBatch::Instrument(data) => catalog.write_instruments(data.as_ref().to_vec()).map(|_| ()),
189                $(DataBatch::$batch(data) => catalog.write_grouped_to_parquet(data.as_ref(), start, end, skip_disjoint_check),)+
190                _ => anyhow::bail!("Unsupported catalog data batch: {}", batch.data_type_name()),
191            }
192        }
193    };
194}
195
196nautilus_model::for_each_data_type!(define_builtin_data_dispatch);
197
198/// A high-performance data catalog for storing and retrieving financial market data using Apache Parquet format.
199///
200/// The `ParquetDataCatalog` provides a solution for managing large volumes of financial
201/// market data with efficient storage, querying, and consolidation capabilities. It supports various
202/// object store backends including local filesystems, AWS S3, and other cloud storage providers.
203///
204/// # Features
205///
206/// - **Efficient Storage**: Uses Apache Parquet format with configurable compression.
207/// - **Object Store Backend**: Supports multiple storage backends through the `object_store` crate.
208/// - **Time-based Organization**: Organizes data by timestamp ranges for optimal query performance.
209/// - **Data Validation**: Ensures timestamp ordering and interval consistency.
210/// - **Consolidation**: Merges multiple files to reduce storage overhead and improve query speed.
211/// - **Type Safety**: Strongly typed data handling with compile-time guarantees.
212///
213/// # Data Organization
214///
215/// Data is organized hierarchically by data type and instrument:
216/// - `data/{data_type}/{instrument_id}/{start_ts}-{end_ts}.parquet`.
217/// - Files are named with their timestamp ranges for efficient range queries.
218/// - Intervals are validated to be disjoint to prevent data overlap.
219///
220/// # Performance Considerations
221///
222/// - **Batch Size**: Controls memory usage during data processing.
223/// - **Compression**: SNAPPY compression provides good balance of speed and size.
224/// - **Row Group Size**: Affects query performance and memory usage.
225/// - **File Consolidation**: Reduces the number of files for better query performance.
226pub struct ParquetDataCatalog {
227    /// The base path for data storage within the object store.
228    pub base_path: String,
229    /// The original URI provided when creating the catalog.
230    pub original_uri: String,
231    /// The object store backend for data persistence.
232    pub object_store: Arc<dyn ObjectStore>,
233    /// The DataFusion session for query execution.
234    pub session: DataBackendSession,
235    /// The number of records to process in each batch.
236    pub batch_size: usize,
237    /// The compression algorithm used for Parquet files.
238    pub compression: parquet::basic::Compression,
239    /// The maximum number of rows in each Parquet row group.
240    pub max_row_group_size: usize,
241}
242
243impl Debug for ParquetDataCatalog {
244    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245        f.debug_struct(stringify!(ParquetDataCatalog))
246            .field("base_path", &self.base_path)
247            .finish()
248    }
249}
250
251mod coverage;
252mod query;
253mod session;
254mod store;
255mod write;
256
257impl ParquetDataCatalog {
258    /// Creates a new [`ParquetDataCatalog`] instance from a local file path.
259    ///
260    /// This is a convenience constructor that converts a local path to a URI format
261    /// and delegates to [`Self::from_uri`].
262    ///
263    /// # Parameters
264    ///
265    /// - `base_path`: The base directory path for data storage.
266    /// - `storage_options`: Optional `HashMap` containing storage-specific configuration options.
267    /// - `batch_size`: Number of records to process in each batch (default: 5000).
268    /// - `compression`: Parquet compression algorithm (default: SNAPPY).
269    /// - `max_row_group_size`: Maximum rows per Parquet row group (default: 131,072).
270    ///
271    /// # Panics
272    ///
273    /// Panics if the path cannot be converted to a valid URI or if the object store
274    /// cannot be created from the path.
275    ///
276    /// # Examples
277    ///
278    /// ```rust,no_run
279    /// use std::path::Path;
280    ///
281    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
282    ///
283    /// let catalog = ParquetDataCatalog::new(
284    ///     Path::new("/tmp/nautilus_data"),
285    ///     None,       // no storage options
286    ///     Some(1000), // smaller batch size
287    ///     None,       // default compression
288    ///     None,       // default row group size
289    /// );
290    /// ```
291    #[must_use]
292    pub fn new(
293        base_path: &Path,
294        storage_options: Option<AHashMap<String, String>>,
295        batch_size: Option<usize>,
296        compression: Option<parquet::basic::Compression>,
297        max_row_group_size: Option<usize>,
298    ) -> Self {
299        let path_str = base_path.to_string_lossy().to_string();
300        Self::from_uri(
301            &path_str,
302            storage_options,
303            batch_size,
304            compression,
305            max_row_group_size,
306        )
307        .expect("Failed to create catalog from path")
308    }
309
310    /// Creates a new [`ParquetDataCatalog`] instance from a URI with optional storage options.
311    ///
312    /// Supports various URI schemes including local file paths and multiple cloud storage backends
313    /// supported by the `object_store` crate.
314    ///
315    /// # Supported URI Schemes
316    ///
317    /// - **AWS S3**: `s3://bucket/path`.
318    /// - **Google Cloud Storage**: `gs://bucket/path` or `gcs://bucket/path`.
319    /// - **Azure Blob Storage**: `az://container/path` or `abfs://container@account.dfs.core.windows.net/path`.
320    /// - **HTTP/WebDAV**: `http://` or `https://`.
321    /// - **Local files**: `file://path` or plain paths.
322    ///
323    /// # Parameters
324    ///
325    /// - `uri`: The URI for the data storage location.
326    /// - `storage_options`: Optional `HashMap` containing storage-specific configuration options:
327    ///   - For S3: `endpoint_url`, region, `access_key_id`, `secret_access_key`, `session_token`, etc.
328    ///   - For GCS: `service_account_path`, `service_account_key`, `project_id`, etc.
329    ///   - For Azure: `account_name`, `account_key`, `sas_token`, etc.
330    /// - `batch_size`: Number of records to process in each batch (default: 5000).
331    /// - `compression`: Parquet compression algorithm (default: SNAPPY).
332    /// - `max_row_group_size`: Maximum rows per Parquet row group (default: 131,072).
333    ///
334    /// # Errors
335    ///
336    /// Returns an error if:
337    /// - The URI format is invalid or unsupported.
338    /// - The object store cannot be created or accessed.
339    /// - Authentication fails for cloud storage backends.
340    ///
341    /// # Examples
342    ///
343    /// ```rust,no_run
344    /// use ahash::AHashMap;
345    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
346    ///
347    /// // Local filesystem
348    /// let local_catalog = ParquetDataCatalog::from_uri("/tmp/nautilus_data", None, None, None, None)?;
349    ///
350    /// // S3 bucket
351    /// let s3_catalog =
352    ///     ParquetDataCatalog::from_uri("s3://my-bucket/nautilus-data", None, None, None, None)?;
353    ///
354    /// // Google Cloud Storage
355    /// let gcs_catalog =
356    ///     ParquetDataCatalog::from_uri("gs://my-bucket/nautilus-data", None, None, None, None)?;
357    ///
358    /// // Azure Blob Storage
359    /// let azure_catalog =
360    ///     ParquetDataCatalog::from_uri("az://container/nautilus-data", None, None, None, None)?;
361    ///
362    /// // S3 with custom endpoint and credentials
363    /// let mut storage_options = AHashMap::new();
364    /// storage_options.insert(
365    ///     "endpoint_url".to_string(),
366    ///     "https://my-s3-endpoint.com".to_string(),
367    /// );
368    /// storage_options.insert("access_key_id".to_string(), "my-key".to_string());
369    /// storage_options.insert("secret_access_key".to_string(), "my-secret".to_string());
370    ///
371    /// let custom_s3_catalog = ParquetDataCatalog::from_uri(
372    ///     "s3://my-bucket/nautilus-data",
373    ///     Some(storage_options),
374    ///     None,
375    ///     None,
376    ///     None,
377    /// )?;
378    /// # Ok::<(), anyhow::Error>(())
379    /// ```
380    pub fn from_uri(
381        uri: &str,
382        storage_options: Option<AHashMap<String, String>>,
383        batch_size: Option<usize>,
384        compression: Option<parquet::basic::Compression>,
385        max_row_group_size: Option<usize>,
386    ) -> anyhow::Result<Self> {
387        let batch_size = batch_size.unwrap_or(DEFAULT_DATA_BATCH_CHUNK_SIZE);
388        let compression = compression.unwrap_or(parquet::basic::Compression::ZSTD(
389            parquet::basic::ZstdLevel::default(),
390        ));
391        let max_row_group_size =
392            max_row_group_size.unwrap_or(crate::backend::parquet::DEFAULT_ROW_GROUP_SIZE);
393
394        let location = crate::backend::parquet::io::create_object_store_location_from_path(
395            uri,
396            storage_options,
397        )?;
398
399        Ok(Self {
400            base_path: location.base_path,
401            original_uri: location.original_uri,
402            object_store: location.object_store,
403            session: DataBackendSession::new(batch_size),
404            batch_size,
405            compression,
406            max_row_group_size,
407        })
408    }
409
410    /// Returns the base path of the catalog for testing purposes.
411    #[must_use]
412    pub fn get_base_path(&self) -> String {
413        self.base_path.clone()
414    }
415
416    /// Clears cached table registrations so a later query re-reads files that changed.
417    ///
418    /// Catalog operations that modify files call this before querying again.
419    pub(crate) fn clear_session_tables(&mut self) {
420        self.session.clear_registered_tables();
421    }
422}
423
424impl CatalogReader for ParquetDataCatalog {
425    fn fork_query_catalog(&self) -> anyhow::Result<Option<DataCatalog>> {
426        Ok(Some(Box::new(Self {
427            base_path: self.base_path.clone(),
428            original_uri: self.original_uri.clone(),
429            object_store: self.object_store.clone(),
430            session: DataBackendSession::new(self.batch_size),
431            batch_size: self.batch_size,
432            compression: self.compression,
433            max_row_group_size: self.max_row_group_size,
434        })))
435    }
436
437    fn query_batch_session(
438        &mut self,
439        query: &CatalogQuery,
440        chunk_size: Option<usize>,
441    ) -> anyhow::Result<DataBatchQueryResult> {
442        ensure_latest_query(query)?;
443        macro_rules! batch_session {
444            ((Instrument, InstrumentAny, Instrument, Instrument, $instrument_prefix:literal), $(($variant:ident, $type:ident, $data:ident, $batch:ident, $prefix:literal)),+ $(,)?) => {
445                match &query.data_type {
446                    $(NautilusDataType::$variant => {
447                        let pages = self.query_typed_pages::<$type>(
448                            query.identifiers.clone(), query.start, query.end, query.where_clause.as_deref(), None,
449                            query.params.as_ref().and_then(|params| params.get_bool(QUERY_OPTIMIZE_FILE_LOADING)).unwrap_or(true),
450                        )?;
451                        Ok(Box::new(TypedDataBatchSession::new(pages, chunk_size)) as DataBatchQueryResult)
452                    },)+
453                    _ => match self.query_batch(query)? {
454                        DataBatch::Instrument(data) => Ok(Box::new(TypedDataBatchSession::from_vec(data.as_ref().to_vec(), chunk_size))),
455                        DataBatch::Custom(data) => Ok(Box::new(TypedDataBatchSession::from_vec(data.as_ref().to_vec(), chunk_size))),
456                        _ => anyhow::bail!("Unsupported Parquet query family"),
457                    },
458                }
459            };
460        }
461        nautilus_model::for_each_data_type!(batch_session)
462    }
463
464    fn reset_session(&mut self) {
465        self.clear_session_tables();
466    }
467
468    fn instruments(
469        &mut self,
470        query: &CatalogInstrumentQuery,
471    ) -> anyhow::Result<Vec<InstrumentAny>> {
472        let CatalogInstrumentQuery {
473            instrument_ids,
474            start,
475            end,
476            where_clause,
477            instrument_type,
478        } = query.clone();
479        let instrument_ids = instrument_ids.as_deref();
480        self.query_instruments_filtered_with_where_and_type(
481            instrument_ids,
482            start,
483            end,
484            where_clause.as_deref(),
485            instrument_type.as_ref(),
486        )
487    }
488
489    fn query_batch(&mut self, query: &CatalogQuery) -> anyhow::Result<DataBatch> {
490        ensure_latest_query(query)?;
491        let CatalogQuery {
492            data_type,
493            identifiers,
494            start,
495            end,
496            where_clause,
497            params,
498            instrument_type,
499            ..
500        } = query.clone();
501        let where_clause = where_clause.as_deref();
502        let optimize_file_loading = params
503            .as_ref()
504            .and_then(|params| params.get_bool(QUERY_OPTIMIZE_FILE_LOADING))
505            .unwrap_or(true);
506
507        match data_type {
508            NautilusDataType::Instrument => {
509                let data = self.query_instruments_filtered_with_where_and_type(
510                    identifiers.as_deref(),
511                    start,
512                    end,
513                    where_clause,
514                    instrument_type.as_ref(),
515                )?;
516                Ok(DataBatch::Instrument(
517                    filter_instrument_query_result(data, start, params.as_ref()).into(),
518                ))
519            }
520            NautilusDataType::Custom { type_name } => {
521                let data = self.query_custom_data_dynamic(
522                    &type_name,
523                    identifiers.as_deref(),
524                    start,
525                    end,
526                    where_clause,
527                    None,
528                    optimize_file_loading,
529                )?;
530                Ok(DataBatch::Custom(
531                    data.into_iter()
532                        .filter_map(|item| match item {
533                            Data::Custom(custom) => Some(custom),
534                            _ => None,
535                        })
536                        .collect::<Vec<_>>()
537                        .into(),
538                ))
539            }
540            #[cfg(feature = "defi")]
541            NautilusDataType::Defi => Err(anyhow::Error::from(
542                crate::errors::PersistenceError::unsupported("Parquet catalog DeFi data"),
543            )),
544            data_type => query_builtin_batch(
545                self,
546                &data_type,
547                identifiers,
548                start,
549                end,
550                where_clause,
551                optimize_file_loading,
552            )
553            .expect("built-in data type dispatch is exhaustive"),
554        }
555    }
556
557    fn query_identifiers(&mut self, query: &CatalogQuery) -> anyhow::Result<Vec<String>> {
558        ensure_latest_query(query)?;
559        let CatalogQuery {
560            data_type,
561            identifiers,
562            start,
563            end,
564            where_clause,
565            params,
566            instrument_type,
567            ..
568        } = query.clone();
569
570        if data_type == NautilusDataType::Instrument {
571            let mut identifiers = self
572                .query_instruments_filtered_with_where_and_type(
573                    identifiers.as_deref(),
574                    start,
575                    end,
576                    where_clause.as_deref(),
577                    instrument_type.as_ref(),
578                )?
579                .into_iter()
580                .map(|instrument| instrument.id().to_string())
581                .collect::<Vec<_>>();
582            identifiers.sort();
583            identifiers.dedup();
584            return Ok(identifiers);
585        }
586
587        let optimize_file_loading = params
588            .as_ref()
589            .and_then(|params| params.get_bool(QUERY_OPTIMIZE_FILE_LOADING))
590            .unwrap_or(true);
591        Self::query_identifiers(
592            self,
593            &CatalogDataType::Data(data_type),
594            identifiers,
595            start,
596            end,
597            where_clause.as_deref(),
598            optimize_file_loading,
599        )
600    }
601
602    fn query_display_record_batches(
603        &mut self,
604        query: &CatalogQuery,
605    ) -> anyhow::Result<Vec<RecordBatch>> {
606        ensure_latest_query(query)?;
607        let CatalogQuery {
608            data_type,
609            identifiers,
610            start,
611            end,
612            where_clause,
613            params,
614            instrument_type,
615            ..
616        } = query.clone();
617
618        if data_type == NautilusDataType::Instrument {
619            let instruments = self.query_instruments_filtered_with_where_and_type(
620                identifiers.as_deref(),
621                start,
622                end,
623                where_clause.as_deref(),
624                instrument_type.as_ref(),
625            )?;
626            return if instruments.is_empty() {
627                Ok(Vec::new())
628            } else {
629                Ok(vec![encode_instruments(&instruments)?])
630            };
631        }
632
633        let optimize_file_loading = params
634            .as_ref()
635            .and_then(|params| params.get_bool(QUERY_OPTIMIZE_FILE_LOADING))
636            .unwrap_or(true);
637        Self::query_display_record_batches(
638            self,
639            &data_type,
640            identifiers,
641            start,
642            end,
643            where_clause.as_deref(),
644            optimize_file_loading,
645        )
646    }
647
648    fn query_record_batches(
649        &mut self,
650        query: &CatalogRecordQuery,
651    ) -> anyhow::Result<Vec<RecordBatch>> {
652        anyhow::ensure!(
653            query.as_of == CatalogAsOf::Latest,
654            "Parquet catalog does not support historical queries"
655        );
656        let CatalogRecordQuery {
657            record_type,
658            identifier,
659            start,
660            end,
661            where_clause,
662            params,
663            ..
664        } = query.clone();
665        let optimize_file_loading = params
666            .as_ref()
667            .and_then(|params| params.get_bool(QUERY_OPTIMIZE_FILE_LOADING))
668            .unwrap_or(true);
669        Self::query_record_batches(
670            self,
671            &CatalogDataType::Record(record_type),
672            identifier,
673            start,
674            end,
675            where_clause.as_deref(),
676            optimize_file_loading,
677        )
678    }
679
680    fn query_record_display_batches(
681        &mut self,
682        query: &CatalogRecordQuery,
683    ) -> anyhow::Result<Vec<RecordBatch>> {
684        anyhow::ensure!(
685            query.as_of == CatalogAsOf::Latest,
686            "Parquet catalog does not support historical queries"
687        );
688        CatalogReader::query_record_batches(self, query)
689    }
690
691    fn query_metadata(&mut self, query: &CatalogQuery) -> anyhow::Result<Vec<CatalogMetadata>> {
692        ensure_latest_query(query)?;
693        let CatalogQuery {
694            data_type,
695            identifiers,
696            start,
697            end,
698            where_clause,
699            instrument_type,
700            ..
701        } = query.clone();
702        let data_type = match (data_type, instrument_type) {
703            (NautilusDataType::Instrument, Some(instrument_type)) => {
704                CatalogDataType::Instrument(instrument_type)
705            }
706            (data_type, _) => CatalogDataType::Data(data_type),
707        };
708
709        Self::query_metadata(
710            self,
711            &data_type,
712            identifiers,
713            start,
714            end,
715            where_clause.as_deref(),
716        )
717    }
718
719    fn get_missing_intervals_for_request(
720        &mut self,
721        start: UnixNanos,
722        end: UnixNanos,
723        data_type: NautilusDataType,
724        identifier: Option<&str>,
725    ) -> anyhow::Result<Vec<(u64, u64)>> {
726        match data_type {
727            NautilusDataType::Instrument => {
728                let identifiers = identifier.map(|value| vec![value.to_string()]);
729                let data = self.query_instruments_filtered(
730                    identifiers.as_deref(),
731                    Some(start),
732                    Some(end),
733                )?;
734                Ok(if data.is_empty() {
735                    vec![(start.as_u64(), end.as_u64())]
736                } else {
737                    Vec::new()
738                })
739            }
740            NautilusDataType::Custom { type_name } => {
741                if let Some(identifier) = identifier {
742                    let directory = self.make_path_custom_data(&type_name, Some(identifier))?;
743                    let intervals = self.get_directory_intervals(&directory)?;
744
745                    Ok(query_interval_diff(
746                        start.as_u64(),
747                        end.as_u64(),
748                        &intervals,
749                    ))
750                } else {
751                    Self::get_missing_intervals_for_request(
752                        self,
753                        start.as_u64(),
754                        end.as_u64(),
755                        &CatalogDataType::Data(NautilusDataType::Custom { type_name }),
756                        None,
757                    )
758                }
759            }
760            _ => Self::get_missing_intervals_for_request(
761                self,
762                start.as_u64(),
763                end.as_u64(),
764                &CatalogDataType::Data(data_type),
765                identifier,
766            ),
767        }
768    }
769
770    fn query_last_timestamp(
771        &mut self,
772        data_type: NautilusDataType,
773        identifier: Option<&str>,
774    ) -> anyhow::Result<Option<u64>> {
775        match data_type {
776            NautilusDataType::Instrument => {
777                let identifiers = identifier.map(|value| vec![value.to_string()]);
778                Ok(self
779                    .query_instruments(identifiers.as_deref())?
780                    .into_iter()
781                    .map(|instrument| HasTsInit::ts_init(&instrument).as_u64())
782                    .max())
783            }
784            NautilusDataType::Custom { type_name } => {
785                if let Some(identifier) = identifier {
786                    let directory = self.make_path_custom_data(&type_name, Some(identifier))?;
787                    let intervals = self.get_directory_intervals(&directory)?;
788
789                    Ok(intervals.into_iter().map(|(_, end)| end).max())
790                } else {
791                    Self::query_last_timestamp(
792                        self,
793                        &CatalogDataType::Data(NautilusDataType::Custom { type_name }),
794                        None,
795                    )
796                }
797            }
798            _ => Self::query_last_timestamp(self, &CatalogDataType::Data(data_type), identifier),
799        }
800    }
801}
802
803impl CatalogWriter for ParquetDataCatalog {
804    fn write_instruments(&mut self, instruments: &[InstrumentAny]) -> anyhow::Result<()> {
805        Self::write_instruments(self, instruments.to_vec()).map(|_| ())
806    }
807
808    fn write_data(
809        &mut self,
810        data: &[Data],
811        start: Option<UnixNanos>,
812        end: Option<UnixNanos>,
813        params: Option<Params>,
814    ) -> anyhow::Result<()> {
815        if data.is_empty() {
816            return Ok(());
817        }
818        let skip_disjoint_check = params
819            .as_ref()
820            .and_then(|params| params.get_bool(WRITE_SKIP_DISJOINT_CHECK));
821        self.write_data_enum(data, start, end, skip_disjoint_check)
822    }
823
824    fn write_data_batch(
825        &mut self,
826        batch: &DataBatch,
827        start: Option<UnixNanos>,
828        end: Option<UnixNanos>,
829        params: Option<Params>,
830    ) -> anyhow::Result<()> {
831        let skip_disjoint_check = params
832            .as_ref()
833            .and_then(|params| params.get_bool(WRITE_SKIP_DISJOINT_CHECK));
834
835        write_catalog_batch(self, batch, start, end, skip_disjoint_check)
836    }
837
838    fn write_records(
839        &mut self,
840        record_type: NautilusRecordType,
841        batches: &[RecordBatch],
842        params: Option<Params>,
843    ) -> anyhow::Result<()> {
844        let params = params.unwrap_or_default();
845        let identifier = params.get_str("identifier").map(str::to_owned);
846        self.write_record_batches(&record_type, identifier.as_deref(), batches, &params)
847    }
848
849    fn record_empty_coverage(
850        &mut self,
851        data_type: NautilusDataType,
852        identifier: Option<&str>,
853        start: UnixNanos,
854        end: UnixNanos,
855    ) -> anyhow::Result<()> {
856        match &data_type {
857            NautilusDataType::Instrument => {
858                anyhow::bail!(
859                    "Cannot record empty instrument coverage without a concrete instrument type"
860                )
861            }
862            NautilusDataType::Custom { type_name } => {
863                let directory = self.make_path_custom_data(type_name, identifier)?;
864                self.extend_file_name_in_directory(&directory, start, end)
865            }
866            _ => self.extend_file_name(&CatalogDataType::Data(data_type), identifier, start, end),
867        }
868    }
869}
870
871// Re-export public items from sibling modules so historical
872// `crate::backend::parquet::catalog::...` imports continue to resolve.
873pub use crate::backend::parquet::{
874    intervals::{are_intervals_contiguous, are_intervals_disjoint},
875    paths::{
876        CatalogPathPrefix, extract_identifier_from_path, extract_path_components,
877        extract_sql_safe_filename, local_to_object_store_path, make_local_path,
878        make_object_store_path, make_sql_safe_identifier, parse_filename_timestamps,
879        safe_directory_identifier, timestamps_to_filename, urisafe_instrument_id,
880    },
881};
882
883fn ensure_latest_query(query: &CatalogQuery) -> anyhow::Result<()> {
884    anyhow::ensure!(
885        query.as_of == CatalogAsOf::Latest,
886        "Parquet catalog does not support historical queries"
887    );
888    Ok(())
889}
nautilus_persistence/backend/parquet/catalog/mod.rs

nautilus_persistence/backend/parquet/catalog/
mod.rs
