Skip to main content

nautilus_persistence/backend/parquet/
consolidation.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//! Period-based and bulk consolidation of parquet files in a catalog directory.
17
18#![expect(
19    clippy::missing_errors_doc,
20    clippy::missing_panics_doc,
21    reason = "consolidation operations forward catalog/storage errors and operate on validated batches"
22)]
23
24use nautilus_core::UnixNanos;
25use nautilus_model::data::{
26    Bar, CustomData, Data, HasTsInit, IndexPriceUpdate, MarkPriceUpdate, NautilusDataType,
27    OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick, close::InstrumentClose,
28};
29use nautilus_serialization::arrow::{DecodeTypedFromRecordBatch, EncodeToRecordBatch};
30use object_store::path::Path as ObjectPath;
31
32use crate::{
33    backend::parquet::{
34        catalog::ParquetDataCatalog,
35        intervals::{are_intervals_contiguous, are_intervals_disjoint},
36        io::combine_parquet_files_from_object_store,
37        paths::{
38            extract_path_components, make_object_store_path, normalize_path_separators,
39            parse_filename_timestamps, timestamps_to_filename,
40        },
41    },
42    catalog::types::{
43        CatalogDataType, HasCatalogDataType, parquet_catalog_data_type_path_prefixes,
44        parquet_data_path_prefix,
45    },
46    common::custom::group_custom_data_by_type,
47};
48
49/// Information about a consolidation query to be executed.
50#[derive(Debug, Clone)]
51pub struct ConsolidationQuery {
52    /// Start timestamp for the query range (inclusive, in nanoseconds).
53    pub query_start: u64,
54    /// End timestamp for the query range (inclusive, in nanoseconds).
55    pub query_end: u64,
56    /// Whether to use period boundaries for file naming (true) or actual data timestamps (false).
57    pub use_period_boundaries: bool,
58}
59
60impl ParquetDataCatalog {
61    /// Consolidates all data files in the catalog.
62    ///
63    /// This method identifies all leaf directories in the catalog that contain parquet files
64    /// and consolidates them. A leaf directory is one that contains files but no subdirectories.
65    /// This is a convenience method that effectively calls `consolidate_data` for all data types
66    /// and instrument IDs in the catalog.
67    ///
68    /// # Parameters
69    ///
70    /// - `start`: Optional start timestamp for the consolidation range. Only files with timestamps
71    ///   greater than or equal to this value will be consolidated. If None, all files
72    ///   from the beginning of time will be considered.
73    /// - `end`: Optional end timestamp for the consolidation range. Only files with timestamps
74    ///   less than or equal to this value will be consolidated. If None, all files
75    ///   up to the end of time will be considered.
76    /// - `ensure_contiguous_files`: Whether to validate that consolidated intervals are contiguous (default: true).
77    ///
78    /// # Returns
79    ///
80    /// Returns `Ok(())` on success, or an error if consolidation fails for any directory.
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if:
85    /// - Directory listing fails.
86    /// - File consolidation operations fail.
87    /// - Interval validation fails (when `ensure_contiguous_files` is true).
88    ///
89    /// # Examples
90    ///
91    /// ```rust,no_run
92    /// use nautilus_core::UnixNanos;
93    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
94    ///
95    /// let mut catalog = ParquetDataCatalog::new(
96    ///     std::path::Path::new("/tmp/nautilus_data"),
97    ///     None,
98    ///     None,
99    ///     None,
100    ///     None,
101    /// );
102    ///
103    /// // Consolidate all files in the catalog
104    /// catalog.consolidate_catalog(None, None, None, None)?;
105    ///
106    /// // Consolidate only files within a specific time range
107    /// catalog.consolidate_catalog(
108    ///     Some(UnixNanos::from(1609459200000000000)),
109    ///     Some(UnixNanos::from(1609545600000000000)),
110    ///     Some(true),
111    ///     None,
112    /// )?;
113    /// # Ok::<(), anyhow::Error>(())
114    /// ```
115    pub fn consolidate_catalog(
116        &self,
117        start: Option<UnixNanos>,
118        end: Option<UnixNanos>,
119        ensure_contiguous_files: Option<bool>,
120        deduplicate: Option<bool>,
121    ) -> anyhow::Result<()> {
122        let leaf_directories = self.find_leaf_data_directories()?;
123
124        for directory in leaf_directories {
125            self.consolidate_directory(
126                &directory,
127                start,
128                end,
129                ensure_contiguous_files,
130                deduplicate,
131            )?;
132        }
133
134        Ok(())
135    }
136
137    /// Consolidates data files for a specific data type and identifier.
138    ///
139    /// This method consolidates Parquet files within a specific directory (defined by data type
140    /// and optional identifier) by merging multiple files into a single file. This improves
141    /// query performance and can reduce storage overhead.
142    ///
143    /// # Parameters
144    ///
145    /// - `data_type`: The stored family to consolidate.
146    /// - `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").
147    /// - `start`: Optional start timestamp to limit consolidation to files within this range.
148    /// - `end`: Optional end timestamp to limit consolidation to files within this range.
149    /// - `ensure_contiguous_files`: Whether to validate that consolidated intervals are contiguous (default: true).
150    ///
151    /// # Returns
152    ///
153    /// Returns `Ok(())` on success, or an error if consolidation fails.
154    ///
155    /// # Errors
156    ///
157    /// Returns an error if:
158    /// - The directory path cannot be constructed.
159    /// - File consolidation operations fail.
160    /// - Interval validation fails (when `ensure_contiguous_files` is true).
161    ///
162    /// # Examples
163    ///
164    /// ```rust,no_run
165    /// use nautilus_core::UnixNanos;
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    /// // Consolidate all quote files for a specific instrument
178    /// catalog.consolidate_data(
179    ///     &NautilusDataType::QuoteTick.into(),
180    ///     Some("BTCUSD"),
181    ///     None,
182    ///     None,
183    ///     None,
184    ///     None,
185    /// )?;
186    ///
187    /// // Consolidate trade files within a time range
188    /// catalog.consolidate_data(
189    ///     &NautilusDataType::TradeTick.into(),
190    ///     None,
191    ///     Some(UnixNanos::from(1609459200000000000)),
192    ///     Some(UnixNanos::from(1609545600000000000)),
193    ///     Some(true),
194    ///     None,
195    /// )?;
196    /// # Ok::<(), anyhow::Error>(())
197    /// ```
198    pub fn consolidate_data(
199        &mut self,
200        data_type: &CatalogDataType,
201        identifier: Option<&str>,
202        start: Option<UnixNanos>,
203        end: Option<UnixNanos>,
204        ensure_contiguous_files: Option<bool>,
205        deduplicate: Option<bool>,
206    ) -> anyhow::Result<()> {
207        for type_name in parquet_catalog_data_type_path_prefixes(data_type) {
208            self.consolidate_prefix_data(
209                type_name.as_ref(),
210                identifier,
211                start,
212                end,
213                ensure_contiguous_files,
214                deduplicate,
215            )?;
216        }
217
218        Ok(())
219    }
220
221    fn consolidate_prefix_data(
222        &mut self,
223        type_name: &str,
224        identifier: Option<&str>,
225        start: Option<UnixNanos>,
226        end: Option<UnixNanos>,
227        ensure_contiguous_files: Option<bool>,
228        deduplicate: Option<bool>,
229    ) -> anyhow::Result<()> {
230        let directory = self.make_path(type_name, identifier)?;
231        let raw_result = self.consolidate_directory(
232            &directory,
233            start,
234            end,
235            ensure_contiguous_files,
236            deduplicate,
237        );
238
239        match raw_result {
240            Ok(()) => Ok(()),
241            Err(raw_error)
242                if can_rewrite_consolidation_by_period(type_name)
243                    && is_schema_incompatibility(&raw_error) =>
244            {
245                if deduplicate.unwrap_or(false) {
246                    anyhow::bail!(
247                        "Raw parquet consolidation failed due to incompatible file schemas, \
248                         but typed period consolidation cannot preserve deduplicate=true. \
249                         Raw consolidation error: {raw_error}"
250                    );
251                }
252
253                log::warn!(
254                    "Raw parquet consolidation failed due to incompatible file schemas for \
255                     {type_name}; retrying with typed period consolidation. Raw error: {raw_error}"
256                );
257
258                self.consolidate_prefix_data_by_period(
259                    type_name,
260                    identifier,
261                    None,
262                    start,
263                    end,
264                    ensure_contiguous_files,
265                )
266                .map_err(|typed_error| {
267                    anyhow::anyhow!(
268                        "Raw parquet consolidation failed due to incompatible file schemas, \
269                         and typed period consolidation also failed. Raw error: {raw_error}; \
270                         typed period error: {typed_error}"
271                    )
272                })
273            }
274            Err(e) => Err(e),
275        }
276    }
277
278    /// Consolidates Parquet files within a specific directory by merging them into a single file.
279    ///
280    /// This internal method performs the actual consolidation work for a single directory.
281    /// It identifies files within the specified time range, validates their intervals,
282    /// and combines them into a single Parquet file with optimized storage.
283    ///
284    /// # Parameters
285    ///
286    /// - `directory`: The directory path containing Parquet files to consolidate.
287    /// - `start`: Optional start timestamp to limit consolidation to files within this range.
288    /// - `end`: Optional end timestamp to limit consolidation to files within this range.
289    /// - `ensure_contiguous_files`: Whether to validate that consolidated intervals are contiguous.
290    ///
291    /// # Returns
292    ///
293    /// Returns `Ok(())` on success, or an error if consolidation fails.
294    ///
295    /// # Behavior
296    ///
297    /// - Skips consolidation if directory contains 1 or fewer files.
298    /// - Filters files by timestamp range if start/end are specified.
299    /// - Sorts intervals by start timestamp before consolidation.
300    /// - Creates a new file spanning the entire time range of input files.
301    /// - Validates interval disjointness after consolidation (if enabled).
302    ///
303    /// # Errors
304    ///
305    /// Returns an error if:
306    /// - Directory listing fails.
307    /// - File combination operations fail.
308    /// - Interval validation fails (when `ensure_contiguous_files` is true).
309    /// - Object store operations fail.
310    fn consolidate_directory(
311        &self,
312        directory: &str,
313        start: Option<UnixNanos>,
314        end: Option<UnixNanos>,
315        ensure_contiguous_files: Option<bool>,
316        deduplicate: Option<bool>,
317    ) -> anyhow::Result<()> {
318        let parquet_files = self.list_parquet_files(directory)?;
319
320        if parquet_files.len() <= 1 {
321            return Ok(());
322        }
323
324        let mut files_to_consolidate = Vec::new();
325        let start = start.map(|t| t.as_u64());
326        let end = end.map(|t| t.as_u64());
327
328        for file in parquet_files {
329            let Some(interval) = parse_filename_timestamps(&file) else {
330                continue;
331            };
332
333            if start.is_none_or(|s| interval.0 >= s) && end.is_none_or(|e| interval.1 <= e) {
334                files_to_consolidate.push((file, interval));
335            }
336        }
337
338        files_to_consolidate.sort_by_key(|&(_, (interval_start, _))| interval_start);
339        let intervals: Vec<(u64, u64)> = files_to_consolidate
340            .iter()
341            .map(|&(_, interval)| interval)
342            .collect();
343
344        // Validate disjointness before merging so source files are left untouched on failure
345        if ensure_contiguous_files.unwrap_or(true) && !are_intervals_disjoint(&intervals) {
346            anyhow::bail!("Intervals are not disjoint before consolidating a directory");
347        }
348
349        let Some(last_end) = intervals.iter().map(|interval| interval.1).max() else {
350            return Ok(());
351        };
352
353        let file_name =
354            timestamps_to_filename(UnixNanos::from(intervals[0].0), UnixNanos::from(last_end));
355        let path = make_object_store_path(directory, [&file_name]);
356
357        let object_paths = files_to_consolidate
358            .iter()
359            .map(|(path, _)| self.to_object_path_parsed(path))
360            .collect::<anyhow::Result<Vec<_>>>()?;
361
362        self.execute_async(|| async {
363            combine_parquet_files_from_object_store(
364                self.object_store.clone(),
365                object_paths,
366                &ObjectPath::from(path),
367                Some(self.compression),
368                Some(self.max_row_group_size),
369                deduplicate,
370            )
371            .await
372        })
373    }
374
375    /// Consolidates all data files in the catalog by splitting them into fixed time periods.
376    ///
377    /// This method identifies all leaf directories in the catalog that contain parquet files
378    /// and consolidates them by period. A leaf directory is one that contains files but no subdirectories.
379    /// This is a convenience method that effectively calls `consolidate_data_by_period` for all data types
380    /// and instrument IDs in the catalog.
381    ///
382    /// # Parameters
383    ///
384    /// - `period_nanos`: The period duration for consolidation in nanoseconds. Default is 1 day (86400000000000).
385    ///   Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
386    /// - `start`: Optional start timestamp for the consolidation range. Only files with timestamps
387    ///   greater than or equal to this value will be consolidated. If None, all files
388    ///   from the beginning of time will be considered.
389    /// - `end`: Optional end timestamp for the consolidation range. Only files with timestamps
390    ///   less than or equal to this value will be consolidated. If None, all files
391    ///   up to the end of time will be considered.
392    /// - `ensure_contiguous_files`: If true, uses period boundaries for file naming.
393    ///   If false, uses actual data timestamps for file naming.
394    ///
395    /// # Returns
396    ///
397    /// Returns `Ok(())` on success, or an error if consolidation fails for any directory.
398    ///
399    /// # Errors
400    ///
401    /// Returns an error if:
402    /// - `period_nanos` is zero and a directory holds data to consolidate.
403    /// - Directory listing fails.
404    /// - Data type extraction from path fails.
405    /// - Period-based consolidation operations fail.
406    ///
407    /// # Notes
408    ///
409    /// - This operation can be resource-intensive for large catalogs with many data types.
410    ///   and instruments.
411    /// - The consolidation process splits data into fixed time periods rather than combining.
412    ///   all files into a single file per directory.
413    /// - Uses the same period-based consolidation logic as `consolidate_data_by_period`.
414    /// - Original files are removed and replaced with period-based consolidated files.
415    /// - This method is useful for periodic maintenance of the catalog to standardize.
416    ///   file organization by time periods.
417    ///
418    /// # Examples
419    ///
420    /// ```rust,no_run
421    /// use nautilus_core::UnixNanos;
422    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
423    ///
424    /// let mut catalog = ParquetDataCatalog::new(
425    ///     std::path::Path::new("/tmp/nautilus_data"),
426    ///     None,
427    ///     None,
428    ///     None,
429    ///     None,
430    /// );
431    ///
432    /// // Consolidate all files in the catalog by 1-day periods
433    /// catalog.consolidate_catalog_by_period(
434    ///     Some(86400000000000), // 1 day in nanoseconds
435    ///     None,
436    ///     None,
437    ///     Some(true),
438    /// )?;
439    ///
440    /// // Consolidate only files within a specific time range by 1-hour periods
441    /// catalog.consolidate_catalog_by_period(
442    ///     Some(3600000000000), // 1 hour in nanoseconds
443    ///     Some(UnixNanos::from(1609459200000000000)),
444    ///     Some(UnixNanos::from(1609545600000000000)),
445    ///     Some(false),
446    /// )?;
447    /// # Ok::<(), anyhow::Error>(())
448    /// ```
449    pub fn consolidate_catalog_by_period(
450        &mut self,
451        period_nanos: Option<u64>,
452        start: Option<UnixNanos>,
453        end: Option<UnixNanos>,
454        ensure_contiguous_files: Option<bool>,
455    ) -> anyhow::Result<()> {
456        let leaf_directories = self.find_leaf_data_directories()?;
457
458        for directory in leaf_directories {
459            let (Some(data_cls_name), identifier) =
460                self.extract_data_cls_and_identifier_from_path(&directory)?
461            else {
462                continue;
463            };
464
465            if !self.dispatch_consolidate_data_by_period(
466                &data_cls_name,
467                identifier.as_deref(),
468                period_nanos,
469                start,
470                end,
471                ensure_contiguous_files,
472            )? {
473                // Skip unknown data types
474                log::warn!("Unknown data type for consolidation: {data_cls_name}");
475            }
476        }
477
478        Ok(())
479    }
480
481    /// Extracts data class and identifier from a directory path.
482    ///
483    /// This method parses a directory path to extract the data type and optional
484    /// instrument identifier. It's used to determine what type of data consolidation
485    /// to perform for each directory.
486    ///
487    /// # Parameters
488    ///
489    /// - `path`: The directory path to parse.
490    ///
491    /// # Returns
492    ///
493    /// Returns a tuple of (`data_class`, identifier) where both are optional strings.
494    pub fn extract_data_cls_and_identifier_from_path(
495        &self,
496        path: &str,
497    ) -> anyhow::Result<(Option<String>, Option<String>)> {
498        // Parse from the catalog root so a base path containing a `data` segment is skipped
499        let path = self.path_without_local_base(&normalize_path_separators(path));
500        let path_components = extract_path_components(&path);
501
502        // Find the "data" directory in the path
503        if let Some(data_index) = path_components.iter().position(|part| part == "data")
504            && data_index + 1 < path_components.len()
505        {
506            let second = &path_components[data_index + 1];
507
508            if second == "custom" {
509                let Some(type_name) = path_components.get(data_index + 2) else {
510                    return Ok((None, None));
511                };
512
513                let identifier = (path_components.len() > data_index + 3)
514                    .then(|| path_components[data_index + 3..].join("/"));
515                return Ok((Some(format!("custom/{type_name}")), identifier));
516            }
517
518            let data_cls = second.clone();
519            let identifier = path_components.get(data_index + 2).cloned();
520
521            return Ok((Some(data_cls), identifier));
522        }
523
524        // If we can't parse the path, return None for both
525        Ok((None, None))
526    }
527
528    /// Consolidates data files by splitting them into fixed time periods.
529    ///
530    /// This method queries data by period and writes consolidated files immediately,
531    /// using efficient period-based consolidation logic. When start/end boundaries intersect existing files,
532    /// the function automatically splits those files to preserve all data.
533    ///
534    /// # Parameters
535    ///
536    /// - `data_type`: The stored family to consolidate.
537    /// - `identifier`: Optional instrument ID to consolidate. If None, consolidates all instruments.
538    /// - `period_nanos`: The period duration for consolidation in nanoseconds. Default is 1 day (86400000000000).
539    ///   Examples: 3600000000000 (1 hour), 604800000000000 (7 days), 1800000000000 (30 minutes)
540    /// - `start`: Optional start timestamp for consolidation range. If None, uses earliest available data.
541    ///   If specified and intersects existing files, those files will be split to preserve
542    ///   data outside the consolidation range.
543    /// - `end`: Optional end timestamp for consolidation range. If None, uses latest available data.
544    ///   If specified and intersects existing files, those files will be split to preserve
545    ///   data outside the consolidation range.
546    /// - `ensure_contiguous_files`: If true, uses period boundaries for file naming.
547    ///   If false, uses actual data timestamps for file naming.
548    ///
549    /// # Returns
550    ///
551    /// Returns `Ok(())` on success, or an error if consolidation fails.
552    ///
553    /// # Errors
554    ///
555    /// Returns an error if:
556    /// - `period_nanos` is zero and the directory holds data to consolidate.
557    /// - `data_type` is a record family or an instrument selector, which have no
558    ///   period-typed rewrite; use [`Self::consolidate_data`] for those.
559    /// - The directory path cannot be constructed.
560    /// - File operations fail.
561    /// - Data querying or writing fails.
562    ///
563    /// # Notes
564    ///
565    /// - Uses two-phase approach: first determines all queries, then executes them.
566    /// - Groups intervals into contiguous groups to preserve holes between groups.
567    /// - Allows consolidation across multiple files within each contiguous group.
568    /// - Skips queries if target files already exist for efficiency.
569    /// - Original files are removed immediately after querying each period.
570    /// - When `ensure_contiguous_files=false`, file timestamps match actual data range.
571    /// - When `ensure_contiguous_files=true`, file timestamps use period boundaries.
572    /// - Uses modulo arithmetic for efficient period boundary calculation.
573    /// - Preserves holes in data by preventing queries from spanning across gaps.
574    /// - Automatically splits files at start/end boundaries to preserve all data.
575    /// - Split operations are executed before consolidation to ensure data preservation.
576    ///
577    /// # Examples
578    ///
579    /// ```rust,no_run
580    /// use nautilus_core::UnixNanos;
581    /// use nautilus_model::data::NautilusDataType;
582    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
583    ///
584    /// let mut catalog = ParquetDataCatalog::new(
585    ///     std::path::Path::new("/tmp/nautilus_data"),
586    ///     None,
587    ///     None,
588    ///     None,
589    ///     None,
590    /// );
591    ///
592    /// // Consolidate all quote files by 1-day periods
593    /// catalog.consolidate_data_by_period(
594    ///     &NautilusDataType::QuoteTick.into(),
595    ///     None,
596    ///     Some(86400000000000), // 1 day in nanoseconds
597    ///     None,
598    ///     None,
599    ///     Some(true),
600    /// )?;
601    ///
602    /// // Consolidate specific instrument by 1-hour periods
603    /// catalog.consolidate_data_by_period(
604    ///     &NautilusDataType::TradeTick.into(),
605    ///     Some("BTCUSD"),
606    ///     Some(3600000000000), // 1 hour in nanoseconds
607    ///     Some(UnixNanos::from(1609459200000000000)),
608    ///     Some(UnixNanos::from(1609545600000000000)),
609    ///     Some(false),
610    /// )?;
611    /// # Ok::<(), anyhow::Error>(())
612    /// ```
613    pub fn consolidate_data_by_period(
614        &mut self,
615        data_type: &CatalogDataType,
616        identifier: Option<&str>,
617        period_nanos: Option<u64>,
618        start: Option<UnixNanos>,
619        end: Option<UnixNanos>,
620        ensure_contiguous_files: Option<bool>,
621    ) -> anyhow::Result<()> {
622        anyhow::ensure!(
623            matches!(
624                data_type,
625                CatalogDataType::Data(data_type) if *data_type != NautilusDataType::Instrument
626            ),
627            "Period consolidation applies to data families only, not {data_type}; \
628             use consolidate_data",
629        );
630
631        for type_name in parquet_catalog_data_type_path_prefixes(data_type) {
632            self.consolidate_prefix_data_by_period(
633                type_name.as_ref(),
634                identifier,
635                period_nanos,
636                start,
637                end,
638                ensure_contiguous_files,
639            )?;
640        }
641
642        Ok(())
643    }
644
645    fn consolidate_prefix_data_by_period(
646        &mut self,
647        type_name: &str,
648        identifier: Option<&str>,
649        period_nanos: Option<u64>,
650        start: Option<UnixNanos>,
651        end: Option<UnixNanos>,
652        ensure_contiguous_files: Option<bool>,
653    ) -> anyhow::Result<()> {
654        if !self.dispatch_consolidate_data_by_period(
655            type_name,
656            identifier,
657            period_nanos,
658            start,
659            end,
660            ensure_contiguous_files,
661        )? {
662            anyhow::bail!("Unknown data type for consolidation: {type_name}");
663        }
664
665        Ok(())
666    }
667
668    /// Dispatches period-based consolidation for `type_name`, returning `Ok(false)` for
669    /// unknown data types so callers choose whether to warn or fail.
670    fn dispatch_consolidate_data_by_period(
671        &mut self,
672        type_name: &str,
673        identifier: Option<&str>,
674        period_nanos: Option<u64>,
675        start: Option<UnixNanos>,
676        end: Option<UnixNanos>,
677        ensure_contiguous_files: Option<bool>,
678    ) -> anyhow::Result<bool> {
679        match type_name {
680            "quotes" => {
681                self.consolidate_data_by_period_generic::<QuoteTick>(
682                    identifier,
683                    period_nanos,
684                    start,
685                    end,
686                    ensure_contiguous_files,
687                )?;
688            }
689            "trades" => {
690                self.consolidate_data_by_period_generic::<TradeTick>(
691                    identifier,
692                    period_nanos,
693                    start,
694                    end,
695                    ensure_contiguous_files,
696                )?;
697            }
698            "order_book_deltas" => {
699                self.consolidate_data_by_period_generic::<OrderBookDelta>(
700                    identifier,
701                    period_nanos,
702                    start,
703                    end,
704                    ensure_contiguous_files,
705                )?;
706            }
707            "order_book_depths" => {
708                self.consolidate_data_by_period_generic::<OrderBookDepth>(
709                    identifier,
710                    period_nanos,
711                    start,
712                    end,
713                    ensure_contiguous_files,
714                )?;
715            }
716            "bars" => {
717                self.consolidate_data_by_period_generic::<Bar>(
718                    identifier,
719                    period_nanos,
720                    start,
721                    end,
722                    ensure_contiguous_files,
723                )?;
724            }
725            "index_prices" => {
726                self.consolidate_data_by_period_generic::<IndexPriceUpdate>(
727                    identifier,
728                    period_nanos,
729                    start,
730                    end,
731                    ensure_contiguous_files,
732                )?;
733            }
734            "mark_prices" => {
735                self.consolidate_data_by_period_generic::<MarkPriceUpdate>(
736                    identifier,
737                    period_nanos,
738                    start,
739                    end,
740                    ensure_contiguous_files,
741                )?;
742            }
743            "instrument_closes" => {
744                self.consolidate_data_by_period_generic::<InstrumentClose>(
745                    identifier,
746                    period_nanos,
747                    start,
748                    end,
749                    ensure_contiguous_files,
750                )?;
751            }
752            _ => {
753                if let Some(custom_type_name) = type_name.strip_prefix("custom/") {
754                    self.consolidate_custom_data_by_period(
755                        custom_type_name,
756                        identifier,
757                        period_nanos,
758                        start,
759                        end,
760                        ensure_contiguous_files,
761                    )?;
762                } else {
763                    return Ok(false);
764                }
765            }
766        }
767
768        Ok(true)
769    }
770
771    /// Generic consolidate data files by splitting them into fixed time periods.
772    ///
773    /// This is a type-safe version of `consolidate_data_by_period` that uses generic types
774    /// to ensure compile-time correctness and enable reuse across different data types.
775    ///
776    /// # Type Parameters
777    ///
778    /// - `T`: The data type to consolidate, must implement required traits for serialization.
779    ///
780    /// # Parameters
781    ///
782    /// - `identifier`: Optional instrument ID to target a specific instrument's data.
783    /// - `period_nanos`: Optional period size in nanoseconds (default: 1 day).
784    /// - `start`: Optional start timestamp for consolidation range.
785    /// - `end`: Optional end timestamp for consolidation range.
786    /// - `ensure_contiguous_files`: Optional flag to control file naming strategy.
787    ///
788    /// # Returns
789    ///
790    /// Returns `Ok(())` on success, or an error if consolidation fails.
791    pub fn consolidate_data_by_period_generic<T>(
792        &mut self,
793        identifier: Option<&str>,
794        period_nanos: Option<u64>,
795        start: Option<UnixNanos>,
796        end: Option<UnixNanos>,
797        ensure_contiguous_files: Option<bool>,
798    ) -> anyhow::Result<()>
799    where
800        T: DecodeTypedFromRecordBatch
801            + HasCatalogDataType
802            + EncodeToRecordBatch
803            + HasTsInit
804            + TryFrom<Data>
805            + Clone,
806    {
807        let period_nanos = period_nanos.unwrap_or(86_400_000_000_000); // Default: 1 day
808        let ensure_contiguous_files = ensure_contiguous_files.unwrap_or(true);
809
810        // Use get_intervals for cleaner implementation
811        let data_type = T::catalog_data_type();
812        let path_prefix = parquet_data_path_prefix(&data_type);
813        let intervals =
814            self.get_intervals(&CatalogDataType::Data(data_type.clone()), identifier)?;
815
816        if intervals.is_empty() {
817            return Ok(()); // No files to consolidate
818        }
819
820        // Use auxiliary function to prepare all queries for execution
821        let queries_to_execute = self.prepare_consolidation_queries(
822            path_prefix.as_ref(),
823            identifier,
824            &intervals,
825            period_nanos,
826            start,
827            end,
828            ensure_contiguous_files,
829        )?;
830
831        if queries_to_execute.is_empty() {
832            return Ok(()); // No queries to execute
833        }
834
835        // Get directory for file operations
836        let directory = self.make_path(path_prefix.as_ref(), identifier)?;
837        let mut existing_files = self.list_parquet_files(&directory)?;
838        existing_files.sort();
839
840        // Capture the overall window's left bound before the loop consumes queries_to_execute,
841        // a source file is only deleted when its interval is fully consumed by the consolidation.
842        let overall_query_start = queries_to_execute[0].query_start;
843
844        // Phase 2: Execute queries, write, and delete
845        let mut file_start_ns: Option<u64> = None; // Track contiguity across periods
846
847        for query_info in queries_to_execute {
848            // Query data for this period
849            let instrument_ids = identifier.map(|id| vec![id.to_string()]);
850
851            // Use optimize_file_loading=false to match Python behavior:
852            // During consolidation, we want to read only the specific files being consolidated,
853            // not the entire directory. This ensures precise file control during consolidation.
854            let period_data = self.query::<T>(
855                instrument_ids,
856                Some(UnixNanos::from(query_info.query_start)),
857                Some(UnixNanos::from(query_info.query_end)),
858                None,
859                Some(existing_files.clone()),
860                false, // optimize_file_loading=false for precise file control during consolidation
861            )?;
862
863            if period_data.is_empty() {
864                // Skip if no data found, but maintain contiguity by using query start
865                if file_start_ns.is_none() {
866                    file_start_ns = Some(query_info.query_start);
867                }
868
869                continue;
870            }
871
872            // Determine final file timestamps
873            let (final_start_ns, final_end_ns) = if query_info.use_period_boundaries {
874                // Use period boundaries for file naming, maintaining contiguity
875                if file_start_ns.is_none() {
876                    file_start_ns = Some(query_info.query_start);
877                }
878
879                let start = file_start_ns.unwrap();
880                (start, query_info.query_end)
881            } else {
882                // Use actual data timestamps for file naming
883                let first_ts = period_data.first().unwrap().ts_init().as_u64();
884                let last_ts = period_data.last().unwrap().ts_init().as_u64();
885                (first_ts, last_ts)
886            };
887
888            // Check again if target file exists (in case it was created during this process)
889            let target_filename = format!(
890                "{}/{}",
891                directory,
892                timestamps_to_filename(
893                    UnixNanos::from(final_start_ns),
894                    UnixNanos::from(final_end_ns)
895                )
896            );
897
898            if self.file_exists(&target_filename)? {
899                // This period is already consolidated; do not let a later cleanup delete it.
900                let target_object_path = self.to_object_path(&target_filename)?.to_string();
901                existing_files.retain(|f| f != &target_object_path);
902                // Reset so the next period starts a new segment after the existing file
903                file_start_ns = None;
904                continue;
905            }
906
907            // Write consolidated data for this period using write_to_parquet
908            // Use skip_disjoint_check since we're managing file removal carefully
909            let start_ts = UnixNanos::from(final_start_ns);
910            let end_ts = UnixNanos::from(final_end_ns);
911            self.write_to_parquet(&period_data, Some(start_ts), Some(end_ts), Some(true))?;
912
913            // Delete files fully consumed by this period; keep straddlers so no data is lost
914            for file in existing_files.clone() {
915                if let Some(interval) = parse_filename_timestamps(&file)
916                    && interval.1 <= query_info.query_end
917                    && interval.0 >= overall_query_start
918                {
919                    existing_files.retain(|f| f != &file);
920                    self.delete_file(&file)?;
921                }
922            }
923
924            // Reset so next period starts a new contiguous segment
925            file_start_ns = None;
926        }
927
928        Ok(())
929    }
930
931    /// Consolidates custom data files by splitting them into fixed time periods.
932    ///
933    /// This method provides consolidation for custom data types that don't have compile-time
934    /// type information. It uses dynamic querying and writing methods.
935    ///
936    /// # Parameters
937    ///
938    /// - `type_name`: The custom data type name (without "custom/" prefix).
939    /// - `identifier`: Optional instrument ID to consolidate.
940    /// - `period_nanos`: Optional period size in nanoseconds (default: 1 day).
941    /// - `start`: Optional start timestamp for consolidation range.
942    /// - `end`: Optional end timestamp for consolidation range.
943    /// - `ensure_contiguous_files`: Optional flag to control file naming strategy.
944    ///
945    /// # Returns
946    ///
947    /// Returns `Ok(())` on success, or an error if consolidation fails.
948    fn consolidate_custom_data_by_period(
949        &mut self,
950        type_name: &str,
951        identifier: Option<&str>,
952        period_nanos: Option<u64>,
953        start: Option<UnixNanos>,
954        end: Option<UnixNanos>,
955        ensure_contiguous_files: Option<bool>,
956    ) -> anyhow::Result<()> {
957        let period_nanos = period_nanos.unwrap_or(86_400_000_000_000); // Default: 1 day
958        let ensure_contiguous_files = ensure_contiguous_files.unwrap_or(true);
959
960        // Get intervals for the custom data type
961        let data_type = NautilusDataType::Custom {
962            type_name: type_name.to_string(),
963        };
964
965        let path_prefix = parquet_data_path_prefix(&data_type);
966        let intervals = self.get_intervals(&CatalogDataType::Data(data_type), identifier)?;
967
968        if intervals.is_empty() {
969            return Ok(()); // No files to consolidate
970        }
971
972        // Use auxiliary function to prepare all queries for execution
973        let queries_to_execute = self.prepare_consolidation_queries(
974            path_prefix.as_ref(),
975            identifier,
976            &intervals,
977            period_nanos,
978            start,
979            end,
980            ensure_contiguous_files,
981        )?;
982
983        if queries_to_execute.is_empty() {
984            return Ok(()); // No queries to execute
985        }
986
987        // Get directory for file operations
988        let directory = self.make_path(path_prefix.as_ref(), identifier)?;
989        let mut existing_files = self.list_parquet_files(&directory)?;
990        existing_files.sort();
991
992        // Capture the overall window's left bound before the loop consumes queries_to_execute,
993        // a source file is only deleted when its interval is fully consumed by the consolidation.
994        let overall_query_start = queries_to_execute[0].query_start;
995
996        // Phase 2: Execute queries, write, and delete
997        let mut file_start_ns: Option<u64> = None; // Track contiguity across periods
998
999        for query_info in queries_to_execute {
1000            // Query custom data for this period using query_custom_data_dynamic
1001            let instrument_ids = identifier.map(|id| vec![id.to_string()]);
1002
1003            let period_data = self.query_custom_data_dynamic(
1004                type_name,
1005                instrument_ids.as_deref(),
1006                Some(UnixNanos::from(query_info.query_start)),
1007                Some(UnixNanos::from(query_info.query_end)),
1008                None,
1009                Some(existing_files.clone()),
1010                false, // optimize_file_loading=false for precise file control during consolidation
1011            )?;
1012
1013            if period_data.is_empty() {
1014                // Skip if no data found, but maintain contiguity by using query start
1015                if file_start_ns.is_none() {
1016                    file_start_ns = Some(query_info.query_start);
1017                }
1018
1019                continue;
1020            }
1021
1022            // Determine final file timestamps
1023            let (final_start_ns, final_end_ns) = if query_info.use_period_boundaries {
1024                // Use period boundaries for file naming, maintaining contiguity
1025                if file_start_ns.is_none() {
1026                    file_start_ns = Some(query_info.query_start);
1027                }
1028
1029                let start = file_start_ns.unwrap();
1030                (start, query_info.query_end)
1031            } else {
1032                // Use actual data timestamps for file naming
1033                let first_ts = period_data.first().unwrap().ts_init().as_u64();
1034                let last_ts = period_data.last().unwrap().ts_init().as_u64();
1035                (first_ts, last_ts)
1036            };
1037
1038            // Check again if target file exists (in case it was created during this process)
1039            let target_filename = format!(
1040                "{}/{}",
1041                directory,
1042                timestamps_to_filename(
1043                    UnixNanos::from(final_start_ns),
1044                    UnixNanos::from(final_end_ns)
1045                )
1046            );
1047
1048            if self.file_exists(&target_filename)? {
1049                // This period is already consolidated; do not let a later cleanup delete it.
1050                let target_object_path = self.to_object_path(&target_filename)?.to_string();
1051                existing_files.retain(|f| f != &target_object_path);
1052                // Reset so the next period starts a new segment after the existing file
1053                file_start_ns = None;
1054                continue;
1055            }
1056
1057            let custom_items: Vec<CustomData> = period_data
1058                .into_iter()
1059                .filter_map(|data| match data {
1060                    Data::Custom(c) => Some(c),
1061                    _ => None,
1062                })
1063                .collect();
1064
1065            // Write consolidated data for each type
1066            let start_ts = UnixNanos::from(final_start_ns);
1067            let end_ts = UnixNanos::from(final_end_ns);
1068
1069            for items in group_custom_data_by_type(custom_items.iter()) {
1070                self.write_custom_data_refs_batch(
1071                    &items,
1072                    Some(start_ts),
1073                    Some(end_ts),
1074                    Some(true),
1075                )?;
1076            }
1077
1078            // Delete files fully consumed by this period; keep straddlers so no data is lost
1079            for file in existing_files.clone() {
1080                if let Some(interval) = parse_filename_timestamps(&file)
1081                    && interval.1 <= query_info.query_end
1082                    && interval.0 >= overall_query_start
1083                {
1084                    existing_files.retain(|f| f != &file);
1085                    self.delete_file(&file)?;
1086                }
1087            }
1088
1089            // Reset so next period starts a new contiguous segment
1090            file_start_ns = None;
1091        }
1092
1093        Ok(())
1094    }
1095
1096    /// Prepares all queries for consolidation by filtering, grouping, and handling splits.
1097    ///
1098    /// This auxiliary function handles all the preparation logic for consolidation:
1099    /// 1. Filters intervals by time range.
1100    /// 2. Groups intervals into contiguous groups.
1101    /// 3. Identifies and creates split operations for data preservation.
1102    /// 4. Generates period-based consolidation queries.
1103    /// 5. Checks for existing target files.
1104    ///
1105    /// # Errors
1106    ///
1107    /// Returns an error if:
1108    /// - `period_nanos` is zero.
1109    /// - The intervals are not contiguous when `ensure_contiguous_files` is true.
1110    /// - A target file existence check fails.
1111    #[expect(
1112        clippy::too_many_arguments,
1113        reason = "Consolidation keeps its window and policy arguments explicit"
1114    )]
1115    pub fn prepare_consolidation_queries(
1116        &self,
1117        type_name: &str,
1118        identifier: Option<&str>,
1119        intervals: &[(u64, u64)],
1120        period_nanos: u64,
1121        start: Option<UnixNanos>,
1122        end: Option<UnixNanos>,
1123        ensure_contiguous_files: bool,
1124    ) -> anyhow::Result<Vec<ConsolidationQuery>> {
1125        anyhow::ensure!(period_nanos > 0, "period_nanos must be positive");
1126
1127        // Filter intervals by time range if specified
1128        let used_start = start.map(|s| s.as_u64());
1129        let used_end = end.map(|e| e.as_u64());
1130
1131        let mut filtered_intervals = Vec::new();
1132
1133        for &(interval_start, interval_end) in intervals {
1134            // Check if interval overlaps with the specified range
1135            if used_start.is_none_or(|start| start <= interval_end)
1136                && used_end.is_none_or(|end| interval_start <= end)
1137            {
1138                filtered_intervals.push((interval_start, interval_end));
1139            }
1140        }
1141
1142        if filtered_intervals.is_empty() {
1143            return Ok(Vec::new()); // No intervals in the specified range
1144        }
1145
1146        // Check contiguity of filtered intervals if required
1147        if ensure_contiguous_files && !are_intervals_contiguous(&filtered_intervals) {
1148            anyhow::bail!(
1149                "Intervals are not contiguous. When ensure_contiguous_files=true, \
1150                 all files in the consolidation range must have contiguous timestamps."
1151            );
1152        }
1153
1154        // Group intervals by the target period: split only when the gap between files
1155        // exceeds one period, since sub-period gaps land in the same consolidated file.
1156        let contiguous_groups = self.group_contiguous_intervals(&filtered_intervals, period_nanos);
1157
1158        let mut queries_to_execute = Vec::new();
1159
1160        // Handle interval splitting by creating split operations for data preservation
1161        if !filtered_intervals.is_empty() {
1162            if let Some(start_ts) = used_start {
1163                let first_interval = filtered_intervals[0];
1164                if first_interval.0 < start_ts && start_ts <= first_interval.1 {
1165                    // Split before start: preserve data from interval_start to start-1
1166                    queries_to_execute.push(ConsolidationQuery {
1167                        query_start: first_interval.0,
1168                        query_end: start_ts - 1,
1169                        use_period_boundaries: false,
1170                    });
1171                }
1172            }
1173
1174            if let Some(end_ts) = used_end {
1175                let last_interval = filtered_intervals[filtered_intervals.len() - 1];
1176                if last_interval.0 <= end_ts && end_ts < last_interval.1 {
1177                    // Split after end: preserve data from end+1 to interval_end
1178                    queries_to_execute.push(ConsolidationQuery {
1179                        query_start: end_ts + 1,
1180                        query_end: last_interval.1,
1181                        use_period_boundaries: false,
1182                    });
1183                }
1184            }
1185        }
1186
1187        // Generate period-based consolidation queries for each contiguous group
1188        for group in contiguous_groups {
1189            let group_start = group[0].0;
1190            let group_end = group[group.len() - 1].1;
1191
1192            // Apply start/end filtering to the group
1193            let effective_start = used_start.map_or(group_start, |s| s.max(group_start));
1194            let effective_end = used_end.map_or(group_end, |e| e.min(group_end));
1195
1196            if effective_start > effective_end {
1197                continue; // Skip if no overlap
1198            }
1199
1200            // Generate period-based queries within this contiguous group
1201            let mut current_start_ns = (effective_start / period_nanos) * period_nanos;
1202
1203            loop {
1204                let current_end_ns = current_start_ns
1205                    .saturating_add(period_nanos - 1)
1206                    .min(effective_end);
1207
1208                // Check if target file already exists (only when ensure_contiguous_files is true)
1209                let target_exists = if ensure_contiguous_files {
1210                    let directory = self.make_path(type_name, identifier)?;
1211                    let target_filename = format!(
1212                        "{}/{}",
1213                        directory,
1214                        timestamps_to_filename(
1215                            UnixNanos::from(current_start_ns),
1216                            UnixNanos::from(current_end_ns)
1217                        )
1218                    );
1219                    self.file_exists(&target_filename)?
1220                } else {
1221                    false
1222                };
1223
1224                if !target_exists {
1225                    queries_to_execute.push(ConsolidationQuery {
1226                        query_start: current_start_ns,
1227                        query_end: current_end_ns,
1228                        use_period_boundaries: ensure_contiguous_files,
1229                    });
1230                }
1231
1232                match current_start_ns.checked_add(period_nanos) {
1233                    Some(next_start_ns) if next_start_ns <= effective_end => {
1234                        current_start_ns = next_start_ns;
1235                    }
1236                    _ => break,
1237                }
1238            }
1239        }
1240
1241        // Sort queries by start date to enable efficient file removal
1242        // Files can be removed when interval[1] <= query_info["query_end"]
1243        // and processing in chronological order ensures optimal cleanup
1244        queries_to_execute.sort_by_key(|q| q.query_start);
1245
1246        Ok(queries_to_execute)
1247    }
1248
1249    /// Groups intervals for period-based consolidation.
1250    ///
1251    /// Groups adjacent intervals into the same bucket unless the gap between them exceeds
1252    /// `period_nanos`. Sub-period gaps land in the same consolidated file anyway, so they
1253    /// do not warrant a split. Gaps larger than one period represent genuine data holes.
1254    ///
1255    /// # Parameters
1256    ///
1257    /// - `intervals`: A slice of timestamp intervals as (start, end) tuples, sorted by start.
1258    /// - `period_nanos`: The target consolidation period; gaps larger than this split groups.
1259    ///
1260    /// # Returns
1261    ///
1262    /// Returns a vector of groups. Returns an empty vector if the input is empty.
1263    ///
1264    /// # Examples
1265    ///
1266    /// ```text
1267    /// Legacy chunked files with period=86_400_000_000_000 (1 day):
1268    ///   [(1,5), (6,10), (11,15)] -> [[(1,5), (6,10), (11,15)]]
1269    ///
1270    /// Small period=1 with mixed gaps:
1271    ///   [(1,5), (8,10), (12,15)] -> [[(1,5)], [(8,10)], [(12,15)]]
1272    /// ```
1273    #[must_use]
1274    pub fn group_contiguous_intervals(
1275        &self,
1276        intervals: &[(u64, u64)],
1277        period_nanos: u64,
1278    ) -> Vec<Vec<(u64, u64)>> {
1279        if intervals.is_empty() {
1280            return Vec::new();
1281        }
1282
1283        // Split groups only when the gap between files exceeds one period,
1284        // since sub-period gaps land in the same consolidated file anyway.
1285        // This works for both legacy chunked files (gap ~1ns) and fragment-per-flush
1286        // catalogs (gap ~bar interval) without inferring spacing from the data.
1287        let mut contiguous_groups = Vec::new();
1288        let mut current_group = vec![intervals[0]];
1289
1290        for i in 1..intervals.len() {
1291            let prev_end = intervals[i - 1].1;
1292            let curr_start = intervals[i].0;
1293
1294            if curr_start.saturating_sub(prev_end) > period_nanos {
1295                contiguous_groups.push(current_group);
1296                current_group = vec![intervals[i]];
1297            } else {
1298                current_group.push(intervals[i]);
1299            }
1300        }
1301
1302        contiguous_groups.push(current_group);
1303
1304        contiguous_groups
1305    }
1306}
1307
1308fn can_rewrite_consolidation_by_period(type_name: &str) -> bool {
1309    matches!(
1310        type_name,
1311        "quotes"
1312            | "trades"
1313            | "order_book_deltas"
1314            | "order_book_depths"
1315            | "bars"
1316            | "index_prices"
1317            | "mark_prices"
1318    )
1319}
1320
1321fn is_schema_incompatibility(error: &anyhow::Error) -> bool {
1322    let message = error.to_string().to_lowercase();
1323    [
1324        "schema",
1325        "field",
1326        "column",
1327        "data type",
1328        "datatype",
1329        "not compatible",
1330        "mismatch",
1331    ]
1332    .iter()
1333    .any(|needle| message.contains(needle))
1334}
1335
1336#[cfg(test)]
1337mod tests {
1338    use rstest::rstest;
1339
1340    use super::*;
1341
1342    #[rstest]
1343    #[case("quotes", true)]
1344    #[case("bars", true)]
1345    #[case("instrument_closes", false)]
1346    #[case("custom/signal", false)]
1347    fn can_rewrite_consolidation_by_period_only_for_supported_types(
1348        #[case] type_name: &str,
1349        #[case] expected: bool,
1350    ) {
1351        assert_eq!(can_rewrite_consolidation_by_period(type_name), expected);
1352    }
1353
1354    #[rstest]
1355    #[case("schema mismatch while writing record batch", true)]
1356    #[case("object store request timed out", false)]
1357    fn schema_incompatibility_detection_matches_schema_errors(
1358        #[case] message: &str,
1359        #[case] expected: bool,
1360    ) {
1361        assert_eq!(
1362            is_schema_incompatibility(&anyhow::anyhow!(message.to_string())),
1363            expected
1364        );
1365    }
1366}