Skip to main content

nautilus_serialization/arrow/
legacy.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 Arrow schema detection and conversion to current Rust schemas.
17
18use std::{
19    collections::{BTreeMap, HashMap},
20    fmt::{Display, Write},
21    str::FromStr,
22    sync::Arc,
23};
24
25use arrow::{
26    array::{
27        Array, ArrayRef, BinaryArray, Decimal128Array, FixedSizeBinaryArray, FixedSizeListArray,
28        StringArray, StringBuilder, TimestampNanosecondArray, UInt8Array, UInt16Array, UInt64Array,
29        UInt64Builder, new_null_array,
30    },
31    compute::cast,
32    datatypes::{DataType, Field, Schema, TimeUnit},
33    error::ArrowError,
34    record_batch::RecordBatch,
35};
36use nautilus_model::{
37    data::{
38        Bar, DEPTH10_LEN, FundingRateUpdate, IndexPriceUpdate, InstrumentClose, InstrumentStatus,
39        MarkPriceUpdate, OptionGreeks, OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick,
40    },
41    enums::{AggressorSide, BookAction, FromU8, InstrumentCloseType, OrderSide},
42    types::Price,
43};
44
45use super::{
46    ArrowSchemaProvider, FIXED_DECIMAL_PRECISION, FIXED_DECIMAL_SCALE, KEY_IDENTIFIER,
47    KEY_INSTRUMENT_ID, KEY_PRICE_PRECISION, STANDARD_TO_DECIMAL_SCALE, enum_dictionary_array,
48    enum_dictionary_data_type, fixed_decimal_data_type, price_decimal_array, timestamp_column,
49    timestamp_data_type,
50};
51
52/// Stable identity for Arrow field names, types, and nullability.
53#[derive(Clone, Debug, Eq, Hash, PartialEq)]
54pub struct SchemaFingerprint(String);
55
56impl Display for SchemaFingerprint {
57    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58        f.write_str(&self.0)
59    }
60}
61
62/// Registry decision applied to a source batch.
63#[derive(Clone, Copy, Debug, Eq, PartialEq)]
64pub enum LegacyTranscodeKind {
65    PassThrough,
66    InstrumentStatusV1,
67    FundingRateUpdateV1,
68    InstrumentCloseV1,
69}
70
71/// Record batches resolved to a current Rust Arrow schema.
72#[derive(Debug)]
73pub struct LegacyTranscodeResult {
74    pub batches: Vec<RecordBatch>,
75    pub kind: LegacyTranscodeKind,
76}
77
78/// File-scoped state for legacy Arrow transcoding.
79#[derive(Debug, Default)]
80pub struct LegacyTranscodeState {
81    instrument_close_precisions: BTreeMap<String, u8>,
82}
83
84/// Preflight schema resolution for a source file.
85#[derive(Clone, Debug, Eq, PartialEq)]
86pub struct LegacySchemaResolution {
87    pub kind: LegacyTranscodeKind,
88    pub source_fingerprint: SchemaFingerprint,
89    pub target_fingerprint: SchemaFingerprint,
90}
91
92/// Failure to resolve or convert a legacy Arrow schema.
93#[derive(Debug, thiserror::Error)]
94pub enum LegacyArrowError {
95    #[error(
96        "No Arrow transcoder is registered for type {type_name}, file {file_path}, \
97         and schema fingerprint {fingerprint}"
98    )]
99    UnknownSchema {
100        type_name: String,
101        file_path: String,
102        fingerprint: SchemaFingerprint,
103    },
104    #[error("Failed to transcode type {type_name} from {file_path}: {message}")]
105    Transcode {
106        type_name: String,
107        file_path: String,
108        message: String,
109    },
110}
111
112/// Returns a schema fingerprint that excludes schema metadata.
113#[must_use]
114pub fn schema_fingerprint(schema: &Schema) -> SchemaFingerprint {
115    let mut value = String::new();
116
117    for field in schema.fields() {
118        write!(
119            value,
120            "{}:{:?}:{};",
121            field.name(),
122            field.data_type(),
123            field.is_nullable()
124        )
125        .expect("writing a schema fingerprint to a string cannot fail");
126    }
127    SchemaFingerprint(value)
128}
129
130/// Resolves a normalized source batch through the legacy schema registry.
131///
132/// # Errors
133///
134/// Returns an error if a registered Nautilus type matches neither its current schema nor a known v1
135/// schema. Returns an error if a known v1 batch contains values that cannot be converted without
136/// loss.
137pub fn transcode_legacy_record_batch(
138    type_name: &str,
139    file_path: &str,
140    batch: RecordBatch,
141) -> Result<LegacyTranscodeResult, LegacyArrowError> {
142    transcode_legacy_record_batch_with_state(
143        type_name,
144        file_path,
145        batch,
146        &mut LegacyTranscodeState::default(),
147    )
148}
149
150/// Resolves a normalized source batch through the legacy schema registry with file-scoped state.
151///
152/// # Errors
153///
154/// Returns an error if a registered Nautilus type matches neither its current schema nor a known v1
155/// schema. Returns an error if a known v1 batch contains values that cannot be converted without
156/// loss.
157pub fn transcode_legacy_record_batch_with_state(
158    type_name: &str,
159    file_path: &str,
160    batch: RecordBatch,
161    state: &mut LegacyTranscodeState,
162) -> Result<LegacyTranscodeResult, LegacyArrowError> {
163    let resolution = resolve_legacy_schema(type_name, file_path, batch.schema().as_ref())?;
164    if resolution.kind == LegacyTranscodeKind::PassThrough {
165        return Ok(pass_through(batch));
166    }
167    let batches = match resolution.kind {
168        LegacyTranscodeKind::InstrumentStatusV1 => {
169            vec![transcode_instrument_status(&batch, file_path)?]
170        }
171        LegacyTranscodeKind::FundingRateUpdateV1 => {
172            vec![transcode_funding_rate(&batch, file_path)?]
173        }
174        LegacyTranscodeKind::InstrumentCloseV1 => {
175            transcode_instrument_close(&batch, file_path, state)?
176        }
177        LegacyTranscodeKind::PassThrough => {
178            unreachable!("pass-through schemas are handled before legacy registry lookup")
179        }
180    };
181
182    Ok(LegacyTranscodeResult {
183        batches,
184        kind: resolution.kind,
185    })
186}
187
188/// Resolves a normalized source schema without decoding its record batches.
189///
190/// # Errors
191///
192/// Returns an error if the schema of a built-in type is unregistered.
193pub fn resolve_legacy_schema(
194    type_name: &str,
195    file_path: &str,
196    schema: &Schema,
197) -> Result<LegacySchemaResolution, LegacyArrowError> {
198    let source_fingerprint = schema_fingerprint(schema);
199    let Some(current_schema) = current_schema(type_name, schema.metadata().clone()) else {
200        return Ok(LegacySchemaResolution {
201            kind: LegacyTranscodeKind::PassThrough,
202            target_fingerprint: source_fingerprint.clone(),
203            source_fingerprint,
204        });
205    };
206    let target_fingerprint = schema_fingerprint(&current_schema);
207
208    // Exact v1 shapes take their transcoder before the looser by-name match can claim them
209    let kind = if let Some(kind) = registered_legacy_kind(type_name, &source_fingerprint) {
210        kind
211    } else if matches_current_schema(schema, &current_schema) {
212        LegacyTranscodeKind::PassThrough
213    } else {
214        return Err(LegacyArrowError::UnknownSchema {
215            type_name: type_name.to_string(),
216            file_path: file_path.to_string(),
217            fingerprint: source_fingerprint,
218        });
219    };
220
221    Ok(LegacySchemaResolution {
222        kind,
223        source_fingerprint,
224        target_fingerprint,
225    })
226}
227
228// Migration re-encodes pass-through batches, so a source needs every current column by name (a
229// dictionary column may be plain `Utf8`), and any other columns must be identity columns
230fn matches_current_schema(source: &Schema, current: &Schema) -> bool {
231    let current_columns_present = current
232        .fields()
233        .iter()
234        .filter(|field| field.name() != KEY_IDENTIFIER)
235        .all(|expected| {
236            source.field_with_name(expected.name()).is_ok_and(|field| {
237                field.data_type() == expected.data_type()
238                    || (matches!(expected.data_type(), DataType::Dictionary(_, value_type)
239                        if value_type.as_ref() == &DataType::Utf8)
240                        && field.data_type() == &DataType::Utf8)
241            })
242        });
243
244    current_columns_present
245        && source.fields().iter().all(|field| {
246            current.field_with_name(field.name()).is_ok()
247                || matches!(field.name().as_str(), KEY_IDENTIFIER | KEY_INSTRUMENT_ID)
248        })
249}
250
251fn pass_through(batch: RecordBatch) -> LegacyTranscodeResult {
252    LegacyTranscodeResult {
253        batches: vec![batch],
254        kind: LegacyTranscodeKind::PassThrough,
255    }
256}
257
258fn current_schema(type_name: &str, metadata: HashMap<String, String>) -> Option<Schema> {
259    let metadata = Some(metadata);
260    match type_name {
261        "quotes" => Some(QuoteTick::get_schema(metadata)),
262        "trades" => Some(TradeTick::get_schema(metadata)),
263        "bars" => Some(Bar::get_schema(metadata)),
264        "order_book_deltas" => Some(OrderBookDelta::get_schema(metadata)),
265        "order_book_depths" => Some(OrderBookDepth::get_schema(metadata)),
266        "mark_prices" => Some(MarkPriceUpdate::get_schema(metadata)),
267        "index_prices" => Some(IndexPriceUpdate::get_schema(metadata)),
268        "funding_rates" => Some(FundingRateUpdate::get_schema(metadata)),
269        "instrument_status" => Some(InstrumentStatus::get_schema(metadata)),
270        "option_greeks" => Some(OptionGreeks::get_schema(metadata)),
271        "instrument_closes" => Some(InstrumentClose::get_schema(metadata)),
272        _ => None,
273    }
274}
275
276fn registered_legacy_kind(
277    type_name: &str,
278    fingerprint: &SchemaFingerprint,
279) -> Option<LegacyTranscodeKind> {
280    let entries = [
281        (
282            "instrument_status",
283            legacy_instrument_status_schema(),
284            LegacyTranscodeKind::InstrumentStatusV1,
285        ),
286        (
287            "funding_rates",
288            legacy_funding_rate_schema(),
289            LegacyTranscodeKind::FundingRateUpdateV1,
290        ),
291        (
292            "instrument_closes",
293            legacy_instrument_close_schema(),
294            LegacyTranscodeKind::InstrumentCloseV1,
295        ),
296    ];
297
298    entries
299        .into_iter()
300        .find(|(entry_type, schema, _)| {
301            type_name == *entry_type
302                && (schema_fingerprint(schema) == *fingerprint
303                    || schema_fingerprint(&schema_with_normalized_legacy_types(schema))
304                        == *fingerprint)
305        })
306        .map(|(_, _, kind)| kind)
307}
308
309fn schema_with_normalized_legacy_types(schema: &Schema) -> Schema {
310    let fields = schema
311        .fields()
312        .iter()
313        .map(|field| {
314            Arc::new(
315                field
316                    .as_ref()
317                    .clone()
318                    .with_data_type(normalized_legacy_data_type(field.name(), field.data_type())),
319            )
320        })
321        .collect::<Vec<_>>();
322    Schema::new_with_metadata(fields, schema.metadata().clone())
323}
324
325fn legacy_instrument_status_schema() -> Schema {
326    Schema::new(vec![
327        Field::new("instrument_id", DataType::Utf8, true),
328        Field::new("action", DataType::Utf8, true),
329        Field::new("reason", DataType::Utf8, true),
330        Field::new("trading_event", DataType::Utf8, true),
331        Field::new("is_trading", DataType::Boolean, true),
332        Field::new("is_quoting", DataType::Boolean, true),
333        Field::new("is_short_sell_restricted", DataType::Boolean, true),
334        Field::new("ts_event", DataType::UInt64, true),
335        Field::new("ts_init", DataType::UInt64, true),
336    ])
337}
338
339fn legacy_funding_rate_schema() -> Schema {
340    Schema::new(vec![
341        Field::new("rate", DataType::Binary, false),
342        Field::new("interval", DataType::UInt16, true),
343        Field::new("next_funding_ns", DataType::UInt64, true),
344        Field::new("ts_event", DataType::UInt64, false),
345        Field::new("ts_init", DataType::UInt64, false),
346    ])
347}
348
349fn legacy_instrument_close_schema() -> Schema {
350    Schema::new(vec![
351        Field::new("instrument_id", DataType::Utf8, true),
352        Field::new("close_type", DataType::Utf8, true),
353        Field::new("close_price", DataType::Utf8, true),
354        Field::new("ts_event", DataType::UInt64, true),
355        Field::new("ts_init", DataType::UInt64, true),
356    ])
357}
358
359fn transcode_instrument_status(
360    batch: &RecordBatch,
361    file_path: &str,
362) -> Result<RecordBatch, LegacyArrowError> {
363    let schema = InstrumentStatus::get_schema(Some(batch.schema().metadata().clone()));
364    let instrument_id = required_column(batch, "instrument_id", file_path)?;
365    let mut columns = Vec::with_capacity(schema.fields().len());
366
367    for field in schema.fields() {
368        let column = source_status_column(batch, field, &instrument_id, file_path)?;
369        let convertible_timestamp = field.data_type() == &super::timestamp_data_type()
370            && column.data_type() == &DataType::UInt64;
371        if column.data_type() != field.data_type() && !convertible_timestamp {
372            return Err(transcode_error(
373                "instrument_status",
374                file_path,
375                format!(
376                    "field {} expected {:?}, found {:?}",
377                    field.name(),
378                    field.data_type(),
379                    column.data_type()
380                ),
381            ));
382        }
383
384        if !field.is_nullable() && column.null_count() > 0 {
385            return Err(transcode_error(
386                "instrument_status",
387                file_path,
388                format!("required field {} contains nulls", field.name()),
389            ));
390        }
391        columns.push(column);
392    }
393
394    super::record_batch_with_timestamps(Arc::new(schema), columns)
395        .map_err(|e| transcode_error("instrument_status", file_path, e.to_string()))
396}
397
398fn source_status_column(
399    batch: &RecordBatch,
400    field: &Field,
401    instrument_id: &ArrayRef,
402    file_path: &str,
403) -> Result<ArrayRef, LegacyArrowError> {
404    if field.name() == KEY_IDENTIFIER {
405        return Ok(Arc::clone(instrument_id));
406    }
407
408    let Some(column) = batch.column_by_name(field.name()) else {
409        if field.is_nullable() {
410            return Ok(new_null_array(field.data_type(), batch.num_rows()));
411        }
412
413        return Err(transcode_error(
414            "instrument_status",
415            file_path,
416            format!("required field {} is absent", field.name()),
417        ));
418    };
419
420    if field.data_type() == &enum_dictionary_data_type() && column.data_type() == &DataType::Utf8 {
421        return cast(column, field.data_type())
422            .map_err(|e| transcode_error("instrument_status", file_path, e.to_string()));
423    }
424
425    Ok(Arc::clone(column))
426}
427
428fn transcode_funding_rate(
429    batch: &RecordBatch,
430    file_path: &str,
431) -> Result<RecordBatch, LegacyArrowError> {
432    let instrument_id = batch
433        .schema()
434        .metadata()
435        .get(KEY_INSTRUMENT_ID)
436        .cloned()
437        .ok_or_else(|| {
438            transcode_error(
439                "funding_rates",
440                file_path,
441                "instrument_id schema metadata is absent",
442            )
443        })?;
444    let rate = required_typed_column::<BinaryArray>(batch, "rate", file_path, "funding_rates")?;
445    let interval =
446        required_typed_column::<UInt16Array>(batch, "interval", file_path, "funding_rates")?;
447    let mut rate_builder = StringBuilder::with_capacity(batch.num_rows(), rate.value_data().len());
448
449    for row in 0..batch.num_rows() {
450        if rate.is_null(row) {
451            return Err(transcode_error(
452                "funding_rates",
453                file_path,
454                format!("required field rate is null at row {row}"),
455            ));
456        }
457        let value = serde_json::from_slice::<String>(rate.value(row)).map_err(|e| {
458            transcode_error(
459                "funding_rates",
460                file_path,
461                format!("rate at row {row} is not a msgspec JSON string: {e}"),
462            )
463        })?;
464        rate_builder.append_value(value);
465    }
466
467    let schema = FundingRateUpdate::get_schema(Some(batch.schema().metadata().clone()));
468    let identifiers = StringArray::from(vec![instrument_id; batch.num_rows()]);
469    let columns: Vec<ArrayRef> = vec![
470        Arc::new(rate_builder.finish()),
471        cast(interval, &DataType::UInt64)
472            .map_err(|e| transcode_error("funding_rates", file_path, e.to_string()))?,
473        required_column(batch, "next_funding_ns", file_path)?,
474        required_column(batch, "ts_event", file_path)?,
475        required_column(batch, "ts_init", file_path)?,
476        Arc::new(identifiers),
477    ];
478
479    super::record_batch_with_timestamps(Arc::new(schema), columns)
480        .map_err(|e| transcode_error("funding_rates", file_path, e.to_string()))
481}
482
483fn transcode_instrument_close(
484    batch: &RecordBatch,
485    file_path: &str,
486    state: &mut LegacyTranscodeState,
487) -> Result<Vec<RecordBatch>, LegacyArrowError> {
488    let normalized = super::record_batch_with_u64_timestamps(batch)
489        .map_err(|e| transcode_error("instrument_closes", file_path, e.to_string()))?;
490    let batch = &normalized;
491    let instrument_ids = required_typed_column::<StringArray>(
492        batch,
493        "instrument_id",
494        file_path,
495        "instrument_closes",
496    )?;
497    let close_types =
498        required_typed_column::<StringArray>(batch, "close_type", file_path, "instrument_closes")?;
499    let close_prices =
500        required_typed_column::<StringArray>(batch, "close_price", file_path, "instrument_closes")?;
501    let ts_events =
502        required_typed_column::<UInt64Array>(batch, "ts_event", file_path, "instrument_closes")?;
503    let ts_inits =
504        required_typed_column::<UInt64Array>(batch, "ts_init", file_path, "instrument_closes")?;
505    let mut rows_by_instrument: BTreeMap<String, Vec<usize>> = BTreeMap::new();
506
507    for row in 0..batch.num_rows() {
508        if instrument_ids.is_null(row) {
509            return Err(transcode_error(
510                "instrument_closes",
511                file_path,
512                format!("instrument_id is null at row {row}"),
513            ));
514        }
515        rows_by_instrument
516            .entry(instrument_ids.value(row).to_string())
517            .or_default()
518            .push(row);
519    }
520
521    let mut batches = Vec::with_capacity(rows_by_instrument.len());
522    for (instrument_id, rows) in rows_by_instrument {
523        let (batch, precision) = transcode_instrument_close_rows(
524            file_path,
525            &instrument_id,
526            &rows,
527            close_types,
528            close_prices,
529            ts_events,
530            ts_inits,
531            state
532                .instrument_close_precisions
533                .get(&instrument_id)
534                .copied(),
535        )?;
536        state
537            .instrument_close_precisions
538            .insert(instrument_id, precision);
539        batches.push(batch);
540    }
541    Ok(batches)
542}
543
544#[expect(
545    clippy::too_many_arguments,
546    reason = "the arguments are the validated source columns for one legacy close batch"
547)]
548fn transcode_instrument_close_rows(
549    file_path: &str,
550    instrument_id: &str,
551    rows: &[usize],
552    close_types: &StringArray,
553    close_prices: &StringArray,
554    ts_events: &UInt64Array,
555    ts_inits: &UInt64Array,
556    expected_precision: Option<u8>,
557) -> Result<(RecordBatch, u8), LegacyArrowError> {
558    let mut prices = Vec::with_capacity(rows.len());
559    let mut types = Vec::with_capacity(rows.len());
560    let mut event_builder = UInt64Builder::with_capacity(rows.len());
561    let mut init_builder = UInt64Builder::with_capacity(rows.len());
562    let mut precision = expected_precision;
563
564    for &row in rows {
565        if close_types.is_null(row)
566            || close_prices.is_null(row)
567            || ts_events.is_null(row)
568            || ts_inits.is_null(row)
569        {
570            return Err(transcode_error(
571                "instrument_closes",
572                file_path,
573                format!("required close field is null at row {row}"),
574            ));
575        }
576
577        let value = close_prices.value(row);
578        let row_precision = decimal_precision(value).map_err(|message| {
579            transcode_error(
580                "instrument_closes",
581                file_path,
582                format!("invalid close_price {value:?} at row {row}: {message}"),
583            )
584        })?;
585
586        if let Some(existing) = precision
587            && existing != row_precision
588        {
589            return Err(transcode_error(
590                "instrument_closes",
591                file_path,
592                format!(
593                    "close_price precision conflict at row {row}: found {row_precision}, \
594                     expected {existing}"
595                ),
596            ));
597        }
598        precision = Some(row_precision);
599
600        let price = Price::from_str(value).map_err(|e| {
601            transcode_error(
602                "instrument_closes",
603                file_path,
604                format!("invalid close_price {value:?} at row {row}: {e}"),
605            )
606        })?;
607        let close_type = InstrumentCloseType::from_str(close_types.value(row)).map_err(|e| {
608            transcode_error(
609                "instrument_closes",
610                file_path,
611                format!(
612                    "unknown close_type {:?} at row {row}: {e}",
613                    close_types.value(row)
614                ),
615            )
616        })?;
617
618        prices.push(price.raw());
619        types.push(close_type);
620        event_builder.append_value(ts_events.value(row));
621        init_builder.append_value(ts_inits.value(row));
622    }
623
624    let precision = precision.unwrap_or(0);
625    let metadata = HashMap::from([
626        (KEY_INSTRUMENT_ID.to_string(), instrument_id.to_string()),
627        (KEY_PRICE_PRECISION.to_string(), precision.to_string()),
628    ]);
629    let schema = InstrumentClose::get_schema(Some(metadata));
630    let identifiers = StringArray::from(vec![instrument_id; rows.len()]);
631
632    let batch = super::record_batch_with_timestamps(
633        Arc::new(schema),
634        vec![
635            Arc::new(
636                price_decimal_array(prices, "close_price")
637                    .map_err(|e| transcode_error("instrument_closes", file_path, e.to_string()))?,
638            ),
639            Arc::new(
640                enum_dictionary_array(types)
641                    .map_err(|e| transcode_error("instrument_closes", file_path, e.to_string()))?,
642            ),
643            Arc::new(event_builder.finish()),
644            Arc::new(init_builder.finish()),
645            Arc::new(identifiers),
646        ],
647    )
648    .map_err(|e| transcode_error("instrument_closes", file_path, e.to_string()))?;
649    Ok((batch, precision))
650}
651
652fn decimal_precision(value: &str) -> Result<u8, String> {
653    let precision = value
654        .split_once('.')
655        .map_or(0, |(_, fractional)| fractional.len());
656    u8::try_from(precision).map_err(|e| e.to_string())
657}
658
659fn required_column(
660    batch: &RecordBatch,
661    name: &str,
662    file_path: &str,
663) -> Result<ArrayRef, LegacyArrowError> {
664    batch.column_by_name(name).cloned().ok_or_else(|| {
665        transcode_error(
666            "legacy_arrow",
667            file_path,
668            format!("required field {name} is absent"),
669        )
670    })
671}
672
673fn required_typed_column<'a, T: Array + 'static>(
674    batch: &'a RecordBatch,
675    name: &str,
676    file_path: &str,
677    type_name: &str,
678) -> Result<&'a T, LegacyArrowError> {
679    let column = batch.column_by_name(name).ok_or_else(|| {
680        transcode_error(
681            type_name,
682            file_path,
683            format!("required field {name} is absent"),
684        )
685    })?;
686    column.as_any().downcast_ref::<T>().ok_or_else(|| {
687        transcode_error(
688            type_name,
689            file_path,
690            format!("field {name} has unexpected type {:?}", column.data_type()),
691        )
692    })
693}
694
695fn transcode_error(
696    type_name: &str,
697    file_path: &str,
698    message: impl Into<String>,
699) -> LegacyArrowError {
700    LegacyArrowError::Transcode {
701        type_name: type_name.to_string(),
702        file_path: file_path.to_string(),
703        message: message.into(),
704    }
705}
706
707/// Returns the open Arrow type for a recognized legacy catalog field.
708#[must_use]
709pub fn normalized_legacy_data_type(name: &str, data_type: &DataType) -> DataType {
710    match data_type {
711        DataType::FixedSizeBinary(8 | 16) => fixed_decimal_data_type(),
712        DataType::UInt64 if is_timestamp_field(name) => timestamp_data_type(),
713        DataType::Timestamp(TimeUnit::Nanosecond, None) if is_timestamp_field(name) => {
714            timestamp_data_type()
715        }
716        DataType::UInt8 if is_legacy_enum_field(name) => enum_dictionary_data_type(),
717        _ => data_type.clone(),
718    }
719}
720
721/// Normalizes legacy fixed-point byte columns to the open decimal representation.
722///
723/// # Errors
724///
725/// Returns an error if a column has an invalid physical array or value.
726pub fn normalize_legacy_fixed_columns(batch: &RecordBatch) -> Result<RecordBatch, ArrowError> {
727    let mut changed = false;
728    let mut fields = Vec::with_capacity(batch.num_columns());
729    let mut columns = Vec::with_capacity(batch.num_columns());
730    let normalize_named_fields = is_nautilus_legacy_schema(batch.schema_ref());
731    let normalize_timestamps = is_nautilus_timestamp_schema(batch.schema_ref());
732
733    for (field, column) in batch.schema().fields().iter().zip(batch.columns()) {
734        let custom_timestamp = batch.schema().metadata().contains_key("type_name")
735            && matches!(field.name().as_str(), "ts_event" | "ts_init");
736        if field.data_type() == &DataType::UInt64
737            && ((normalize_named_fields && is_timestamp_field(field.name())) || custom_timestamp)
738        {
739            let timestamps = timestamp_column(field, column.as_ref())?;
740            fields.push(Arc::new(field.as_ref().clone().with_data_type(
741                normalized_legacy_data_type(field.name(), field.data_type()),
742            )));
743            columns.push(timestamps);
744            changed = true;
745            continue;
746        }
747
748        if normalize_timestamps
749            && normalized_timestamp_type(field.data_type()) != *field.data_type()
750        {
751            fields.push(Arc::new(
752                field.as_ref().clone().with_data_type(timestamp_data_type()),
753            ));
754            let timestamps = column
755                .as_any()
756                .downcast_ref::<TimestampNanosecondArray>()
757                .ok_or_else(|| {
758                    ArrowError::CastError(format!(
759                        "Column '{}' is not a nanosecond timestamp",
760                        field.name()
761                    ))
762                })?;
763            columns.push(Arc::new(
764                timestamps.clone().with_data_type(timestamp_data_type()),
765            ));
766            changed = true;
767            continue;
768        }
769
770        if normalize_named_fields
771            && field.data_type() == &DataType::UInt8
772            && is_legacy_enum_field(field.name())
773        {
774            fields.push(Arc::new(field.as_ref().clone().with_data_type(
775                normalized_legacy_data_type(field.name(), field.data_type()),
776            )));
777            columns.push(legacy_enum_dictionary_column(field, column.as_ref())?);
778            changed = true;
779            continue;
780        }
781
782        if normalize_named_fields
783            && let DataType::FixedSizeList(item, length) = field.data_type()
784            && let DataType::FixedSizeBinary(width @ (8 | 16)) = item.data_type()
785        {
786            let list = column
787                .as_any()
788                .downcast_ref::<FixedSizeListArray>()
789                .ok_or_else(|| {
790                    ArrowError::CastError(format!("Column '{}' is not FixedSizeList", field.name()))
791                })?;
792            let values = list
793                .values()
794                .as_any()
795                .downcast_ref::<FixedSizeBinaryArray>()
796                .ok_or_else(|| {
797                    ArrowError::CastError(format!(
798                        "Column '{}' values are not FixedSizeBinary",
799                        field.name()
800                    ))
801                })?;
802            let decimal = normalize_legacy_fixed_array(field.name(), values, *width)?;
803            let item = Arc::new(
804                item.as_ref()
805                    .clone()
806                    .with_data_type(fixed_decimal_data_type())
807                    .with_nullable(true),
808            );
809            let list = FixedSizeListArray::try_new(
810                Arc::clone(&item),
811                *length,
812                Arc::new(decimal),
813                list.nulls().cloned(),
814            )?;
815            fields.push(Arc::new(
816                field
817                    .as_ref()
818                    .clone()
819                    .with_data_type(DataType::FixedSizeList(item, *length)),
820            ));
821            columns.push(Arc::new(list) as ArrayRef);
822            changed = true;
823            continue;
824        }
825
826        let DataType::FixedSizeBinary(width @ (8 | 16)) = field.data_type() else {
827            fields.push(Arc::clone(field));
828            columns.push(Arc::clone(column));
829            continue;
830        };
831
832        if !normalize_named_fields {
833            fields.push(Arc::clone(field));
834            columns.push(Arc::clone(column));
835            continue;
836        }
837        let values = column
838            .as_any()
839            .downcast_ref::<FixedSizeBinaryArray>()
840            .ok_or_else(|| {
841                ArrowError::CastError(format!("Column '{}' is not FixedSizeBinary", field.name()))
842            })?;
843        let decimal = normalize_legacy_fixed_array(field.name(), values, *width)?;
844        fields.push(Arc::new(
845            field
846                .as_ref()
847                .clone()
848                .with_data_type(normalized_legacy_data_type(field.name(), field.data_type()))
849                .with_nullable(true),
850        ));
851        columns.push(Arc::new(decimal) as ArrayRef);
852        changed = true;
853    }
854
855    if !changed {
856        return Ok(batch.clone());
857    }
858
859    let schema = Arc::new(Schema::new_with_metadata(
860        fields,
861        batch.schema().metadata().clone(),
862    ));
863    RecordBatch::try_new(schema, columns)
864}
865
866pub(super) fn legacy_enum_dictionary_column(
867    field: &Field,
868    column: &dyn Array,
869) -> Result<ArrayRef, ArrowError> {
870    let values = column
871        .as_any()
872        .downcast_ref::<UInt8Array>()
873        .ok_or_else(|| ArrowError::CastError(format!("Column '{}' is not UInt8", field.name())))?;
874
875    let names = (0..values.len())
876        .map(|row| {
877            if values.is_null(row) {
878                return Err(ArrowError::CastError(format!(
879                    "Legacy enum column '{}' is null at row {row}",
880                    field.name()
881                )));
882            }
883
884            legacy_enum_name(field.name(), values.value(row))
885        })
886        .collect::<Result<Vec<_>, _>>()?;
887    Ok(Arc::new(enum_dictionary_array(names)?) as ArrayRef)
888}
889
890fn normalize_legacy_fixed_array(
891    name: &str,
892    values: &FixedSizeBinaryArray,
893    width: i32,
894) -> Result<Decimal128Array, ArrowError> {
895    let quantity = is_legacy_quantity_field(name);
896    let mut decimals = Vec::with_capacity(values.len());
897
898    for row in 0..values.len() {
899        if values.is_null(row) {
900            decimals.push(None);
901            continue;
902        }
903
904        let bytes = values.value(row);
905        let decimal = match (width, quantity) {
906            (8, false) => {
907                let raw = i64::from_le_bytes(bytes.try_into().map_err(|e| {
908                    ArrowError::CastError(format!(
909                        "Invalid legacy price column '{name}' at row {row}: {e}"
910                    ))
911                })?);
912
913                if raw == i64::MIN {
914                    return Err(ArrowError::CastError(format!(
915                        "Legacy price column '{name}' contains PRICE_ERROR raw value {raw} at row {row}",
916                    )));
917                }
918                (raw != i64::MAX).then_some(i128::from(raw) * STANDARD_TO_DECIMAL_SCALE)
919            }
920            (8, true) => {
921                let raw = u64::from_le_bytes(bytes.try_into().map_err(|e| {
922                    ArrowError::CastError(format!(
923                        "Invalid legacy quantity column '{name}' at row {row}: {e}"
924                    ))
925                })?);
926                (raw != u64::MAX).then_some(i128::from(raw) * STANDARD_TO_DECIMAL_SCALE)
927            }
928            (16, false) => {
929                let raw = i128::from_le_bytes(bytes.try_into().map_err(|e| {
930                    ArrowError::CastError(format!(
931                        "Invalid legacy price column '{name}' at row {row}: {e}"
932                    ))
933                })?);
934
935                if raw == i128::MIN {
936                    return Err(ArrowError::CastError(format!(
937                        "Legacy price column '{name}' contains PRICE_ERROR raw value {raw} at row {row}",
938                    )));
939                }
940                (raw != i128::MAX).then_some(raw)
941            }
942            (16, true) => {
943                let raw = u128::from_le_bytes(bytes.try_into().map_err(|e| {
944                    ArrowError::CastError(format!(
945                        "Invalid legacy quantity column '{name}' at row {row}: {e}"
946                    ))
947                })?);
948
949                if raw == u128::MAX {
950                    None
951                } else {
952                    Some(i128::try_from(raw).map_err(|_| {
953                        ArrowError::CastError(format!(
954                            "Legacy quantity column '{name}' exceeds Decimal128 at row {row}"
955                        ))
956                    })?)
957                }
958            }
959            _ => unreachable!("legacy fixed width checked above"),
960        };
961        decimals.push(decimal);
962    }
963
964    Decimal128Array::from(decimals)
965        .with_precision_and_scale(FIXED_DECIMAL_PRECISION, FIXED_DECIMAL_SCALE)
966}
967
968type LegacySchemaFields = &'static [(&'static str, LegacyFieldType, bool)];
969
970const LEGACY_SCHEMA_FAMILIES: &[(&[&str], LegacySchemaFields)] = &[
971    (
972        &["OrderBookDelta"],
973        &[
974            ("action", LegacyFieldType::UInt8, false),
975            ("side", LegacyFieldType::UInt8, false),
976            ("price", LegacyFieldType::Fixed, false),
977            ("size", LegacyFieldType::Fixed, false),
978            ("order_id", LegacyFieldType::UInt64, false),
979            ("flags", LegacyFieldType::UInt8, false),
980            ("sequence", LegacyFieldType::UInt64, false),
981            ("ts_event", LegacyFieldType::UInt64, false),
982            ("ts_init", LegacyFieldType::UInt64, false),
983        ],
984    ),
985    (
986        &["TradeTick"],
987        &[
988            ("price", LegacyFieldType::Fixed, false),
989            ("size", LegacyFieldType::Fixed, false),
990            ("aggressor_side", LegacyFieldType::UInt8, false),
991            ("trade_id", LegacyFieldType::Utf8, false),
992            ("ts_event", LegacyFieldType::UInt64, false),
993            ("ts_init", LegacyFieldType::UInt64, false),
994        ],
995    ),
996    (
997        &["InstrumentClose"],
998        &[
999            ("close_price", LegacyFieldType::Fixed, false),
1000            ("close_type", LegacyFieldType::UInt8, false),
1001            ("ts_event", LegacyFieldType::UInt64, false),
1002            ("ts_init", LegacyFieldType::UInt64, false),
1003        ],
1004    ),
1005    (
1006        &["InstrumentClose"],
1007        &[
1008            ("instrument_id", LegacyFieldType::Utf8, true),
1009            ("close_type", LegacyFieldType::Utf8, true),
1010            ("close_price", LegacyFieldType::Utf8, true),
1011            ("ts_event", LegacyFieldType::UInt64, true),
1012            ("ts_init", LegacyFieldType::UInt64, true),
1013        ],
1014    ),
1015    (
1016        &["QuoteTick"],
1017        &[
1018            ("bid_price", LegacyFieldType::Fixed, false),
1019            ("ask_price", LegacyFieldType::Fixed, false),
1020            ("bid_size", LegacyFieldType::Fixed, false),
1021            ("ask_size", LegacyFieldType::Fixed, false),
1022            ("ts_event", LegacyFieldType::UInt64, false),
1023            ("ts_init", LegacyFieldType::UInt64, false),
1024        ],
1025    ),
1026    (
1027        &["Bar"],
1028        &[
1029            ("open", LegacyFieldType::Fixed, false),
1030            ("high", LegacyFieldType::Fixed, false),
1031            ("low", LegacyFieldType::Fixed, false),
1032            ("close", LegacyFieldType::Fixed, false),
1033            ("volume", LegacyFieldType::Fixed, false),
1034            ("ts_event", LegacyFieldType::UInt64, false),
1035            ("ts_init", LegacyFieldType::UInt64, false),
1036        ],
1037    ),
1038    (
1039        &["MarkPriceUpdate", "IndexPriceUpdate"],
1040        &[
1041            ("value", LegacyFieldType::Fixed, false),
1042            ("ts_event", LegacyFieldType::UInt64, false),
1043            ("ts_init", LegacyFieldType::UInt64, false),
1044        ],
1045    ),
1046    (
1047        &["FundingRateUpdate"],
1048        &[
1049            ("rate", LegacyFieldType::Binary, false),
1050            ("interval", LegacyFieldType::UInt16, true),
1051            ("next_funding_ns", LegacyFieldType::UInt64, true),
1052            ("ts_event", LegacyFieldType::UInt64, false),
1053            ("ts_init", LegacyFieldType::UInt64, false),
1054        ],
1055    ),
1056    (
1057        &["InstrumentStatus"],
1058        &[
1059            ("instrument_id", LegacyFieldType::Utf8, true),
1060            ("action", LegacyFieldType::Utf8, true),
1061            ("reason", LegacyFieldType::Utf8, true),
1062            ("trading_event", LegacyFieldType::Utf8, true),
1063            ("is_trading", LegacyFieldType::Boolean, true),
1064            ("is_quoting", LegacyFieldType::Boolean, true),
1065            ("is_short_sell_restricted", LegacyFieldType::Boolean, true),
1066            ("ts_event", LegacyFieldType::UInt64, true),
1067            ("ts_init", LegacyFieldType::UInt64, true),
1068        ],
1069    ),
1070    (
1071        &["OptionGreeks"],
1072        &[
1073            ("instrument_id", LegacyFieldType::Utf8, false),
1074            ("delta", LegacyFieldType::Float64, false),
1075            ("gamma", LegacyFieldType::Float64, false),
1076            ("vega", LegacyFieldType::Float64, false),
1077            ("theta", LegacyFieldType::Float64, false),
1078            ("rho", LegacyFieldType::Float64, false),
1079            ("mark_iv", LegacyFieldType::Float64, true),
1080            ("bid_iv", LegacyFieldType::Float64, true),
1081            ("ask_iv", LegacyFieldType::Float64, true),
1082            ("underlying_price", LegacyFieldType::Float64, true),
1083            ("open_interest", LegacyFieldType::Float64, true),
1084            ("ts_event", LegacyFieldType::UInt64, false),
1085            ("ts_init", LegacyFieldType::UInt64, false),
1086            ("convention", LegacyFieldType::Utf8, false),
1087        ],
1088    ),
1089];
1090
1091/// Returns whether a schema is a recognized Nautilus legacy family.
1092#[must_use]
1093pub fn is_nautilus_legacy_schema(schema: &Schema) -> bool {
1094    if let Some(type_name) = schema
1095        .metadata()
1096        .get("type_name")
1097        .or_else(|| schema.metadata().get("type"))
1098    {
1099        return legacy_metadata_family_matches(schema, type_name);
1100    }
1101
1102    LEGACY_SCHEMA_FAMILIES
1103        .iter()
1104        .any(|(_, fields)| schema_fingerprint_matches(schema, fields))
1105        || legacy_flat_depth_fingerprint_matches(schema)
1106        || legacy_fixed_list_depth_fingerprint_matches(schema)
1107}
1108
1109/// Returns whether timestamp annotations belong to a recognized Nautilus schema.
1110#[must_use]
1111pub fn is_nautilus_timestamp_schema(schema: &Schema) -> bool {
1112    if schema.metadata().contains_key("type_name") || schema.metadata().contains_key("type") {
1113        return true;
1114    }
1115
1116    if is_nautilus_legacy_schema(schema) {
1117        return true;
1118    }
1119    let fields = schema
1120        .fields()
1121        .iter()
1122        .filter(|field| field.name() != KEY_IDENTIFIER)
1123        .collect::<Vec<_>>();
1124    LEGACY_SCHEMA_FAMILIES.iter().any(|(_, expected)| {
1125        fields.len() == expected.len()
1126            && fields
1127                .iter()
1128                .zip(*expected)
1129                .all(|(field, (name, expected, _))| {
1130                    field.name() == *name
1131                        && (legacy_field_type_matches(field.data_type(), *expected)
1132                            || (is_timestamp_field(name)
1133                                && matches!(
1134                                    field.data_type(),
1135                                    DataType::Timestamp(TimeUnit::Nanosecond, _)
1136                                ))
1137                            || (matches!(expected, LegacyFieldType::Fixed)
1138                                && field.data_type() == &fixed_decimal_data_type())
1139                            || (is_legacy_enum_field(name)
1140                                && field.data_type() == &enum_dictionary_data_type()))
1141                })
1142    })
1143}
1144
1145/// Adds the UTC annotation to a nanosecond instant without a timezone.
1146#[must_use]
1147pub fn normalized_timestamp_type(data_type: &DataType) -> DataType {
1148    match data_type {
1149        DataType::Timestamp(TimeUnit::Nanosecond, None) => timestamp_data_type(),
1150        _ => data_type.clone(),
1151    }
1152}
1153
1154fn legacy_metadata_family_matches(schema: &Schema, type_name: &str) -> bool {
1155    match type_name {
1156        "OrderBookDepth10" | "OrderBookDepth" => {
1157            return legacy_flat_depth_fingerprint_matches(schema)
1158                || legacy_fixed_list_depth_fingerprint_matches(schema);
1159        }
1160        _ => {}
1161    }
1162
1163    LEGACY_SCHEMA_FAMILIES
1164        .iter()
1165        .filter(|(names, _)| names.contains(&type_name))
1166        .any(|(_, fields)| schema_fingerprint_matches(schema, fields))
1167}
1168
1169#[derive(Clone, Copy)]
1170enum LegacyFieldType {
1171    Binary,
1172    Float64,
1173    Fixed,
1174    UInt8,
1175    UInt16,
1176    UInt32,
1177    UInt64,
1178    Utf8,
1179    Boolean,
1180}
1181
1182fn schema_fingerprint_matches(schema: &Schema, expected: &[(&str, LegacyFieldType, bool)]) -> bool {
1183    let fields = schema
1184        .fields()
1185        .iter()
1186        .filter(|field| field.name() != KEY_IDENTIFIER)
1187        .collect::<Vec<_>>();
1188
1189    let fields_match = fields.len() == expected.len()
1190        && fields
1191            .iter()
1192            .zip(expected)
1193            .all(|(field, (name, data_type, _))| {
1194                field.name() == *name && legacy_field_type_matches(field.data_type(), *data_type)
1195            });
1196    let nullability_matches = fields
1197        .iter()
1198        .zip(expected)
1199        .all(|(field, (_, _, nullable))| field.is_nullable() == *nullable)
1200        || fields.iter().all(|field| field.is_nullable());
1201
1202    fields_match
1203        && nullability_matches
1204        && fixed_width_is_consistent(fields.iter().map(|field| field.data_type()))
1205}
1206
1207fn legacy_field_type_matches(data_type: &DataType, expected: LegacyFieldType) -> bool {
1208    match expected {
1209        LegacyFieldType::Binary => data_type == &DataType::Binary,
1210        LegacyFieldType::Float64 => data_type == &DataType::Float64,
1211        LegacyFieldType::Fixed => matches!(data_type, DataType::FixedSizeBinary(8 | 16)),
1212        LegacyFieldType::UInt8 => data_type == &DataType::UInt8,
1213        LegacyFieldType::UInt16 => data_type == &DataType::UInt16,
1214        LegacyFieldType::UInt32 => data_type == &DataType::UInt32,
1215        LegacyFieldType::UInt64 => data_type == &DataType::UInt64,
1216        LegacyFieldType::Utf8 => match data_type {
1217            DataType::Utf8 | DataType::Utf8View => true,
1218            DataType::Dictionary(_, value) => value.as_ref() == &DataType::Utf8,
1219            _ => false,
1220        },
1221        LegacyFieldType::Boolean => data_type == &DataType::Boolean,
1222    }
1223}
1224
1225fn fixed_width_is_consistent<'a>(data_types: impl Iterator<Item = &'a DataType>) -> bool {
1226    let mut width = None;
1227    data_types
1228        .filter_map(|data_type| match data_type {
1229            DataType::FixedSizeBinary(width) => Some(*width),
1230            _ => None,
1231        })
1232        .all(|current| *width.get_or_insert(current) == current)
1233}
1234
1235fn legacy_flat_depth_fingerprint_matches(schema: &Schema) -> bool {
1236    let fields = schema
1237        .fields()
1238        .iter()
1239        .filter(|field| field.name() != KEY_IDENTIFIER)
1240        .collect::<Vec<_>>();
1241    let market_fields = fields
1242        .iter()
1243        .filter(|field| {
1244            matches!(
1245                field.name().as_str(),
1246                "flags" | "sequence" | "ts_event" | "ts_init"
1247            )
1248        })
1249        .count();
1250
1251    let order_id_fields = fields
1252        .iter()
1253        .filter(|field| {
1254            field.name().starts_with("bid_order_id_") || field.name().starts_with("ask_order_id_")
1255        })
1256        .count();
1257    let all_nullable = fields.iter().all(|field| field.is_nullable());
1258
1259    fields.len() == 6 * DEPTH10_LEN + order_id_fields + market_fields
1260        && market_fields <= 4
1261        && matches!(order_id_fields, 0 | 20)
1262        && fixed_width_is_consistent(fields.iter().map(|field| field.data_type()))
1263        && ["bid", "ask"].iter().all(|side| {
1264            (0..DEPTH10_LEN).all(|level| {
1265                [
1266                    ("price", LegacyFieldType::Fixed),
1267                    ("size", LegacyFieldType::Fixed),
1268                    ("count", LegacyFieldType::UInt32),
1269                ]
1270                .iter()
1271                .all(|(value, data_type)| {
1272                    schema
1273                        .field_with_name(&format!("{side}_{value}_{level}"))
1274                        .is_ok_and(|field| {
1275                            (all_nullable || !field.is_nullable())
1276                                && legacy_field_type_matches(field.data_type(), *data_type)
1277                        })
1278                })
1279            })
1280        })
1281        && (order_id_fields == 0
1282            || ["bid", "ask"].iter().all(|side| {
1283                (0..DEPTH10_LEN).all(|level| {
1284                    schema
1285                        .field_with_name(&format!("{side}_order_id_{level}"))
1286                        .is_ok_and(|field| {
1287                            (all_nullable || !field.is_nullable())
1288                                && legacy_field_type_matches(
1289                                    field.data_type(),
1290                                    LegacyFieldType::UInt64,
1291                                )
1292                        })
1293                })
1294            }))
1295        && [
1296            ("flags", LegacyFieldType::UInt8),
1297            ("sequence", LegacyFieldType::UInt64),
1298            ("ts_event", LegacyFieldType::UInt64),
1299            ("ts_init", LegacyFieldType::UInt64),
1300        ]
1301        .iter()
1302        .all(|(name, data_type)| {
1303            let Ok(field) = schema.field_with_name(name) else {
1304                return true;
1305            };
1306            (all_nullable || !field.is_nullable())
1307                && legacy_field_type_matches(field.data_type(), *data_type)
1308        })
1309}
1310
1311fn legacy_fixed_list_depth_fingerprint_matches(schema: &Schema) -> bool {
1312    let fixed_list_matches = |name: &str, expected: &DataType| {
1313        schema.field_with_name(name).is_ok_and(|field| {
1314            matches!(
1315                field.data_type(),
1316                DataType::FixedSizeList(item, length)
1317                    if *length == i32::try_from(DEPTH10_LEN).expect("depth length fits i32")
1318                        && item.data_type() == expected
1319            )
1320        })
1321    };
1322    let widths = ["bid_price", "ask_price", "bid_size", "ask_size"]
1323        .iter()
1324        .filter_map(|name| {
1325            let field = schema.field_with_name(name).ok()?;
1326            let DataType::FixedSizeList(item, _) = field.data_type() else {
1327                return None;
1328            };
1329            let DataType::FixedSizeBinary(width @ (8 | 16)) = item.data_type() else {
1330                return None;
1331            };
1332            Some(*width)
1333        })
1334        .collect::<Vec<_>>();
1335    let order_id_fields = ["bid_order_id", "ask_order_id"]
1336        .iter()
1337        .filter(|name| schema.field_with_name(name).is_ok())
1338        .count();
1339
1340    widths.len() == 4
1341        && widths.iter().all(|width| *width == widths[0])
1342        && ["bid_count", "ask_count"]
1343            .iter()
1344            .all(|name| fixed_list_matches(name, &DataType::UInt32))
1345        && matches!(order_id_fields, 0 | 2)
1346        && (order_id_fields == 0
1347            || ["bid_order_id", "ask_order_id"]
1348                .iter()
1349                .all(|name| fixed_list_matches(name, &DataType::UInt64)))
1350}
1351
1352/// Returns whether `name` identifies a legacy catalog timestamp field.
1353#[must_use]
1354pub fn is_timestamp_field(name: &str) -> bool {
1355    name.starts_with("ts_")
1356        || matches!(
1357            name,
1358            "activation_ns"
1359                | "expiration_ns"
1360                | "next_funding_ns"
1361                | "expire_time"
1362                | "event_open_date"
1363                | "market_start_time"
1364        )
1365}
1366
1367/// Returns whether `name` identifies a legacy catalog enum field.
1368#[must_use]
1369pub fn is_legacy_enum_field(name: &str) -> bool {
1370    matches!(name, "action" | "side" | "aggressor_side" | "close_type")
1371}
1372
1373fn legacy_enum_name(name: &str, value: u8) -> Result<String, ArrowError> {
1374    let name_value = match name {
1375        "action" => BookAction::from_u8(value).map(|value| value.to_string()),
1376        "side" => match value {
1377            0 => Some("NO_ORDER_SIDE".to_string()),
1378            1 => Some(OrderSide::Buy.to_string()),
1379            2 => Some(OrderSide::Sell.to_string()),
1380            _ => None,
1381        },
1382        "aggressor_side" => AggressorSide::from_u8(value).map(|value| value.to_string()),
1383        "close_type" => InstrumentCloseType::from_u8(value).map(|value| value.to_string()),
1384        _ => None,
1385    };
1386    name_value.ok_or_else(|| {
1387        ArrowError::CastError(format!(
1388            "Invalid legacy enum value {value} for column '{name}'"
1389        ))
1390    })
1391}
1392
1393// Exact field inventory emitted by the legacy fixed-point schemas.
1394const LEGACY_QUANTITY_FIELDS: &[&str] = &[
1395    "size",
1396    "quantity",
1397    "qty",
1398    "volume",
1399    "bid_size",
1400    "ask_size",
1401    "bid_size_0",
1402    "bid_size_1",
1403    "bid_size_2",
1404    "bid_size_3",
1405    "bid_size_4",
1406    "bid_size_5",
1407    "bid_size_6",
1408    "bid_size_7",
1409    "bid_size_8",
1410    "bid_size_9",
1411    "ask_size_0",
1412    "ask_size_1",
1413    "ask_size_2",
1414    "ask_size_3",
1415    "ask_size_4",
1416    "ask_size_5",
1417    "ask_size_6",
1418    "ask_size_7",
1419    "ask_size_8",
1420    "ask_size_9",
1421];
1422
1423fn is_legacy_quantity_field(name: &str) -> bool {
1424    LEGACY_QUANTITY_FIELDS.contains(&name)
1425}
1426
1427#[cfg(test)]
1428mod tests {
1429    use std::collections::HashMap;
1430
1431    use arrow::{
1432        array::{BinaryArray, BooleanArray},
1433        datatypes::Schema,
1434    };
1435    use nautilus_model::{
1436        data::{FundingRateUpdate, InstrumentClose},
1437        enums::{InstrumentCloseType, MarketStatusAction},
1438    };
1439    use rstest::rstest;
1440
1441    use super::*;
1442    use crate::arrow::{
1443        DecodeFromRecordBatch, DecodeTypedFromRecordBatch, EncodeToRecordBatch, StringColumnRef,
1444    };
1445
1446    #[rstest]
1447    #[case("size", true)]
1448    #[case("volume", true)]
1449    #[case("bid_size_0", true)]
1450    #[case("ask_size_9", true)]
1451    #[case("price", false)]
1452    #[case("quantity_hint", false)]
1453    #[case("bid_size_10", false)]
1454    fn legacy_quantity_fields_use_exact_schema_names(#[case] name: &str, #[case] expected: bool) {
1455        assert_eq!(is_legacy_quantity_field(name), expected);
1456    }
1457
1458    #[rstest]
1459    fn legacy_market_columns_normalize_enums_without_changing_flags() {
1460        let price = 1_i64.to_le_bytes();
1461        let size = 2_u64.to_le_bytes();
1462        let batch = RecordBatch::try_new(
1463            Arc::new(Schema::new(vec![
1464                Field::new("action", DataType::UInt8, false),
1465                Field::new("side", DataType::UInt8, false),
1466                Field::new("price", DataType::FixedSizeBinary(8), false),
1467                Field::new("size", DataType::FixedSizeBinary(8), false),
1468                Field::new("order_id", DataType::UInt64, false),
1469                Field::new("flags", DataType::UInt8, false),
1470                Field::new("sequence", DataType::UInt64, false),
1471                Field::new("ts_event", DataType::UInt64, false),
1472                Field::new("ts_init", DataType::UInt64, false),
1473            ])),
1474            vec![
1475                Arc::new(UInt8Array::from(vec![BookAction::Add as u8])),
1476                Arc::new(UInt8Array::from(vec![OrderSide::Buy as u8])),
1477                Arc::new(
1478                    FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1479                        [Some(price.as_slice())].into_iter(),
1480                        8,
1481                    )
1482                    .unwrap(),
1483                ),
1484                Arc::new(
1485                    FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1486                        [Some(size.as_slice())].into_iter(),
1487                        8,
1488                    )
1489                    .unwrap(),
1490                ),
1491                Arc::new(UInt64Array::from(vec![11])),
1492                Arc::new(UInt8Array::from(vec![7])),
1493                Arc::new(UInt64Array::from(vec![12])),
1494                Arc::new(UInt64Array::from(vec![13])),
1495                Arc::new(UInt64Array::from(vec![1])),
1496            ],
1497        )
1498        .unwrap();
1499
1500        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1501        let action =
1502            StringColumnRef::try_from_array(normalized.column_by_name("action").unwrap().as_ref())
1503                .unwrap();
1504        let side =
1505            StringColumnRef::try_from_array(normalized.column_by_name("side").unwrap().as_ref())
1506                .unwrap();
1507
1508        assert_eq!(action.value(0), "ADD");
1509        assert_eq!(side.value(0), "BUY");
1510        assert_eq!(
1511            normalized.column_by_name("price").unwrap().data_type(),
1512            &fixed_decimal_data_type(),
1513        );
1514        assert_eq!(
1515            normalized.column_by_name("size").unwrap().data_type(),
1516            &fixed_decimal_data_type(),
1517        );
1518        assert_eq!(
1519            normalized.column_by_name("flags").unwrap().data_type(),
1520            &DataType::UInt8,
1521        );
1522        assert_eq!(
1523            normalized.column_by_name("ts_init").unwrap().data_type(),
1524            &timestamp_data_type(),
1525        );
1526    }
1527
1528    #[rstest]
1529    fn legacy_enum_normalization_rejects_null_with_field_and_row() {
1530        let field = Field::new("aggressor_side", DataType::UInt8, true);
1531        let values = UInt8Array::from(vec![Some(AggressorSide::Buy as u8), None]);
1532
1533        let error = legacy_enum_dictionary_column(&field, &values).unwrap_err();
1534
1535        assert_eq!(
1536            error.to_string(),
1537            "Cast error: Legacy enum column 'aggressor_side' is null at row 1"
1538        );
1539    }
1540
1541    #[rstest]
1542    fn unrelated_named_columns_are_not_retyped() {
1543        let batch = RecordBatch::try_new(
1544            Arc::new(Schema::new(vec![
1545                Field::new("side", DataType::UInt8, false),
1546                Field::new("ts_recv", DataType::UInt64, false),
1547            ])),
1548            vec![
1549                Arc::new(UInt8Array::from(vec![127])),
1550                Arc::new(UInt64Array::from(vec![u64::MAX])),
1551            ],
1552        )
1553        .unwrap();
1554
1555        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1556
1557        assert_eq!(normalized, batch);
1558    }
1559
1560    #[rstest]
1561    fn unrelated_fixed_binary_column_is_not_retyped() {
1562        let value = i64::MIN.to_le_bytes();
1563        let batch = RecordBatch::try_new(
1564            Arc::new(Schema::new(vec![Field::new(
1565                "price",
1566                DataType::FixedSizeBinary(8),
1567                false,
1568            )])),
1569            vec![Arc::new(
1570                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1571                    [Some(value.as_slice())].into_iter(),
1572                    8,
1573                )
1574                .unwrap(),
1575            )],
1576        )
1577        .unwrap();
1578
1579        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1580
1581        assert_eq!(normalized, batch);
1582    }
1583
1584    #[rstest]
1585    fn legacy_family_fingerprints_cover_unpinned_schemas() {
1586        let fixed = DataType::FixedSizeBinary(8);
1587        let schemas = [
1588            Schema::new(vec![
1589                Field::new("close_price", fixed.clone(), false),
1590                Field::new("close_type", DataType::UInt8, false),
1591                Field::new("ts_event", DataType::UInt64, false),
1592                Field::new("ts_init", DataType::UInt64, false),
1593            ]),
1594            Schema::new(vec![
1595                Field::new("instrument_id", DataType::Utf8, true),
1596                Field::new("close_type", DataType::Utf8, true),
1597                Field::new("close_price", DataType::Utf8, true),
1598                Field::new("ts_event", DataType::UInt64, true),
1599                Field::new("ts_init", DataType::UInt64, true),
1600            ]),
1601            Schema::new(vec![
1602                Field::new("value", fixed, false),
1603                Field::new("ts_event", DataType::UInt64, false),
1604                Field::new("ts_init", DataType::UInt64, false),
1605            ]),
1606            Schema::new(vec![
1607                Field::new("rate", DataType::Binary, false),
1608                Field::new("interval", DataType::UInt16, true),
1609                Field::new("next_funding_ns", DataType::UInt64, true),
1610                Field::new("ts_event", DataType::UInt64, false),
1611                Field::new("ts_init", DataType::UInt64, false),
1612            ]),
1613            Schema::new(vec![
1614                Field::new("instrument_id", DataType::Utf8, true),
1615                Field::new("action", DataType::Utf8, true),
1616                Field::new("reason", DataType::Utf8, true),
1617                Field::new("trading_event", DataType::Utf8, true),
1618                Field::new("is_trading", DataType::Boolean, true),
1619                Field::new("is_quoting", DataType::Boolean, true),
1620                Field::new("is_short_sell_restricted", DataType::Boolean, true),
1621                Field::new("ts_event", DataType::UInt64, true),
1622                Field::new("ts_init", DataType::UInt64, true),
1623            ]),
1624        ];
1625
1626        assert!(schemas.iter().all(is_nautilus_legacy_schema));
1627    }
1628
1629    #[rstest]
1630    fn legacy_family_fingerprint_accepts_dictionary_string_fields() {
1631        let schema = Schema::new(vec![
1632            Field::new("price", DataType::FixedSizeBinary(8), false),
1633            Field::new("size", DataType::FixedSizeBinary(8), false),
1634            Field::new("aggressor_side", DataType::UInt8, false),
1635            Field::new(
1636                "trade_id",
1637                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
1638                false,
1639            ),
1640            Field::new("ts_event", DataType::UInt64, false),
1641            Field::new("ts_init", DataType::UInt64, false),
1642            Field::new(
1643                KEY_IDENTIFIER,
1644                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
1645                false,
1646            ),
1647        ]);
1648
1649        assert!(is_nautilus_legacy_schema(&schema));
1650    }
1651
1652    #[rstest]
1653    fn legacy_flat_depth_fingerprint_accepts_all_nullable_fields() {
1654        let fixed = DataType::FixedSizeBinary(8);
1655        let mut fields = Vec::new();
1656
1657        for side in ["bid", "ask"] {
1658            for level in 0..DEPTH10_LEN {
1659                fields.extend([
1660                    Field::new(format!("{side}_price_{level}"), fixed.clone(), true),
1661                    Field::new(format!("{side}_size_{level}"), fixed.clone(), true),
1662                    Field::new(format!("{side}_count_{level}"), DataType::UInt32, true),
1663                ]);
1664            }
1665        }
1666        fields.extend([
1667            Field::new("flags", DataType::UInt8, true),
1668            Field::new("sequence", DataType::UInt64, true),
1669            Field::new("ts_event", DataType::UInt64, true),
1670            Field::new("ts_init", DataType::UInt64, true),
1671        ]);
1672        let schema = Schema::new(fields);
1673
1674        assert!(super::legacy_flat_depth_fingerprint_matches(&schema));
1675    }
1676
1677    #[rstest]
1678    fn legacy_fixed_list_depth_fingerprint_accepts_missing_order_ids() {
1679        let fixed = Arc::new(Field::new("item", DataType::FixedSizeBinary(8), false));
1680        let count = Arc::new(Field::new("item", DataType::UInt32, false));
1681        let schema = Schema::new(vec![
1682            Field::new(
1683                "bid_price",
1684                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1685                false,
1686            ),
1687            Field::new(
1688                "ask_price",
1689                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1690                false,
1691            ),
1692            Field::new(
1693                "bid_size",
1694                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1695                false,
1696            ),
1697            Field::new("ask_size", DataType::FixedSizeList(fixed, 10), false),
1698            Field::new(
1699                "bid_count",
1700                DataType::FixedSizeList(Arc::clone(&count), 10),
1701                false,
1702            ),
1703            Field::new("ask_count", DataType::FixedSizeList(count, 10), false),
1704        ]);
1705
1706        assert!(is_nautilus_legacy_schema(&schema));
1707    }
1708
1709    #[rstest]
1710    fn metadata_only_schema_is_not_a_legacy_family() {
1711        let schema = Schema::new_with_metadata(
1712            vec![
1713                Field::new("side", DataType::UInt8, false),
1714                Field::new("ts_recv", DataType::UInt64, false),
1715            ],
1716            [("type_name".to_string(), "CustomData".to_string())].into(),
1717        );
1718
1719        assert!(!is_nautilus_legacy_schema(&schema));
1720    }
1721
1722    #[rstest]
1723    fn known_metadata_families_require_and_normalize_their_legacy_shape() {
1724        let schemas = [
1725            Schema::new_with_metadata(
1726                vec![
1727                    Field::new("rate", DataType::Binary, false),
1728                    Field::new("interval", DataType::UInt16, true),
1729                    Field::new("next_funding_ns", DataType::UInt64, true),
1730                    Field::new("ts_event", DataType::UInt64, false),
1731                    Field::new("ts_init", DataType::UInt64, false),
1732                ],
1733                [("type".to_string(), "FundingRateUpdate".to_string())].into(),
1734            ),
1735            Schema::new_with_metadata(
1736                vec![
1737                    Field::new("instrument_id", DataType::Utf8, true),
1738                    Field::new("action", DataType::Utf8, true),
1739                    Field::new("reason", DataType::Utf8, true),
1740                    Field::new("trading_event", DataType::Utf8, true),
1741                    Field::new("is_trading", DataType::Boolean, true),
1742                    Field::new("is_quoting", DataType::Boolean, true),
1743                    Field::new("is_short_sell_restricted", DataType::Boolean, true),
1744                    Field::new("ts_event", DataType::UInt64, true),
1745                    Field::new("ts_init", DataType::UInt64, true),
1746                ],
1747                [("type".to_string(), "InstrumentStatus".to_string())].into(),
1748            ),
1749            Schema::new_with_metadata(
1750                vec![
1751                    Field::new("instrument_id", DataType::Utf8, false),
1752                    Field::new("delta", DataType::Float64, false),
1753                    Field::new("gamma", DataType::Float64, false),
1754                    Field::new("vega", DataType::Float64, false),
1755                    Field::new("theta", DataType::Float64, false),
1756                    Field::new("rho", DataType::Float64, false),
1757                    Field::new("mark_iv", DataType::Float64, true),
1758                    Field::new("bid_iv", DataType::Float64, true),
1759                    Field::new("ask_iv", DataType::Float64, true),
1760                    Field::new("underlying_price", DataType::Float64, true),
1761                    Field::new("open_interest", DataType::Float64, true),
1762                    Field::new("ts_event", DataType::UInt64, false),
1763                    Field::new("ts_init", DataType::UInt64, false),
1764                    Field::new("convention", DataType::Utf8, false),
1765                ],
1766                [("type".to_string(), "OptionGreeks".to_string())].into(),
1767            ),
1768        ];
1769
1770        for schema in schemas {
1771            assert!(is_nautilus_legacy_schema(&schema));
1772            let normalized =
1773                normalize_legacy_fixed_columns(&RecordBatch::new_empty(Arc::new(schema))).unwrap();
1774
1775            assert_eq!(
1776                normalized
1777                    .schema()
1778                    .field_with_name("ts_init")
1779                    .unwrap()
1780                    .data_type(),
1781                &timestamp_data_type(),
1782            );
1783        }
1784    }
1785
1786    #[rstest]
1787    fn near_match_with_foreign_field_type_is_not_retyped() {
1788        let schema = Arc::new(Schema::new(vec![
1789            Field::new("action", DataType::UInt8, false),
1790            Field::new("side", DataType::UInt8, false),
1791            Field::new("price", DataType::FixedSizeBinary(8), false),
1792            Field::new("size", DataType::FixedSizeBinary(8), false),
1793            Field::new("order_id", DataType::UInt64, false),
1794            Field::new("flags", DataType::UInt8, false),
1795            Field::new("sequence", DataType::Int64, false),
1796            Field::new("ts_event", DataType::UInt64, false),
1797            Field::new("ts_init", DataType::UInt64, false),
1798        ]));
1799        let batch = RecordBatch::new_empty(schema);
1800
1801        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1802
1803        assert_eq!(normalized, batch);
1804    }
1805
1806    #[rstest]
1807    fn legacy_price_error_fails_with_field_and_row() {
1808        let fixed = |raw: [u8; 8]| -> ArrayRef {
1809            Arc::new(
1810                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1811                    [Some(raw.as_slice())].into_iter(),
1812                    8,
1813                )
1814                .unwrap(),
1815            )
1816        };
1817        let batch = RecordBatch::try_new(
1818            Arc::new(Schema::new_with_metadata(
1819                vec![
1820                    Field::new("bid_price", DataType::FixedSizeBinary(8), false),
1821                    Field::new("ask_price", DataType::FixedSizeBinary(8), false),
1822                    Field::new("bid_size", DataType::FixedSizeBinary(8), false),
1823                    Field::new("ask_size", DataType::FixedSizeBinary(8), false),
1824                    Field::new("ts_event", DataType::UInt64, false),
1825                    Field::new("ts_init", DataType::UInt64, false),
1826                ],
1827                [("type".to_string(), "QuoteTick".to_string())].into(),
1828            )),
1829            vec![
1830                fixed(i64::MIN.to_le_bytes()),
1831                fixed(2_i64.to_le_bytes()),
1832                fixed(3_u64.to_le_bytes()),
1833                fixed(4_u64.to_le_bytes()),
1834                Arc::new(UInt64Array::from(vec![5])),
1835                Arc::new(UInt64Array::from(vec![6])),
1836            ],
1837        )
1838        .unwrap();
1839
1840        let error = normalize_legacy_fixed_columns(&batch).unwrap_err();
1841
1842        assert_eq!(
1843            error.to_string(),
1844            format!(
1845                "Cast error: Legacy price column 'bid_price' contains PRICE_ERROR raw value {} at row 0",
1846                i64::MIN,
1847            ),
1848        );
1849    }
1850
1851    #[rstest]
1852    fn metadata_family_near_match_passes_through_unchanged() {
1853        let raw = i64::MIN.to_le_bytes();
1854        let batch = RecordBatch::try_new(
1855            Arc::new(Schema::new_with_metadata(
1856                vec![Field::new("bid_price", DataType::FixedSizeBinary(8), false)],
1857                [("type".to_string(), "QuoteTick".to_string())].into(),
1858            )),
1859            vec![Arc::new(
1860                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1861                    [Some(raw.as_slice())].into_iter(),
1862                    8,
1863                )
1864                .unwrap(),
1865            )],
1866        )
1867        .unwrap();
1868
1869        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1870
1871        assert_eq!(normalized, batch);
1872    }
1873
1874    #[rstest]
1875    fn schema_fingerprint_ignores_metadata_and_includes_nullability() {
1876        let fields = vec![Field::new("value", DataType::Utf8, false)];
1877        let schema1 = Schema::new_with_metadata(
1878            fields.clone(),
1879            HashMap::from([("a".to_string(), "1".to_string())]),
1880        );
1881        let schema2 =
1882            Schema::new_with_metadata(fields, HashMap::from([("b".to_string(), "2".to_string())]));
1883        let nullable = Schema::new(vec![Field::new("value", DataType::Utf8, true)]);
1884
1885        assert_eq!(schema_fingerprint(&schema1), schema_fingerprint(&schema2));
1886        assert_ne!(schema_fingerprint(&schema1), schema_fingerprint(&nullable));
1887    }
1888
1889    #[rstest]
1890    fn current_schema_passes_through() {
1891        let batch = RecordBatch::new_empty(Arc::new(QuoteTick::get_schema(None)));
1892        let result = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap();
1893
1894        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1895        assert_eq!(result.batches.len(), 1);
1896    }
1897
1898    #[rstest]
1899    fn parquet_schema_with_plain_dictionary_strings_passes_through() {
1900        let fields = TradeTick::get_schema(None)
1901            .fields()
1902            .iter()
1903            .filter(|field| field.name() != KEY_IDENTIFIER)
1904            .map(|field| match field.data_type() {
1905                DataType::Dictionary(..) => field.as_ref().clone().with_data_type(DataType::Utf8),
1906                _ => field.as_ref().clone(),
1907            })
1908            .collect::<Vec<_>>();
1909
1910        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1911        let result = transcode_legacy_record_batch("trades", "trades.parquet", batch).unwrap();
1912
1913        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1914        assert_eq!(result.batches.len(), 1);
1915    }
1916
1917    #[rstest]
1918    fn reordered_nullable_schema_with_identity_column_passes_through() {
1919        let mut fields = QuoteTick::get_schema(None)
1920            .fields()
1921            .iter()
1922            .rev()
1923            .map(|field| field.as_ref().clone().with_nullable(true))
1924            .collect::<Vec<_>>();
1925        fields.push(Field::new(KEY_INSTRUMENT_ID, DataType::Utf8, false));
1926        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1927        let result = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap();
1928
1929        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1930        assert_eq!(result.batches.len(), 1);
1931    }
1932
1933    #[rstest]
1934    #[case::extra_column(Some(Field::new("venue", DataType::Utf8, true)), None)]
1935    #[case::missing_column(None, Some("ask_size"))]
1936    fn schema_with_unexpected_columns_is_rejected(
1937        #[case] extra: Option<Field>,
1938        #[case] missing: Option<&str>,
1939    ) {
1940        let mut fields = QuoteTick::get_schema(None)
1941            .fields()
1942            .iter()
1943            .filter(|field| Some(field.name().as_str()) != missing)
1944            .map(|field| field.as_ref().clone())
1945            .collect::<Vec<_>>();
1946        fields.extend(extra);
1947        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1948        let error = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap_err();
1949
1950        assert!(matches!(error, LegacyArrowError::UnknownSchema { .. }));
1951    }
1952
1953    #[rstest]
1954    fn unknown_registered_schema_is_rejected() {
1955        let batch = RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new(
1956            "unexpected",
1957            DataType::UInt64,
1958            false,
1959        )])));
1960        let error = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap_err();
1961
1962        assert!(matches!(error, LegacyArrowError::UnknownSchema { .. }));
1963        assert!(error.to_string().contains("quotes.parquet"));
1964    }
1965
1966    #[rstest]
1967    #[case(false)]
1968    #[case(true)]
1969    fn instrument_status_fields_are_reordered(#[case] normalized: bool) {
1970        let schema = Arc::new(legacy_instrument_status_schema());
1971        let batch = RecordBatch::try_new(
1972            schema,
1973            vec![
1974                Arc::new(StringArray::from(vec!["AAPL.XNAS"])),
1975                Arc::new(StringArray::from(vec!["TRADING"])),
1976                Arc::new(StringArray::from(vec![Some("Normal")])),
1977                Arc::new(StringArray::from(vec![Some("OPEN")])),
1978                Arc::new(BooleanArray::from(vec![Some(true)])),
1979                Arc::new(BooleanArray::from(vec![Some(true)])),
1980                Arc::new(BooleanArray::from(vec![Some(false)])),
1981                Arc::new(UInt64Array::from(vec![1])),
1982                Arc::new(UInt64Array::from(vec![2])),
1983            ],
1984        )
1985        .unwrap();
1986
1987        // Migration normalizes legacy timestamps before resolving the schema
1988        let batch = if normalized {
1989            normalize_legacy_fixed_columns(&batch).unwrap()
1990        } else {
1991            batch
1992        };
1993
1994        let result =
1995            transcode_legacy_record_batch("instrument_status", "status.parquet", batch).unwrap();
1996        let output = &result.batches[0];
1997
1998        assert_eq!(result.kind, LegacyTranscodeKind::InstrumentStatusV1);
1999        assert_eq!(
2000            output
2001                .schema()
2002                .fields()
2003                .iter()
2004                .map(|field| field.name().as_str())
2005                .collect::<Vec<_>>(),
2006            vec![
2007                "action",
2008                "ts_event",
2009                "ts_init",
2010                "reason",
2011                "trading_event",
2012                "is_trading",
2013                "is_quoting",
2014                "is_short_sell_restricted",
2015                "identifier",
2016            ]
2017        );
2018        assert_eq!(
2019            output
2020                .schema()
2021                .field_with_name("action")
2022                .unwrap()
2023                .data_type(),
2024            &enum_dictionary_data_type()
2025        );
2026        let identifiers = output
2027            .column_by_name(KEY_IDENTIFIER)
2028            .unwrap()
2029            .as_any()
2030            .downcast_ref::<StringArray>()
2031            .unwrap();
2032        assert_eq!(identifiers.value(0), "AAPL.XNAS");
2033
2034        // Migration attaches the row identifier as instrument metadata before decoding
2035        let mut metadata = output.schema().metadata().clone();
2036        metadata.insert(
2037            KEY_INSTRUMENT_ID.to_string(),
2038            identifiers.value(0).to_string(),
2039        );
2040        let decoded = InstrumentStatus::decode_typed_batch(&metadata, output.clone()).unwrap();
2041        assert_eq!(decoded[0].action, MarketStatusAction::Trading);
2042        assert_eq!(decoded[0].instrument_id.to_string(), "AAPL.XNAS");
2043        assert_eq!(decoded[0].ts_event.as_u64(), 1);
2044        assert_eq!(decoded[0].ts_init.as_u64(), 2);
2045    }
2046
2047    #[rstest]
2048    fn funding_rate_is_decoded_and_interval_is_widened() {
2049        let schema = Arc::new(Schema::new_with_metadata(
2050            legacy_funding_rate_schema()
2051                .fields()
2052                .iter()
2053                .cloned()
2054                .collect::<Vec<_>>(),
2055            HashMap::from([(
2056                KEY_INSTRUMENT_ID.to_string(),
2057                "BTCUSDT-PERP.BINANCE".to_string(),
2058            )]),
2059        ));
2060        let batch = RecordBatch::try_new(
2061            schema,
2062            vec![
2063                Arc::new(BinaryArray::from(vec![b"\"0.0001\"".as_slice()])),
2064                Arc::new(UInt16Array::from(vec![Some(480)])),
2065                Arc::new(UInt64Array::from(vec![Some(9)])),
2066                Arc::new(UInt64Array::from(vec![1])),
2067                Arc::new(UInt64Array::from(vec![2])),
2068            ],
2069        )
2070        .unwrap();
2071
2072        let result =
2073            transcode_legacy_record_batch("funding_rates", "funding.parquet", batch).unwrap();
2074        let output = &result.batches[0];
2075        let decoded =
2076            FundingRateUpdate::decode_typed_batch(output.schema().metadata(), output.clone())
2077                .unwrap();
2078
2079        assert_eq!(result.kind, LegacyTranscodeKind::FundingRateUpdateV1);
2080        assert_eq!(decoded[0].instrument_id.to_string(), "BTCUSDT-PERP.BINANCE");
2081        assert_eq!(decoded[0].rate.to_string(), "0.0001");
2082        assert_eq!(decoded[0].interval, Some(480));
2083        assert_eq!(
2084            decoded[0].next_funding_ns.map(|value| value.as_u64()),
2085            Some(9)
2086        );
2087    }
2088
2089    #[rstest]
2090    #[case(false)]
2091    #[case(true)]
2092    fn instrument_close_is_split_and_decoded(#[case] normalized: bool) {
2093        let schema = Arc::new(legacy_instrument_close_schema());
2094        let batch = RecordBatch::try_new(
2095            schema,
2096            vec![
2097                Arc::new(StringArray::from(vec![
2098                    "AUD/USD.SIM",
2099                    "GBP/USD.SIM",
2100                    "AUD/USD.SIM",
2101                ])),
2102                Arc::new(StringArray::from(vec![
2103                    "END_OF_SESSION",
2104                    "CONTRACT_EXPIRED",
2105                    "END_OF_SESSION",
2106                ])),
2107                Arc::new(StringArray::from(vec!["1.0500", "2.1000", "1.0600"])),
2108                Arc::new(UInt64Array::from(vec![1, 2, 3])),
2109                Arc::new(UInt64Array::from(vec![4, 5, 6])),
2110            ],
2111        )
2112        .unwrap();
2113
2114        let batch = if normalized {
2115            normalize_legacy_fixed_columns(&batch).unwrap()
2116        } else {
2117            batch
2118        };
2119
2120        let result =
2121            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap();
2122        let decoded = result
2123            .batches
2124            .iter()
2125            .flat_map(|batch| {
2126                InstrumentClose::decode_batch(batch.schema().metadata(), batch.clone()).unwrap()
2127            })
2128            .collect::<Vec<_>>();
2129
2130        assert_eq!(
2131            decoded
2132                .iter()
2133                .map(|close| (close.ts_event.as_u64(), close.ts_init.as_u64()))
2134                .collect::<Vec<_>>(),
2135            vec![(1, 4), (3, 6), (2, 5)]
2136        );
2137        assert_eq!(result.kind, LegacyTranscodeKind::InstrumentCloseV1);
2138        assert_eq!(result.batches.len(), 2);
2139        assert_eq!(
2140            decoded
2141                .iter()
2142                .map(|close| close.instrument_id.to_string())
2143                .collect::<Vec<_>>(),
2144            vec!["AUD/USD.SIM", "AUD/USD.SIM", "GBP/USD.SIM"]
2145        );
2146        assert_eq!(
2147            decoded
2148                .iter()
2149                .map(|close| close.close_price.to_string())
2150                .collect::<Vec<_>>(),
2151            vec!["1.0500", "1.0600", "2.1000"]
2152        );
2153        assert_eq!(
2154            decoded
2155                .iter()
2156                .map(|close| close.close_type)
2157                .collect::<Vec<_>>(),
2158            vec![
2159                InstrumentCloseType::EndOfSession,
2160                InstrumentCloseType::EndOfSession,
2161                InstrumentCloseType::ContractExpired,
2162            ]
2163        );
2164
2165        // Transcoded files must consolidate with files written by the current encoder
2166        for batch in &result.batches {
2167            let schema = batch.schema();
2168            let closes = InstrumentClose::decode_batch(schema.metadata(), batch.clone()).unwrap();
2169            let fresh = InstrumentClose::encode_batch(&closes[0].metadata(), &closes).unwrap();
2170            assert_eq!(fresh.schema(), schema);
2171        }
2172    }
2173
2174    #[rstest]
2175    fn instrument_close_rejects_mixed_precision() {
2176        let batch = RecordBatch::try_new(
2177            Arc::new(legacy_instrument_close_schema()),
2178            vec![
2179                Arc::new(StringArray::from(vec!["AUD/USD.SIM", "AUD/USD.SIM"])),
2180                Arc::new(StringArray::from(vec!["END_OF_SESSION", "END_OF_SESSION"])),
2181                Arc::new(StringArray::from(vec!["1.0500", "1.060"])),
2182                Arc::new(UInt64Array::from(vec![1, 2])),
2183                Arc::new(UInt64Array::from(vec![3, 4])),
2184            ],
2185        )
2186        .unwrap();
2187
2188        let error =
2189            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap_err();
2190
2191        assert!(error.to_string().contains("close.parquet"));
2192        assert!(error.to_string().contains("precision conflict at row 1"));
2193    }
2194
2195    #[rstest]
2196    fn instrument_close_rejects_unknown_close_type() {
2197        let batch = RecordBatch::try_new(
2198            Arc::new(legacy_instrument_close_schema()),
2199            vec![
2200                Arc::new(StringArray::from(vec!["AUD/USD.SIM"])),
2201                Arc::new(StringArray::from(vec!["UNKNOWN"])),
2202                Arc::new(StringArray::from(vec!["1.0500"])),
2203                Arc::new(UInt64Array::from(vec![1])),
2204                Arc::new(UInt64Array::from(vec![2])),
2205            ],
2206        )
2207        .unwrap();
2208
2209        let error =
2210            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap_err();
2211
2212        assert!(error.to_string().contains("close.parquet"));
2213        assert!(
2214            error
2215                .to_string()
2216                .contains("unknown close_type \"UNKNOWN\" at row 0")
2217        );
2218    }
2219}