Struct ParquetDataCatalog 

pub struct ParquetDataCatalog {
    pub base_path: String,
    pub original_uri: String,
    pub object_store: Arc<dyn ObjectStore>,
    pub session: DataBackendSession,
    pub batch_size: usize,
    pub compression: Compression,
    pub max_row_group_size: usize,
}

A high-performance data catalog for storing and retrieving financial market data using Apache Parquet format.

The ParquetDataCatalog provides a solution for managing large volumes of financial market data with efficient storage, querying, and consolidation capabilities. It supports various object store backends including local filesystems, AWS S3, and other cloud storage providers.

Features

• Efficient Storage: Uses Apache Parquet format with configurable compression.
• Object Store Backend: Supports multiple storage backends through the object_store crate.
• Time-based Organization: Organizes data by timestamp ranges for optimal query performance.
• Data Validation: Ensures timestamp ordering and interval consistency.
• Consolidation: Merges multiple files to reduce storage overhead and improve query speed.
• Type Safety: Strongly typed data handling with compile-time guarantees.

Data Organization

Data is organized hierarchically by data type and instrument:

• data/{data_type}/{instrument_id}/{start_ts}-{end_ts}.parquet.
• Files are named with their timestamp ranges for efficient range queries.
• Intervals are validated to be disjoint to prevent data overlap.

Performance Considerations

• Batch Size: Controls memory usage during data processing.
• Compression: SNAPPY compression provides good balance of speed and size.
• Row Group Size: Affects query performance and memory usage.
• File Consolidation: Reduces the number of files for better query performance.

Fields

base_path: String

The base path for data storage within the object store.

original_uri: String

The original URI provided when creating the catalog.

object_store: Arc<dyn ObjectStore>

The object store backend for data persistence.

session: DataBackendSession

The DataFusion session for query execution.

batch_size: usize

The number of records to process in each batch.

compression: Compression

The compression algorithm used for Parquet files.

max_row_group_size: usize

The maximum number of rows in each Parquet row group.

Implementations

impl ParquetDataCatalog

pub fn get_missing_intervals_for_request( &self, start: u64, end: u64, data_type: &CatalogDataType, identifier: Option<&str>, ) -> Result<Vec<(u64, u64)>>

Finds the missing time intervals for a specific data type and instrument ID.

This method compares a requested time range against the existing data coverage and returns the gaps that need to be filled. This is useful for determining what data needs to be fetched or backfilled.

Parameters

• start: Start timestamp of the requested range (Unix nanoseconds).
• end: End timestamp of the requested range (Unix nanoseconds).
• data_type: The stored family to inspect.
• instrument_id: Optional instrument ID to target a specific instrument’s data.

Returns

Returns a vector of (start, end) tuples representing the missing intervals, or an error if the operation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• Interval retrieval fails.
• Gap calculation fails.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Find missing intervals for quote data
let missing = catalog.get_missing_intervals_for_request(
    1609459200000000000, // start
    1609545600000000000, // end
    &NautilusDataType::QuoteTick.into(),
    Some("BTCUSD"),
)?;

for (start, end) in missing {
    println!("Missing data from {} to {}", start, end);
}

pub fn query_first_timestamp( &self, data_type: &CatalogDataType, identifier: Option<&str>, ) -> Result<Option<u64>>

Gets the first (earliest) timestamp for a specific data type and identifier.

This method finds the earliest timestamp covered by existing data files for the specified data type and identifier. This is useful for determining the oldest data available or for incremental data updates.

Parameters

• data_type: The stored family to inspect.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).

Returns

Returns Some(timestamp) if data exists, None if no data is found, or an error if the operation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• Interval retrieval fails.

Note

Unlike the Python implementation, this method does not check subclasses of the data type. The Python version checks [data_cls, *data_cls.__subclasses__()] to handle cases where subclasses might use different directory names. Since Rust works with string names rather than types, subclass checking is not possible. In practice, most subclasses map to the same directory name via class_to_filename, so this difference is typically not significant.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Get the first timestamp for quote data
if let Some(first_ts) =
    catalog.query_first_timestamp(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?
{
    println!("First quote timestamp: {}", first_ts);
} else {
    println!("No quote data found");
}

pub fn query_last_timestamp( &self, data_type: &CatalogDataType, identifier: Option<&str>, ) -> Result<Option<u64>>

Gets the last (most recent) timestamp for a specific data type and identifier.

This method finds the latest timestamp covered by existing data files for the specified data type and identifier. This is useful for determining the most recent data available or for incremental data updates.

Parameters

• data_type: The stored family to inspect.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).

Returns

Returns Some(timestamp) if data exists, None if no data is found, or an error if the operation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• Interval retrieval fails.

Note

Unlike the Python implementation, this method does not check subclasses of the data type. The Python version checks [data_cls, *data_cls.__subclasses__()] to handle cases where subclasses might use different directory names. Since Rust works with string names rather than types, subclass checking is not possible. In practice, most subclasses map to the same directory name via class_to_filename, so this difference is typically not significant.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Get the last timestamp for quote data
if let Some(last_ts) =
    catalog.query_last_timestamp(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?
{
    println!("Last quote timestamp: {}", last_ts);
} else {
    println!("No quote data found");
}

pub fn get_intervals( &self, data_type: &CatalogDataType, identifier: Option<&str>, ) -> Result<Vec<(u64, u64)>>

Gets the time intervals covered by Parquet files for a specific data type and identifier.

This method returns all time intervals covered by existing data files for the specified data type and identifier. The intervals are sorted by start time and represent the complete data coverage available.

Parameters

• data_type: The stored family to inspect.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).

Returns

