nautilus_persistence/backend/parquet/catalog/store.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//! Object-store paths, listing, and file administration for the Parquet catalog.
17
18#![expect(
19 clippy::missing_errors_doc,
20 reason = "catalog store functions forward object-store errors"
21)]
22
23use anyhow::Context;
24use nautilus_common::live::block_on_nautilus_with;
25use nautilus_model::data::NautilusDataType;
26use object_store::ObjectMeta;
27
28use super::{
29 HashSet, ObjectPath, ObjectStore, ObjectStoreExt, ParquetDataCatalog, PathBuf, StreamExt,
30 UnixNanos, append_path_to_file_uri, are_intervals_disjoint, decode_object_store_segment,
31 is_remote_uri_scheme, make_object_store_path, parse_filename_timestamps,
32 query::{filter_identifier_files, is_parquet_bar_prefix},
33 query_intersects_filename, remote_full_uri, remote_store_root_url, timestamps_to_filename,
34 urisafe_instrument_id,
35};
36use crate::{
37 catalog::types::{
38 CatalogDataType, catalog_data_type_from_path, parquet_catalog_data_type_path_prefixes,
39 },
40 common::{paths::normalize_path_separators, storage::StorageBackend},
41};
42
43impl ParquetDataCatalog {
44 /// Extends the timestamp range of an existing Parquet file by renaming it.
45 ///
46 /// This method finds an existing file that is adjacent to the specified time range
47 /// and renames it to include the new range. This is useful when appending data
48 /// that extends the time coverage of existing files.
49 /// The proposed extension is validated against the other files before renaming,
50 /// so a rejected extension leaves existing files unchanged.
51 ///
52 /// If no file is adjacent to the specified range, this method does nothing and
53 /// returns `Ok(())` after confirming the existing intervals are disjoint.
54 ///
55 /// # Parameters
56 ///
57 /// - `data_type`: The stored family to target.
58 /// - `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").
59 /// - `start`: Start timestamp of the new range to extend to.
60 /// - `end`: End timestamp of the new range to extend to.
61 ///
62 /// # Returns
63 ///
64 /// Returns `Ok(())` on success, or an error if the operation fails.
65 ///
66 /// # Errors
67 ///
68 /// Returns an error if:
69 /// - The range is reversed (`start` is after `end`).
70 /// - The directory path cannot be constructed.
71 /// - The proposed extension would overlap another file.
72 /// - The existing intervals are already overlapping.
73 /// - File rename operations fail.
74 ///
75 /// # Examples
76 ///
77 /// ```rust,no_run
78 /// use nautilus_core::UnixNanos;
79 /// use nautilus_model::data::NautilusDataType;
80 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
81 ///
82 /// let mut catalog = ParquetDataCatalog::new(
83 /// std::path::Path::new("/tmp/nautilus_data"),
84 /// None,
85 /// None,
86 /// None,
87 /// None,
88 /// );
89 ///
90 /// // Extend a file's range backwards or forwards
91 /// catalog.extend_file_name(
92 /// &NautilusDataType::QuoteTick.into(),
93 /// Some("BTC/USD.SIM"),
94 /// UnixNanos::from(1609459200000000000),
95 /// UnixNanos::from(1609545600000000000),
96 /// )?;
97 /// # Ok::<(), anyhow::Error>(())
98 /// ```
99 pub fn extend_file_name(
100 &self,
101 data_type: &CatalogDataType,
102 identifier: Option<&str>,
103 start: UnixNanos,
104 end: UnixNanos,
105 ) -> anyhow::Result<()> {
106 let prefixes = parquet_catalog_data_type_path_prefixes(data_type);
107
108 if let [data_cls] = prefixes.as_slice() {
109 let directory = self.make_path(data_cls.as_ref(), identifier)?;
110 return self.extend_file_name_in_directory(&directory, start, end);
111 }
112
113 // The aggregate instrument family spans every class directory, and one identifier can be
114 // stored under several classes, so extend each directory that already holds it rather
115 // than inventing a class for it.
116 let mut extended = false;
117
118 for data_cls in &prefixes {
119 let directory = self.make_path(data_cls.as_ref(), identifier)?;
120 if !self.get_directory_intervals(&directory)?.is_empty() {
121 self.extend_file_name_in_directory(&directory, start, end)?;
122 extended = true;
123 }
124 }
125
126 anyhow::ensure!(
127 extended,
128 "Cannot extend file name for {data_type}: no instrument class holds {}; \
129 name the class with a NautilusInstrumentType",
130 identifier.unwrap_or("any identifier"),
131 );
132
133 Ok(())
134 }
135
136 pub(super) fn extend_file_name_in_directory(
137 &self,
138 directory: &str,
139 start: UnixNanos,
140 end: UnixNanos,
141 ) -> anyhow::Result<()> {
142 let start = start.as_u64();
143 let end = end.as_u64();
144
145 anyhow::ensure!(
146 start <= end,
147 "Cannot extend file in directory '{directory}': reversed range ({start}, {end})",
148 );
149
150 let intervals = self.get_directory_intervals(directory)?;
151
152 anyhow::ensure!(
153 are_intervals_disjoint(&intervals),
154 "Intervals are not disjoint in directory '{directory}': {intervals:?}",
155 );
156
157 let adjacent = intervals.iter().enumerate().find_map(|(index, interval)| {
158 if end.checked_add(1) == Some(interval.0) {
159 // Extend backwards: new file covers [start, interval.1]
160 Some((index, *interval, (start, interval.1)))
161 } else if start.checked_sub(1) == Some(interval.1) {
162 // Extend forwards: new file covers [interval.0, end]
163 Some((index, *interval, (interval.0, end)))
164 } else {
165 None
166 }
167 });
168
169 let Some((index, original, proposed)) = adjacent else {
170 return Ok(());
171 };
172
173 let mut extended = intervals.clone();
174 extended[index] = proposed;
175
176 anyhow::ensure!(
177 are_intervals_disjoint(&extended),
178 "Extending file interval {original:?} to {proposed:?} in directory '{directory}' \
179 with range ({start}, {end}) would create non-disjoint intervals. \
180 Existing intervals: {intervals:?}",
181 );
182
183 self.rename_parquet_file(directory, original.0, original.1, proposed.0, proposed.1)
184 }
185
186 /// Helper method to rename a parquet file by moving it via object store operations
187 fn rename_parquet_file(
188 &self,
189 directory: &str,
190 old_start: u64,
191 old_end: u64,
192 new_start: u64,
193 new_end: u64,
194 ) -> anyhow::Result<()> {
195 let new_filename =
196 timestamps_to_filename(UnixNanos::from(new_start), UnixNanos::from(new_end));
197 let new_path = format!("{directory}/{new_filename}");
198 let matches = self
199 .list_parquet_files(directory)?
200 .into_iter()
201 .filter(|file| parse_filename_timestamps(file) == Some((old_start, old_end)))
202 .collect::<Vec<_>>();
203
204 anyhow::ensure!(
205 matches.len() == 1,
206 "expected one parquet file for interval ({old_start}, {old_end}) in {directory}, \
207 found {}",
208 matches.len()
209 );
210
211 let old_path = &matches[0];
212 if old_path.ends_with(&new_filename) {
213 return Ok(());
214 }
215
216 let old_object_path = self.to_object_path_parsed(old_path)?;
217 let new_object_path = self.to_object_path(&new_path)?;
218 self.move_file(&old_object_path, &new_object_path)
219 }
220
221 /// Lists all Parquet files under a directory, including its subdirectories.
222 ///
223 /// This method scans a directory and returns the full paths of all files with the `.parquet`
224 /// extension. It works with both local filesystems and remote object stores, making it
225 /// suitable for various storage backends.
226 ///
227 /// # Parameters
228 ///
229 /// - `directory`: The directory path to scan for Parquet files, unencoded as
230 /// [`Self::make_path`] returns it.
231 ///
232 /// # Returns
233 ///
234 /// Returns a vector of full file paths (as strings) for all Parquet files under the directory.
235 /// The paths are percent-encoded object-store keys relative to the store root; convert one with
236 /// [`Self::to_object_path_parsed`].
237 /// Returns an empty vector if the directory doesn't exist or contains no Parquet files.
238 ///
239 /// # Errors
240 ///
241 /// Returns an error if:
242 /// - Object store listing operations fail.
243 /// - Directory access is denied.
244 /// - Network issues occur (for remote object stores).
245 ///
246 /// # Notes
247 ///
248 /// - Only files ending with `.parquet` are included.
249 /// - Subdirectories are scanned recursively, so a type directory such as `data/quotes`
250 /// returns the files of every identifier under it.
251 /// - File paths are returned in the order provided by the object store.
252 /// - Works with all supported object store backends (local, S3, GCS, Azure, etc.).
253 ///
254 /// # Examples
255 ///
256 /// ```rust,no_run
257 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
258 ///
259 /// let mut catalog = ParquetDataCatalog::new(
260 /// std::path::Path::new("/tmp/nautilus_data"),
261 /// None,
262 /// None,
263 /// None,
264 /// None,
265 /// );
266 /// let files = catalog.list_parquet_files("data/quotes/EURUSD")?;
267 ///
268 /// for file in files {
269 /// println!("Found Parquet file: {}", file);
270 /// }
271 /// # Ok::<(), anyhow::Error>(())
272 /// ```
273 pub fn list_parquet_files(&self, directory: &str) -> anyhow::Result<Vec<String>> {
274 let files = self
275 .list_objects(directory)?
276 .into_iter()
277 .filter(|object| object.location.as_ref().ends_with(".parquet"))
278 .map(|object| object.location.to_string())
279 .collect();
280
281 Ok(files)
282 }
283
284 /// Lists all instrument identifiers for a specific data type.
285 ///
286 /// This method scans the data directory for a given data type and extracts
287 /// all unique instrument identifiers from the directory structure.
288 ///
289 /// # Parameters
290 ///
291 /// - `data_type`: The stored family to target.
292 ///
293 /// # Returns
294 ///
295 /// Returns a vector of instrument identifier strings.
296 ///
297 /// # Errors
298 ///
299 /// Returns an error if directory listing fails.
300 pub fn list_instruments(&self, data_type: &CatalogDataType) -> anyhow::Result<Vec<String>> {
301 let mut instruments = Vec::new();
302
303 for prefix in parquet_catalog_data_type_path_prefixes(data_type) {
304 instruments.extend(self.list_prefix_instruments(prefix.as_ref())?);
305 }
306
307 // The same identifier can live under more than one instrument class.
308 instruments.sort();
309 instruments.dedup();
310
311 Ok(instruments)
312 }
313
314 fn list_prefix_instruments(&self, data_type: &str) -> anyhow::Result<Vec<String>> {
315 let directory = self.make_path(data_type, None)?;
316 let prefix = ObjectPath::from(directory.as_str());
317 let mut instruments = HashSet::new();
318
319 for object in self.list_objects(&directory)? {
320 // Relative to the prefix a datum is `{identifier}/{filename}.parquet`
321 let Some(relative) = object.location.prefix_match(&prefix) else {
322 continue;
323 };
324 let segments: Vec<_> = relative.collect();
325
326 if let [identifier, filename] = segments.as_slice()
327 && filename.as_ref().ends_with(".parquet")
328 {
329 instruments.insert(decode_object_store_segment(identifier.as_ref()));
330 }
331 }
332
333 Ok(instruments.into_iter().collect())
334 }
335
336 /// Lists Parquet files matching specific criteria (data type, identifiers, time range).
337 ///
338 /// This method finds all Parquet files that match the specified criteria by filtering
339 /// files based on their directory structure and filename timestamps.
340 ///
341 /// # Parameters
342 ///
343 /// - `data_type`: The stored family to target.
344 /// - `identifiers`: Optional list of identifiers to filter by.
345 /// - `start`: Optional start timestamp to filter files by their time range.
346 /// - `end`: Optional end timestamp to filter files by their time range.
347 ///
348 /// # Returns
349 ///
350 /// Returns a vector of file paths that match the criteria.
351 ///
352 /// # Errors
353 ///
354 /// Returns an error if directory listing or file filtering fails.
355 pub fn list_parquet_files_with_criteria(
356 &self,
357 data_type: &CatalogDataType,
358 identifiers: Option<&[String]>,
359 start: Option<UnixNanos>,
360 end: Option<UnixNanos>,
361 ) -> anyhow::Result<Vec<String>> {
362 let mut all_files = Vec::new();
363
364 for data_cls in parquet_catalog_data_type_path_prefixes(data_type) {
365 all_files.extend(self.list_prefix_files_with_criteria(
366 data_cls.as_ref(),
367 identifiers,
368 start,
369 end,
370 )?);
371 }
372
373 Ok(all_files)
374 }
375
376 fn list_prefix_files_with_criteria(
377 &self,
378 data_cls: &str,
379 identifiers: Option<&[String]>,
380 start: Option<UnixNanos>,
381 end: Option<UnixNanos>,
382 ) -> anyhow::Result<Vec<String>> {
383 let start_u64 = start.map(|s| s.as_u64());
384 let end_u64 = end.map(|e| e.as_u64());
385
386 let base_dir = self.make_path(data_cls, None)?;
387
388 // Use recursive listing to match Python's glob behavior
389 let mut all_files = self.list_parquet_files(&base_dir)?;
390
391 if let Some(identifiers) = identifiers {
392 all_files =
393 filter_identifier_files(all_files, identifiers, is_parquet_bar_prefix(data_cls));
394 }
395
396 all_files.retain(|path| query_intersects_filename(path, start_u64, end_u64));
397 Ok(all_files)
398 }
399
400 /// Recursively lists all objects under `{dir}/` in the object store.
401 pub(super) fn list_objects(&self, dir: &str) -> anyhow::Result<Vec<ObjectMeta>> {
402 self.execute_async(|| async {
403 let prefix = ObjectPath::from(format!("{dir}/"));
404 let mut stream = self.object_store.list(Some(&prefix));
405 let mut objects = Vec::new();
406 while let Some(object) = stream.next().await {
407 objects.push(object?);
408 }
409
410 Ok(objects)
411 })
412 }
413
414 /// Helper method to reconstruct full URI for remote object store paths
415 #[must_use]
416 pub fn reconstruct_full_uri(&self, path_str: &str) -> String {
417 if path_str.contains("://") {
418 return path_str.to_string();
419 }
420
421 // Check if this is a remote URI scheme that needs reconstruction
422 if self.is_remote_uri() {
423 let path = self.path_under_base(path_str);
424 if let Ok(uri) = remote_full_uri(&self.original_uri, &path) {
425 return uri;
426 }
427 }
428
429 // For local paths, extract the directory from the original URI
430 if self.original_uri.starts_with("file://") {
431 // Extract the path from the file:// URI
432 if let Ok(url) = url::Url::parse(&self.original_uri)
433 && let Ok(base_path) = url.to_file_path()
434 {
435 // Use platform-appropriate path separator for display
436 // but object store paths always use forward slashes
437 let base_str = base_path.to_string_lossy();
438 return make_object_store_path(&base_str, [path_str]);
439 }
440 }
441
442 // For local paths without file:// prefix, use the original URI as base
443 if self.base_path.is_empty() {
444 // If base_path is empty and not a file URI, try using original_uri as base
445 if self.original_uri.contains("://") {
446 // Fallback: return the path as-is
447 path_str.to_string()
448 } else {
449 make_object_store_path(self.original_uri.trim_end_matches('/'), [path_str])
450 }
451 } else {
452 let base = self.base_path.trim_end_matches('/');
453 make_object_store_path(base, [path_str])
454 }
455 }
456
457 /// Resolves a path for use with DataFusion (avoiding Windows path doubling for file://).
458 /// A remote path resolves to a URL under the store root whose path decodes to the object
459 /// key. A local path is returned as-is if it is already a full URI or absolute; otherwise it
460 /// joins the file:// base or goes through `reconstruct_full_uri`.
461 #[must_use]
462 pub(crate) fn resolve_path_for_datafusion(&self, path: &str) -> String {
463 // DataFusion decodes the URL path once, so encode the key again
464 if self.is_remote_uri()
465 && let Ok(key) = self.object_store_path(path)
466 && let Ok(root) = remote_store_root_url(&self.original_uri)
467 {
468 return append_path_to_file_uri(root.as_str(), &key);
469 }
470
471 if path.contains("://") {
472 return path.to_string();
473 }
474
475 if path.starts_with('/') {
476 return path.to_string();
477 }
478
479 if self.original_uri.starts_with("file://") {
480 return append_path_to_file_uri(&self.original_uri, path);
481 }
482
483 self.reconstruct_full_uri(path)
484 }
485
486 /// Like `resolve_path_for_datafusion` but ensures the result ends with a trailing slash.
487 #[must_use]
488 pub(super) fn resolve_directory_for_datafusion(&self, directory: &str) -> String {
489 let mut resolved = self.resolve_path_for_datafusion(directory);
490 if !resolved.ends_with('/') {
491 resolved.push('/');
492 }
493
494 resolved
495 }
496
497 /// Returns the path string to push in `query_files` result list: relative for file://,
498 /// full URI for remote (so callers can pass to `resolve_path_for_datafusion` later).
499 #[must_use]
500 pub(super) fn path_for_query_list(&self, path: &str) -> String {
501 if self.original_uri.starts_with("file://") {
502 path.to_string()
503 } else {
504 self.reconstruct_full_uri(path)
505 }
506 }
507
508 /// Returns the native path string for the catalog root (for `std::fs`). Only valid when
509 /// !`is_remote_uri()`; uses parquet's `file_uri_to_native_path` for file:// URIs.
510 #[must_use]
511 pub(crate) fn native_base_path_string(&self) -> String {
512 if self.original_uri.starts_with("file://") {
513 crate::backend::parquet::io::file_uri_to_native_path(&self.original_uri)
514 } else {
515 self.original_uri.clone()
516 }
517 }
518
519 pub(crate) fn register_remote_object_store(&mut self) -> anyhow::Result<()> {
520 if self.is_remote_uri() {
521 let base_url = remote_store_root_url(&self.original_uri)?;
522 self.session
523 .register_object_store(&base_url, self.object_store.clone());
524 }
525
526 Ok(())
527 }
528
529 /// Helper method to check if the original URI uses a remote object store scheme
530 #[must_use]
531 pub fn is_remote_uri(&self) -> bool {
532 self.original_uri
533 .split_once("://")
534 .is_some_and(|(scheme, _)| is_remote_uri_scheme(scheme))
535 }
536
537 /// Constructs a directory path for storing data of a specific type and instrument.
538 ///
539 /// This method builds the hierarchical directory structure used by the catalog to organize
540 /// data by type and instrument. The path follows the pattern: `{base_path}/data/{type_name}/{instrument_id}`.
541 /// Instrument IDs are automatically converted to URI-safe format by removing forward slashes.
542 ///
543 /// # Parameters
544 ///
545 /// - `type_name`: The data type directory name (e.g., "quotes", "trades", "bars").
546 /// - `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.
547 ///
548 /// # Returns
549 ///
550 /// Returns the constructed directory path as a string, or an error if path construction fails.
551 ///
552 /// # Errors
553 ///
554 /// Returns an error if:
555 /// - The instrument ID contains invalid characters that cannot be made URI-safe.
556 /// - Path construction fails due to system limitations.
557 ///
558 /// # Path Structure
559 ///
560 /// - Without identifier: `{base_path}/data/{type_name}`.
561 /// - With identifier: `{base_path}/data/{type_name}/{safe_identifier}`.
562 /// - If `base_path` is empty: `data/{type_name}[/{safe_identifier}]`.
563 ///
564 /// # Examples
565 ///
566 /// ```rust,no_run
567 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
568 ///
569 /// let mut catalog = ParquetDataCatalog::new(
570 /// std::path::Path::new("/tmp/nautilus_data"),
571 /// None,
572 /// None,
573 /// None,
574 /// None,
575 /// );
576 ///
577 /// // Path for all quote data
578 /// let quotes_path = catalog.make_path("quotes", None)?;
579 /// // Returns: "/base/path/data/quotes"
580 ///
581 /// // Path for specific instrument quotes
582 /// let eurusd_quotes = catalog.make_path("quotes", Some("EUR/USD"))?;
583 /// // Returns: "/base/path/data/quotes/EURUSD" (slash removed)
584 ///
585 /// // Path for bar data with complex instrument ID
586 /// let bars_path = catalog.make_path("bars", Some("BTC/USD-1H"))?;
587 /// // Returns: "/base/path/data/bars/BTCUSD-1H"
588 /// # Ok::<(), anyhow::Error>(())
589 /// ```
590 pub fn make_path(&self, type_name: &str, identifier: Option<&str>) -> anyhow::Result<String> {
591 let mut components = vec!["data".to_string(), type_name.to_string()];
592
593 if let Some(id) = identifier {
594 let safe_id = urisafe_instrument_id(id);
595 components.push(safe_id);
596 }
597
598 let path = make_object_store_path(&self.base_path, components);
599 Ok(path)
600 }
601
602 /// Builds the directory path for custom data: `data/custom/{type_name}[/{identifier}]`.
603 pub fn make_path_custom_data(
604 &self,
605 type_name: &str,
606 identifier: Option<&str>,
607 ) -> anyhow::Result<String> {
608 let mut components = vec![
609 "data".to_string(),
610 "custom".to_string(),
611 type_name.to_string(),
612 ];
613
614 if let Some(id) = identifier {
615 let safe_id = urisafe_instrument_id(id);
616
617 if !safe_id.is_empty() {
618 components.push(safe_id);
619 }
620 }
621
622 let path = make_object_store_path(&self.base_path, components);
623 Ok(path)
624 }
625
626 /// Converts a catalog path string to an [`ObjectPath`] for object store operations.
627 ///
628 /// This method handles the conversion between catalog-relative paths and object store paths,
629 /// taking into account the catalog's base path configuration. It automatically preserves the
630 /// base path prefix for remote catalogs and strips it for local catalog paths.
631 ///
632 /// # Parameters
633 ///
634 /// - `path`: The catalog path string to convert. Can be absolute or relative.
635 ///
636 /// # Returns
637 ///
638 /// Returns an [`ObjectPath`] suitable for use with object store operations.
639 ///
640 /// # Path Handling
641 ///
642 /// - If `base_path` is empty, the path is used as-is.
643 /// - If `base_path` is set for a remote catalog, it's preserved or prepended.
644 /// - If `base_path` is set for a local catalog, it's stripped from the path if present.
645 /// - Trailing slashes and backslashes are automatically handled.
646 /// - The resulting path is relative to the object store root.
647 /// - All paths are normalized to use forward slashes (object store convention).
648 ///
649 /// # Errors
650 ///
651 /// Returns an error if the catalog is remote and `path` is a full URI whose scheme/host does
652 /// not match the catalog's own root (cross-bucket misuse). Without this guard the caller could
653 /// silently write to or read from the wrong bucket.
654 ///
655 /// # Examples
656 ///
657 /// Local catalog paths (absolute or relative) strip the catalog's base directory:
658 ///
659 /// ```rust,no_run
660 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
661 /// # let catalog: ParquetDataCatalog = unimplemented!();
662 /// let object_path = catalog.to_object_path("/base/data/quotes/file.parquet")?;
663 /// // ObjectPath("data/quotes/file.parquet")
664 /// # Ok::<(), anyhow::Error>(())
665 /// ```
666 ///
667 /// Remote catalog paths (relative or full URI) preserve or prepend the base prefix:
668 ///
669 /// ```rust,no_run
670 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
671 /// # let catalog: ParquetDataCatalog = unimplemented!();
672 /// let object_path = catalog.to_object_path("data/trades/file.parquet")?;
673 /// // ObjectPath("base/data/trades/file.parquet")
674 /// # Ok::<(), anyhow::Error>(())
675 /// ```
676 pub fn to_object_path(&self, path: &str) -> anyhow::Result<ObjectPath> {
677 Ok(ObjectPath::from(self.object_store_path(path)?))
678 }
679
680 /// Converts a path string to [`ObjectPath`] using parse (no percent-encoding).
681 ///
682 /// Use this for paths that were returned by the object store (e.g. from `list()`),
683 /// which may already be percent-encoded. Using [`Self::to_object_path`] (which uses
684 /// `Path::from`) on such paths would double-encode (e.g. `%5E` -> `%255E`).
685 ///
686 /// # Errors
687 ///
688 /// Returns an error if the same cross-bucket condition as [`Self::to_object_path`] occurs or
689 /// the resulting string fails [`ObjectPath::parse`].
690 pub fn to_object_path_parsed(&self, path: &str) -> anyhow::Result<ObjectPath> {
691 let to_parse = self.object_store_path(path)?;
692 ObjectPath::parse(&to_parse).map_err(anyhow::Error::from)
693 }
694
695 pub(super) fn object_store_path(&self, path: &str) -> anyhow::Result<String> {
696 let normalized_path = normalize_path_separators(path);
697
698 if self.is_remote_uri() {
699 if normalized_path.contains("://") {
700 let path_under_root = self.remote_uri_object_path(&normalized_path)?;
701 return Ok(self.path_under_base(&path_under_root));
702 }
703
704 return Ok(self.path_under_base(&normalized_path));
705 }
706
707 Ok(self.path_without_local_base(&normalized_path))
708 }
709
710 fn remote_uri_object_path(&self, path: &str) -> anyhow::Result<String> {
711 let path_url = url::Url::parse(path)
712 .map_err(|e| anyhow::anyhow!("Failed to parse object store URI {path}: {e}"))?;
713 if !is_remote_uri_scheme(path_url.scheme()) {
714 anyhow::bail!(
715 "URI {path} uses non-remote scheme {} for remote catalog at {}",
716 path_url.scheme(),
717 self.original_uri,
718 );
719 }
720
721 let catalog_root = remote_store_root_url(&self.original_uri)?;
722 let path_root = remote_store_root_url(path)?;
723 if catalog_root.as_str().trim_end_matches('/') != path_root.as_str().trim_end_matches('/') {
724 anyhow::bail!(
725 "Cross-store URI {path} (root {}) does not belong to catalog rooted at {} ({})",
726 path_root.as_str().trim_end_matches('/'),
727 self.original_uri,
728 catalog_root.as_str().trim_end_matches('/'),
729 );
730 }
731
732 // The URL crate keeps the path component percent-encoded (e.g. `%5E`),
733 // so preserve that encoding for `ObjectPath::parse` round-trips through
734 // `object_store::list`/`get`.
735 Ok(path_url.path().trim_start_matches('/').to_string())
736 }
737
738 pub(crate) fn path_without_local_base(&self, path: &str) -> String {
739 let base_path = if self.base_path.is_empty() {
740 self.native_base_path_string()
741 } else {
742 self.base_path.clone()
743 };
744
745 let normalized_base = normalize_path_separators(&base_path);
746 let base = normalized_base.trim_end_matches('/');
747
748 if base.is_empty() {
749 path.to_string()
750 } else if path == base {
751 String::new()
752 } else if let Some(without_base) = path.strip_prefix(&format!("{base}/")) {
753 without_base.to_string()
754 } else {
755 path.to_string()
756 }
757 }
758
759 fn path_under_base(&self, path: &str) -> String {
760 let normalized_path = normalize_path_separators(path);
761 let path = normalized_path
762 .trim_start_matches('/')
763 .trim_end_matches('/');
764
765 if self.base_path.is_empty() {
766 return path.to_string();
767 }
768
769 let normalized_base = normalize_path_separators(&self.base_path);
770 let base = normalized_base
771 .trim_start_matches('/')
772 .trim_end_matches('/');
773
774 if base.is_empty() || path == base || path.starts_with(&format!("{base}/")) {
775 path.to_string()
776 } else if path.is_empty() {
777 base.to_string()
778 } else {
779 make_object_store_path(base, [path])
780 }
781 }
782
783 /// Helper method to move a file using object store rename operation
784 pub fn move_file(&self, old_path: &ObjectPath, new_path: &ObjectPath) -> anyhow::Result<()> {
785 if old_path == new_path {
786 return Ok(());
787 }
788
789 self.execute_async(|| async {
790 self.object_store
791 .rename(old_path, new_path)
792 .await
793 .map_err(anyhow::Error::from)
794 })
795 }
796
797 /// Helper method to execute async operations with a runtime
798 pub fn execute_async<C, F, R>(&self, create_future: C) -> anyhow::Result<R>
799 where
800 C: FnOnce() -> F + Send,
801 F: std::future::Future<Output = anyhow::Result<R>>,
802 R: Send,
803 {
804 block_on_nautilus_with(create_future)
805 }
806
807 /// Lists directory stems (directory names without path) in a subdirectory.
808 ///
809 /// This method scans a subdirectory and returns the names of all immediate
810 /// subdirectories. It's used to list data types, backtest runs, and live runs.
811 ///
812 /// # Parameters
813 ///
814 /// - `subdirectory`: The subdirectory path to scan (e.g., "data", "backtest", "live").
815 ///
816 /// # Returns
817 ///
818 /// Returns the sorted directory names (stems) found in the subdirectory, decoded from their
819 /// percent-encoded object-store form, or an error if the operation fails.
820 ///
821 /// # Errors
822 ///
823 /// Returns an error if:
824 /// - Object store listing operations fail.
825 /// - Directory access is denied.
826 ///
827 /// # Examples
828 ///
829 /// ```rust,no_run
830 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
831 ///
832 /// let mut catalog = ParquetDataCatalog::new(
833 /// std::path::Path::new("/tmp/nautilus_data"),
834 /// None,
835 /// None,
836 /// None,
837 /// None,
838 /// );
839 ///
840 /// // List all data types
841 /// let data_types = catalog.list_directory_stems("data")?;
842 /// for data_type in data_types {
843 /// println!("Found data type: {}", data_type);
844 /// }
845 /// # Ok::<(), anyhow::Error>(())
846 /// ```
847 pub fn list_directory_stems(&self, subdirectory: &str) -> anyhow::Result<Vec<String>> {
848 // For local filesystem paths, use filesystem operations to detect empty directories
849 // For remote object stores, we can only list directories that contain files
850 if !self.is_remote_uri() {
851 let directory = PathBuf::from(self.native_base_path_string()).join(subdirectory);
852
853 let entries = match std::fs::read_dir(&directory) {
854 Ok(entries) => entries,
855 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
856 Err(e) => {
857 return Err(e)
858 .with_context(|| format!("failed to list {}", directory.display()));
859 }
860 };
861
862 let mut directories = Vec::new();
863
864 for entry in entries {
865 let entry =
866 entry.with_context(|| format!("failed to list {}", directory.display()))?;
867 let file_type = entry
868 .file_type()
869 .with_context(|| format!("failed to read {}", entry.path().display()))?;
870
871 // Use file_name() to get the directory name (not file_stem which removes extension)
872 if file_type.is_dir()
873 && let Some(name) = entry.path().file_name()
874 {
875 directories.push(decode_object_store_segment(&name.to_string_lossy()));
876 }
877 }
878
879 directories.sort();
880 return Ok(directories);
881 }
882
883 let storage = StorageBackend {
884 object_store: self.object_store.clone(),
885 base_path: self.base_path.clone(),
886 original_uri: self.original_uri.clone(),
887 };
888
889 self.execute_async(|| async { storage.list_directory_stems(subdirectory).await })
890 }
891
892 /// Lists the data, record, and instrument types stored in the catalog.
893 ///
894 /// Each `data/<type>` directory maps to its catalog type, and each `data/custom/<type_name>`
895 /// directory to a custom data type. Unrecognized directories are skipped with a warning.
896 ///
897 /// # Returns
898 ///
899 /// Returns the catalog types, or an error if the operation fails.
900 ///
901 /// # Errors
902 ///
903 /// Returns an error if:
904 /// - Object store listing operations fail.
905 /// - Directory access is denied.
906 ///
907 /// # Examples
908 ///
909 /// ```rust,no_run
910 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
911 ///
912 /// let mut catalog = ParquetDataCatalog::new(
913 /// std::path::Path::new("/tmp/nautilus_data"),
914 /// None,
915 /// None,
916 /// None,
917 /// None,
918 /// );
919 ///
920 /// // List all data types
921 /// let data_types = catalog.list_data_types()?;
922 /// for data_type in data_types {
923 /// println!("Available data type: {data_type}");
924 /// }
925 /// # Ok::<(), anyhow::Error>(())
926 /// ```
927 pub fn list_data_types(&self) -> anyhow::Result<Vec<CatalogDataType>> {
928 let mut data_types = Vec::new();
929
930 for stem in self.list_directory_stems("data")? {
931 if stem == "custom" {
932 for type_name in self.list_directory_stems("data/custom")? {
933 data_types.push(CatalogDataType::from(NautilusDataType::Custom {
934 type_name,
935 }));
936 }
937
938 continue;
939 }
940
941 match catalog_data_type_from_path(&stem) {
942 Ok(data_type) => data_types.push(data_type),
943 Err(e) => log::warn!("Skipping catalog directory data/{stem}: {e}"),
944 }
945 }
946
947 Ok(data_types)
948 }
949
950 /// Lists all backtest run IDs available in the catalog.
951 ///
952 /// This method returns the names of all backtest run directories in the catalog.
953 /// Each backtest run corresponds to a specific backtest execution instance.
954 ///
955 /// # Returns
956 ///
957 /// Returns a vector of backtest run IDs, or an error if the operation fails.
958 ///
959 /// # Errors
960 ///
961 /// Returns an error if:
962 /// - Object store listing operations fail.
963 /// - Directory access is denied.
964 ///
965 /// # Examples
966 ///
967 /// ```rust,no_run
968 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
969 ///
970 /// let mut catalog = ParquetDataCatalog::new(
971 /// std::path::Path::new("/tmp/nautilus_data"),
972 /// None,
973 /// None,
974 /// None,
975 /// None,
976 /// );
977 ///
978 /// // List all backtest runs
979 /// let runs = catalog.list_backtest_runs()?;
980 /// for run_id in runs {
981 /// println!("Backtest run: {}", run_id);
982 /// }
983 /// # Ok::<(), anyhow::Error>(())
984 /// ```
985 pub fn list_backtest_runs(&self) -> anyhow::Result<Vec<String>> {
986 self.list_directory_stems("backtest")
987 }
988
989 /// Lists all live run IDs available in the catalog.
990 ///
991 /// This method returns the names of all live run directories in the catalog.
992 /// Each live run corresponds to a specific live trading execution instance.
993 ///
994 /// # Returns
995 ///
996 /// Returns a vector of live run IDs, or an error if the operation fails.
997 ///
998 /// # Errors
999 ///
1000 /// Returns an error if:
1001 /// - Object store listing operations fail.
1002 /// - Directory access is denied.
1003 ///
1004 /// # Examples
1005 ///
1006 /// ```rust,no_run
1007 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
1008 ///
1009 /// let mut catalog = ParquetDataCatalog::new(
1010 /// std::path::Path::new("/tmp/nautilus_data"),
1011 /// None,
1012 /// None,
1013 /// None,
1014 /// None,
1015 /// );
1016 ///
1017 /// // List all live runs
1018 /// let runs = catalog.list_live_runs()?;
1019 /// for run_id in runs {
1020 /// println!("Live run: {}", run_id);
1021 /// }
1022 /// # Ok::<(), anyhow::Error>(())
1023 /// ```
1024 pub fn list_live_runs(&self) -> anyhow::Result<Vec<String>> {
1025 self.list_directory_stems("live")
1026 }
1027}
1028
1029#[cfg(test)]
1030mod tests {
1031 use std::sync::Arc;
1032
1033 use nautilus_model::data::NautilusDataType;
1034 use object_store::{ObjectStoreExt, PutPayload, memory::InMemory, path::Path as ObjectPath};
1035 use rstest::rstest;
1036
1037 use super::ParquetDataCatalog;
1038 use crate::{catalog::types::CatalogDataType, common::datafusion::DataBackendSession};
1039
1040 /// Builds a catalog over an in-memory store. Only remote catalogs carry a non-empty
1041 /// `base_path`, so seeding one here is the sole way to reproduce a bucket sub-prefix.
1042 fn memory_catalog(base_path: &str) -> ParquetDataCatalog {
1043 ParquetDataCatalog {
1044 base_path: base_path.to_string(),
1045 original_uri: "memory://".to_string(),
1046 object_store: Arc::new(InMemory::new()),
1047 session: DataBackendSession::new(5_000),
1048 batch_size: 5_000,
1049 compression: parquet::basic::Compression::SNAPPY,
1050 max_row_group_size: 5_000,
1051 }
1052 }
1053
1054 fn seed(catalog: &ParquetDataCatalog, keys: &[&str]) {
1055 catalog
1056 .execute_async(|| async {
1057 for key in keys {
1058 catalog
1059 .object_store
1060 .put(&ObjectPath::from(*key), PutPayload::from_static(b"x"))
1061 .await?;
1062 }
1063 Ok(())
1064 })
1065 .unwrap();
1066 }
1067
1068 #[rstest]
1069 #[case("nautilus-data", "nautilus-data/")]
1070 #[case("", "")]
1071 fn list_instruments_returns_ids_under_base_path(
1072 #[case] base_path: &str,
1073 #[case] key_prefix: &str,
1074 ) {
1075 let catalog = memory_catalog(base_path);
1076 let keys = [
1077 "data/quotes/EURUSD.SIM/0-1.parquet",
1078 "data/quotes/EURUSD.SIM/2-3.parquet",
1079 "data/quotes/GBPUSD.SIM/0-1.parquet",
1080 "data/trades/AUDUSD.SIM/0-1.parquet",
1081 ]
1082 .map(|key| format!("{key_prefix}{key}"));
1083 seed(&catalog, &keys.each_ref().map(String::as_str));
1084
1085 assert_eq!(
1086 catalog
1087 .list_instruments(&NautilusDataType::QuoteTick.into())
1088 .unwrap(),
1089 ["EURUSD.SIM", "GBPUSD.SIM"]
1090 );
1091 }
1092
1093 #[rstest]
1094 #[case("nautilus-data", "nautilus-data/")]
1095 #[case("", "")]
1096 fn list_parquet_files_lists_type_directory_recursively(
1097 #[case] base_path: &str,
1098 #[case] key_prefix: &str,
1099 ) {
1100 let catalog = memory_catalog(base_path);
1101 let keys = [
1102 "data/quotes/EURUSD.SIM/0-1.parquet",
1103 "data/quotes/EURUSD.SIM/0-1.json",
1104 "data/quotes/BTC€.SIM/2-3.parquet",
1105 "data/trades/AUDUSD.SIM/0-1.parquet",
1106 ]
1107 .map(|key| format!("{key_prefix}{key}"));
1108 seed(&catalog, &keys.each_ref().map(String::as_str));
1109
1110 let directory = catalog.make_path("quotes", None).unwrap();
1111 let mut files = catalog.list_parquet_files(&directory).unwrap();
1112 files.sort();
1113
1114 assert_eq!(
1115 files,
1116 [
1117 format!("{key_prefix}data/quotes/BTC%E2%82%AC.SIM/2-3.parquet"),
1118 format!("{key_prefix}data/quotes/EURUSD.SIM/0-1.parquet"),
1119 ]
1120 );
1121 }
1122
1123 #[rstest]
1124 fn list_instruments_ignores_unpartitioned_files() {
1125 // An empty `base_path` keeps this distinct from the remote prefix defect, so an empty
1126 // result can only come from the layout check.
1127 let catalog = memory_catalog("");
1128 seed(
1129 &catalog,
1130 &[
1131 "data/custom/MyType/1-2.parquet",
1132 "data/custom/MyType/1-2.json",
1133 ],
1134 );
1135
1136 let custom = CatalogDataType::Data(NautilusDataType::Custom {
1137 type_name: "MyType".to_string(),
1138 });
1139
1140 assert!(catalog.list_instruments(&custom).unwrap().is_empty());
1141 }
1142
1143 #[rstest]
1144 fn list_instruments_decodes_percent_encoded_ids() {
1145 let catalog = memory_catalog("nautilus-data");
1146 seed(
1147 &catalog,
1148 &["nautilus-data/data/quotes/BTC€.SIM/0-1.parquet"],
1149 );
1150
1151 assert_eq!(
1152 catalog
1153 .list_instruments(&NautilusDataType::QuoteTick.into())
1154 .unwrap(),
1155 ["BTC€.SIM"]
1156 );
1157 }
1158}