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