nautilus_persistence/backend/parquet/
paths.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Parquet-specific path, filename, and identifier operations.
17//!
18//! Cross-backend primitives (the `CatalogPathPrefix` trait, object-store path constructors,
19//! identifier sanitization) live in [`crate::common::paths`] and are re-exported below for
20//! parquet internal callers.
21
22use std::path::{Path, PathBuf};
23
24use nautilus_core::{
25    UnixNanos,
26    datetime::{iso8601_to_unix_nanos, unix_nanos_to_iso8601},
27};
28
29pub use crate::common::paths::{
30    CatalogPathPrefix, extract_identifier_from_path, extract_path_components,
31    local_to_object_store_path, make_object_store_path, make_sql_safe_identifier,
32    normalize_path_separators, safe_directory_identifier, urisafe_instrument_id,
33};
34
35/// Converts timestamps to a filename using ISO 8601 format.
36///
37/// Returns a filename string in the format: "`iso_timestamp_1_iso_timestamp_2.parquet`".
38#[must_use]
39pub fn timestamps_to_filename(timestamp_1: UnixNanos, timestamp_2: UnixNanos) -> String {
40    let datetime_1 = iso_timestamp_to_file_timestamp(&unix_nanos_to_iso8601(timestamp_1));
41    let datetime_2 = iso_timestamp_to_file_timestamp(&unix_nanos_to_iso8601(timestamp_2));
42
43    format!("{datetime_1}_{datetime_2}.parquet")
44}
45
46/// Converts an ISO 8601 timestamp to a filesystem-safe format.
47pub(crate) fn iso_timestamp_to_file_timestamp(iso_timestamp: &str) -> String {
48    iso_timestamp.replace([':', '.'], "-")
49}
50
51/// Converts a filesystem-safe timestamp back to ISO 8601 format.
52pub(crate) fn file_timestamp_to_iso_timestamp(file_timestamp: &str) -> String {
53    let (date_part, time_part) = file_timestamp
54        .split_once('T')
55        .unwrap_or((file_timestamp, ""));
56    let time_part = time_part.strip_suffix('Z').unwrap_or(time_part);
57
58    // Find the last hyphen to separate nanoseconds
59    if let Some(last_hyphen_idx) = time_part.rfind('-') {
60        let time_with_dot_for_nanos = format!(
61            "{}.{}",
62            &time_part[..last_hyphen_idx],
63            &time_part[last_hyphen_idx + 1..]
64        );
65        let final_time_part = time_with_dot_for_nanos.replace('-', ":");
66        format!("{date_part}T{final_time_part}Z")
67    } else {
68        // Fallback if no nanoseconds part found
69        let final_time_part = time_part.replace('-', ":");
70        format!("{date_part}T{final_time_part}Z")
71    }
72}
73
74/// Converts an ISO 8601 timestamp string to Unix nanoseconds.
75pub(crate) fn iso_to_unix_nanos(iso_timestamp: &str) -> anyhow::Result<u64> {
76    Ok(iso8601_to_unix_nanos(iso_timestamp)?.into())
77}
78
79// Extract the instrument ID portion from a bar type directory name.
80// Handles both standard and composite formats:
81//   {id}-{step}-{agg}-{price}-{source}
82//   {id}-{step}-{agg}-{price}-{source}@{step}-{agg}-{source}
83// Strips the composite suffix before parsing with rsplitn(5, '-').
84pub(crate) fn extract_bar_type_instrument_id(bar_type_dir: &str) -> Option<&str> {
85    let standard = bar_type_dir.split('@').next().unwrap_or(bar_type_dir);
86    let pieces: Vec<&str> = standard.rsplitn(5, '-').collect();
87    // pieces (reversed): [source, price_type, agg, step, instrument_id]
88    if pieces.len() == 5 && pieces[3].chars().all(|c| c.is_ascii_digit()) {
89        Some(pieces[4])
90    } else {
91        None
92    }
93}
94
95/// Extracts the filename from a file path and makes it SQL-safe.
96#[must_use]
97pub fn extract_sql_safe_filename(file_path: &str) -> String {
98    if file_path.is_empty() {
99        return "unknown_file".to_string();
100    }
101
102    let filename = file_path
103        .split(['/', '\\'])
104        .next_back()
105        .unwrap_or("unknown_file");
106
107    // Remove .parquet extension
108    let name_without_ext = if let Some(dot_pos) = filename.rfind(".parquet") {
109        &filename[..dot_pos]
110    } else {
111        filename
112    };
113
114    // Remove characters that can pose problems: hyphens, colons, etc.
115    name_without_ext
116        .replace(['-', ':', '.'], "_")
117        .to_lowercase()
118}
119
120/// Creates a platform-appropriate local path using `PathBuf`.
121pub fn make_local_path<P: AsRef<Path>>(base_path: P, components: &[&str]) -> PathBuf {
122    let mut path = PathBuf::from(base_path.as_ref());
123    for component in components {
124        path.push(component);
125    }
126    path
127}
128
129/// Checks if a filename's timestamp range intersects with a query interval.
130pub(crate) fn query_intersects_filename(
131    filename: &str,
132    start: Option<u64>,
133    end: Option<u64>,
134) -> bool {
135    if let Some((file_start, file_end)) = parse_filename_timestamps(filename) {
136        start.is_none_or(|start| start <= file_end) && end.is_none_or(|end| file_start <= end)
137    } else {
138        true
139    }
140}
141
142/// Parses timestamps from a Parquet filename.
143///
144/// Extracts the start and end timestamps from filenames that follow the ISO 8601 format:
145/// "`iso_timestamp_1_iso_timestamp_2.parquet`".
146#[must_use]
147pub fn parse_filename_timestamps(filename: &str) -> Option<(u64, u64)> {
148    let path = Path::new(filename);
149    let base_name = path.file_name()?.to_str()?;
150    let base_filename = base_name.strip_suffix(".parquet")?;
151    let mut parts = base_filename.split('_');
152    let first_part = parts.next()?;
153    let second_part = parts.next()?;
154    if let Some(replay_identity) = parts.next()
155        && (replay_identity.is_empty()
156            || !replay_identity
157                .bytes()
158                .all(|byte| byte.is_ascii_alphanumeric()))
159    {
160        return None;
161    }
162
163    if parts.next().is_some() {
164        return None;
165    }
166
167    let first_iso = file_timestamp_to_iso_timestamp(first_part);
168    let second_iso = file_timestamp_to_iso_timestamp(second_part);
169
170    let first_ts = iso_to_unix_nanos(&first_iso).ok()?;
171    let second_ts = iso_to_unix_nanos(&second_iso).ok()?;
172
173    Some((first_ts, second_ts))
174}
175
176#[cfg(test)]
177mod tests {
178    use nautilus_core::UnixNanos;
179    use rstest::rstest;
180
181    use super::{parse_filename_timestamps, timestamps_to_filename};
182
183    #[rstest]
184    fn parse_filename_timestamps_accepts_replay_identity_suffix() {
185        let base = timestamps_to_filename(UnixNanos::from(1), UnixNanos::from(2));
186        let filename = base.replace(".parquet", "_replay.parquet");
187
188        assert_eq!(parse_filename_timestamps(&filename), Some((1, 2)));
189    }
190
191    #[rstest]
192    fn parse_filename_timestamps_rejects_non_timestamp_segments() {
193        let base = timestamps_to_filename(UnixNanos::from(1), UnixNanos::from(2));
194        let filename = base.replace(".parquet", "_bad-suffix.parquet");
195
196        assert_eq!(parse_filename_timestamps(&filename), None);
197    }
198}
nautilus_persistence/backend/parquet/paths.rs

nautilus_persistence/backend/parquet/
paths.rs