Returns a vector of (start, end) tuples representing the covered intervals, sorted by start time, or an error if the operation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• Directory listing fails.
• Filename parsing fails.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Get all intervals for quote data
let intervals = catalog.get_intervals(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?;
for (start, end) in intervals {
    println!("Data available from {} to {}", start, end);
}

pub fn get_directory_intervals( &self, directory: &str, ) -> Result<Vec<(u64, u64)>>

Gets the time intervals covered by Parquet files in a specific directory.

This method scans a directory for Parquet files and extracts the timestamp ranges from their filenames. It’s used internally by other methods to determine data coverage and is essential for interval-based operations like gap detection and consolidation.

Parameters

• directory: The directory path to scan for Parquet files.

Returns

Returns a vector of (start, end) tuples representing the time intervals covered by files in the directory, sorted by start timestamp. Returns an empty vector if the directory doesn’t exist or contains no valid Parquet files.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Notes

• Only files with valid timestamp-based filenames are included.
• Files with unparsable names are silently ignored.
• The method works with both local and remote object stores.
• Results are automatically sorted by start timestamp.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let intervals = catalog.get_directory_intervals("data/quotes/EURUSD")?;

for (start, end) in intervals {
    println!("File covers {} to {}", start, end);
}

impl ParquetDataCatalog

pub fn query<T>( &mut self, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, files: Option<Vec<String>>, optimize_file_loading: bool, ) -> Result<QueryResult>

where T: DecodeTypedFromRecordBatch + HasCatalogDataType + HasTsInit + Into<Data> + Send + 'static,

Queries one data family through the existing row iterator API.

pub fn query_instruments( &self, instrument_ids: Option<&[String]>, ) -> Result<Vec<InstrumentAny>>

Queries instruments from the catalog.

Instruments are stored under v1-compatible concrete instrument type folders: data/{instrument_type}/{instrument_id}/.

Parameters

• instrument_ids: Optional list of instrument IDs to filter by. If None, queries all instruments.

Returns

Returns a vector of InstrumentAny instances, or an error if the operation fails.

Errors

Returns an error if:

• File discovery fails.
• File reading fails.
• Data deserialization fails.

Examples

use nautilus_model::instruments::InstrumentAny;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Query all instruments
let instruments = catalog.query_instruments(None)?;

// Query specific instruments
let instrument_ids = vec!["EUR/USD.SIM".to_string()];
let instruments = catalog.query_instruments(Some(&instrument_ids))?;

pub fn query_instruments_filtered( &self, instrument_ids: Option<&[String]>, _start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<InstrumentAny>>

Queries instruments from the catalog with optional timestamp filtering.

This reads all matching parquet files under data/{instrument_type}/{instrument_id}/, decodes the records back to InstrumentAny, and filters them by ts_init when a range is provided.

pub fn query_instruments_filtered_with_where( &mut self, instrument_ids: Option<&[String]>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, ) -> Result<Vec<InstrumentAny>>

Queries instruments from the catalog with optional timestamp and SQL filtering.

When where_clause is provided, the predicate is applied through DataFusion before instrument records are decoded.

pub fn query_instruments_filtered_with_where_and_type( &mut self, instrument_ids: Option<&[String]>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, instrument_type: Option<&NautilusInstrumentType>, ) -> Result<Vec<InstrumentAny>>

pub fn query_typed_data<T>( &mut self, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, files: Option<Vec<String>>, optimize_file_loading: bool, ) -> Result<Vec<T>>

where T: DecodeTypedFromRecordBatch + HasCatalogDataType + HasTsInit,

Queries typed data from the catalog and returns results as a strongly-typed vector.

This is a convenience method that wraps the generic query method and automatically collects and converts the results into a vector of the specific data type. It handles the type conversion from the generic [Data] enum to the concrete type T.

Type Parameters

• T: The specific data type to query and return. Must implement required traits for deserialization, cataloging, and conversion from the [Data] enum.

Parameters

• identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). If None, queries all identifiers. For bars, partial matching is supported (e.g., “EUR/USD.SIM” will match “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).
• start: Optional start timestamp for filtering (inclusive). If None, queries from the beginning.
• end: Optional end timestamp for filtering (inclusive). If None, queries to the end.
• where_clause: Optional SQL WHERE clause for additional filtering. Use standard SQL syntax with column names matching the Parquet schema (e.g., “bid_price > 1.2000”, “volume > 1000”).

Returns

Returns a vector of the specific data type T, sorted by timestamp. The vector will be empty if no data matches the query criteria.

Errors

Returns an error if:

• The underlying query execution fails.
• Data type conversion fails.
• Object store access fails.
• Invalid WHERE clause syntax is provided.

Performance Considerations

• Use specific instrument IDs and time ranges to minimize data scanning.
• WHERE clauses are pushed down to Parquet readers when possible.
• Results are automatically sorted by timestamp during collection.
• Memory usage scales with the amount of data returned.

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::{Bar, QuoteTick, TradeTick};
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Query all quotes for a specific instrument
let quotes: Vec<QuoteTick> = catalog.query_typed_data(
    Some(vec!["EUR/USD.SIM".to_string()]),
    None,
    None,
    None,
    None,
    true,
)?;

// Query trades within a specific time range
let trades: Vec<TradeTick> = catalog.query_typed_data(
    Some(vec!["BTC/USD.SIM".to_string()]),
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
    None,
    None,
    true,
)?;

// Query bars with volume filter (using instrument_id - partial match for bar_type)
let bars: Vec<Bar> = catalog.query_typed_data(
    Some(vec!["AAPL.NASDAQ".to_string()]),
    None,
    None,
    Some("volume > 1000000"),
    None,
    true,
)?;

// Query bars with specific bar_type
let bars: Vec<Bar> = catalog.query_typed_data(
    Some(vec!["AAPL.NASDAQ-1-MINUTE-LAST-EXTERNAL".to_string()]),
    None,
    None,
    None,
    None,
    true,
)?;

// Query multiple instruments with price filter
let quotes: Vec<QuoteTick> = catalog.query_typed_data(
    Some(vec!["EUR/USD.SIM".to_string(), "GBP/USD.SIM".to_string()]),
    None,
    None,
    Some("bid_price > 1.2000 AND ask_price < 1.3000"),
    None,
    true,
)?;

pub fn query_typed<T>( &mut self, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, files: Option<Vec<String>>, optimize_file_loading: bool, ) -> Result<Vec<T>>

where T: DecodeTypedFromRecordBatch + HasCatalogDataType + HasTsInit,

Queries typed records that are not represented by the [Data] enum.

pub fn query_record_batches( &mut self, data_type: &CatalogDataType, identifier: Option<String>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, optimize_file_loading: bool, ) -> Result<Vec<RecordBatch>>

Queries raw catalog Arrow record batches for any supported record table.

Errors

Returns an error if file discovery or DataFusion query execution fails.

pub fn query_display_record_batches( &mut self, data_type: &NautilusDataType, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, optimize_file_loading: bool, ) -> Result<Vec<RecordBatch>>

Queries raw catalog batches and converts them to display-friendly Arrow batches.

Errors

Returns an error if file discovery, DataFusion query execution, or catalog display conversion fails.

pub fn query_identifiers( &mut self, data_type: &CatalogDataType, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, _optimize_file_loading: bool, ) -> Result<Vec<String>>

Queries concrete catalog identifiers for matching data rows.

pub fn query_custom_data_dynamic( &mut self, type_name: &str, identifiers: Option<&[String]>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, files: Option<Vec<String>>, _optimize_file_loading: bool, ) -> Result<Vec<Data>>

Queries custom data dynamically by type name.

This method allows querying custom data types without compile-time knowledge of the type. It uses dynamic schema decoding based on the type name stored in metadata.

Parameters

• type_name: The name of the custom data type to query.
• identifiers: Optional list of instrument identifiers to filter by.
• start: Optional start timestamp for filtering.
• end: Optional end timestamp for filtering.
• where_clause: Optional SQL WHERE clause for additional filtering.
• files: Optional list of specific files to query.
• _optimize_file_loading: Whether to optimize file loading (currently unused).

Returns

Returns a vector of Data enum variants containing the custom data.

Errors

Returns an error if:

• File discovery fails.
• Data decoding fails.
• Query execution fails.

pub fn query_files( &self, data_type: &CatalogDataType, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<String>>

Queries all Parquet files for a specific data type and optional instrument IDs.

This method finds all Parquet files that match the specified criteria and returns their full URIs. The files are filtered by data type, instrument IDs (if provided), and timestamp range (if provided).

Parameters

• data_type: The stored family to read.
• identifiers: Optional list of identifiers to filter by. Can be instrument_id strings (e.g., “EUR/USD.SIM”) or bar_type strings (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). For bars, partial matching is supported.
• start: Optional start timestamp to filter files by their time range.
• end: Optional end timestamp to filter files by their time range.

Returns

Returns a vector of file URI strings that match the query criteria, or an error if the query fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• Object store listing operations fail.
• URI reconstruction fails.

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Query all quote files
let files = catalog.query_files(&NautilusDataType::QuoteTick.into(), None, None, None)?;

// Query trade files for specific instruments within a time range
let files = catalog.query_files(
    &NautilusDataType::TradeTick.into(),
    Some(vec!["BTC/USD.SIM".to_string(), "ETH/USD.SIM".to_string()]),
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
)?;

pub fn quote_ticks( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<QuoteTick>>

pub fn trade_ticks( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<TradeTick>>

Queries trade tick data for the specified instrument(s) and time range.

pub fn bars( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<Bar>>

Queries bar data for the specified instrument(s) and time range.

pub fn order_book_deltas( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<OrderBookDelta>>

Queries order book delta data for the specified instrument(s) and time range.

pub fn order_book_depths( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<OrderBookDepth>>

Queries order book depth data for the specified instrument(s) and time range.

pub fn funding_rates( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<FundingRateUpdate>>

Queries funding rate updates for the specified instrument(s) and time range.

pub fn instrument_closes( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<InstrumentClose>>

Queries instrument close data for the specified instrument(s) and time range.

pub fn option_greeks( &mut self, instrument_ids: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<OptionGreeks>>

Queries option greeks data for the specified instrument(s) and time range.

pub fn instruments( &self, instrument_ids: Option<&[String]>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<InstrumentAny>>

Queries any instrument data for the specified instrument(s) and time range.

pub fn get_file_list_from_data_cls( &self, data_type: &CatalogDataType, ) -> Result<Vec<String>>

Retrieves a list of file paths for a given data type.

This method constructs a path pattern to find all parquet files associated with the specified data type in the catalog’s directory structure.

Parameters

• data_type: The stored family to read.

Returns

Returns a vector of file paths matching the data type, or an error if the operation fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let files = catalog.get_file_list_from_data_cls(&NautilusDataType::QuoteTick.into())?;

for file in files {
    println!("Found file: {}", file);
}

pub fn filter_files( &self, data_type: &CatalogDataType, file_paths: Vec<String>, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<String>>

Filters a list of file paths based on identifiers and time range.

This method filters the provided file paths by:

1. Matching identifiers (exact match for instruments, prefix match for bars)
2. Intersecting with the specified time range

Parameters

• data_type: The stored family to read.
• file_paths: List of file paths to filter.
• identifiers: Optional list of identifiers to match against file paths.
• start: Optional start timestamp for filtering.
• end: Optional end timestamp for filtering.

Returns

Returns a filtered vector of file paths that match the criteria.

Notes

For Bar data types, if exact identifier matching fails, the function attempts partial matching by checking if the file’s identifier starts with the provided identifier followed by a dash (to match bar type patterns).

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let all_files = catalog.get_file_list_from_data_cls(&NautilusDataType::QuoteTick.into())?;

let filtered = catalog.filter_files(
    &NautilusDataType::QuoteTick.into(),
    all_files,
    Some(vec!["EUR/USD.SIM".to_string()]),
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
)?;

impl ParquetDataCatalog

pub fn extend_file_name( &self, data_type: &CatalogDataType, identifier: Option<&str>, start: UnixNanos, end: UnixNanos, ) -> Result<()>

Extends the timestamp range of an existing Parquet file by renaming it.

This method finds an existing file that is adjacent to the specified time range and renames it to include the new range. This is useful when appending data that extends the time coverage of existing files. The proposed extension is validated against the other files before renaming, so a rejected extension leaves existing files unchanged.

If no file is adjacent to the specified range, this method does nothing and returns Ok(()) after confirming the existing intervals are disjoint.

Parameters

• data_type: The stored family to target.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).
• start: Start timestamp of the new range to extend to.
• end: End timestamp of the new range to extend to.

Returns

Returns Ok(()) on success, or an error if the operation fails.

Errors

Returns an error if:

• The range is reversed (start is after end).
• The directory path cannot be constructed.
• The proposed extension would overlap another file.
• The existing intervals are already overlapping.
• File rename operations fail.

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Extend a file's range backwards or forwards
catalog.extend_file_name(
    &NautilusDataType::QuoteTick.into(),
    Some("BTC/USD.SIM"),
    UnixNanos::from(1609459200000000000),
    UnixNanos::from(1609545600000000000),
)?;

pub fn list_parquet_files(&self, directory: &str) -> Result<Vec<String>>

Lists all Parquet files in a specified directory.

This method scans a directory and returns the full paths of all files with the .parquet extension. It works with both local filesystems and remote object stores, making it suitable for various storage backends.

Parameters

• directory: The directory path to scan for Parquet files.

Returns

Returns a vector of full file paths (as strings) for all Parquet files found in the directory. The paths are relative to the object store root and suitable for use with object store operations. Returns an empty vector if the directory doesn’t exist or contains no Parquet files.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.
• Network issues occur (for remote object stores).

Notes

• Only files ending with .parquet are included.
• Subdirectories are not recursively scanned.
• File paths are returned in the order provided by the object store.
• Works with all supported object store backends (local, S3, GCS, Azure, etc.).

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let files = catalog.list_parquet_files("data/quotes/EURUSD")?;

for file in files {
    println!("Found Parquet file: {}", file);
}

pub fn list_instruments( &self, data_type: &CatalogDataType, ) -> Result<Vec<String>>

Lists all instrument identifiers for a specific data type.

This method scans the data directory for a given data type and extracts all unique instrument identifiers from the directory structure.

Parameters

• data_type: The stored family to target.

Returns

Returns a vector of instrument identifier strings.

Errors

Returns an error if directory listing fails.

pub fn list_parquet_files_with_criteria( &self, data_type: &CatalogDataType, identifiers: Option<&[String]>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<String>>

Lists Parquet files matching specific criteria (data type, identifiers, time range).

This method finds all Parquet files that match the specified criteria by filtering files based on their directory structure and filename timestamps.

Parameters

• data_type: The stored family to target.
• identifiers: Optional list of identifiers to filter by.
• start: Optional start timestamp to filter files by their time range.
• end: Optional end timestamp to filter files by their time range.

Returns

Returns a vector of file paths that match the criteria.

Errors

Returns an error if directory listing or file filtering fails.

pub fn reconstruct_full_uri(&self, path_str: &str) -> String

Helper method to reconstruct full URI for remote object store paths

pub fn is_remote_uri(&self) -> bool

Helper method to check if the original URI uses a remote object store scheme

pub fn make_path( &self, type_name: &str, identifier: Option<&str>, ) -> Result<String>

Constructs a directory path for storing data of a specific type and instrument.

This method builds the hierarchical directory structure used by the catalog to organize data by type and instrument. The path follows the pattern: {base_path}/data/{type_name}/{instrument_id}. Instrument IDs are automatically converted to URI-safe format by removing forward slashes.

Parameters

• type_name: The data type directory name (e.g., “quotes”, “trades”, “bars”).
• identifier: Optional identifier. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). If provided, creates a subdirectory for the identifier. If None, returns the path to the data type directory.

Returns

Returns the constructed directory path as a string, or an error if path construction fails.

Errors

Returns an error if:

• The instrument ID contains invalid characters that cannot be made URI-safe.
• Path construction fails due to system limitations.

Path Structure

• Without identifier: {base_path}/data/{type_name}.
• With identifier: {base_path}/data/{type_name}/{safe_identifier}.
• If base_path is empty: data/{type_name}[/{safe_identifier}].

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Path for all quote data
let quotes_path = catalog.make_path("quotes", None)?;
// Returns: "/base/path/data/quotes"

// Path for specific instrument quotes
let eurusd_quotes = catalog.make_path("quotes", Some("EUR/USD"))?;
// Returns: "/base/path/data/quotes/EURUSD" (slash removed)

// Path for bar data with complex instrument ID
let bars_path = catalog.make_path("bars", Some("BTC/USD-1H"))?;
// Returns: "/base/path/data/bars/BTCUSD-1H"

pub fn make_path_custom_data( &self, type_name: &str, identifier: Option<&str>, ) -> Result<String>

Builds the directory path for custom data: data/custom/{type_name}[/{identifier}].

pub fn to_object_path(&self, path: &str) -> Result<ObjectPath>

Converts a catalog path string to an [ObjectPath] for object store operations.

This method handles the conversion between catalog-relative paths and object store paths, taking into account the catalog’s base path configuration. It automatically preserves the base path prefix for remote catalogs and strips it for local catalog paths.

Parameters

• path: The catalog path string to convert. Can be absolute or relative.

Returns

Returns an [ObjectPath] suitable for use with object store operations.

Path Handling

• If base_path is empty, the path is used as-is.
• If base_path is set for a remote catalog, it’s preserved or prepended.
• If base_path is set for a local catalog, it’s stripped from the path if present.
• Trailing slashes and backslashes are automatically handled.
• The resulting path is relative to the object store root.
• All paths are normalized to use forward slashes (object store convention).

Errors

Returns an error for remote catalogs when path is a full URI whose scheme/host does not match the catalog’s own root (cross-bucket misuse). Without this guard the caller could silently write to or read from the wrong bucket.

Examples

Local catalog paths (absolute or relative) strip the catalog’s base directory:

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
let object_path = catalog.to_object_path("/base/data/quotes/file.parquet")?;
// ObjectPath("data/quotes/file.parquet")

Remote catalog paths (relative or full URI) preserve or prepend the base prefix:

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
let object_path = catalog.to_object_path("data/trades/file.parquet")?;
// ObjectPath("base/data/trades/file.parquet")

pub fn to_object_path_parsed(&self, path: &str) -> Result<ObjectPath>

Converts a path string to [ObjectPath] using parse (no percent-encoding).

Use this for paths that were returned by the object store (e.g. from list()), which may already be percent-encoded. Using Self::to_object_path (which uses Path::from) on such paths would double-encode (e.g. %5E -> %255E).

Errors

Returns an error for the same cross-bucket case as Self::to_object_path, or when the resulting string fails [ObjectPath::parse].

pub fn move_file( &self, old_path: &ObjectPath, new_path: &ObjectPath, ) -> Result<()>

Helper method to move a file using object store rename operation

pub fn execute_async<C, F, R>(&self, create_future: C) -> Result<R>

where C: FnOnce() -> F + Send, F: Future<Output = Result<R>>, R: Send,

Helper method to execute async operations with a runtime

pub fn list_directory_stems(&self, subdirectory: &str) -> Result<Vec<String>>

Lists directory stems (directory names without path) in a subdirectory.

This method scans a subdirectory and returns the names of all immediate subdirectories. It’s used to list data types, backtest runs, and live runs.

Parameters

• subdirectory: The subdirectory path to scan (e.g., “data”, “backtest”, “live”).

Returns

Returns a vector of directory names (stems) found in the subdirectory, or an error if the operation fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// List all data types
let data_types = catalog.list_directory_stems("data")?;
for data_type in data_types {
    println!("Found data type: {}", data_type);
}

pub fn list_data_types(&self) -> Result<Vec<String>>

Lists all data types available in the catalog.

This method returns the names of all data type directories in the catalog. Data types correspond to different kinds of market data (e.g., “quotes”, “trades”, “bars”).

Returns

Returns a vector of data type names, or an error if the operation fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// List all data types
let data_types = catalog.list_data_types()?;
for data_type in data_types {
    println!("Available data type: {}", data_type);
}

pub fn list_backtest_runs(&self) -> Result<Vec<String>>

Lists all backtest run IDs available in the catalog.

This method returns the names of all backtest run directories in the catalog. Each backtest run corresponds to a specific backtest execution instance.

Returns

Returns a vector of backtest run IDs, or an error if the operation fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// List all backtest runs
let runs = catalog.list_backtest_runs()?;
for run_id in runs {
    println!("Backtest run: {}", run_id);
}

pub fn list_live_runs(&self) -> Result<Vec<String>>

Lists all live run IDs available in the catalog.

This method returns the names of all live run directories in the catalog. Each live run corresponds to a specific live trading execution instance.

Returns

Returns a vector of live run IDs, or an error if the operation fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory access is denied.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// List all live runs
let runs = catalog.list_live_runs()?;
for run_id in runs {
    println!("Live run: {}", run_id);
}

impl ParquetDataCatalog

pub fn write_data_enum( &self, data: &[Data], start: Option<UnixNanos>, end: Option<UnixNanos>, skip_disjoint_check: Option<bool>, ) -> Result<()>

Writes mixed data types to the catalog by separating them into type-specific collections.

This method takes a heterogeneous collection of market data and separates it by type, then writes each type to its appropriate location in the catalog. This is useful when processing mixed data streams or bulk data imports.

Parameters

• data: A vector of mixed [Data] enum variants.
• start: Optional start timestamp to override the data’s natural range.
• end: Optional end timestamp to override the data’s natural range.

Notes

• Data is automatically sorted by type before writing.
• Each data type is written to its own directory structure.
• Instrument data handling is not yet implemented (TODO).

Examples

use nautilus_model::data::Data;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let mixed_data: Vec<Data> = vec![/* mixed data types */];

catalog.write_data_enum(&mixed_data, None, None, None)?;

pub fn write_record_batches( &mut self, record_type: &NautilusRecordType, identifier: Option<&str>, batches: &[RecordBatch], params: &Params, ) -> Result<()>

Writes Arrow batches into catalog under a record type optional identifier.

Errors

Returns error if batches do not contain ts_init or cannot be persisted.

pub fn write_to_parquet<T>( &self, data: &[T], start: Option<UnixNanos>, end: Option<UnixNanos>, skip_disjoint_check: Option<bool>, ) -> Result<PathBuf>

where T: HasTsInit + EncodeToRecordBatch + HasCatalogDataType,

Writes typed data to a Parquet file in the catalog.

This is the core method for persisting market data to the catalog. It handles data validation, batching, compression, and ensures proper file organization with timestamp-based naming.

Type Parameters

• T: The data type to write, must implement required traits for serialization and cataloging.

Parameters

• data: Vector of data records to write (must be in ascending timestamp order).
• start: Optional start timestamp to override the natural data range.
• end: Optional end timestamp to override the natural data range.

Returns

Returns the PathBuf of the created file, or an empty path if no data was provided. If the target file already exists, returns the path without writing (skips write).

Errors

Returns an error if:

• Data serialization to Arrow record batches fails.
• Object store write operations fail.
• File path construction fails.
• Writing would create non-disjoint timestamp intervals.

Panics

Panics if:

• Data timestamps are not in ascending order.
• Record batches are empty after conversion.
• Required metadata is missing from the schema.

Examples

use nautilus_model::data::QuoteTick;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let quotes: Vec<QuoteTick> = vec![/* quote data */];

let path = catalog.write_to_parquet(&quotes, None, None, None)?;
println!("Data written to: {:?}", path);

pub fn write_custom_data_batch<D>( &self, data: D, start: Option<UnixNanos>, end: Option<UnixNanos>, skip_disjoint_check: Option<bool>, ) -> Result<PathBuf>

where D: AsRef<[CustomData]>,

Writes custom data to a Parquet file in the catalog.

This method handles writing custom data types that implement CustomDataTrait. Custom data is organized by type name in a custom/{type_name}/ directory structure.

Parameters

• data: Vector of custom data items to write (must be in ascending timestamp order).
• start: Optional start timestamp to override the natural data range.
• end: Optional end timestamp to override the natural data range.
• skip_disjoint_check: Whether to skip interval disjointness validation.

Returns

Returns the PathBuf of the created file, or an empty path if no data was provided.

Errors

Returns an error if:

• The registered Arrow schema omits ts_init or uses incompatible timestamp types.
• Data serialization to Arrow record batches fails.
• Object store write operations fail.
• File path construction fails.
• Writing would create non-disjoint timestamp intervals (unless skipped).

pub fn write_instruments( &self, instruments: Vec<InstrumentAny>, ) -> Result<Vec<PathBuf>>

Writes instruments to Parquet files in the catalog.

Instruments are stored under their instrument ID directory using timestamp-ranged file names, allowing multiple historical versions of the same instrument to be appended over time: data/instruments/{instrument_id}/{start_ts}-{end_ts}.parquet

Parameters

• instruments: Vector of instruments to write.

Returns

Returns a vector of paths to the created files.

Errors

Returns an error if:

• Data serialization fails.
• Object store write operations fail.
• File path construction fails.

Examples

use nautilus_model::instruments::InstrumentAny;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let instruments: Vec<InstrumentAny> = vec![/* instruments */];

let paths = catalog.write_instruments(instruments)?;

pub fn write_to_json<T>( &self, data: Vec<T>, path: Option<PathBuf>, write_metadata: bool, ) -> Result<PathBuf>

where T: HasTsInit + Serialize + HasCatalogDataType + EncodeToRecordBatch,

Writes typed data to a JSON file in the catalog.

This method provides an alternative to Parquet format for data export and debugging. JSON files are human-readable but less efficient for large datasets.

Type Parameters

• T: The data type to write, must implement serialization and cataloging traits.

Parameters

• data: Vector of data records to write (must be in ascending timestamp order).
• path: Optional custom directory path (defaults to catalog’s standard structure).
• write_metadata: Whether to write a separate metadata file alongside the data.

Returns

Returns the PathBuf of the created JSON file.

Errors

Returns an error if:

• JSON serialization fails.
• Object store write operations fail.
• File path construction fails.

Panics

Panics if data timestamps are not in ascending order.

Examples

use std::path::PathBuf;
use nautilus_model::data::TradeTick;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);
let trades: Vec<TradeTick> = vec![/* trade data */];

let path = catalog.write_to_json(
    trades,
    Some(PathBuf::from("/custom/path")),
    true  // write metadata
)?;

pub fn check_ascending_timestamps<T: HasTsInit>( data: &[T], type_name: &str, ) -> Result<()>

Validates that data timestamps are in ascending order.

Parameters

• data: Slice of data records to validate.
• type_name: Name of the data type for error messages.

pub fn data_to_record_batches<T>(&self, data: &[T]) -> Result<Vec<RecordBatch>>

where T: HasTsInit + EncodeToRecordBatch,

Converts data into Arrow record batches for Parquet serialization.

This method chunks the data according to the configured batch size and converts each chunk into an Arrow record batch with appropriate metadata.

Type Parameters

• T: The data type to convert, must implement required encoding traits.

Parameters

• data: Vector of data records to convert.

Returns

Returns a vector of Arrow RecordBatch instances ready for Parquet serialization.

Errors

Returns an error if record batch encoding fails for any chunk.

impl ParquetDataCatalog

pub fn new( base_path: &Path, storage_options: Option<AHashMap<String, String>>, batch_size: Option<usize>, compression: Option<Compression>, max_row_group_size: Option<usize>, ) -> Self

Creates a new ParquetDataCatalog instance from a local file path.

This is a convenience constructor that converts a local path to a URI format and delegates to Self::from_uri.

Parameters

• base_path: The base directory path for data storage.
• storage_options: Optional HashMap containing storage-specific configuration options.
• batch_size: Number of records to process in each batch (default: 5000).
• compression: Parquet compression algorithm (default: SNAPPY).
• max_row_group_size: Maximum rows per Parquet row group (default: 131,072).

Panics

Panics if the path cannot be converted to a valid URI or if the object store cannot be created from the path.

Examples

use std::path::Path;

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let catalog = ParquetDataCatalog::new(
    Path::new("/tmp/nautilus_data"),
    None,       // no storage options
    Some(1000), // smaller batch size
    None,       // default compression
    None,       // default row group size
);

pub fn from_uri( uri: &str, storage_options: Option<AHashMap<String, String>>, batch_size: Option<usize>, compression: Option<Compression>, max_row_group_size: Option<usize>, ) -> Result<Self>

Creates a new ParquetDataCatalog instance from a URI with optional storage options.

Supports various URI schemes including local file paths and multiple cloud storage backends supported by the object_store crate.

Supported URI Schemes

• AWS S3: s3://bucket/path.
• Google Cloud Storage: gs://bucket/path or gcs://bucket/path.
• Azure Blob Storage: az://container/path or abfs://container@account.dfs.core.windows.net/path.
• HTTP/WebDAV: http:// or https://.
• Local files: file://path or plain paths.

Parameters

• uri: The URI for the data storage location.
• storage_options: Optional HashMap containing storage-specific configuration options:
  • For S3: endpoint_url, region, access_key_id, secret_access_key, session_token, etc.
  • For GCS: service_account_path, service_account_key, project_id, etc.
  • For Azure: account_name, account_key, sas_token, etc.
• batch_size: Number of records to process in each batch (default: 5000).
• compression: Parquet compression algorithm (default: SNAPPY).
• max_row_group_size: Maximum rows per Parquet row group (default: 131,072).

Errors

Returns an error if:

• The URI format is invalid or unsupported.
• The object store cannot be created or accessed.
• Authentication fails for cloud storage backends.

Examples

use ahash::AHashMap;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

// Local filesystem
let local_catalog = ParquetDataCatalog::from_uri("/tmp/nautilus_data", None, None, None, None)?;

// S3 bucket
let s3_catalog =
    ParquetDataCatalog::from_uri("s3://my-bucket/nautilus-data", None, None, None, None)?;

// Google Cloud Storage
let gcs_catalog =
    ParquetDataCatalog::from_uri("gs://my-bucket/nautilus-data", None, None, None, None)?;

// Azure Blob Storage
let azure_catalog =
    ParquetDataCatalog::from_uri("az://container/nautilus-data", None, None, None, None)?;

// S3 with custom endpoint and credentials
let mut storage_options = AHashMap::new();
storage_options.insert(
    "endpoint_url".to_string(),
    "https://my-s3-endpoint.com".to_string(),
);
storage_options.insert("access_key_id".to_string(), "my-key".to_string());
storage_options.insert("secret_access_key".to_string(), "my-secret".to_string());

let custom_s3_catalog = ParquetDataCatalog::from_uri(
    "s3://my-bucket/nautilus-data",
    Some(storage_options),
    None,
    None,
    None,
)?;

pub fn get_base_path(&self) -> String

Returns the base path of the catalog for testing purposes.

impl ParquetDataCatalog

pub fn consolidate_catalog( &self, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: Option<bool>, deduplicate: Option<bool>, ) -> Result<()>

Consolidates all data files in the catalog.

This method identifies all leaf directories in the catalog that contain parquet files and consolidates them. A leaf directory is one that contains files but no subdirectories. This is a convenience method that effectively calls consolidate_data for all data types and instrument IDs in the catalog.

Parameters

• start: Optional start timestamp for the consolidation range. Only files with timestamps greater than or equal to this value will be consolidated. If None, all files from the beginning of time will be considered.
• end: Optional end timestamp for the consolidation range. Only files with timestamps less than or equal to this value will be consolidated. If None, all files up to the end of time will be considered.
• ensure_contiguous_files: Whether to validate that consolidated intervals are contiguous (default: true).

Returns

Returns Ok(()) on success, or an error if consolidation fails for any directory.

Errors

Returns an error if:

• Directory listing fails.
• File consolidation operations fail.
• Interval validation fails (when ensure_contiguous_files is true).

Examples

use nautilus_core::UnixNanos;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Consolidate all files in the catalog
catalog.consolidate_catalog(None, None, None, None)?;

// Consolidate only files within a specific time range
catalog.consolidate_catalog(
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
    Some(true),
    None,
)?;

pub fn consolidate_data( &mut self, data_type: &CatalogDataType, identifier: Option<&str>, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: Option<bool>, deduplicate: Option<bool>, ) -> Result<()>

Consolidates data files for a specific data type and identifier.

This method consolidates Parquet files within a specific directory (defined by data type and optional identifier) by merging multiple files into a single file. This improves query performance and can reduce storage overhead.

Parameters

• data_type: The stored family to consolidate.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).
• start: Optional start timestamp to limit consolidation to files within this range.
• end: Optional end timestamp to limit consolidation to files within this range.
• ensure_contiguous_files: Whether to validate that consolidated intervals are contiguous (default: true).

Returns

Returns Ok(()) on success, or an error if consolidation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• File consolidation operations fail.
• Interval validation fails (when ensure_contiguous_files is true).

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Consolidate all quote files for a specific instrument
catalog.consolidate_data(
    &NautilusDataType::QuoteTick.into(),
    Some("BTCUSD"),
    None,
    None,
    None,
    None,
)?;

// Consolidate trade files within a time range
catalog.consolidate_data(
    &NautilusDataType::TradeTick.into(),
    None,
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
    Some(true),
    None,
)?;

pub fn consolidate_catalog_by_period( &mut self, period_nanos: Option<u64>, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: Option<bool>, ) -> Result<()>

Consolidates all data files in the catalog by splitting them into fixed time periods.

This method identifies all leaf directories in the catalog that contain parquet files and consolidates them by period. A leaf directory is one that contains files but no subdirectories. This is a convenience method that effectively calls consolidate_data_by_period for all data types and instrument IDs in the catalog.

Parameters

• period_nanos: The period duration for consolidation in nanoseconds. Default is 1 day (86400000000000). Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
• start: Optional start timestamp for the consolidation range. Only files with timestamps greater than or equal to this value will be consolidated. If None, all files from the beginning of time will be considered.
• end: Optional end timestamp for the consolidation range. Only files with timestamps less than or equal to this value will be consolidated. If None, all files up to the end of time will be considered.
• ensure_contiguous_files: If true, uses period boundaries for file naming. If false, uses actual data timestamps for file naming.

Returns

Returns Ok(()) on success, or an error if consolidation fails for any directory.

Errors

Returns an error if:

• Directory listing fails.
• Data type extraction from path fails.
• Period-based consolidation operations fail.

Notes

• This operation can be resource-intensive for large catalogs with many data types. and instruments.
• The consolidation process splits data into fixed time periods rather than combining. all files into a single file per directory.
• Uses the same period-based consolidation logic as consolidate_data_by_period.
• Original files are removed and replaced with period-based consolidated files.
• This method is useful for periodic maintenance of the catalog to standardize. file organization by time periods.

Examples

use nautilus_core::UnixNanos;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Consolidate all files in the catalog by 1-day periods
catalog.consolidate_catalog_by_period(
    Some(86400000000000), // 1 day in nanoseconds
    None,
    None,
    Some(true),
)?;

// Consolidate only files within a specific time range by 1-hour periods
catalog.consolidate_catalog_by_period(
    Some(3600000000000), // 1 hour in nanoseconds
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
    Some(false),
)?;

pub fn extract_data_cls_and_identifier_from_path( &self, path: &str, ) -> Result<(Option<String>, Option<String>)>

Extracts data class and identifier from a directory path.

This method parses a directory path to extract the data type and optional instrument identifier. It’s used to determine what type of data consolidation to perform for each directory.

Parameters

• path: The directory path to parse.

Returns

Returns a tuple of (data_class, identifier) where both are optional strings.

pub fn consolidate_data_by_period( &mut self, data_type: &CatalogDataType, identifier: Option<&str>, period_nanos: Option<u64>, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: Option<bool>, ) -> Result<()>

Consolidates data files by splitting them into fixed time periods.

This method queries data by period and writes consolidated files immediately, using efficient period-based consolidation logic. When start/end boundaries intersect existing files, the function automatically splits those files to preserve all data.

Parameters

• data_type: The stored family to consolidate.
• identifier: Optional instrument ID to consolidate. If None, consolidates all instruments.
• period_nanos: The period duration for consolidation in nanoseconds. Default is 1 day (86400000000000). Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
• start: Optional start timestamp for consolidation range. If None, uses earliest available data. If specified and intersects existing files, those files will be split to preserve data outside the consolidation range.
• end: Optional end timestamp for consolidation range. If None, uses latest available data. If specified and intersects existing files, those files will be split to preserve data outside the consolidation range.
• ensure_contiguous_files: If true, uses period boundaries for file naming. If false, uses actual data timestamps for file naming.

Returns

Returns Ok(()) on success, or an error if consolidation fails.

Errors

Returns an error if:

• data_type is a record family or an instrument selector, which have no period-typed rewrite; use Self::consolidate_data for those.
• The directory path cannot be constructed.
• File operations fail.
• Data querying or writing fails.

Notes

• Uses two-phase approach: first determines all queries, then executes them.
• Groups intervals into contiguous groups to preserve holes between groups.
• Allows consolidation across multiple files within each contiguous group.
• Skips queries if target files already exist for efficiency.
• Original files are removed immediately after querying each period.
• When ensure_contiguous_files=false, file timestamps match actual data range.
• When ensure_contiguous_files=true, file timestamps use period boundaries.
• Uses modulo arithmetic for efficient period boundary calculation.
• Preserves holes in data by preventing queries from spanning across gaps.
• Automatically splits files at start/end boundaries to preserve all data.
• Split operations are executed before consolidation to ensure data preservation.

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Consolidate all quote files by 1-day periods
catalog.consolidate_data_by_period(
    &NautilusDataType::QuoteTick.into(),
    None,
    Some(86400000000000), // 1 day in nanoseconds
    None,
    None,
    Some(true),
)?;

// Consolidate specific instrument by 1-hour periods
catalog.consolidate_data_by_period(
    &NautilusDataType::TradeTick.into(),
    Some("BTCUSD"),
    Some(3600000000000), // 1 hour in nanoseconds
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
    Some(false),
)?;

pub fn consolidate_data_by_period_generic<T>( &mut self, identifier: Option<&str>, period_nanos: Option<u64>, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: Option<bool>, ) -> Result<()>

where T: DecodeTypedFromRecordBatch + HasCatalogDataType + EncodeToRecordBatch + HasTsInit + TryFrom<Data> + Clone,

Generic consolidate data files by splitting them into fixed time periods.

This is a type-safe version of consolidate_data_by_period that uses generic types to ensure compile-time correctness and enable reuse across different data types.

Type Parameters

• T: The data type to consolidate, must implement required traits for serialization.

Parameters

• identifier: Optional instrument ID to target a specific instrument’s data.
• period_nanos: Optional period size in nanoseconds (default: 1 day).
• start: Optional start timestamp for consolidation range.
• end: Optional end timestamp for consolidation range.
• ensure_contiguous_files: Optional flag to control file naming strategy.

Returns

Returns Ok(()) on success, or an error if consolidation fails.

pub fn prepare_consolidation_queries( &self, type_name: &str, identifier: Option<&str>, intervals: &[(u64, u64)], period_nanos: u64, start: Option<UnixNanos>, end: Option<UnixNanos>, ensure_contiguous_files: bool, ) -> Result<Vec<ConsolidationQuery>>

Prepares all queries for consolidation by filtering, grouping, and handling splits.

This auxiliary function handles all the preparation logic for consolidation:

1. Filters intervals by time range.
2. Groups intervals into contiguous groups.
3. Identifies and creates split operations for data preservation.
4. Generates period-based consolidation queries.
5. Checks for existing target files.

pub fn group_contiguous_intervals( &self, intervals: &[(u64, u64)], period_nanos: u64, ) -> Vec<Vec<(u64, u64)>>

Groups intervals for period-based consolidation.

Groups adjacent intervals into the same bucket unless the gap between them exceeds period_nanos. Sub-period gaps land in the same consolidated file anyway, so they do not warrant a split. Gaps larger than one period represent genuine data holes.

Parameters

• intervals: A slice of timestamp intervals as (start, end) tuples, sorted by start.
• period_nanos: The target consolidation period; gaps larger than this split groups.

Returns

Returns a vector of groups. Returns an empty vector if the input is empty.

Examples

Legacy chunked files with period=86_400_000_000_000 (1 day):
  [(1,5), (6,10), (11,15)] -> [[(1,5), (6,10), (11,15)]]

Small period=1 with mixed gaps:
  [(1,5), (8,10), (12,15)] -> [[(1,5)], [(8,10)], [(12,15)]]

impl ParquetDataCatalog

pub fn delete_data_range( &mut self, data_type: &NautilusDataType, identifier: Option<&str>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<()>

Deletes data within a specified time range for a specific data type and identifier.

This method identifies all parquet files that intersect with the specified time range and handles them appropriately:

• Files completely within the range are deleted
• Files partially overlapping the range are split to preserve data outside the range
• The original intersecting files are removed after processing

Parameters

• data_type: The data type to delete from.
• identifier: Optional identifier to delete data for. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”). If None, deletes data across all identifiers.
• start: Optional start timestamp for the deletion range. If None, deletes from the beginning.
• end: Optional end timestamp for the deletion range. If None, deletes to the end.

Returns

Returns Ok(()) on success, or an error if deletion fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• File operations fail.
• Data querying or writing fails.

Notes

• This operation permanently removes data and cannot be undone.
• Files that partially overlap the deletion range are split to preserve data outside the range.
• The method ensures data integrity by using atomic operations where possible.
• Empty directories are not automatically removed after deletion.

Examples

use nautilus_core::UnixNanos;
use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Delete all quote data for a specific instrument
catalog.delete_data_range(&NautilusDataType::QuoteTick, Some("BTCUSD"), None, None)?;

// Delete trade data within a specific time range
catalog.delete_data_range(
    &NautilusDataType::TradeTick,
    None,
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
)?;

pub fn delete_catalog_range( &mut self, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<()>

Deletes data within a specified time range across the entire catalog.

This method identifies all leaf directories in the catalog that contain parquet files and deletes data within the specified time range from each directory. A leaf directory is one that contains files but no subdirectories. This is a convenience method that effectively calls delete_data_range for all data types and instrument IDs in the catalog.

Parameters

• start: Optional start timestamp for the deletion range. If None, deletes from the beginning.
• end: Optional end timestamp for the deletion range. If None, deletes to the end.

Returns

Returns Ok(()) on success, or an error if deletion fails.

Errors

Returns an error if:

• Directory traversal fails.
• Data class extraction from paths fails.
• Individual delete operations fail.

Notes

• This operation permanently removes data and cannot be undone.
• The deletion process handles file intersections intelligently by splitting files when they partially overlap with the deletion range.
• Files completely within the deletion range are removed entirely.
• Files partially overlapping the deletion range are split to preserve data outside the range.
• This method is useful for bulk data cleanup operations across the entire catalog.
• Empty directories are not automatically removed after deletion.

Examples

use nautilus_core::UnixNanos;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Delete all data before a specific date across entire catalog
catalog.delete_catalog_range(None, Some(UnixNanos::from(1609459200000000000)))?;

// Delete all data within a specific range across entire catalog
catalog.delete_catalog_range(
    Some(UnixNanos::from(1609459200000000000)),
    Some(UnixNanos::from(1609545600000000000)),
)?;

// Delete all data after a specific date across entire catalog
catalog.delete_catalog_range(Some(UnixNanos::from(1609459200000000000)), None)?;

pub fn delete_data_range_generic<T>( &mut self, identifier: Option<&str>, start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<()>

where T: DecodeTypedFromRecordBatch + HasCatalogDataType + EncodeToRecordBatch + HasTsInit + TryFrom<Data> + Clone,

Generic implementation for deleting data within a specified time range.

This method provides the core deletion logic that works with any data type that implements the required traits. It handles file intersection analysis, data splitting for partial overlaps, and file cleanup.

Type Parameters

• T: The data type that implements required traits for catalog operations.

Parameters

• identifier: Optional instrument ID to delete data for.
• start: Optional start timestamp for the deletion range.
• end: Optional end timestamp for the deletion range.

Returns

Returns Ok(()) on success, or an error if deletion fails.

pub fn prepare_delete_operations( &self, type_name: &str, identifier: Option<&str>, intervals: &[(u64, u64)], start: Option<UnixNanos>, end: Option<UnixNanos>, ) -> Result<Vec<DeleteOperation>>

Prepares all operations for data deletion by identifying files that need to be split or removed.

This auxiliary function handles all the preparation logic for deletion:

1. Filters intervals by time range
2. Identifies files that intersect with the deletion range
3. Creates split operations for files that partially overlap
4. Generates removal operations for files completely within the range

Parameters

• type_name: The data type directory name for path generation.
• identifier: Optional instrument identifier for path generation.
• intervals: List of (start_ts, end_ts) tuples representing existing file intervals.
• start: Optional start timestamp for deletion range.
• end: Optional end timestamp for deletion range.

Returns

Returns a vector of DeleteOperation structs ready for execution.

impl ParquetDataCatalog

pub fn read_live_run(&self, instance_id: &str) -> Result<Vec<Data>>

Reads data from a live run instance.

This method reads all data associated with a specific live run instance from feather files stored in the catalog.

Parameters

• instance_id: The ID of the live run instance to read.

Returns

Returns a vector of Data objects from the live run, sorted by timestamp, or an error if the operation fails.

Errors

Returns an error if:

• The instance ID doesn’t exist.
• Feather file reading fails.
• Data deserialization fails.

Note

This method reads through the run reader: it lists the run’s data-type directories, reads every Feather file through the Arrow IPC stream reader with staged batch restoration, decodes quotes, trades, order book deltas and depths, bars, index and mark prices, option Greeks, funding rates, instrument status and closes, and custom data files into Data values, skips unknown data types, and sorts the result by ts_init.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Read data from a live run
let data = catalog.read_live_run("instance-123")?;
for item in data {
    println!("Data: {:?}", item);
}

pub fn read_backtest(&self, instance_id: &str) -> Result<Vec<Data>>

Reads data from a backtest run instance.

This method reads all data associated with a specific backtest run instance from feather files stored in the catalog.

Parameters

• instance_id: The ID of the backtest run instance to read.

Returns

Returns a vector of Data objects from the backtest run, sorted by timestamp, or an error if the operation fails.

Errors

Returns an error if:

• The instance ID doesn’t exist.
• Feather file reading fails.
• Data deserialization fails.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Read data from a backtest run
let data = catalog.read_backtest("instance-123")?;
for item in data {
    println!("Data: {:?}", item);
}

pub fn convert_stream_to_data( &mut self, instance_id: &str, data_type: &CatalogDataType, subdirectory: Option<&str>, identifiers: Option<&[String]>, use_ts_event_for_ts_init: bool, ) -> Result<()>

Converts stream data from feather files to parquet files.

This method reads data from feather files generated during a backtest or live run and writes it to the catalog in parquet format. It’s useful for converting temporary stream data into a more permanent and queryable format.

Parameters

• instance_id: The ID of the backtest or live run instance.
• data_cls: The data class name (e.g., “quotes”, “trades”, “bars”), or custom/{TypeName} with the registered type name verbatim for custom data.
• subdirectory: The subdirectory containing the feather files. Either “backtest” or “live” (default: “backtest”).
• identifiers: Optional list of identifiers to filter by (instrument IDs or bar types).
• use_ts_event_for_ts_init: If true, replaces the ts_init column with ts_event column values before deserializing.

Returns

Returns Ok(()) on success, or an error if the operation fails.

Errors

Returns an error if:

• data_type is an instrument class selector, which has no staged stream name.
• data_type is a family streams do not support.
• Feather file listing fails.
• Feather file reading fails.
• Writing to parquet fails.

Note

This method converts directly between Arrow IPC stream batches and Parquet batches without materializing Nautilus data objects. An instance with no staged files for the family converts nothing and returns success. It requires:

• Listing feather files in the specified subdirectory
• Reading feather files (Arrow IPC stream reading)
• Applying table-only stream conversion transforms
• Writing Arrow batches to the catalog

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Convert backtest stream data to parquet
catalog.convert_stream_to_data(
    "instance-123",
    &NautilusDataType::QuoteTick.into(),
    Some("backtest"),
    None,
    false,
)?;

impl ParquetDataCatalog

pub fn reset_all_file_names(&self) -> Result<()>

Resets the filenames of all Parquet files in the catalog to match their actual content timestamps.

This method scans all leaf data directories in the catalog and renames files based on the actual timestamp range of their content. This is useful when files have been modified or when filename conventions have changed.

Returns

Returns Ok(()) on success, or an error if the operation fails.

Errors

Returns an error if:

• Directory listing fails.
• File metadata reading fails.
• File rename operations fail.
• Interval validation fails after renaming.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Reset all filenames in the catalog
catalog.reset_all_file_names()?;

pub fn reset_data_file_names( &self, data_type: &CatalogDataType, identifier: Option<&str>, ) -> Result<()>

Resets the filenames of Parquet files for a specific data type and identifier.

This method renames files in a specific directory based on the actual timestamp range of their content. This is useful for correcting filenames after data modifications or when filename conventions have changed.

Parameters

• data_type: The stored family to target.
• identifier: Optional identifier to target a specific instrument’s data. Can be an instrument_id (e.g., “EUR/USD.SIM”) or a bar_type (e.g., “EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL”).

Returns

Returns Ok(()) on success, or an error if the operation fails.

Errors

Returns an error if:

• The directory path cannot be constructed.
• File metadata reading fails.
• File rename operations fail.
• Interval validation fails after renaming.

Examples

use nautilus_model::data::NautilusDataType;
use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

// Reset filenames for all quote files
catalog.reset_data_file_names(&NautilusDataType::QuoteTick.into(), None)?;

// Reset filenames for a specific instrument's trade files
catalog.reset_data_file_names(&NautilusDataType::TradeTick.into(), Some("BTCUSD"))?;

pub fn find_leaf_data_directories(&self) -> Result<Vec<String>>

Finds all leaf data directories in the catalog.

A leaf directory is one that contains data files but no subdirectories. This method is used to identify directories that can be processed for consolidation or other operations.

Returns

Returns a vector of directory path strings representing leaf directories, or an error if directory traversal fails.

Errors

Returns an error if:

• Object store listing operations fail.
• Directory structure cannot be analyzed.

Examples

use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;

let mut catalog = ParquetDataCatalog::new(
    std::path::Path::new("/tmp/nautilus_data"),
    None,
    None,
    None,
    None,
);

let leaf_dirs = catalog.find_leaf_data_directories()?;
for dir in leaf_dirs {
    println!("Found leaf directory: {}", dir);
}

impl ParquetDataCatalog

pub fn migrate_from_legacy_parquet_catalog( &self, source: &Self, ) -> Result<CatalogMigrationReport>

Rewrites a legacy Parquet catalog into this current Parquet catalog.

Errors

Returns an error if source preflight fails, this catalog contains any leaf object, or a source file cannot be read, converted, or written.

impl ParquetDataCatalog

pub fn query_metadata( &mut self, data_type: &CatalogDataType, identifiers: Option<Vec<String>>, start: Option<UnixNanos>, end: Option<UnixNanos>, where_clause: Option<&str>, ) -> Result<Vec<CatalogMetadata>>

Queries Arrow schema metadata and the first queried timestamp where each metadata is used.

Errors

Returns an error if file discovery, Parquet metadata reading, or query execution fails.

Trait Implementations

impl CatalogReader for ParquetDataCatalog

fn fork_query_catalog(&self) -> Result<Option<DataCatalog>>

Creates an independent catalog for a lazy query while sharing backend resources.

fn query_batch_session( &mut self, query: &CatalogQuery, chunk_size: Option<usize>, ) -> Result<DataBatchQueryResult>

Queries catalog data as a typed batch session.

fn reset_session(&mut self)

Resets any per-query session state.

fn instruments( &mut self, query: &CatalogInstrumentQuery, ) -> Result<Vec<InstrumentAny>>

Queries instruments known by the catalog.

fn query_batch(&mut self, query: &CatalogQuery) -> Result<DataBatch>

Queries catalog data as a typed batch.

fn query_identifiers(&mut self, query: &CatalogQuery) -> Result<Vec<String>>

Queries the concrete catalog row identifiers matched by a data query.

fn query_display_record_batches( &mut self, query: &CatalogQuery, ) -> Result<Vec<RecordBatch>>

Queries catalog data as display-friendly Arrow record batches.

fn query_record_batches( &mut self, query: &CatalogRecordQuery, ) -> Result<Vec<RecordBatch>>

Queries catalog records as raw Arrow record batches.

fn query_record_display_batches( &mut self, query: &CatalogRecordQuery, ) -> Result<Vec<RecordBatch>>

Queries catalog records as display-friendly Arrow record batches.

fn query_metadata( &mut self, query: &CatalogQuery, ) -> Result<Vec<CatalogMetadata>>

Queries Arrow schema metadata and the first queried timestamp where each metadata is used.

fn get_missing_intervals_for_request( &mut self, start: UnixNanos, end: UnixNanos, data_type: NautilusDataType, identifier: Option<&str>, ) -> Result<Vec<(u64, u64)>>

Returns request intervals not covered by catalog data or known-empty coverage.

fn query_last_timestamp( &mut self, data_type: NautilusDataType, identifier: Option<&str>, ) -> Result<Option<u64>>

Returns the last timestamp covered by the catalog for a data type and optional identifier.

fn get_missing_intervals_for_identifiers( &mut self, start: UnixNanos, end: UnixNanos, data_type: NautilusDataType, identifiers: &[String], ) -> Result<AHashMap<String, Vec<(u64, u64)>>>

Returns missing request intervals for each identifier.

fn get_coverage_intervals_for_identifiers( &mut self, start: UnixNanos, end: UnixNanos, data_type: NautilusDataType, identifiers: &[String], ) -> Result<AHashMap<String, CoverageIntervals>>

Returns effective data and known-empty coverage for each identifier.

impl CatalogWriter for ParquetDataCatalog

fn write_instruments(&mut self, instruments: &[InstrumentAny]) -> Result<()>

Writes instrument definitions into the catalog.

fn write_data( &mut self, data: &[Data], start: Option<UnixNanos>, end: Option<UnixNanos>, params: Option<Params>, ) -> Result<()>

Writes mixed built-in data values into the catalog.

fn write_data_batch( &mut self, batch: &DataBatch, start: Option<UnixNanos>, end: Option<UnixNanos>, params: Option<Params>, ) -> Result<()>

Writes a typed data batch into the catalog.

fn write_records( &mut self, record_type: NautilusRecordType, batches: &[RecordBatch], params: Option<Params>, ) -> Result<()>

Writes Arrow record batches for a record family into the catalog.

fn record_empty_coverage( &mut self, data_type: NautilusDataType, identifier: Option<&str>, start: UnixNanos, end: UnixNanos, ) -> Result<()>

Records request coverage for a known-empty interval.

impl Debug for ParquetDataCatalog

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl ParquetCatalogSource for ParquetDataCatalog

fn object_store(&self) -> Arc<dyn ObjectStore>

Returns the object store containing the source catalog.

fn base_path(&self) -> &str

Returns the catalog path within the object store.

fn original_uri(&self) -> &str

Returns the URI identifying the source catalog.

fn to_object_path_parsed(&self, path: &str) -> Result<ObjectPath>

Resolves a plan-relative path within the source catalog.