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