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

nautilus_persistence/backend/parquet/
delete.rs
