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
18use std::{collections::BTreeMap, sync::Arc};
19
20use ahash::AHashSet;
21use arrow::{
22    array::{ArrayRef, StringArray},
23    compute::cast,
24    datatypes::DataType as ArrowDataType,
25    record_batch::RecordBatch,
26    row::{RowConverter, SortField},
27};
28use nautilus_core::{UnixNanos, string::conversions::to_snake_case};
29use nautilus_model::data::{
30    Bar, CustomData, Data, DataType, FundingRateUpdate, HasTsInit, IndexPriceUpdate,
31    InstrumentStatus, MarkPriceUpdate, NautilusDataType, OptionGreeks, OrderBookDelta,
32    OrderBookDepth, QuoteTick, TradeTick, close::InstrumentClose,
33};
34use nautilus_serialization::arrow::{
35    DecodeTypedFromRecordBatch, EncodeToRecordBatch, KEY_BAR_TYPE, KEY_INSTRUMENT_ID,
36    KEY_PRICE_PRECISION, KEY_SIZE_PRECISION, KEY_TYPE_NAME, is_json_string_field,
37};
38
39use crate::{
40    backend::parquet::{
41        catalog::ParquetDataCatalog,
42        io::read_parquet_from_object_store,
43        paths::{make_object_store_path, parse_filename_timestamps, timestamps_to_filename},
44    },
45    catalog::types::{
46        HasCatalogDataType, data_type_from_data_path_prefix, parquet_data_path_prefix,
47    },
48};
49
50/// Kind of deletion operation to execute.
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub enum DeleteOperationKind {
53    /// Remove the files entirely.
54    Remove,
55    /// Rewrite the data preserved before the deleted range, then remove the files.
56    SplitBefore,
57    /// Rewrite the data preserved after the deleted range, then remove the files.
58    SplitAfter,
59}
60
61/// Information about a deletion operation to be executed.
62#[derive(Debug, Clone)]
63pub struct DeleteOperation {
64    /// Kind of deletion operation.
65    pub kind: DeleteOperationKind,
66    /// List of files involved in this operation.
67    pub files: Vec<String>,
68    /// Start timestamp for data query (used for split operations).
69    pub query_start: u64,
70    /// End timestamp for data query (used for split operations).
71    pub query_end: u64,
72    /// Start timestamp for new file naming (used for split operations).
73    pub file_start_ns: u64,
74    /// End timestamp for new file naming (used for split operations).
75    pub file_end_ns: u64,
76}
77
78impl ParquetDataCatalog {
79    /// Deletes data within a specified time range across the entire catalog.
80    ///
81    /// This method identifies all leaf directories in the catalog that contain parquet files
82    /// and deletes data within the specified time range from each directory. A leaf directory
83    /// is one that contains files but no subdirectories. This is a convenience method that
84    /// effectively calls `delete_data_range` for all data types and instrument IDs in the catalog.
85    ///
86    /// # Parameters
87    ///
88    /// - `start`: Optional start timestamp for the deletion range. If None, deletes from the beginning.
89    /// - `end`: Optional end timestamp for the deletion range. If None, deletes to the end.
90    ///
91    /// # Returns
92    ///
93    /// Returns `Ok(())` on success, or an error if deletion fails.
94    ///
95    /// # Errors
96    ///
97    /// Returns an error if:
98    /// - Directory traversal fails.
99    /// - Data class extraction from paths fails.
100    /// - Individual delete operations fail.
101    ///
102    /// # Notes
103    ///
104    /// - This operation permanently removes data and cannot be undone.
105    /// - The deletion process handles file intersections intelligently by splitting files
106    ///   when they partially overlap with the deletion range.
107    /// - Files completely within the deletion range are removed entirely.
108    /// - Files partially overlapping the deletion range are split to preserve data outside the range.
109    /// - This method is useful for bulk data cleanup operations across the entire catalog.
110    /// - Empty directories are not automatically removed after deletion.
111    ///
112    /// # Examples
113    ///
114    /// ```rust,no_run
115    /// use nautilus_core::UnixNanos;
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    /// // Delete all data before a specific date across entire catalog
127    /// catalog.delete_catalog_range(None, Some(UnixNanos::from(1609459200000000000)))?;
128    ///
129    /// // Delete all data within a specific range across entire catalog
130    /// catalog.delete_catalog_range(
131    ///     Some(UnixNanos::from(1609459200000000000)),
132    ///     Some(UnixNanos::from(1609545600000000000)),
133    /// )?;
134    ///
135    /// // Delete all data after a specific date across entire catalog
136    /// catalog.delete_catalog_range(Some(UnixNanos::from(1609459200000000000)), None)?;
137    /// # Ok::<(), anyhow::Error>(())
138    /// ```
139    pub fn delete_catalog_range(
140        &mut self,
141        start: Option<UnixNanos>,
142        end: Option<UnixNanos>,
143    ) -> anyhow::Result<()> {
144        let leaf_directories = self.find_leaf_data_directories()?;
145
146        for directory in leaf_directories {
147            let Ok((Some(data_cls), identifier)) =
148                self.extract_data_cls_and_identifier_from_path(&directory)
149            else {
150                continue;
151            };
152
153            let Ok(data_type) = data_type_from_data_path_prefix(&data_cls) else {
154                log::warn!("Skipping directory {directory}: unknown data class {data_cls}");
155                continue;
156            };
157
158            self.delete_data_range(&data_type, identifier.as_deref(), start, end)?;
159        }
160
161        Ok(())
162    }
163
164    /// Deletes data within a specified time range for a specific data type and identifier.
165    ///
166    /// This method identifies all parquet files that intersect with the specified time range
167    /// and handles them appropriately:
168    /// - Files completely within the range are deleted
169    /// - Files partially overlapping the range are split to preserve data outside the range
170    /// - The original intersecting files are removed after processing
171    ///
172    /// # Parameters
173    ///
174    /// - `data_type`: The data type to delete from.
175    /// - `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.
176    /// - `start`: Optional start timestamp for the deletion range. If None, deletes from the beginning.
177    /// - `end`: Optional end timestamp for the deletion range. If None, deletes to the end.
178    ///
179    /// # Returns
180    ///
181    /// Returns `Ok(())` on success, or an error if deletion fails.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if:
186    /// - `start` exceeds `end` and the directory holds data to delete.
187    /// - The directory path cannot be constructed.
188    /// - File operations fail.
189    /// - Data querying or writing fails.
190    ///
191    /// # Notes
192    ///
193    /// - This operation permanently removes data and cannot be undone.
194    /// - Files that partially overlap the deletion range are split to preserve data outside the range.
195    /// - The method ensures data integrity by using atomic operations where possible.
196    /// - Empty directories are not automatically removed after deletion.
197    ///
198    /// # Examples
199    ///
200    /// ```rust,no_run
201    /// use nautilus_core::UnixNanos;
202    /// use nautilus_model::data::NautilusDataType;
203    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
204    ///
205    /// let mut catalog = ParquetDataCatalog::new(
206    ///     std::path::Path::new("/tmp/nautilus_data"),
207    ///     None,
208    ///     None,
209    ///     None,
210    ///     None,
211    /// );
212    ///
213    /// // Delete all quote data for a specific instrument
214    /// catalog.delete_data_range(&NautilusDataType::QuoteTick, Some("BTCUSD"), None, None)?;
215    ///
216    /// // Delete trade data within a specific time range
217    /// catalog.delete_data_range(
218    ///     &NautilusDataType::TradeTick,
219    ///     None,
220    ///     Some(UnixNanos::from(1609459200000000000)),
221    ///     Some(UnixNanos::from(1609545600000000000)),
222    /// )?;
223    /// # Ok::<(), anyhow::Error>(())
224    /// ```
225    pub fn delete_data_range(
226        &mut self,
227        data_type: &NautilusDataType,
228        identifier: Option<&str>,
229        start: Option<UnixNanos>,
230        end: Option<UnixNanos>,
231    ) -> anyhow::Result<()> {
232        match data_type {
233            NautilusDataType::QuoteTick => {
234                self.delete_data_range_generic::<QuoteTick>(identifier, start, end)
235            }
236            NautilusDataType::TradeTick => {
237                self.delete_data_range_generic::<TradeTick>(identifier, start, end)
238            }
239            NautilusDataType::Bar => self.delete_data_range_generic::<Bar>(identifier, start, end),
240            NautilusDataType::OrderBookDelta => {
241                self.delete_data_range_generic::<OrderBookDelta>(identifier, start, end)
242            }
243            NautilusDataType::OrderBookDepth => {
244                self.delete_data_range_generic::<OrderBookDepth>(identifier, start, end)
245            }
246            NautilusDataType::MarkPriceUpdate => {
247                self.delete_data_range_generic::<MarkPriceUpdate>(identifier, start, end)
248            }
249            NautilusDataType::IndexPriceUpdate => {
250                self.delete_data_range_generic::<IndexPriceUpdate>(identifier, start, end)
251            }
252            NautilusDataType::InstrumentClose => {
253                self.delete_data_range_generic::<InstrumentClose>(identifier, start, end)
254            }
255            NautilusDataType::FundingRateUpdate => {
256                self.delete_data_range_generic::<FundingRateUpdate>(identifier, start, end)
257            }
258            NautilusDataType::OptionGreeks => {
259                self.delete_data_range_generic::<OptionGreeks>(identifier, start, end)
260            }
261            NautilusDataType::InstrumentStatus => {
262                self.delete_data_range_generic::<InstrumentStatus>(identifier, start, end)
263            }
264            NautilusDataType::Custom { type_name } => {
265                self.delete_custom_data_range(type_name, identifier, start, end)
266            }
267            other @ NautilusDataType::Instrument => {
268                anyhow::bail!("Unsupported data type: {other}")
269            }
270            #[cfg(feature = "defi")]
271            other @ NautilusDataType::Defi => {
272                anyhow::bail!("Unsupported data type: {other}")
273            }
274            #[cfg(not(feature = "defi"))]
275            #[allow(
276                unreachable_patterns,
277                reason = "DeFi variants can exist without this crate's defi feature"
278            )]
279            other => anyhow::bail!("Unsupported data type: {other}"),
280        }
281    }
282
283    /// Generic implementation for deleting data within a specified time range.
284    ///
285    /// This method provides the core deletion logic that works with any data type
286    /// that implements the required traits. It handles file intersection analysis,
287    /// data splitting for partial overlaps, and file cleanup.
288    ///
289    /// # Type Parameters
290    ///
291    /// - `T`: The data type that implements required traits for catalog operations.
292    ///
293    /// # Parameters
294    ///
295    /// - `identifier`: Optional instrument ID to delete data for.
296    /// - `start`: Optional start timestamp for the deletion range.
297    /// - `end`: Optional end timestamp for the deletion range.
298    ///
299    /// # Returns
300    ///
301    /// Returns `Ok(())` on success, or an error if deletion fails.
302    ///
303    /// # Errors
304    ///
305    /// Returns an error if:
306    /// - `start` exceeds `end` and the directory holds data to delete.
307    /// - Data querying, encoding, or file operations fail.
308    pub fn delete_data_range_generic<T>(
309        &mut self,
310        identifier: Option<&str>,
311        start: Option<UnixNanos>,
312        end: Option<UnixNanos>,
313    ) -> anyhow::Result<()>
314    where
315        T: DecodeTypedFromRecordBatch
316            + HasCatalogDataType
317            + EncodeToRecordBatch
318            + HasTsInit
319            + TryFrom<Data>
320            + Clone,
321    {
322        // Get intervals for cleaner implementation
323        let data_type = T::catalog_data_type();
324        let files = self.files_for_delete(&data_type, identifier)?;
325
326        if files.is_empty() {
327            return Ok(());
328        }
329
330        let operations_to_execute = operations_for_files(&files, start, end)?;
331
332        if operations_to_execute.is_empty() {
333            return Ok(()); // No operations to execute
334        }
335
336        let file_paths: AHashSet<&str> = files.iter().map(|(path, ..)| path.as_str()).collect();
337        let (targets, files_to_remove) = group_operations_by_target(operations_to_execute);
338        let type_name = to_snake_case(std::any::type_name::<T>());
339        let mut writes = Vec::new();
340
341        // Splits of overlapping files can target the same file, so each target is written once
342        // with the rows of all its splits; writing them in turn would skip all but the first.
343        // Every target is checked before any file is written or removed. Each pass holds the
344        // rows of one target only, so the write pass queries them again.
345        for (target, sources) in &targets {
346            let rows = self.query_split_rows::<T>(identifier, target, sources)?;
347
348            if rows.is_empty() {
349                continue;
350            }
351
352            Self::check_chunk_identity(&rows, &type_name)?;
353            if file_paths.contains(target.0.as_str()) {
354                let batches = self.data_to_record_batches(&rows)?;
355                self.ensure_split_target_holds_rows(&target.0, &batches)?;
356                continue;
357            }
358
359            writes.push((target, sources));
360        }
361
362        for (target, sources) in writes {
363            let rows = self.query_split_rows::<T>(identifier, target, sources)?;
364            self.write_to_parquet(
365                &rows,
366                Some(UnixNanos::from(target.1)),
367                Some(UnixNanos::from(target.2)),
368                Some(true),
369            )?;
370        }
371
372        for file in files_to_remove {
373            self.delete_file(&file)?;
374        }
375
376        Ok(())
377    }
378
379    /// Deletes custom data within a specified time range.
380    ///
381    /// This method provides deletion for custom data types that don't have compile-time
382    /// type information. It uses dynamic querying and writing methods.
383    ///
384    /// # Parameters
385    ///
386    /// - `type_name`: The custom data type name (without "custom/" prefix).
387    /// - `identifier`: Optional instrument ID to delete data for.
388    /// - `start`: Optional start timestamp for the deletion range.
389    /// - `end`: Optional end timestamp for the deletion range.
390    ///
391    /// # Returns
392    ///
393    /// Returns `Ok(())` on success, or an error if deletion fails.
394    fn delete_custom_data_range(
395        &mut self,
396        type_name: &str,
397        identifier: Option<&str>,
398        start: Option<UnixNanos>,
399        end: Option<UnixNanos>,
400    ) -> anyhow::Result<()> {
401        let data_type = NautilusDataType::Custom {
402            type_name: type_name.to_string(),
403        };
404
405        let files = self.files_for_delete(&data_type, identifier)?;
406
407        if files.is_empty() {
408            return Ok(());
409        }
410
411        let operations_to_execute = operations_for_files(&files, start, end)?;
412
413        if operations_to_execute.is_empty() {
414            return Ok(()); // No operations to execute
415        }
416
417        let file_paths: AHashSet<&str> = files.iter().map(|(path, ..)| path.as_str()).collect();
418        let (targets, files_to_remove) = group_operations_by_target(operations_to_execute);
419        let mut writes = Vec::new();
420
421        // Splits of overlapping files can target the same file, so each target is written once
422        // with the rows of all its splits. Every target and group is checked before any file is
423        // written or removed, so groups with different metadata that would share a file fail the
424        // delete without changes. Each pass holds the rows of one target only, so the write pass
425        // queries them again.
426        for (target, sources) in &targets {
427            let rows = self.query_custom_split_rows(type_name, identifier, target, sources)?;
428
429            if rows.is_empty() {
430                continue;
431            }
432
433            let target_writes = self.prepare_custom_data_groups(
434                &rows,
435                Some(UnixNanos::from(target.1)),
436                Some(UnixNanos::from(target.2)),
437                Some(true),
438            )?;
439
440            if file_paths.contains(target.0.as_str()) {
441                let batches: Vec<RecordBatch> = target_writes
442                    .into_iter()
443                    .flat_map(|write| write.batches)
444                    .collect();
445                self.ensure_split_target_holds_rows(&target.0, &batches)?;
446                continue;
447            }
448
449            writes.push((target, sources));
450        }
451
452        for (target, sources) in writes {
453            let rows = self.query_custom_split_rows(type_name, identifier, target, sources)?;
454            self.write_custom_data_groups(
455                &rows,
456                Some(UnixNanos::from(target.1)),
457                Some(UnixNanos::from(target.2)),
458                Some(true),
459            )?;
460        }
461
462        for file in files_to_remove {
463            self.delete_file(&file)?;
464        }
465
466        Ok(())
467    }
468
469    // Queries the rows a split target preserves from its source files, sorted by `ts_init`.
470    fn query_split_rows<T>(
471        &mut self,
472        identifier: Option<&str>,
473        target: &SplitTarget,
474        sources: &[String],
475    ) -> anyhow::Result<Vec<T>>
476    where
477        T: DecodeTypedFromRecordBatch + HasCatalogDataType + HasTsInit,
478    {
479        // Reset the session so each query loads the source files afresh
480        self.clear_session_tables();
481        // Use optimize_file_loading=false for precise file control during split operations
482        let instrument_ids = identifier.map(|id| vec![id.to_string()]);
483        let mut rows = self.query::<T>(
484            instrument_ids,
485            Some(UnixNanos::from(target.1)),
486            Some(UnixNanos::from(target.2)),
487            None,
488            Some(sources.to_vec()),
489            false,
490        )?;
491        rows.sort_by_key(HasTsInit::ts_init);
492        Ok(rows)
493    }
494
495    // Queries the custom rows a split target preserves from its source files, sorted by
496    // `ts_init`.
497    fn query_custom_split_rows(
498        &mut self,
499        type_name: &str,
500        identifier: Option<&str>,
501        target: &SplitTarget,
502        sources: &[String],
503    ) -> anyhow::Result<Vec<CustomData>> {
504        self.clear_session_tables();
505        let instrument_ids = identifier.map(|id| vec![id.to_string()]);
506        let mut rows: Vec<CustomData> = self
507            .query_custom_data_dynamic(
508                type_name,
509                instrument_ids.as_deref(),
510                Some(UnixNanos::from(target.1)),
511                Some(UnixNanos::from(target.2)),
512                None,
513                Some(sources.to_vec()),
514                false,
515            )?
516            .into_iter()
517            .filter_map(|data| match data {
518                Data::Custom(c) => Some(c),
519                _ => None,
520            })
521            .collect();
522        rows.sort_by_key(|custom| custom.data.ts_init());
523        Ok(rows)
524    }
525
526    // An existing split target is skipped by the write, and the split's sources are then removed,
527    // so it must already hold every preserved row, as after a rerun of an interrupted delete.
528    // Rows are compared in their Arrow row encoding, where equal values match, NaN included,
529    // unlike `PartialEq` on decoded rows.
530    fn ensure_split_target_holds_rows(
531        &self,
532        target: &str,
533        rows: &[RecordBatch],
534    ) -> anyhow::Result<()> {
535        anyhow::ensure!(
536            self.target_holds_rows(target, rows)?,
537            "Cannot preserve the rows outside the deleted range: their target file {target} \
538             already exists without them"
539        );
540        Ok(())
541    }
542
543    pub(crate) fn target_holds_rows(
544        &self,
545        target: &str,
546        rows: &[RecordBatch],
547    ) -> anyhow::Result<bool> {
548        let object_path = self.to_object_path_parsed(target)?;
549
550        let (existing, schema) = self.execute_async(|| async {
551            read_parquet_from_object_store(self.object_store.clone(), &object_path).await
552        })?;
553
554        for batch in rows {
555            for key in [
556                KEY_INSTRUMENT_ID,
557                KEY_BAR_TYPE,
558                KEY_PRICE_PRECISION,
559                KEY_SIZE_PRECISION,
560                KEY_TYPE_NAME,
561            ] {
562                if schema.metadata().get(key) != batch.schema().metadata().get(key) {
563                    return Ok(false);
564                }
565            }
566        }
567
568        batches_hold_rows(&existing, rows)
569    }
570
571    fn files_for_delete(
572        &self,
573        data_type: &NautilusDataType,
574        identifier: Option<&str>,
575    ) -> anyhow::Result<Vec<(String, u64, u64)>> {
576        let directory = if let NautilusDataType::Custom { type_name } = data_type {
577            self.make_path_custom_data(type_name, identifier)?
578        } else {
579            let prefix = parquet_data_path_prefix(data_type);
580            self.make_path(prefix.as_ref(), identifier)?
581        };
582
583        let files = self.list_parquet_files(&directory)?;
584
585        Ok(files
586            .into_iter()
587            .filter_map(|path| {
588                parse_filename_timestamps(&path).map(|(start, end)| (path, start, end))
589            })
590            .collect())
591    }
592
593    /// Prepares all operations for data deletion by identifying files that need to be
594    /// split or removed.
595    ///
596    /// This auxiliary function handles all the preparation logic for deletion:
597    /// 1. Filters intervals by time range
598    /// 2. Identifies files that intersect with the deletion range
599    /// 3. Creates split operations for files that partially overlap
600    /// 4. Generates removal operations for files completely within the range
601    ///
602    /// # Parameters
603    ///
604    /// - `type_name`: The data type directory name for path generation.
605    /// - `identifier`: Optional instrument identifier for path generation.
606    /// - `intervals`: List of (`start_ts`, `end_ts`) tuples representing existing file intervals.
607    /// - `start`: Optional start timestamp for deletion range.
608    /// - `end`: Optional end timestamp for deletion range.
609    ///
610    /// # Returns
611    ///
612    /// Returns a vector of `DeleteOperation` structs ready for execution.
613    ///
614    /// Plans from the supplied intervals. It does not look up stored filenames, so a
615    /// replay-identity suffix is not included. Catalog deletion resolves stored paths itself.
616    ///
617    /// # Errors
618    ///
619    /// Returns an error if `start` exceeds `end` or the directory path cannot be constructed.
620    pub fn prepare_delete_operations(
621        &self,
622        type_name: &str,
623        identifier: Option<&str>,
624        intervals: &[(u64, u64)],
625        start: Option<UnixNanos>,
626        end: Option<UnixNanos>,
627    ) -> anyhow::Result<Vec<DeleteOperation>> {
628        let directory = self.make_path(type_name, identifier)?;
629
630        let files = intervals
631            .iter()
632            .map(|&(file_start_ns, file_end_ns)| {
633                let filename = timestamps_to_filename(
634                    UnixNanos::from(file_start_ns),
635                    UnixNanos::from(file_end_ns),
636                );
637                let file_path = make_object_store_path(&directory, [&filename]);
638                (file_path, file_start_ns, file_end_ns)
639            })
640            .collect::<Vec<_>>();
641
642        operations_for_files(&files, start, end)
643    }
644}
645
646fn operations_for_files(
647    files: &[(String, u64, u64)],
648    start: Option<UnixNanos>,
649    end: Option<UnixNanos>,
650) -> anyhow::Result<Vec<DeleteOperation>> {
651    if let (Some(start), Some(end)) = (start, end) {
652        anyhow::ensure!(start <= end, "deletion start must not exceed end");
653    }
654
655    let delete_start_ns = start.map(|s| s.as_u64());
656    let delete_end_ns = end.map(|e| e.as_u64());
657    let mut operations = Vec::new();
658
659    for (file_path, file_start_ns, file_end_ns) in files {
660        let intersects = delete_start_ns.is_none_or(|start| start <= *file_end_ns)
661            && delete_end_ns.is_none_or(|end| *file_start_ns <= end);
662
663        if !intersects {
664            continue;
665        }
666
667        let file_completely_within_range = delete_start_ns
668            .is_none_or(|start| start <= *file_start_ns)
669            && delete_end_ns.is_none_or(|end| *file_end_ns <= end);
670
671        if file_completely_within_range {
672            operations.push(DeleteOperation {
673                kind: DeleteOperationKind::Remove,
674                files: vec![file_path.clone()],
675                query_start: 0,
676                query_end: 0,
677                file_start_ns: 0,
678                file_end_ns: 0,
679            });
680
681            continue;
682        }
683
684        if let Some(delete_start) = delete_start_ns
685            && *file_start_ns < delete_start
686        {
687            operations.push(DeleteOperation {
688                kind: DeleteOperationKind::SplitBefore,
689                files: vec![file_path.clone()],
690                query_start: *file_start_ns,
691                query_end: delete_start.saturating_sub(1),
692                file_start_ns: *file_start_ns,
693                file_end_ns: delete_start.saturating_sub(1),
694            });
695        }
696
697        if let Some(delete_end) = delete_end_ns
698            && delete_end < *file_end_ns
699        {
700            operations.push(DeleteOperation {
701                kind: DeleteOperationKind::SplitAfter,
702                files: vec![file_path.clone()],
703                query_start: delete_end.saturating_add(1),
704                query_end: *file_end_ns,
705                file_start_ns: delete_end.saturating_add(1),
706                file_end_ns: *file_end_ns,
707            });
708        }
709    }
710
711    Ok(operations)
712}
713
714// A file a split operation writes: its source directory and file name, with the file range.
715type SplitTarget = (String, u64, u64);
716
717// Groups split operations by the file they write, with the source files of each, and collects
718// every file the delete removes. All splits of one target share its range as their query range.
719fn group_operations_by_target(
720    operations: Vec<DeleteOperation>,
721) -> (BTreeMap<SplitTarget, Vec<String>>, AHashSet<String>) {
722    let mut targets = BTreeMap::<SplitTarget, Vec<String>>::new();
723    let mut files_to_remove = AHashSet::new();
724
725    for operation in operations {
726        if operation.kind != DeleteOperationKind::Remove {
727            targets
728                .entry(split_target(&operation))
729                .or_default()
730                .extend(operation.files.iter().cloned());
731        }
732        files_to_remove.extend(operation.files);
733    }
734
735    (targets, files_to_remove)
736}
737
738fn split_target(operation: &DeleteOperation) -> SplitTarget {
739    let source = &operation.files[0];
740    let filename = timestamps_to_filename(
741        UnixNanos::from(operation.file_start_ns),
742        UnixNanos::from(operation.file_end_ns),
743    );
744    let target = match source.rsplit_once('/') {
745        Some((directory, _)) => make_object_store_path(directory, [&filename]),
746        None => filename,
747    };
748
749    (target, operation.file_start_ns, operation.file_end_ns)
750}
751
752// Returns whether every row of `rows` occurs in `existing`, comparing the columns both share by
753// name, so a column only one side carries, such as a stored identifier column, is ignored.
754// JSON columns compare by value, since an equal map can serialize its keys in any order: custom
755// JSON-backed fields and the plain UTF-8 `data_type` column that holds custom data metadata.
756fn batches_hold_rows(existing: &[RecordBatch], rows: &[RecordBatch]) -> anyhow::Result<bool> {
757    let Some(first) = rows.first() else {
758        return Ok(true);
759    };
760    let Some(existing_first) = existing.first() else {
761        return Ok(rows.iter().all(|batch| batch.num_rows() == 0));
762    };
763
764    let existing_schema = existing_first.schema();
765    let names: Vec<String> = first
766        .schema()
767        .fields()
768        .iter()
769        .map(|field| field.name().clone())
770        .filter(|name| existing_schema.field_with_name(name).is_ok())
771        .collect();
772    let fields = names
773        .iter()
774        .map(|name| {
775            let data_type = first.schema().field_with_name(name)?.data_type().clone();
776            Ok(SortField::new(data_type))
777        })
778        .collect::<anyhow::Result<Vec<_>>>()?;
779    let converter = RowConverter::new(fields)?;
780    let json_names: AHashSet<&str> = names
781        .iter()
782        .filter(|name| {
783            *name == "data_type"
784                || [first.schema(), existing_schema.clone()]
785                    .iter()
786                    .any(|schema| schema.field_with_name(name).is_ok_and(is_json_string_field))
787        })
788        .map(String::as_str)
789        .collect();
790    let columns = |batch: &RecordBatch| {
791        names
792            .iter()
793            .map(|name| {
794                let column = batch
795                    .column_by_name(name)
796                    .ok_or_else(|| anyhow::anyhow!("Missing column {name} in Parquet batch"))?;
797
798                if json_names.contains(name.as_str()) {
799                    canonical_json_column(column, name == "data_type")
800                } else {
801                    Ok(column.clone())
802                }
803            })
804            .collect::<anyhow::Result<Vec<_>>>()
805    };
806
807    let mut existing_rows = AHashSet::new();
808
809    for batch in existing {
810        for row in &converter.convert_columns(&columns(batch)?)? {
811            existing_rows.insert(row.as_ref().to_vec());
812        }
813    }
814
815    for batch in rows {
816        for row in &converter.convert_columns(&columns(batch)?)? {
817            if !existing_rows.contains(row.as_ref()) {
818                return Ok(false);
819            }
820        }
821    }
822
823    Ok(true)
824}
825
826// Rewrites each JSON value with object keys sorted, so equal values compare equal as strings.
827fn canonical_json_column(column: &ArrayRef, is_data_type: bool) -> anyhow::Result<ArrayRef> {
828    let column = cast(column, &ArrowDataType::Utf8)?;
829    let values = column
830        .as_any()
831        .downcast_ref::<StringArray>()
832        .ok_or_else(|| anyhow::anyhow!("JSON column is not a string array"))?;
833    let canonical = values
834        .iter()
835        .map(|value| {
836            value
837                .map(|value| {
838                    let value: serde_json::Value = if is_data_type {
839                        let data_type = DataType::from_persistence_json(value)?;
840                        let json = data_type.to_persistence_json()?;
841                        serde_json::from_str(&json)?
842                    } else {
843                        serde_json::from_str(value)?
844                    };
845
846                    Ok::<_, anyhow::Error>(canonical_json_value(value).to_string())
847                })
848                .transpose()
849        })
850        .collect::<Result<StringArray, _>>()?;
851    Ok(Arc::new(canonical))
852}
853
854fn canonical_json_value(value: serde_json::Value) -> serde_json::Value {
855    match value {
856        serde_json::Value::Object(map) => {
857            let mut entries: Vec<_> = map.into_iter().collect();
858            entries.sort_by(|a, b| a.0.cmp(&b.0));
859            serde_json::Value::Object(
860                entries
861                    .into_iter()
862                    .map(|(key, value)| (key, canonical_json_value(value)))
863                    .collect(),
864            )
865        }
866        serde_json::Value::Array(values) => {
867            serde_json::Value::Array(values.into_iter().map(canonical_json_value).collect())
868        }
869        value => value,
870    }
871}
nautilus_persistence/backend/parquet/delete.rs

nautilus_persistence/backend/parquet/
delete.rs
