nautilus_persistence/backend/parquet/
migration.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//! Legacy Parquet catalog rewrite into the current Parquet schema and layout.
17
18use arrow::record_batch::RecordBatch;
19use futures::StreamExt;
20use nautilus_core::UnixNanos;
21use nautilus_model::instruments::NautilusInstrumentType;
22use nautilus_serialization::arrow::record_batch_without_identifier_column;
23use object_store::{PutMode, PutOptions, path::Path as ObjectPath};
24
25use super::{
26    catalog::ParquetDataCatalog,
27    io::write_batches_to_object_store_create,
28    paths::{parse_filename_timestamps, timestamps_to_filename},
29};
30use crate::{
31    backend::migration::{
32        CatalogMigrationPlan, CatalogMigrationReport, IdentifierSource, ParquetCatalogSource,
33        PlannedMigrationFile, build_catalog_migration_plan, ensure_distinct_migration_locations,
34        ensure_planned_file_unchanged, prepare_migration_parts, read_planned_migration_file,
35    },
36    catalog::types::instrument_path_prefix,
37    common::{metadata::record_batch_ts_init_range, storage::normalize_storage_location},
38};
39
40/// Settings for converting a legacy Parquet catalog into a separate native catalog.
41#[derive(Debug)]
42pub struct ParquetMigrationConfig {
43    pub source_uri: String,
44    pub target_uri: String,
45    pub source_options: Vec<(String, String)>,
46    pub target_options: Vec<(String, String)>,
47    pub dry_run: bool,
48}
49
50/// Converts the current or legacy Arrow representation of a Parquet catalog.
51///
52/// The source remains unchanged. The destination must be empty, and preflight rejects
53/// unsupported schemas before any destination files are written.
54///
55/// # Errors
56///
57/// Returns an error if locations overlap, schemas are invalid, the destination is nonempty, or an
58/// object-store read or write fails.
59pub fn migrate_parquet_catalog(
60    config: ParquetMigrationConfig,
61) -> anyhow::Result<CatalogMigrationReport> {
62    let source_uri = normalize_storage_location(&config.source_uri)?;
63    let target_uri = normalize_storage_location(&config.target_uri)?;
64    ensure_distinct_migration_locations(&source_uri, &target_uri)?;
65    let source = ParquetDataCatalog::from_uri(
66        &source_uri,
67        Some(config.source_options.into_iter().collect()),
68        None,
69        None,
70        None,
71    )?;
72    let plan = build_catalog_migration_plan(&source)?;
73    plan.ensure_ready()?;
74    if config.dry_run {
75        return Ok(CatalogMigrationReport::from_plan(&plan, true));
76    }
77
78    let url = url::Url::parse(&target_uri)?;
79    if url.scheme() == "file" {
80        let path = url
81            .to_file_path()
82            .map_err(|()| anyhow::anyhow!("Invalid destination file URI"))?;
83        std::fs::create_dir_all(path)?;
84    }
85
86    let target = ParquetDataCatalog::from_uri(
87        &target_uri,
88        Some(config.target_options.into_iter().collect()),
89        None,
90        None,
91        None,
92    )?;
93    target.migrate_from_legacy_parquet_catalog_plan(&source, &plan)
94}
95
96impl ParquetCatalogSource for ParquetDataCatalog {
97    fn object_store(&self) -> std::sync::Arc<dyn object_store::ObjectStore> {
98        self.object_store.clone()
99    }
100    fn base_path(&self) -> &str {
101        &self.base_path
102    }
103    fn original_uri(&self) -> &str {
104        &self.original_uri
105    }
106
107    fn to_object_path_parsed(&self, path: &str) -> anyhow::Result<ObjectPath> {
108        Self::to_object_path_parsed(self, path)
109    }
110}
111
112impl ParquetDataCatalog {
113    /// Rewrites a legacy Parquet catalog into this current Parquet catalog.
114    ///
115    /// # Errors
116    ///
117    /// Returns an error if source preflight fails, this catalog contains any leaf object, or a
118    /// source file cannot be read, converted, or written.
119    pub fn migrate_from_legacy_parquet_catalog(
120        &self,
121        source: &Self,
122    ) -> anyhow::Result<CatalogMigrationReport> {
123        ensure_distinct_migration_locations(&source.original_uri, &self.original_uri)?;
124        let plan = build_catalog_migration_plan(source)?;
125        plan.ensure_ready()?;
126        self.migrate_from_legacy_parquet_catalog_plan(source, &plan)
127    }
128
129    fn migrate_from_legacy_parquet_catalog_plan(
130        &self,
131        source: &Self,
132        plan: &CatalogMigrationPlan,
133    ) -> anyhow::Result<CatalogMigrationReport> {
134        self.ensure_migration_target_empty()?;
135        let mut report = CatalogMigrationReport::from_plan(plan, false);
136
137        for file in &plan.files {
138            if file.size == 0 {
139                let source_path = source.to_object_path_parsed(&file.path)?;
140                ensure_planned_file_unchanged(source, file, &source_path)?;
141
142                let target_prefix = match file.target_type_name.as_str() {
143                    "instruments" => file.source_type_name.clone(),
144                    "custom" => file
145                        .source_type_name
146                        .strip_prefix("custom_")
147                        .map_or_else(|| "custom".to_string(), |name| format!("custom/{name}")),
148                    _ => file.target_type_name.clone(),
149                };
150
151                let relative_path = file.relative_path.replacen(
152                    &format!("data/{}/", file.source_type_name),
153                    &format!("data/{target_prefix}/"),
154                    1,
155                );
156                let target_path = self.to_object_path_parsed(&relative_path)?;
157                self.execute_async(|| async {
158                    self.object_store
159                        .put_opts(
160                            &target_path,
161                            Vec::new().into(),
162                            PutOptions {
163                                mode: PutMode::Create,
164                                ..Default::default()
165                            },
166                        )
167                        .await?;
168
169                    Ok(())
170                })?;
171
172                report.record_migrated_file(file, 0, 0);
173                continue;
174            }
175
176            let batches = read_planned_migration_file(source, file)?;
177            let (migrated_rows, path_identifier_rows) =
178                self.write_migration_batches(file, batches)?;
179            if migrated_rows == 0 {
180                report.record_skipped_file(file);
181            } else {
182                report.record_migrated_file(file, migrated_rows, path_identifier_rows);
183            }
184        }
185
186        Ok(report)
187    }
188
189    fn write_migration_batches(
190        &self,
191        file: &PlannedMigrationFile,
192        batches: Vec<RecordBatch>,
193    ) -> anyhow::Result<(usize, usize)> {
194        let mut migrated_rows = 0;
195        let mut path_identifier_rows = 0;
196
197        let parts = prepare_migration_parts(file, batches)?;
198        let coverage = parse_filename_timestamps(&file.relative_path);
199
200        for part in parts {
201            if part.row_count == 0 {
202                continue;
203            }
204
205            let directory = if let Some(custom_type_name) =
206                file.target_type_name.strip_prefix("custom/")
207            {
208                self.make_path_custom_data(custom_type_name, part.identifier.as_deref())?
209            } else {
210                let prefix = if let Some(class) = file.target_table.strip_prefix("instruments/") {
211                    instrument_path_prefix(&class.parse::<NautilusInstrumentType>()?)
212                } else {
213                    &file.target_type_name
214                };
215
216                self.make_path(prefix, part.identifier.as_deref())?
217            };
218
219            let (row_start, row_end) = record_batch_ts_init_range(&part.batches)?;
220            let (start_ts, end_ts) = coverage.unwrap_or((row_start, row_end));
221            anyhow::ensure!(
222                start_ts <= row_start && row_end <= end_ts,
223                "Parquet migration source coverage does not contain its rows: {}",
224                file.relative_path
225            );
226            let filename =
227                timestamps_to_filename(UnixNanos::from(start_ts), UnixNanos::from(end_ts));
228            let object_path = self.to_object_path(&format!("{directory}/{filename}"))?;
229            let batches = part
230                .batches
231                .into_iter()
232                .map(record_batch_without_identifier_column)
233                .collect::<Result<Vec<_>, _>>()?;
234
235            self.execute_async(|| async {
236                write_batches_to_object_store_create(
237                    &batches,
238                    self.object_store.clone(),
239                    &object_path,
240                    Some(self.compression),
241                    Some(self.max_row_group_size),
242                    None,
243                )
244                .await
245            })
246            .map_err(|e| {
247                anyhow::anyhow!(
248                    "Parquet migration target object already exists or cannot be created: \
249                     {object_path}: {e}"
250                )
251            })?;
252
253            migrated_rows += part.row_count;
254            if part.identifier_source == IdentifierSource::Path {
255                path_identifier_rows += part.row_count;
256            }
257        }
258
259        Ok((migrated_rows, path_identifier_rows))
260    }
261
262    fn ensure_migration_target_empty(&self) -> anyhow::Result<()> {
263        let prefix = (!self.base_path.is_empty()).then(|| ObjectPath::from(self.base_path.clone()));
264        self.execute_async(|| async {
265            let mut objects = self.object_store.list(prefix.as_ref());
266            anyhow::ensure!(
267                objects.next().await.transpose()?.is_none(),
268                "Parquet migration target must be new or empty",
269            );
270            Ok(())
271        })
272    }
273}
nautilus_persistence/backend/parquet/migration.rs

nautilus_persistence/backend/parquet/
migration.rs
