nautilus_persistence/backend/parquet/delete.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//! Range-based deletion and split operations for the parquet catalog.
17
18#![expect(
19 clippy::missing_errors_doc,
20 reason = "delete operations forward catalog/storage errors and operate on validated batches"
21)]
22
23use ahash::AHashSet;
24use nautilus_core::UnixNanos;
25use nautilus_model::data::{
26 Bar, CustomData, Data, FundingRateUpdate, HasTsInit, IndexPriceUpdate, InstrumentStatus,
27 MarkPriceUpdate, NautilusDataType, OptionGreeks, OrderBookDelta, OrderBookDepth, QuoteTick,
28 TradeTick, close::InstrumentClose,
29};
30use nautilus_serialization::arrow::{DecodeTypedFromRecordBatch, EncodeToRecordBatch};
31
32use crate::{
33 backend::parquet::{
34 catalog::ParquetDataCatalog,
35 paths::{make_object_store_path, parse_filename_timestamps, timestamps_to_filename},
36 },
37 catalog::types::{
38 HasCatalogDataType, data_type_from_data_path_prefix, parquet_data_path_prefix,
39 },
40 common::custom::group_custom_data_by_type,
41};
42
43/// Kind of deletion operation to execute.
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum DeleteOperationKind {
46 /// Remove the files entirely.
47 Remove,
48 /// Rewrite the data preserved before the deleted range, then remove the files.
49 SplitBefore,
50 /// Rewrite the data preserved after the deleted range, then remove the files.
51 SplitAfter,
52}
53
54/// Information about a deletion operation to be executed.
55#[derive(Debug, Clone)]
56pub struct DeleteOperation {
57 /// Kind of deletion operation.
58 pub kind: DeleteOperationKind,
59 /// List of files involved in this operation.
60 pub files: Vec<String>,
61 /// Start timestamp for data query (used for split operations).
62 pub query_start: u64,
63 /// End timestamp for data query (used for split operations).
64 pub query_end: u64,
65 /// Start timestamp for new file naming (used for split operations).
66 pub file_start_ns: u64,
67 /// End timestamp for new file naming (used for split operations).
68 pub file_end_ns: u64,
69}
70
71impl ParquetDataCatalog {
72 /// Deletes data within a specified time range across the entire catalog.
73 ///
74 /// This method identifies all leaf directories in the catalog that contain parquet files
75 /// and deletes data within the specified time range from each directory. A leaf directory
76 /// is one that contains files but no subdirectories. This is a convenience method that
77 /// effectively calls `delete_data_range` for all data types and instrument IDs in the catalog.
78 ///
79 /// # Parameters
80 ///
81 /// - `start`: Optional start timestamp for the deletion range. If None, deletes from the beginning.
82 /// - `end`: Optional end timestamp for the deletion range. If None, deletes to the end.
83 ///
84 /// # Returns
85 ///
86 /// Returns `Ok(())` on success, or an error if deletion fails.
87 ///
88 /// # Errors
89 ///
90 /// Returns an error if:
91 /// - Directory traversal fails.
92 /// - Data class extraction from paths fails.
93 /// - Individual delete operations fail.
94 ///
95 /// # Notes
96 ///
97 /// - This operation permanently removes data and cannot be undone.
98 /// - The deletion process handles file intersections intelligently by splitting files
99 /// when they partially overlap with the deletion range.
100 /// - Files completely within the deletion range are removed entirely.
101 /// - Files partially overlapping the deletion range are split to preserve data outside the range.
102 /// - This method is useful for bulk data cleanup operations across the entire catalog.
103 /// - Empty directories are not automatically removed after deletion.
104 ///
105 /// # Examples
106 ///
107 /// ```rust,no_run
108 /// use nautilus_core::UnixNanos;
109 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
110 ///
111 /// let mut catalog = ParquetDataCatalog::new(
112 /// std::path::Path::new("/tmp/nautilus_data"),
113 /// None,
114 /// None,
115 /// None,
116 /// None,
117 /// );
118 ///
119 /// // Delete all data before a specific date across entire catalog
120 /// catalog.delete_catalog_range(None, Some(UnixNanos::from(1609459200000000000)))?;
121 ///
122 /// // Delete all data within a specific range across entire catalog
123 /// catalog.delete_catalog_range(
124 /// Some(UnixNanos::from(1609459200000000000)),
125 /// Some(UnixNanos::from(1609545600000000000)),
126 /// )?;
127 ///
128 /// // Delete all data after a specific date across entire catalog
129 /// catalog.delete_catalog_range(Some(UnixNanos::from(1609459200000000000)), None)?;
130 /// # Ok::<(), anyhow::Error>(())
131 /// ```
132 pub fn delete_catalog_range(
133 &mut self,
134 start: Option<UnixNanos>,
135 end: Option<UnixNanos>,
136 ) -> anyhow::Result<()> {
137 let leaf_directories = self.find_leaf_data_directories()?;
138
139 for directory in leaf_directories {
140 let Ok((Some(data_cls), identifier)) =
141 self.extract_data_cls_and_identifier_from_path(&directory)
142 else {
143 continue;
144 };
145
146 let Ok(data_type) = data_type_from_data_path_prefix(&data_cls) else {
147 log::warn!("Skipping directory {directory}: unknown data class {data_cls}");
148 continue;
149 };
150
151 self.delete_data_range(&data_type, identifier.as_deref(), start, end)?;
152 }
153
154 Ok(())
155 }
156
157 /// Deletes data within a specified time range for a specific data type and identifier.
158 ///
159 /// This method identifies all parquet files that intersect with the specified time range
160 /// and handles them appropriately:
161 /// - Files completely within the range are deleted
162 /// - Files partially overlapping the range are split to preserve data outside the range
163 /// - The original intersecting files are removed after processing
164 ///
165 /// # Parameters
166 ///
167 /// - `data_type`: The data type to delete from.
168 /// - `identifier`: Optional identifier to delete data for. Can be an `instrument_id` (e.g., "EUR/USD.SIM") or a `bar_type` (e.g., "EUR/USD.SIM-1-MINUTE-LAST-EXTERNAL"). If None, deletes data across all identifiers.
169 /// - `start`: Optional start timestamp for the deletion range. If None, deletes from the beginning.
170 /// - `end`: Optional end timestamp for the deletion range. If None, deletes to the end.
171 ///
172 /// # Returns
173 ///
174 /// Returns `Ok(())` on success, or an error if deletion fails.
175 ///
176 /// # Errors
177 ///
178 /// Returns an error if:
179 /// - The directory path cannot be constructed.
180 /// - File operations fail.
181 /// - Data querying or writing fails.
182 ///
183 /// # Notes
184 ///
185 /// - This operation permanently removes data and cannot be undone.
186 /// - Files that partially overlap the deletion range are split to preserve data outside the range.
187 /// - The method ensures data integrity by using atomic operations where possible.
188 /// - Empty directories are not automatically removed after deletion.
189 ///
190 /// # Examples
191 ///
192 /// ```rust,no_run
193 /// use nautilus_core::UnixNanos;
194 /// use nautilus_model::data::NautilusDataType;
195 /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
196 ///
197 /// let mut catalog = ParquetDataCatalog::new(
198 /// std::path::Path::new("/tmp/nautilus_data"),
199 /// None,
200 /// None,
201 /// None,
202 /// None,
203 /// );
204 ///
205 /// // Delete all quote data for a specific instrument
206 /// catalog.delete_data_range(&NautilusDataType::QuoteTick, Some("BTCUSD"), None, None)?;
207 ///
208 /// // Delete trade data within a specific time range
209 /// catalog.delete_data_range(
210 /// &NautilusDataType::TradeTick,
211 /// None,
212 /// Some(UnixNanos::from(1609459200000000000)),
213 /// Some(UnixNanos::from(1609545600000000000)),
214 /// )?;
215 /// # Ok::<(), anyhow::Error>(())
216 /// ```
217 pub fn delete_data_range(
218 &mut self,
219 data_type: &NautilusDataType,
220 identifier: Option<&str>,
221 start: Option<UnixNanos>,
222 end: Option<UnixNanos>,
223 ) -> anyhow::Result<()> {
224 match data_type {
225 NautilusDataType::QuoteTick => {
226 self.delete_data_range_generic::<QuoteTick>(identifier, start, end)
227 }
228 NautilusDataType::TradeTick => {
229 self.delete_data_range_generic::<TradeTick>(identifier, start, end)
230 }
231 NautilusDataType::Bar => self.delete_data_range_generic::<Bar>(identifier, start, end),
232 NautilusDataType::OrderBookDelta => {
233 self.delete_data_range_generic::<OrderBookDelta>(identifier, start, end)
234 }
235 NautilusDataType::OrderBookDepth => {
236 self.delete_data_range_generic::<OrderBookDepth>(identifier, start, end)
237 }
238 NautilusDataType::MarkPriceUpdate => {
239 self.delete_data_range_generic::<MarkPriceUpdate>(identifier, start, end)
240 }
241 NautilusDataType::IndexPriceUpdate => {
242 self.delete_data_range_generic::<IndexPriceUpdate>(identifier, start, end)
243 }
244 NautilusDataType::InstrumentClose => {
245 self.delete_data_range_generic::<InstrumentClose>(identifier, start, end)
246 }
247 NautilusDataType::FundingRateUpdate => {
248 self.delete_data_range_generic::<FundingRateUpdate>(identifier, start, end)
249 }
250 NautilusDataType::OptionGreeks => {
251 self.delete_data_range_generic::<OptionGreeks>(identifier, start, end)
252 }
253 NautilusDataType::InstrumentStatus => {
254 self.delete_data_range_generic::<InstrumentStatus>(identifier, start, end)
255 }
256 NautilusDataType::Custom { type_name } => {
257 self.delete_custom_data_range(type_name, identifier, start, end)
258 }
259 other @ NautilusDataType::Instrument => {
260 anyhow::bail!("Unsupported data type: {other}")
261 }
262 #[cfg(feature = "defi")]
263 other @ NautilusDataType::Defi => {
264 anyhow::bail!("Unsupported data type: {other}")
265 }
266 #[cfg(not(feature = "defi"))]
267 #[allow(
268 unreachable_patterns,
269 reason = "DeFi variants can exist without this crate's defi feature"
270 )]
271 other => anyhow::bail!("Unsupported data type: {other}"),
272 }
273 }
274
275 /// Generic implementation for deleting data within a specified time range.
276 ///
277 /// This method provides the core deletion logic that works with any data type
278 /// that implements the required traits. It handles file intersection analysis,
279 /// data splitting for partial overlaps, and file cleanup.
280 ///
281 /// # Type Parameters
282 ///
283 /// - `T`: The data type that implements required traits for catalog operations.
284 ///
285 /// # Parameters
286 ///
287 /// - `identifier`: Optional instrument ID to delete data for.
288 /// - `start`: Optional start timestamp for the deletion range.
289 /// - `end`: Optional end timestamp for the deletion range.
290 ///
291 /// # Returns
292 ///
293 /// Returns `Ok(())` on success, or an error if deletion fails.
294 pub fn delete_data_range_generic<T>(
295 &mut self,
296 identifier: Option<&str>,
297 start: Option<UnixNanos>,
298 end: Option<UnixNanos>,
299 ) -> anyhow::Result<()>
300 where
301 T: DecodeTypedFromRecordBatch
302 + HasCatalogDataType
303 + EncodeToRecordBatch
304 + HasTsInit
305 + TryFrom<Data>
306 + Clone,
307 {
308 // Get intervals for cleaner implementation
309 let data_type = T::catalog_data_type();
310 let files = self.files_for_delete(&data_type, identifier)?;
311
312 if files.is_empty() {
313 return Ok(());
314 }
315
316 let operations_to_execute = operations_for_files(&files, start, end);
317
318 if operations_to_execute.is_empty() {
319 return Ok(()); // No operations to execute
320 }
321
322 // Execute all operations
323 let mut files_to_remove = AHashSet::<String>::new();
324
325 for operation in operations_to_execute {
326 // Reset the session before each operation to ensure fresh data is loaded
327 // This clears any cached table registrations that might interfere with file operations
328 self.clear_session_tables();
329
330 match operation.kind {
331 DeleteOperationKind::SplitBefore | DeleteOperationKind::SplitAfter => {
332 // Query the data preserved by the split and write it
333 // Use optimize_file_loading=false for precise file control during split operations
334 let instrument_ids = identifier.map(|id| vec![id.to_string()]);
335 let preserved_data = self.query::<T>(
336 instrument_ids,
337 Some(UnixNanos::from(operation.query_start)),
338 Some(UnixNanos::from(operation.query_end)),
339 None,
340 Some(operation.files.clone()),
341 false, // optimize_file_loading=false for precise file control
342 )?;
343
344 if !preserved_data.is_empty() {
345 let start_ts = UnixNanos::from(operation.file_start_ns);
346 let end_ts = UnixNanos::from(operation.file_end_ns);
347 self.write_to_parquet(
348 &preserved_data,
349 Some(start_ts),
350 Some(end_ts),
351 Some(true),
352 )?;
353 }
354 }
355 DeleteOperationKind::Remove => {}
356 }
357
358 // Mark files for removal (applies to all operation types)
359 for file in operation.files {
360 files_to_remove.insert(file);
361 }
362 }
363
364 for file in files_to_remove {
365 self.delete_file(&file)?;
366 }
367
368 Ok(())
369 }
370
371 /// Deletes custom data within a specified time range.
372 ///
373 /// This method provides deletion for custom data types that don't have compile-time
374 /// type information. It uses dynamic querying and writing methods.
375 ///
376 /// # Parameters
377 ///
378 /// - `type_name`: The custom data type name (without "custom/" prefix).
379 /// - `identifier`: Optional instrument ID to delete data for.
380 /// - `start`: Optional start timestamp for the deletion range.
381 /// - `end`: Optional end timestamp for the deletion range.
382 ///
383 /// # Returns
384 ///
385 /// Returns `Ok(())` on success, or an error if deletion fails.
386 fn delete_custom_data_range(
387 &mut self,
388 type_name: &str,
389 identifier: Option<&str>,
390 start: Option<UnixNanos>,
391 end: Option<UnixNanos>,
392 ) -> anyhow::Result<()> {
393 let data_type = NautilusDataType::Custom {
394 type_name: type_name.to_string(),
395 };
396
397 let files = self.files_for_delete(&data_type, identifier)?;
398
399 if files.is_empty() {
400 return Ok(());
401 }
402
403 let operations_to_execute = operations_for_files(&files, start, end);
404
405 if operations_to_execute.is_empty() {
406 return Ok(()); // No operations to execute
407 }
408
409 // Execute all operations
410 let mut files_to_remove = AHashSet::<String>::new();
411
412 for operation in operations_to_execute {
413 // Reset the session before each operation
414 self.clear_session_tables();
415
416 match operation.kind {
417 DeleteOperationKind::SplitBefore | DeleteOperationKind::SplitAfter => {
418 // Query the custom data preserved by the split and write it
419 let instrument_ids = identifier.map(|id| vec![id.to_string()]);
420 let preserved_data = self.query_custom_data_dynamic(
421 type_name,
422 instrument_ids.as_deref(),
423 Some(UnixNanos::from(operation.query_start)),
424 Some(UnixNanos::from(operation.query_end)),
425 None,
426 Some(operation.files.clone()),
427 false,
428 )?;
429
430 if !preserved_data.is_empty() {
431 let custom_items: Vec<CustomData> = preserved_data
432 .into_iter()
433 .filter_map(|data| match data {
434 Data::Custom(c) => Some(c),
435 _ => None,
436 })
437 .collect();
438
439 let start_ts = UnixNanos::from(operation.file_start_ns);
440 let end_ts = UnixNanos::from(operation.file_end_ns);
441
442 for items in group_custom_data_by_type(custom_items.iter()) {
443 self.write_custom_data_refs_batch(
444 &items,
445 Some(start_ts),
446 Some(end_ts),
447 Some(true),
448 )?;
449 }
450 }
451 }
452 DeleteOperationKind::Remove => {}
453 }
454
455 // Mark files for removal (applies to all operation types)
456 for file in operation.files {
457 files_to_remove.insert(file);
458 }
459 }
460
461 for file in files_to_remove {
462 self.delete_file(&file)?;
463 }
464
465 Ok(())
466 }
467
468 fn files_for_delete(
469 &self,
470 data_type: &NautilusDataType,
471 identifier: Option<&str>,
472 ) -> anyhow::Result<Vec<(String, u64, u64)>> {
473 let directory = if let NautilusDataType::Custom { type_name } = data_type {
474 self.make_path_custom_data(type_name, identifier)?
475 } else {
476 let prefix = parquet_data_path_prefix(data_type);
477 self.make_path(prefix.as_ref(), identifier)?
478 };
479
480 let files = self.list_parquet_files(&directory)?;
481
482 Ok(files
483 .into_iter()
484 .filter_map(|path| {
485 parse_filename_timestamps(&path).map(|(start, end)| (path, start, end))
486 })
487 .collect())
488 }
489
490 /// Prepares all operations for data deletion by identifying files that need to be
491 /// split or removed.
492 ///
493 /// This auxiliary function handles all the preparation logic for deletion:
494 /// 1. Filters intervals by time range
495 /// 2. Identifies files that intersect with the deletion range
496 /// 3. Creates split operations for files that partially overlap
497 /// 4. Generates removal operations for files completely within the range
498 ///
499 /// # Parameters
500 ///
501 /// - `type_name`: The data type directory name for path generation.
502 /// - `identifier`: Optional instrument identifier for path generation.
503 /// - `intervals`: List of (`start_ts`, `end_ts`) tuples representing existing file intervals.
504 /// - `start`: Optional start timestamp for deletion range.
505 /// - `end`: Optional end timestamp for deletion range.
506 ///
507 /// # Returns
508 ///
509 /// Returns a vector of `DeleteOperation` structs ready for execution.
510 ///
511 /// Plans from the supplied intervals. It does not look up stored filenames, so a
512 /// replay-identity suffix is not included. Catalog deletion resolves stored paths itself.
513 pub fn prepare_delete_operations(
514 &self,
515 type_name: &str,
516 identifier: Option<&str>,
517 intervals: &[(u64, u64)],
518 start: Option<UnixNanos>,
519 end: Option<UnixNanos>,
520 ) -> anyhow::Result<Vec<DeleteOperation>> {
521 let directory = self.make_path(type_name, identifier)?;
522
523 let files = intervals
524 .iter()
525 .map(|&(file_start_ns, file_end_ns)| {
526 let filename = timestamps_to_filename(
527 UnixNanos::from(file_start_ns),
528 UnixNanos::from(file_end_ns),
529 );
530 let file_path = make_object_store_path(&directory, [&filename]);
531 (file_path, file_start_ns, file_end_ns)
532 })
533 .collect::<Vec<_>>();
534
535 Ok(operations_for_files(&files, start, end))
536 }
537}
538
539fn operations_for_files(
540 files: &[(String, u64, u64)],
541 start: Option<UnixNanos>,
542 end: Option<UnixNanos>,
543) -> Vec<DeleteOperation> {
544 let delete_start_ns = start.map(|s| s.as_u64());
545 let delete_end_ns = end.map(|e| e.as_u64());
546 let mut operations = Vec::new();
547
548 for (file_path, file_start_ns, file_end_ns) in files {
549 let intersects = delete_start_ns.is_none_or(|start| start <= *file_end_ns)
550 && delete_end_ns.is_none_or(|end| *file_start_ns <= end);
551
552 if !intersects {
553 continue;
554 }
555
556 let file_completely_within_range = delete_start_ns
557 .is_none_or(|start| start <= *file_start_ns)
558 && delete_end_ns.is_none_or(|end| *file_end_ns <= end);
559
560 if file_completely_within_range {
561 operations.push(DeleteOperation {
562 kind: DeleteOperationKind::Remove,
563 files: vec![file_path.clone()],
564 query_start: 0,
565 query_end: 0,
566 file_start_ns: 0,
567 file_end_ns: 0,
568 });
569
570 continue;
571 }
572
573 if let Some(delete_start) = delete_start_ns
574 && *file_start_ns < delete_start
575 {
576 operations.push(DeleteOperation {
577 kind: DeleteOperationKind::SplitBefore,
578 files: vec![file_path.clone()],
579 query_start: *file_start_ns,
580 query_end: delete_start.saturating_sub(1),
581 file_start_ns: *file_start_ns,
582 file_end_ns: delete_start.saturating_sub(1),
583 });
584 }
585
586 if let Some(delete_end) = delete_end_ns
587 && delete_end < *file_end_ns
588 {
589 operations.push(DeleteOperation {
590 kind: DeleteOperationKind::SplitAfter,
591 files: vec![file_path.clone()],
592 query_start: delete_end.saturating_add(1),
593 query_end: *file_end_ns,
594 file_start_ns: delete_end.saturating_add(1),
595 file_end_ns: *file_end_ns,
596 });
597 }
598 }
599
600 operations
601}