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