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/// Builds a catalog parquet filename, appending a promotion-identity hash when present.
47#[must_use]
48pub(crate) fn catalog_filename(
49    start_ts: UnixNanos,
50    end_ts: UnixNanos,
51    replay_identity: Option<&str>,
52) -> String {
53    let filename = timestamps_to_filename(start_ts, end_ts);
54
55    let Some(identity) = replay_identity else {
56        return filename;
57    };
58
59    let stem = filename.strip_suffix(".parquet").unwrap_or(&filename);
60    let digest = blake3::hash(identity.as_bytes()).to_hex();
61    format!("{stem}_{}.parquet", &digest[..16])
62}
63
64/// Converts an ISO 8601 timestamp to a filesystem-safe format.
65pub(crate) fn iso_timestamp_to_file_timestamp(iso_timestamp: &str) -> String {
66    iso_timestamp.replace([':', '.'], "-")
67}
68
69/// Converts a filesystem-safe timestamp back to ISO 8601 format.
70pub(crate) fn file_timestamp_to_iso_timestamp(file_timestamp: &str) -> String {
71    let (date_part, time_part) = file_timestamp
72        .split_once('T')
73        .unwrap_or((file_timestamp, ""));
74    let time_part = time_part.strip_suffix('Z').unwrap_or(time_part);
75
76    // Find the last hyphen to separate nanoseconds
77    let final_time_part = match time_part.rsplit_once('-') {
78        Some((time, nanos)) => format!("{}.{nanos}", time.replace('-', ":")),
79        // Fallback if no nanoseconds part found
80        None => time_part.replace('-', ":"),
81    };
82
83    format!("{date_part}T{final_time_part}Z")
84}
85
86// Extract the instrument ID portion from a bar type directory name.
87// Handles both standard and composite formats:
88//   {id}-{step}-{agg}-{price}-{source}
89//   {id}-{step}-{agg}-{price}-{source}@{step}-{agg}-{source}
90// Strips the composite suffix before parsing with rsplitn(5, '-').
91pub(crate) fn extract_bar_type_instrument_id(bar_type_dir: &str) -> Option<&str> {
92    let standard = bar_type_dir.split('@').next().unwrap_or(bar_type_dir);
93    let pieces: Vec<&str> = standard.rsplitn(5, '-').collect();
94    // pieces (reversed): [source, price_type, agg, step, instrument_id]
95    if pieces.len() == 5 && pieces[3].chars().all(|c| c.is_ascii_digit()) {
96        Some(pieces[4])
97    } else {
98        None
99    }
100}
101
102/// Extracts the filename from a file path and makes it SQL-safe.
103#[must_use]
104pub fn extract_sql_safe_filename(file_path: &str) -> String {
105    if file_path.is_empty() {
106        return "unknown_file".to_string();
107    }
108
109    let filename = file_path
110        .split(['/', '\\'])
111        .next_back()
112        .unwrap_or("unknown_file");
113
114    // Remove .parquet extension
115    let name_without_ext = if let Some(dot_pos) = filename.rfind(".parquet") {
116        &filename[..dot_pos]
117    } else {
118        filename
119    };
120
121    // Remove characters that can pose problems: hyphens, colons, etc.
122    name_without_ext
123        .replace(['-', ':', '.'], "_")
124        .to_lowercase()
125}
126
127/// Creates a platform-appropriate local path using `PathBuf`.
128pub fn make_local_path<P: AsRef<Path>>(base_path: P, components: &[&str]) -> PathBuf {
129    let mut path = base_path.as_ref().to_path_buf();
130    path.extend(components);
131    path
132}
133
134/// Checks if a filename's timestamp range intersects with a query interval.
135pub(crate) fn query_intersects_filename(
136    filename: &str,
137    start: Option<u64>,
138    end: Option<u64>,
139) -> bool {
140    let Some((file_start, file_end)) = parse_filename_timestamps(filename) else {
141        return true;
142    };
143
144    start.is_none_or(|start| start <= file_end) && end.is_none_or(|end| file_start <= end)
145}
146
147/// Parses timestamps from a Parquet filename.
148///
149/// Extracts the start and end timestamps from filenames that follow the ISO 8601 format:
150/// "`iso_timestamp_1_iso_timestamp_2.parquet`".
151#[must_use]
152pub fn parse_filename_timestamps(filename: &str) -> Option<(u64, u64)> {
153    let path = Path::new(filename);
154    let base_name = path.file_name()?.to_str()?;
155    let base_filename = base_name.strip_suffix(".parquet")?;
156    let mut parts = base_filename.split('_');
157    let first_part = parts.next()?;
158    let second_part = parts.next()?;
159    if let Some(replay_identity) = parts.next()
160        && (replay_identity.is_empty()
161            || !replay_identity
162                .bytes()
163                .all(|byte| byte.is_ascii_alphanumeric()))
164    {
165        return None;
166    }
167
168    if parts.next().is_some() {
169        return None;
170    }
171
172    let first_iso = file_timestamp_to_iso_timestamp(first_part);
173    let second_iso = file_timestamp_to_iso_timestamp(second_part);
174
175    let first_ts = iso8601_to_unix_nanos(&first_iso).ok()?.as_u64();
176    let second_ts = iso8601_to_unix_nanos(&second_iso).ok()?.as_u64();
177
178    Some((first_ts, second_ts))
179}
180
181#[cfg(test)]
182mod tests {
183    use nautilus_core::UnixNanos;
184    use rstest::rstest;
185
186    use super::{parse_filename_timestamps, query_intersects_filename, timestamps_to_filename};
187
188    #[rstest]
189    fn parse_filename_timestamps_accepts_replay_identity_suffix() {
190        let base = timestamps_to_filename(UnixNanos::from(1), UnixNanos::from(2));
191        let filename = base.replace(".parquet", "_replay.parquet");
192
193        assert_eq!(parse_filename_timestamps(&filename), Some((1, 2)));
194    }
195
196    #[rstest]
197    fn parse_filename_timestamps_rejects_non_timestamp_segments() {
198        let base = timestamps_to_filename(UnixNanos::from(1), UnixNanos::from(2));
199        let filename = base.replace(".parquet", "_bad-suffix.parquet");
200
201        assert_eq!(parse_filename_timestamps(&filename), None);
202    }
203
204    #[rstest]
205    #[case::unparsable_name_is_kept(None, Some(10), Some(20), true)]
206    #[case::start_on_file_end(Some((10, 20)), Some(20), None, true)]
207    #[case::start_after_file_end(Some((10, 20)), Some(21), None, false)]
208    #[case::end_on_file_start(Some((10, 20)), None, Some(10), true)]
209    #[case::end_before_file_start(Some((10, 20)), None, Some(9), false)]
210    #[case::unbounded(Some((10, 20)), None, None, true)]
211    fn query_intersects_filename_cases(
212        #[case] interval: Option<(u64, u64)>,
213        #[case] start: Option<u64>,
214        #[case] end: Option<u64>,
215        #[case] expected: bool,
216    ) {
217        let filename = interval.map_or_else(
218            || "value.parquet".to_string(),
219            |(file_start, file_end)| {
220                timestamps_to_filename(UnixNanos::from(file_start), UnixNanos::from(file_end))
221            },
222        );
223
224        assert_eq!(query_intersects_filename(&filename, start, end), expected);
225    }
226}
nautilus_persistence/backend/parquet/paths.rs

nautilus_persistence/backend/parquet/
paths.rs
