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

nautilus_persistence/backend/parquet/
migration.rs
