nautilus_persistence/backend/parquet/
mod.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//! Parquet catalog backend.
17
18use std::sync::Arc;
19
20use parquet::basic::{BrotliLevel, Compression, GzipLevel, ZstdLevel};
21
22use crate::{
23    catalog::{factory as catalog_factory, traits as catalog_traits},
24    config::CatalogCompression,
25};
26
27pub mod catalog;
28pub mod consolidation;
29pub mod delete;
30pub mod feather_session;
31pub mod file_admin;
32pub mod intervals;
33pub mod io;
34pub mod migration;
35pub mod paths;
36pub mod writer;
37
38pub(crate) mod metadata;
39
40/// Default number of rows in a Parquet row group.
41pub const DEFAULT_ROW_GROUP_SIZE: usize = 131_072;
42
43pub(crate) fn register_catalog_factory(registry: &mut catalog_factory::CatalogFactoryRegistry) {
44    registry.insert(
45        catalog_factory::PARQUET_CATALOG_FACTORY_NAME.to_string(),
46        Arc::new(|config: &catalog_factory::CatalogConnectConfig| {
47            Ok(Box::new(open_catalog(config)?) as catalog_traits::DataCatalog)
48        }),
49    );
50}
51
52// Streaming promotion opens its catalog here too, so both honor the catalog settings
53pub(crate) fn open_catalog(
54    config: &catalog_factory::CatalogConnectConfig,
55) -> anyhow::Result<catalog::ParquetDataCatalog> {
56    // Parquet reads only the typed settings, so every `params` key is unknown
57    catalog_factory::validate_catalog_params(
58        catalog_factory::PARQUET_CATALOG_FACTORY_NAME,
59        config.params.as_ref(),
60        &[],
61    )?;
62
63    catalog::ParquetDataCatalog::from_uri(
64        &config.uri,
65        config.storage_options.clone(),
66        config.batch_size,
67        config.compression.map(Compression::from),
68        config.max_row_group_size,
69    )
70}
71
72impl From<CatalogCompression> for Compression {
73    fn from(compression: CatalogCompression) -> Self {
74        match compression {
75            CatalogCompression::Uncompressed => Self::UNCOMPRESSED,
76            CatalogCompression::Snappy => Self::SNAPPY,
77            CatalogCompression::Gzip => Self::GZIP(GzipLevel::default()),
78            CatalogCompression::Brotli => Self::BROTLI(BrotliLevel::default()),
79            CatalogCompression::Lz4 => Self::LZ4_RAW,
80            CatalogCompression::Zstd => Self::ZSTD(ZstdLevel::default()),
81        }
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use std::fs::File;
88
89    use arrow::record_batch::RecordBatch;
90    use nautilus_core::{Params, UnixNanos};
91    use nautilus_model::data::{NautilusDataType, QuoteTick, stubs::quote_audusd};
92    use parquet::file::{
93        metadata::{ParquetMetaData, RowGroupMetaData},
94        reader::{FileReader, SerializedFileReader},
95    };
96    use rstest::rstest;
97    use serde_json::json;
98    use tempfile::TempDir;
99
100    use super::*;
101    use crate::config::DataCatalogConfig;
102
103    #[rstest]
104    fn open_catalog_writes_with_typed_settings() {
105        let directory = TempDir::new().unwrap();
106        let config = DataCatalogConfig::builder()
107            .path(directory.path().to_string_lossy().to_string())
108            .batch_size(2)
109            .compression(CatalogCompression::Uncompressed)
110            .max_row_group_size(3)
111            .build()
112            .unwrap();
113        let catalog = open_catalog(&config.connect_config()).unwrap();
114        let quotes = (1..=5).map(quote).collect::<Vec<_>>();
115
116        catalog.write_to_parquet(&quotes, None, None, None).unwrap();
117
118        let batches = catalog.data_to_record_batches(&quotes).unwrap();
119        let metadata = quote_file_metadata(&catalog, &directory);
120
121        assert_eq!(catalog.batch_size, 2);
122        assert_eq!(
123            batches
124                .iter()
125                .map(RecordBatch::num_rows)
126                .collect::<Vec<_>>(),
127            vec![2, 2, 1]
128        );
129        assert_eq!(
130            metadata
131                .row_groups()
132                .iter()
133                .map(RowGroupMetaData::num_rows)
134                .collect::<Vec<_>>(),
135            vec![3, 2]
136        );
137        assert!(
138            metadata
139                .row_groups()
140                .iter()
141                .flat_map(RowGroupMetaData::columns)
142                .all(|column| column.compression() == Compression::UNCOMPRESSED)
143        );
144    }
145
146    #[rstest]
147    #[case::uncompressed(CatalogCompression::Uncompressed, Compression::UNCOMPRESSED)]
148    #[case::snappy(CatalogCompression::Snappy, Compression::SNAPPY)]
149    #[case::gzip(CatalogCompression::Gzip, Compression::GZIP(GzipLevel::default()))]
150    #[case::brotli(
151        CatalogCompression::Brotli,
152        Compression::BROTLI(BrotliLevel::default())
153    )]
154    #[case::lz4(CatalogCompression::Lz4, Compression::LZ4_RAW)]
155    #[case::zstd(CatalogCompression::Zstd, Compression::ZSTD(ZstdLevel::default()))]
156    fn open_catalog_writes_each_compression(
157        #[case] compression: CatalogCompression,
158        #[case] expected: Compression,
159    ) {
160        let directory = TempDir::new().unwrap();
161        let mut config =
162            catalog_factory::CatalogConnectConfig::new(directory.path().to_string_lossy(), None);
163        config.compression = Some(compression);
164        let catalog = open_catalog(&config).unwrap();
165
166        catalog
167            .write_to_parquet(&[quote(1)], None, None, None)
168            .unwrap();
169
170        let metadata = quote_file_metadata(&catalog, &directory);
171
172        assert_eq!(catalog.compression, expected);
173        assert!(
174            metadata
175                .row_group(0)
176                .columns()
177                .iter()
178                .all(|column| column.compression() == expected)
179        );
180    }
181
182    #[rstest]
183    #[case::moved_setting("batch_size")]
184    #[case::unknown("no_such_param")]
185    fn open_catalog_rejects_params_naming_key(#[case] key: &str) {
186        let directory = TempDir::new().unwrap();
187        let mut params = Params::new();
188        params.insert(key.to_string(), json!(1024));
189        let config = DataCatalogConfig::builder()
190            .path(directory.path().to_string_lossy().to_string())
191            .params(params)
192            .build()
193            .unwrap();
194
195        let error = open_catalog(&config.connect_config()).unwrap_err();
196
197        assert_eq!(
198            error.to_string(),
199            format!("Unknown Parquet catalog param '{key}': this catalog takes no params")
200        );
201    }
202
203    // A deserialized config skips `DataCatalogConfig::validate`, so the factory checks counts too
204    #[rstest]
205    #[case::batch_size(Some(0), None, "batch_size")]
206    #[case::max_row_group_size(None, Some(0), "max_row_group_size")]
207    fn open_catalog_rejects_zero_count(
208        #[case] batch_size: Option<usize>,
209        #[case] max_row_group_size: Option<usize>,
210        #[case] field: &str,
211    ) {
212        let directory = TempDir::new().unwrap();
213        let mut config =
214            catalog_factory::CatalogConnectConfig::new(directory.path().to_string_lossy(), None);
215        config.batch_size = batch_size;
216        config.max_row_group_size = max_row_group_size;
217
218        let error = open_catalog(&config).unwrap_err();
219
220        assert_eq!(
221            error.to_string(),
222            format!(
223                "invalid {field}: must be a positive number of rows; omit the field for the \
224                 backend default"
225            )
226        );
227    }
228
229    #[rstest]
230    #[case::batch_size(Some(0), None, "batch_size")]
231    #[case::max_row_group_size(None, Some(0), "max_row_group_size")]
232    fn from_uri_rejects_zero_count(
233        #[case] batch_size: Option<usize>,
234        #[case] max_row_group_size: Option<usize>,
235        #[case] field: &str,
236    ) {
237        let directory = TempDir::new().unwrap();
238
239        let error = catalog::ParquetDataCatalog::from_uri(
240            &directory.path().to_string_lossy(),
241            None,
242            batch_size,
243            None,
244            max_row_group_size,
245        )
246        .unwrap_err();
247
248        assert_eq!(
249            error.to_string(),
250            format!(
251                "invalid {field}: must be a positive number of rows; omit the field for the \
252                 backend default"
253            )
254        );
255    }
256
257    fn quote_file_metadata(
258        catalog: &catalog::ParquetDataCatalog,
259        directory: &TempDir,
260    ) -> ParquetMetaData {
261        let files = catalog
262            .query_files(&NautilusDataType::QuoteTick.into(), None, None, None)
263            .unwrap();
264        assert_eq!(files.len(), 1);
265
266        SerializedFileReader::new(File::open(directory.path().join(&files[0])).unwrap())
267            .unwrap()
268            .metadata()
269            .clone()
270    }
271
272    fn quote(ts: u64) -> QuoteTick {
273        QuoteTick {
274            ts_event: UnixNanos::from(ts),
275            ts_init: UnixNanos::from(ts),
276            ..quote_audusd()
277        }
278    }
279}
nautilus_persistence/backend/parquet/mod.rs

nautilus_persistence/backend/parquet/
mod.rs
