nautilus_persistence/backend/parquet/file_admin.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//! File-level admin operations: existence checks, deletion, name resets, leaf-directory walk.
17
18use std::collections::BTreeMap;
19
20use ahash::AHashSet;
21use futures::StreamExt;
22use nautilus_core::UnixNanos;
23use object_store::{ObjectStoreExt, path::Path as ObjectPath};
24
25use crate::{
26 backend::parquet::{
27 catalog::ParquetDataCatalog,
28 intervals::are_intervals_disjoint,
29 io::{decode_object_store_segment, min_max_from_parquet_metadata_object_store},
30 paths::{make_object_store_path, timestamps_to_filename},
31 },
32 catalog::types::{CatalogDataType, parquet_catalog_data_type_path_prefixes},
33};
34
35impl ParquetDataCatalog {
36 /// Checks if a file exists in the object store.
37 ///
38 /// This method performs a HEAD operation on the object store to determine if a file
39 /// exists without downloading its content. It works with both local and remote object stores.
40 ///
41 /// # Parameters
42 ///
43 /// - `path`: The file path to check, relative to the catalog structure.
44 ///
45 /// # Returns
46 ///
47 /// Returns `true` if the file exists, `false` if it doesn't exist.
48 ///
49 /// # Errors
50 ///
51 /// Returns an error if the object store operation fails due to network issues,
52 /// authentication problems, or other I/O errors.
53 pub(crate) fn file_exists(&self, path: &str) -> anyhow::Result<bool> {
54 let object_path = self.to_object_path(path)?;
55
56 self.execute_async(|| async {
57 match self.object_store.head(&object_path).await {
58 Ok(_) => Ok(true),
59 Err(object_store::Error::NotFound { .. }) => Ok(false),
60 Err(e) => Err(e.into()),
61 }
62 })
63 }
64
65 /// Deletes a file from the object store.
66 ///
67 /// This method removes a file from the object store. The operation is permanent
68 /// and cannot be undone. It works with both local filesystems and remote object stores.
69 ///
70 /// # Parameters
71 ///
72 /// - `path`: The file path to delete, as an object-store listing returns it.
73 ///
74 /// # Returns
75 ///
76 /// Returns `Ok(())` on successful deletion.
77 ///
78 /// # Errors
79 ///
80 /// Returns an error if:
81 /// - The file doesn't exist.
82 /// - Permission is denied.
83 /// - Network issues occur (for remote stores).
84 /// - The object store operation fails.
85 ///
86 /// # Safety
87 ///
88 /// This operation is irreversible. Ensure the file is no longer needed before deletion.
89 pub(crate) fn delete_file(&self, path: &str) -> anyhow::Result<()> {
90 let object_path = self.to_object_path_parsed(path)?;
91 self.execute_async(|| async {
92 self.object_store
93 .delete(&object_path)
94 .await
95 .map_err(anyhow::Error::from)
96 })
97 }
98
99 /// Resets the filenames of all Parquet files in the catalog to match their actual content timestamps.
100 ///
101 /// This method scans all leaf data directories in the catalog and renames files based on
102 /// the actual timestamp range of their content. This is useful when files have been
103 /// modified or when filename conventions have changed.
104 ///
105 /// # Returns
106 ///
107 /// Returns `Ok(())` on success, or an error if the operation fails.
108 ///
109 /// # Errors
110 ///
111 /// Returns an error if:
112 /// - Directory listing fails.
113 /// - File metadata reading fails.
114 /// - File rename operations fail.
115 /// - The content intervals in a directory are not disjoint, or a new name matches another
116 /// file's current name. Both are checked before any file in that directory is renamed.
117 ///
118 /// # Examples
119 ///
120 /// ```rust,no_run
121 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
122 ///
123 /// let mut catalog = ParquetDataCatalog::new(
124 /// std::path::Path::new("/tmp/nautilus_data"),
125 /// None,
126 /// None,
127 /// None,
128 /// None,
129 /// );
130 ///
131 /// // Reset all filenames in the catalog
132 /// catalog.reset_all_file_names()?;
133 /// # Ok::<(), anyhow::Error>(())
134 /// ```
135 pub fn reset_all_file_names(&self) -> anyhow::Result<()> {
136 let leaf_directories = self.find_leaf_data_directories()?;
137
138 for directory in leaf_directories {
139 self.reset_file_names(&directory)?;
140 }
141
142 Ok(())
143 }
144
145 /// Resets the filenames of Parquet files for a specific data type and identifier.
146 ///
147 /// This method renames files in a specific directory based on the actual timestamp
148 /// range of their content. This is useful for correcting filenames after data
149 /// modifications or when filename conventions have changed.
150 ///
151 /// # Parameters
152 ///
153 /// - `data_type`: The stored family to target.
154 /// - `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").
155 ///
156 /// # Returns
157 ///
158 /// Returns `Ok(())` on success, or an error if the operation fails.
159 ///
160 /// # Errors
161 ///
162 /// Returns an error if:
163 /// - The directory path cannot be constructed.
164 /// - File metadata reading fails.
165 /// - File rename operations fail.
166 /// - The content intervals in a directory are not disjoint, or a new name matches another
167 /// file's current name. Both are checked before any file in that directory is renamed.
168 ///
169 /// # Examples
170 ///
171 /// ```rust,no_run
172 /// use nautilus_model::data::NautilusDataType;
173 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
174 ///
175 /// let mut catalog = ParquetDataCatalog::new(
176 /// std::path::Path::new("/tmp/nautilus_data"),
177 /// None,
178 /// None,
179 /// None,
180 /// None,
181 /// );
182 ///
183 /// // Reset filenames for all quote files
184 /// catalog.reset_data_file_names(&NautilusDataType::QuoteTick.into(), None)?;
185 ///
186 /// // Reset filenames for a specific instrument's trade files
187 /// catalog.reset_data_file_names(&NautilusDataType::TradeTick.into(), Some("BTCUSD"))?;
188 /// # Ok::<(), anyhow::Error>(())
189 /// ```
190 pub fn reset_data_file_names(
191 &self,
192 data_type: &CatalogDataType,
193 identifier: Option<&str>,
194 ) -> anyhow::Result<()> {
195 for type_name in parquet_catalog_data_type_path_prefixes(data_type) {
196 let directory = self.make_path(type_name.as_ref(), identifier)?;
197 self.reset_file_names(&directory)?;
198 }
199
200 Ok(())
201 }
202
203 /// Resets the filenames of Parquet files in a directory to match their actual content timestamps.
204 ///
205 /// This internal method scans all Parquet files in a directory, reads their metadata to
206 /// determine the actual timestamp range of their content, and renames the files accordingly.
207 /// This ensures that filenames accurately reflect the data they contain. The listing includes
208 /// subdirectories, so each file keeps its own directory and is validated against the files
209 /// that share it.
210 ///
211 /// # Parameters
212 ///
213 /// - `directory`: The directory path containing Parquet files to rename.
214 ///
215 /// # Returns
216 ///
217 /// Returns `Ok(())` on success, or an error if the operation fails.
218 ///
219 /// # Process
220 ///
221 /// 1. Lists all Parquet files in the directory
222 /// 2. For each file, reads metadata to extract min/max timestamps
223 /// 3. Generates a new filename based on actual timestamp range
224 /// 4. Validates the new names before moving any file
225 /// 5. Moves each file to its new name using object store operations
226 ///
227 /// # Errors
228 ///
229 /// Returns an error if:
230 /// - Directory listing fails.
231 /// - Metadata reading fails for any file.
232 /// - The content intervals within a file's directory are not disjoint.
233 /// - A new name matches the current name of another file.
234 /// - File move operations fail.
235 /// - Object store operations fail.
236 ///
237 /// # Notes
238 ///
239 /// - This operation can be time-consuming for directories with many files.
240 /// - A validation error leaves every file name unchanged.
241 /// - Files are renamed one at a time. Object stores without a native rename copy then
242 /// delete, so a failure can leave a file under both names.
243 fn reset_file_names(&self, directory: &str) -> anyhow::Result<()> {
244 let parquet_files = self.list_parquet_files(directory)?;
245 let mut intervals_by_directory = BTreeMap::<ObjectPath, Vec<(u64, u64)>>::new();
246 let mut moves = Vec::with_capacity(parquet_files.len());
247
248 for file in parquet_files {
249 let object_path = self.to_object_path_parsed(&file)?;
250
251 let (first_ts, last_ts) = self.execute_async(|| async {
252 min_max_from_parquet_metadata_object_store(
253 self.object_store.clone(),
254 &object_path,
255 "ts_init",
256 )
257 .await
258 })?;
259
260 let new_filename =
261 timestamps_to_filename(UnixNanos::from(first_ts), UnixNanos::from(last_ts));
262 let file_directory = object_path.parent().unwrap_or_default();
263 let new_object_path = file_directory.clone().join(new_filename.as_str());
264
265 intervals_by_directory
266 .entry(file_directory)
267 .or_default()
268 .push((first_ts, last_ts));
269 moves.push((object_path, new_object_path));
270 }
271
272 for (file_directory, intervals) in &intervals_by_directory {
273 anyhow::ensure!(
274 are_intervals_disjoint(intervals),
275 "Cannot reset file names in directory '{file_directory}': content intervals are \
276 not disjoint: {intervals:?}",
277 );
278 }
279
280 // Moves overwrite, so a new name held by another file would replace that file
281 let current_paths = moves
282 .iter()
283 .map(|(old_path, _)| old_path)
284 .collect::<AHashSet<_>>();
285
286 if let Some((old_path, new_path)) = moves
287 .iter()
288 .find(|(old_path, new_path)| old_path != new_path && current_paths.contains(new_path))
289 {
290 anyhow::bail!(
291 "Cannot reset file names in directory '{directory}': new name {new_path} for \
292 {old_path} is held by another file; rename that file to an unused interval \
293 name and retry"
294 );
295 }
296
297 for (old_path, new_path) in &moves {
298 self.move_file(old_path, new_path)?;
299 }
300
301 Ok(())
302 }
303
304 /// Finds all leaf data directories in the catalog.
305 ///
306 /// A leaf directory is one that contains data files but no subdirectories.
307 /// This method is used to identify directories that can be processed for
308 /// consolidation or other operations.
309 ///
310 /// # Returns
311 ///
312 /// Returns a vector of directory path strings representing leaf directories,
313 /// or an error if directory traversal fails.
314 ///
315 /// # Errors
316 ///
317 /// Returns an error if:
318 /// - Object store listing operations fail.
319 /// - Directory structure cannot be analyzed.
320 ///
321 /// # Examples
322 ///
323 /// ```rust,no_run
324 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
325 ///
326 /// let mut catalog = ParquetDataCatalog::new(
327 /// std::path::Path::new("/tmp/nautilus_data"),
328 /// None,
329 /// None,
330 /// None,
331 /// None,
332 /// );
333 ///
334 /// let leaf_dirs = catalog.find_leaf_data_directories()?;
335 /// for dir in leaf_dirs {
336 /// println!("Found leaf directory: {}", dir);
337 /// }
338 /// # Ok::<(), anyhow::Error>(())
339 /// ```
340 pub fn find_leaf_data_directories(&self) -> anyhow::Result<Vec<String>> {
341 let data_dir = make_object_store_path(&self.base_path, ["data"]);
342
343 self.execute_async(|| async {
344 let mut directories = AHashSet::new();
345
346 // List all objects under the data directory
347 let prefix = ObjectPath::from(format!("{data_dir}/"));
348 let mut stream = self.object_store.list(Some(&prefix));
349
350 while let Some(object) = stream.next().await {
351 let object = object?;
352 let parts = object.location.parts().collect::<Vec<_>>();
353
354 // Decode to match `make_path` output, which callers encode again
355 if let Some((_, parent)) = parts.split_last() {
356 let directory = parent
357 .iter()
358 .map(|part| decode_object_store_segment(part.as_ref()))
359 .collect::<Vec<_>>()
360 .join("/");
361 directories.insert(directory);
362 }
363 }
364
365 // Find leaf directories (every listed directory contains at least one file,
366 // so a leaf is one without subdirectories)
367 let mut leaf_dirs = Vec::new();
368
369 for dir in &directories {
370 let prefix = format!("{dir}/");
371 let has_subdirs = directories.iter().any(|d| d.starts_with(&prefix));
372
373 if !has_subdirs {
374 leaf_dirs.push(dir.clone());
375 }
376 }
377
378 leaf_dirs.sort();
379
380 Ok::<Vec<String>, anyhow::Error>(leaf_dirs)
381 })
382 }
383}