nautilus_persistence/backend/parquet/catalog/
coverage.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//! Interval coverage and missing-interval checks for the Parquet catalog.
17
18use super::{
19    Cow, ParquetDataCatalog, extract_bar_type_instrument_id, parse_filename_timestamps,
20    query::is_parquet_bar_prefix, query_interval_diff, urisafe_instrument_id, urlencoding,
21};
22use crate::catalog::types::{CatalogDataType, parquet_catalog_data_type_path_prefixes};
23
24impl ParquetDataCatalog {
25    /// Finds the missing time intervals for a specific data type and instrument ID.
26    ///
27    /// This method compares a requested time range against the existing data coverage
28    /// and returns the gaps that need to be filled. This is useful for determining
29    /// what data needs to be fetched or backfilled.
30    ///
31    /// # Parameters
32    ///
33    /// - `start`: Start timestamp of the requested range (Unix nanoseconds).
34    /// - `end`: End timestamp of the requested range (Unix nanoseconds).
35    /// - `data_type`: The stored family to inspect.
36    /// - `instrument_id`: Optional instrument ID to target a specific instrument's data.
37    ///
38    /// # Returns
39    ///
40    /// Returns a vector of (start, end) tuples representing the missing intervals,
41    /// or an error if the operation fails.
42    ///
43    /// # Errors
44    ///
45    /// Returns an error if:
46    /// - The directory path cannot be constructed.
47    /// - Interval retrieval fails.
48    /// - Gap calculation fails.
49    ///
50    /// # Examples
51    ///
52    /// ```rust,no_run
53    /// use nautilus_model::data::NautilusDataType;
54    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
55    ///
56    /// let mut catalog = ParquetDataCatalog::new(
57    ///     std::path::Path::new("/tmp/nautilus_data"),
58    ///     None,
59    ///     None,
60    ///     None,
61    ///     None,
62    /// );
63    ///
64    /// // Find missing intervals for quote data
65    /// let missing = catalog.get_missing_intervals_for_request(
66    ///     1609459200000000000, // start
67    ///     1609545600000000000, // end
68    ///     &NautilusDataType::QuoteTick.into(),
69    ///     Some("BTCUSD"),
70    /// )?;
71    ///
72    /// for (start, end) in missing {
73    ///     println!("Missing data from {} to {}", start, end);
74    /// }
75    /// # Ok::<(), anyhow::Error>(())
76    /// ```
77    pub fn get_missing_intervals_for_request(
78        &self,
79        start: u64,
80        end: u64,
81        data_type: &CatalogDataType,
82        identifier: Option<&str>,
83    ) -> anyhow::Result<Vec<(u64, u64)>> {
84        let intervals = self.get_intervals(data_type, identifier)?;
85
86        Ok(query_interval_diff(start, end, &intervals))
87    }
88
89    /// Gets the first (earliest) timestamp for a specific data type and identifier.
90    ///
91    /// This method finds the earliest timestamp covered by existing data files for
92    /// the specified data type and identifier. This is useful for determining
93    /// the oldest data available or for incremental data updates.
94    ///
95    /// # Parameters
96    ///
97    /// - `data_type`: The stored family to inspect.
98    /// - `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").
99    ///
100    /// # Returns
101    ///
102    /// Returns `Some(timestamp)` if data exists, `None` if no data is found,
103    /// or an error if the operation fails.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error if:
108    /// - The directory path cannot be constructed.
109    /// - Interval retrieval fails.
110    ///
111    /// # Note
112    ///
113    /// Unlike the Python implementation, this method does not check subclasses of the
114    /// data type. The Python version checks `[data_cls, *data_cls.__subclasses__()]` to
115    /// handle cases where subclasses might use different directory names. Since Rust
116    /// works with string names rather than types, subclass checking is not possible.
117    /// In practice, most subclasses map to the same directory name via `class_to_filename`,
118    /// so this difference is typically not significant.
119    ///
120    /// # Examples
121    ///
122    /// ```rust,no_run
123    /// use nautilus_model::data::NautilusDataType;
124    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
125    ///
126    /// let mut catalog = ParquetDataCatalog::new(
127    ///     std::path::Path::new("/tmp/nautilus_data"),
128    ///     None,
129    ///     None,
130    ///     None,
131    ///     None,
132    /// );
133    ///
134    /// // Get the first timestamp for quote data
135    /// if let Some(first_ts) =
136    ///     catalog.query_first_timestamp(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?
137    /// {
138    ///     println!("First quote timestamp: {}", first_ts);
139    /// } else {
140    ///     println!("No quote data found");
141    /// }
142    /// # Ok::<(), anyhow::Error>(())
143    /// ```
144    pub fn query_first_timestamp(
145        &self,
146        data_type: &CatalogDataType,
147        identifier: Option<&str>,
148    ) -> anyhow::Result<Option<u64>> {
149        let intervals = self.get_intervals(data_type, identifier)?;
150
151        Ok(intervals.first().map(|interval| interval.0))
152    }
153
154    /// Gets the last (most recent) timestamp for a specific data type and identifier.
155    ///
156    /// This method finds the latest timestamp covered by existing data files for
157    /// the specified data type and identifier. This is useful for determining
158    /// the most recent data available or for incremental data updates.
159    ///
160    /// # Parameters
161    ///
162    /// - `data_type`: The stored family to inspect.
163    /// - `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").
164    ///
165    /// # Returns
166    ///
167    /// Returns `Some(timestamp)` if data exists, `None` if no data is found,
168    /// or an error if the operation fails.
169    ///
170    /// # Errors
171    ///
172    /// Returns an error if:
173    /// - The directory path cannot be constructed.
174    /// - Interval retrieval fails.
175    ///
176    /// # Note
177    ///
178    /// Unlike the Python implementation, this method does not check subclasses of the
179    /// data type. The Python version checks `[data_cls, *data_cls.__subclasses__()]` to
180    /// handle cases where subclasses might use different directory names. Since Rust
181    /// works with string names rather than types, subclass checking is not possible.
182    /// In practice, most subclasses map to the same directory name via `class_to_filename`,
183    /// so this difference is typically not significant.
184    ///
185    /// # Examples
186    ///
187    /// ```rust,no_run
188    /// use nautilus_model::data::NautilusDataType;
189    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
190    ///
191    /// let mut catalog = ParquetDataCatalog::new(
192    ///     std::path::Path::new("/tmp/nautilus_data"),
193    ///     None,
194    ///     None,
195    ///     None,
196    ///     None,
197    /// );
198    ///
199    /// // Get the last timestamp for quote data
200    /// if let Some(last_ts) =
201    ///     catalog.query_last_timestamp(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?
202    /// {
203    ///     println!("Last quote timestamp: {}", last_ts);
204    /// } else {
205    ///     println!("No quote data found");
206    /// }
207    /// # Ok::<(), anyhow::Error>(())
208    /// ```
209    pub fn query_last_timestamp(
210        &self,
211        data_type: &CatalogDataType,
212        identifier: Option<&str>,
213    ) -> anyhow::Result<Option<u64>> {
214        let intervals = self.get_intervals(data_type, identifier)?;
215
216        Ok(intervals.last().map(|interval| interval.1))
217    }
218
219    /// Gets the time intervals covered by Parquet files for a specific data type and identifier.
220    ///
221    /// This method returns all time intervals covered by existing data files for the
222    /// specified data type and identifier. The intervals are sorted by start time and
223    /// represent the complete data coverage available.
224    ///
225    /// # Parameters
226    ///
227    /// - `data_type`: The stored family to inspect.
228    /// - `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").
229    ///
230    /// # Returns
231    ///
232    /// Returns a vector of (start, end) tuples representing the covered intervals,
233    /// sorted by start time, or an error if the operation fails.
234    ///
235    /// # Errors
236    ///
237    /// Returns an error if:
238    /// - The directory path cannot be constructed.
239    /// - Directory listing fails.
240    /// - Filename parsing fails.
241    ///
242    /// # Examples
243    ///
244    /// ```rust,no_run
245    /// use nautilus_model::data::NautilusDataType;
246    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
247    ///
248    /// let mut catalog = ParquetDataCatalog::new(
249    ///     std::path::Path::new("/tmp/nautilus_data"),
250    ///     None,
251    ///     None,
252    ///     None,
253    ///     None,
254    /// );
255    ///
256    /// // Get all intervals for quote data
257    /// let intervals = catalog.get_intervals(&NautilusDataType::QuoteTick.into(), Some("BTCUSD"))?;
258    /// for (start, end) in intervals {
259    ///     println!("Data available from {} to {}", start, end);
260    /// }
261    /// # Ok::<(), anyhow::Error>(())
262    /// ```
263    pub fn get_intervals(
264        &self,
265        data_type: &CatalogDataType,
266        identifier: Option<&str>,
267    ) -> anyhow::Result<Vec<(u64, u64)>> {
268        let prefixes = parquet_catalog_data_type_path_prefixes(data_type);
269
270        if let [data_cls] = prefixes.as_slice() {
271            return self.get_prefix_intervals(data_cls.as_ref(), identifier);
272        }
273
274        let mut intervals = Vec::new();
275        for data_cls in &prefixes {
276            intervals.extend(self.get_prefix_intervals(data_cls.as_ref(), identifier)?);
277        }
278
279        intervals.sort_by_key(|&(start, _)| start);
280        Ok(merge_overlapping(intervals))
281    }
282
283    fn get_prefix_intervals(
284        &self,
285        data_cls: &str,
286        identifier: Option<&str>,
287    ) -> anyhow::Result<Vec<(u64, u64)>> {
288        let directory = self.make_path(data_cls, identifier)?;
289        let intervals = self.get_directory_intervals(&directory)?;
290
291        if identifier.is_none() {
292            // `get_directory_intervals` already recursed through every per-identifier
293            // subdirectory via `object_store.list`, so intervals from different
294            // identifiers can overlap. Merge overlaps into a disjoint sorted union
295            // so callers like `query_last_timestamp` see the true max end and
296            // `consolidate_data_by_period` sees contiguous coverage.
297            return Ok(merge_overlapping(intervals));
298        }
299
300        // For bars, fall back to partial matching when the exact directory
301        // doesn't exist (callers may pass an instrument_id like "EUR/USD.SIM"
302        // but bars are stored under bar_type dirs like "EURUSD.SIM-1-MINUTE-...")
303
304        if !intervals.is_empty() || !is_parquet_bar_prefix(data_cls) {
305            return Ok(intervals);
306        }
307
308        let safe_id = urisafe_instrument_id(identifier.unwrap());
309
310        // Use relative path so list_directory_stems doesn't double-prefix
311        // for remote catalogs (make_path already includes base_path)
312        let bars_subdir = format!("data/{data_cls}");
313        let subdirs = self.list_directory_stems(&bars_subdir)?;
314
315        let mut all_intervals = Vec::new();
316
317        for subdir in &subdirs {
318            let decoded = urlencoding::decode(subdir).unwrap_or(Cow::Borrowed(subdir));
319
320            if extract_bar_type_instrument_id(&decoded) == Some(safe_id.as_str()) {
321                // Use decoded name to avoid double percent-encoding
322                // (to_object_path uses Path::from which re-encodes)
323                let subdir_path = self.make_path(data_cls, Some(&decoded))?;
324                all_intervals.extend(self.get_directory_intervals(&subdir_path)?);
325            }
326        }
327
328        all_intervals.sort_by_key(|&(start, _)| start);
329
330        // Merge overlapping intervals from different bar types so that
331        // last().1 reliably gives the maximum end timestamp
332        Ok(merge_overlapping(all_intervals))
333    }
334
335    /// Gets the time intervals covered by Parquet files in a specific directory.
336    ///
337    /// This method scans a directory for Parquet files and extracts the timestamp ranges
338    /// from their filenames. It's used internally by other methods to determine data coverage
339    /// and is essential for interval-based operations like gap detection and consolidation.
340    ///
341    /// # Parameters
342    ///
343    /// - `directory`: The directory path to scan for Parquet files.
344    ///
345    /// # Returns
346    ///
347    /// Returns a vector of (start, end) tuples representing the time intervals covered
348    /// by files in the directory, sorted by start timestamp. Returns an empty vector
349    /// if the directory doesn't exist or contains no valid Parquet files.
350    ///
351    /// # Errors
352    ///
353    /// Returns an error if:
354    /// - Object store listing operations fail.
355    /// - Directory access is denied.
356    ///
357    /// # Notes
358    ///
359    /// - Only files with valid timestamp-based filenames are included.
360    /// - Files with unparsable names are silently ignored.
361    /// - The method works with both local and remote object stores.
362    /// - Results are automatically sorted by start timestamp.
363    ///
364    /// # Examples
365    ///
366    /// ```rust,no_run
367    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
368    ///
369    /// let mut catalog = ParquetDataCatalog::new(
370    ///     std::path::Path::new("/tmp/nautilus_data"),
371    ///     None,
372    ///     None,
373    ///     None,
374    ///     None,
375    /// );
376    /// let intervals = catalog.get_directory_intervals("data/quotes/EURUSD")?;
377    ///
378    /// for (start, end) in intervals {
379    ///     println!("File covers {} to {}", start, end);
380    /// }
381    /// # Ok::<(), anyhow::Error>(())
382    /// ```
383    pub fn get_directory_intervals(&self, directory: &str) -> anyhow::Result<Vec<(u64, u64)>> {
384        // Use object store for all operations
385        // Convert directory to object path format (consistent with how files are written)
386        // For local stores with empty base_path, to_object_path returns path as-is.
387        // For remote stores, to_object_path preserves or prepends the catalog base path.
388        let object_dir = self.to_object_path(directory)?;
389        let list_result = self.list_objects(object_dir.as_ref())?;
390
391        let mut intervals = Vec::new();
392
393        for object in list_result {
394            let path_str = object.location.to_string();
395            if path_str.ends_with(".parquet")
396                && let Some(interval) = parse_filename_timestamps(&path_str)
397            {
398                intervals.push(interval);
399            }
400        }
401
402        intervals.sort_by_key(|&(start, _)| start);
403
404        Ok(intervals)
405    }
406}
407
408/// Merges overlapping intervals (sorted by start) into a disjoint sorted union.
409///
410/// Adjacent intervals stay separate, unlike [`crate::common::coverage::merge_closed_intervals`]:
411/// these intervals describe stored files, and `are_intervals_contiguous` checks that consecutive
412/// files abut exactly, which merging them away would hide.
413fn merge_overlapping(intervals: Vec<(u64, u64)>) -> Vec<(u64, u64)> {
414    let mut merged: Vec<(u64, u64)> = Vec::new();
415
416    for interval in intervals {
417        if let Some(last) = merged.last_mut()
418            && interval.0 <= last.1
419        {
420            last.1 = last.1.max(interval.1);
421            continue;
422        }
423        merged.push(interval);
424    }
425
426    merged
427}
nautilus_persistence/backend/parquet/catalog/coverage.rs

nautilus_persistence/backend/parquet/catalog/
coverage.rs
