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    let names = (0..values.len())
875        .map(|row| legacy_enum_name(field.name(), values.value(row)))
876        .collect::<Result<Vec<_>, _>>()?;
877    Ok(Arc::new(enum_dictionary_array(names)?) as ArrayRef)
878}
879
880fn normalize_legacy_fixed_array(
881    name: &str,
882    values: &FixedSizeBinaryArray,
883    width: i32,
884) -> Result<Decimal128Array, ArrowError> {
885    let quantity = is_legacy_quantity_field(name);
886    let mut decimals = Vec::with_capacity(values.len());
887
888    for row in 0..values.len() {
889        if values.is_null(row) {
890            decimals.push(None);
891            continue;
892        }
893
894        let bytes = values.value(row);
895        let decimal = match (width, quantity) {
896            (8, false) => {
897                let raw = i64::from_le_bytes(bytes.try_into().map_err(|e| {
898                    ArrowError::CastError(format!(
899                        "Invalid legacy price column '{name}' at row {row}: {e}"
900                    ))
901                })?);
902
903                if raw == i64::MIN {
904                    return Err(ArrowError::CastError(format!(
905                        "Legacy price column '{name}' contains PRICE_ERROR raw value {raw} at row {row}",
906                    )));
907                }
908                (raw != i64::MAX).then_some(i128::from(raw) * STANDARD_TO_DECIMAL_SCALE)
909            }
910            (8, true) => {
911                let raw = u64::from_le_bytes(bytes.try_into().map_err(|e| {
912                    ArrowError::CastError(format!(
913                        "Invalid legacy quantity column '{name}' at row {row}: {e}"
914                    ))
915                })?);
916                (raw != u64::MAX).then_some(i128::from(raw) * STANDARD_TO_DECIMAL_SCALE)
917            }
918            (16, false) => {
919                let raw = i128::from_le_bytes(bytes.try_into().map_err(|e| {
920                    ArrowError::CastError(format!(
921                        "Invalid legacy price column '{name}' at row {row}: {e}"
922                    ))
923                })?);
924
925                if raw == i128::MIN {
926                    return Err(ArrowError::CastError(format!(
927                        "Legacy price column '{name}' contains PRICE_ERROR raw value {raw} at row {row}",
928                    )));
929                }
930                (raw != i128::MAX).then_some(raw)
931            }
932            (16, true) => {
933                let raw = u128::from_le_bytes(bytes.try_into().map_err(|e| {
934                    ArrowError::CastError(format!(
935                        "Invalid legacy quantity column '{name}' at row {row}: {e}"
936                    ))
937                })?);
938
939                if raw == u128::MAX {
940                    None
941                } else {
942                    Some(i128::try_from(raw).map_err(|_| {
943                        ArrowError::CastError(format!(
944                            "Legacy quantity column '{name}' exceeds Decimal128 at row {row}"
945                        ))
946                    })?)
947                }
948            }
949            _ => unreachable!("legacy fixed width checked above"),
950        };
951        decimals.push(decimal);
952    }
953
954    Decimal128Array::from(decimals)
955        .with_precision_and_scale(FIXED_DECIMAL_PRECISION, FIXED_DECIMAL_SCALE)
956}
957
958type LegacySchemaFields = &'static [(&'static str, LegacyFieldType, bool)];
959
960const LEGACY_SCHEMA_FAMILIES: &[(&[&str], LegacySchemaFields)] = &[
961    (
962        &["OrderBookDelta"],
963        &[
964            ("action", LegacyFieldType::UInt8, false),
965            ("side", LegacyFieldType::UInt8, false),
966            ("price", LegacyFieldType::Fixed, false),
967            ("size", LegacyFieldType::Fixed, false),
968            ("order_id", LegacyFieldType::UInt64, false),
969            ("flags", LegacyFieldType::UInt8, false),
970            ("sequence", LegacyFieldType::UInt64, false),
971            ("ts_event", LegacyFieldType::UInt64, false),
972            ("ts_init", LegacyFieldType::UInt64, false),
973        ],
974    ),
975    (
976        &["TradeTick"],
977        &[
978            ("price", LegacyFieldType::Fixed, false),
979            ("size", LegacyFieldType::Fixed, false),
980            ("aggressor_side", LegacyFieldType::UInt8, false),
981            ("trade_id", LegacyFieldType::Utf8, false),
982            ("ts_event", LegacyFieldType::UInt64, false),
983            ("ts_init", LegacyFieldType::UInt64, false),
984        ],
985    ),
986    (
987        &["InstrumentClose"],
988        &[
989            ("close_price", LegacyFieldType::Fixed, false),
990            ("close_type", LegacyFieldType::UInt8, false),
991            ("ts_event", LegacyFieldType::UInt64, false),
992            ("ts_init", LegacyFieldType::UInt64, false),
993        ],
994    ),
995    (
996        &["InstrumentClose"],
997        &[
998            ("instrument_id", LegacyFieldType::Utf8, true),
999            ("close_type", LegacyFieldType::Utf8, true),
1000            ("close_price", LegacyFieldType::Utf8, true),
1001            ("ts_event", LegacyFieldType::UInt64, true),
1002            ("ts_init", LegacyFieldType::UInt64, true),
1003        ],
1004    ),
1005    (
1006        &["QuoteTick"],
1007        &[
1008            ("bid_price", LegacyFieldType::Fixed, false),
1009            ("ask_price", LegacyFieldType::Fixed, false),
1010            ("bid_size", LegacyFieldType::Fixed, false),
1011            ("ask_size", LegacyFieldType::Fixed, false),
1012            ("ts_event", LegacyFieldType::UInt64, false),
1013            ("ts_init", LegacyFieldType::UInt64, false),
1014        ],
1015    ),
1016    (
1017        &["Bar"],
1018        &[
1019            ("open", LegacyFieldType::Fixed, false),
1020            ("high", LegacyFieldType::Fixed, false),
1021            ("low", LegacyFieldType::Fixed, false),
1022            ("close", LegacyFieldType::Fixed, false),
1023            ("volume", LegacyFieldType::Fixed, false),
1024            ("ts_event", LegacyFieldType::UInt64, false),
1025            ("ts_init", LegacyFieldType::UInt64, false),
1026        ],
1027    ),
1028    (
1029        &["MarkPriceUpdate", "IndexPriceUpdate"],
1030        &[
1031            ("value", LegacyFieldType::Fixed, false),
1032            ("ts_event", LegacyFieldType::UInt64, false),
1033            ("ts_init", LegacyFieldType::UInt64, false),
1034        ],
1035    ),
1036    (
1037        &["FundingRateUpdate"],
1038        &[
1039            ("rate", LegacyFieldType::Binary, false),
1040            ("interval", LegacyFieldType::UInt16, true),
1041            ("next_funding_ns", LegacyFieldType::UInt64, true),
1042            ("ts_event", LegacyFieldType::UInt64, false),
1043            ("ts_init", LegacyFieldType::UInt64, false),
1044        ],
1045    ),
1046    (
1047        &["InstrumentStatus"],
1048        &[
1049            ("instrument_id", LegacyFieldType::Utf8, true),
1050            ("action", LegacyFieldType::Utf8, true),
1051            ("reason", LegacyFieldType::Utf8, true),
1052            ("trading_event", LegacyFieldType::Utf8, true),
1053            ("is_trading", LegacyFieldType::Boolean, true),
1054            ("is_quoting", LegacyFieldType::Boolean, true),
1055            ("is_short_sell_restricted", LegacyFieldType::Boolean, true),
1056            ("ts_event", LegacyFieldType::UInt64, true),
1057            ("ts_init", LegacyFieldType::UInt64, true),
1058        ],
1059    ),
1060    (
1061        &["OptionGreeks"],
1062        &[
1063            ("instrument_id", LegacyFieldType::Utf8, false),
1064            ("delta", LegacyFieldType::Float64, false),
1065            ("gamma", LegacyFieldType::Float64, false),
1066            ("vega", LegacyFieldType::Float64, false),
1067            ("theta", LegacyFieldType::Float64, false),
1068            ("rho", LegacyFieldType::Float64, false),
1069            ("mark_iv", LegacyFieldType::Float64, true),
1070            ("bid_iv", LegacyFieldType::Float64, true),
1071            ("ask_iv", LegacyFieldType::Float64, true),
1072            ("underlying_price", LegacyFieldType::Float64, true),
1073            ("open_interest", LegacyFieldType::Float64, true),
1074            ("ts_event", LegacyFieldType::UInt64, false),
1075            ("ts_init", LegacyFieldType::UInt64, false),
1076            ("convention", LegacyFieldType::Utf8, false),
1077        ],
1078    ),
1079];
1080
1081/// Returns whether a schema is a recognized Nautilus legacy family.
1082#[must_use]
1083pub fn is_nautilus_legacy_schema(schema: &Schema) -> bool {
1084    if let Some(type_name) = schema
1085        .metadata()
1086        .get("type_name")
1087        .or_else(|| schema.metadata().get("type"))
1088    {
1089        return legacy_metadata_family_matches(schema, type_name);
1090    }
1091
1092    LEGACY_SCHEMA_FAMILIES
1093        .iter()
1094        .any(|(_, fields)| schema_fingerprint_matches(schema, fields))
1095        || legacy_flat_depth_fingerprint_matches(schema)
1096        || legacy_fixed_list_depth_fingerprint_matches(schema)
1097}
1098
1099/// Returns whether timestamp annotations belong to a recognized Nautilus schema.
1100#[must_use]
1101pub fn is_nautilus_timestamp_schema(schema: &Schema) -> bool {
1102    if schema.metadata().contains_key("type_name") || schema.metadata().contains_key("type") {
1103        return true;
1104    }
1105
1106    if is_nautilus_legacy_schema(schema) {
1107        return true;
1108    }
1109    let fields = schema
1110        .fields()
1111        .iter()
1112        .filter(|field| field.name() != KEY_IDENTIFIER)
1113        .collect::<Vec<_>>();
1114    LEGACY_SCHEMA_FAMILIES.iter().any(|(_, expected)| {
1115        fields.len() == expected.len()
1116            && fields
1117                .iter()
1118                .zip(*expected)
1119                .all(|(field, (name, expected, _))| {
1120                    field.name() == *name
1121                        && (legacy_field_type_matches(field.data_type(), *expected)
1122                            || (is_timestamp_field(name)
1123                                && matches!(
1124                                    field.data_type(),
1125                                    DataType::Timestamp(TimeUnit::Nanosecond, _)
1126                                ))
1127                            || (matches!(expected, LegacyFieldType::Fixed)
1128                                && field.data_type() == &fixed_decimal_data_type())
1129                            || (is_legacy_enum_field(name)
1130                                && field.data_type() == &enum_dictionary_data_type()))
1131                })
1132    })
1133}
1134
1135/// Adds the UTC annotation to a nanosecond instant without a timezone.
1136#[must_use]
1137pub fn normalized_timestamp_type(data_type: &DataType) -> DataType {
1138    match data_type {
1139        DataType::Timestamp(TimeUnit::Nanosecond, None) => timestamp_data_type(),
1140        _ => data_type.clone(),
1141    }
1142}
1143
1144fn legacy_metadata_family_matches(schema: &Schema, type_name: &str) -> bool {
1145    match type_name {
1146        "OrderBookDepth10" | "OrderBookDepth" => {
1147            return legacy_flat_depth_fingerprint_matches(schema)
1148                || legacy_fixed_list_depth_fingerprint_matches(schema);
1149        }
1150        _ => {}
1151    }
1152
1153    LEGACY_SCHEMA_FAMILIES
1154        .iter()
1155        .filter(|(names, _)| names.contains(&type_name))
1156        .any(|(_, fields)| schema_fingerprint_matches(schema, fields))
1157}
1158
1159#[derive(Clone, Copy)]
1160enum LegacyFieldType {
1161    Binary,
1162    Float64,
1163    Fixed,
1164    UInt8,
1165    UInt16,
1166    UInt32,
1167    UInt64,
1168    Utf8,
1169    Boolean,
1170}
1171
1172fn schema_fingerprint_matches(schema: &Schema, expected: &[(&str, LegacyFieldType, bool)]) -> bool {
1173    let fields = schema
1174        .fields()
1175        .iter()
1176        .filter(|field| field.name() != KEY_IDENTIFIER)
1177        .collect::<Vec<_>>();
1178
1179    let fields_match = fields.len() == expected.len()
1180        && fields
1181            .iter()
1182            .zip(expected)
1183            .all(|(field, (name, data_type, _))| {
1184                field.name() == *name && legacy_field_type_matches(field.data_type(), *data_type)
1185            });
1186    let nullability_matches = fields
1187        .iter()
1188        .zip(expected)
1189        .all(|(field, (_, _, nullable))| field.is_nullable() == *nullable)
1190        || fields.iter().all(|field| field.is_nullable());
1191
1192    fields_match
1193        && nullability_matches
1194        && fixed_width_is_consistent(fields.iter().map(|field| field.data_type()))
1195}
1196
1197fn legacy_field_type_matches(data_type: &DataType, expected: LegacyFieldType) -> bool {
1198    match expected {
1199        LegacyFieldType::Binary => data_type == &DataType::Binary,
1200        LegacyFieldType::Float64 => data_type == &DataType::Float64,
1201        LegacyFieldType::Fixed => matches!(data_type, DataType::FixedSizeBinary(8 | 16)),
1202        LegacyFieldType::UInt8 => data_type == &DataType::UInt8,
1203        LegacyFieldType::UInt16 => data_type == &DataType::UInt16,
1204        LegacyFieldType::UInt32 => data_type == &DataType::UInt32,
1205        LegacyFieldType::UInt64 => data_type == &DataType::UInt64,
1206        LegacyFieldType::Utf8 => match data_type {
1207            DataType::Utf8 | DataType::Utf8View => true,
1208            DataType::Dictionary(_, value) => value.as_ref() == &DataType::Utf8,
1209            _ => false,
1210        },
1211        LegacyFieldType::Boolean => data_type == &DataType::Boolean,
1212    }
1213}
1214
1215fn fixed_width_is_consistent<'a>(data_types: impl Iterator<Item = &'a DataType>) -> bool {
1216    let mut width = None;
1217    data_types
1218        .filter_map(|data_type| match data_type {
1219            DataType::FixedSizeBinary(width) => Some(*width),
1220            _ => None,
1221        })
1222        .all(|current| *width.get_or_insert(current) == current)
1223}
1224
1225fn legacy_flat_depth_fingerprint_matches(schema: &Schema) -> bool {
1226    let fields = schema
1227        .fields()
1228        .iter()
1229        .filter(|field| field.name() != KEY_IDENTIFIER)
1230        .collect::<Vec<_>>();
1231    let market_fields = fields
1232        .iter()
1233        .filter(|field| {
1234            matches!(
1235                field.name().as_str(),
1236                "flags" | "sequence" | "ts_event" | "ts_init"
1237            )
1238        })
1239        .count();
1240
1241    let order_id_fields = fields
1242        .iter()
1243        .filter(|field| {
1244            field.name().starts_with("bid_order_id_") || field.name().starts_with("ask_order_id_")
1245        })
1246        .count();
1247    let all_nullable = fields.iter().all(|field| field.is_nullable());
1248
1249    fields.len() == 6 * DEPTH10_LEN + order_id_fields + market_fields
1250        && market_fields <= 4
1251        && matches!(order_id_fields, 0 | 20)
1252        && fixed_width_is_consistent(fields.iter().map(|field| field.data_type()))
1253        && ["bid", "ask"].iter().all(|side| {
1254            (0..DEPTH10_LEN).all(|level| {
1255                [
1256                    ("price", LegacyFieldType::Fixed),
1257                    ("size", LegacyFieldType::Fixed),
1258                    ("count", LegacyFieldType::UInt32),
1259                ]
1260                .iter()
1261                .all(|(value, data_type)| {
1262                    schema
1263                        .field_with_name(&format!("{side}_{value}_{level}"))
1264                        .is_ok_and(|field| {
1265                            (all_nullable || !field.is_nullable())
1266                                && legacy_field_type_matches(field.data_type(), *data_type)
1267                        })
1268                })
1269            })
1270        })
1271        && (order_id_fields == 0
1272            || ["bid", "ask"].iter().all(|side| {
1273                (0..DEPTH10_LEN).all(|level| {
1274                    schema
1275                        .field_with_name(&format!("{side}_order_id_{level}"))
1276                        .is_ok_and(|field| {
1277                            (all_nullable || !field.is_nullable())
1278                                && legacy_field_type_matches(
1279                                    field.data_type(),
1280                                    LegacyFieldType::UInt64,
1281                                )
1282                        })
1283                })
1284            }))
1285        && [
1286            ("flags", LegacyFieldType::UInt8),
1287            ("sequence", LegacyFieldType::UInt64),
1288            ("ts_event", LegacyFieldType::UInt64),
1289            ("ts_init", LegacyFieldType::UInt64),
1290        ]
1291        .iter()
1292        .all(|(name, data_type)| {
1293            let Ok(field) = schema.field_with_name(name) else {
1294                return true;
1295            };
1296            (all_nullable || !field.is_nullable())
1297                && legacy_field_type_matches(field.data_type(), *data_type)
1298        })
1299}
1300
1301fn legacy_fixed_list_depth_fingerprint_matches(schema: &Schema) -> bool {
1302    let fixed_list_matches = |name: &str, expected: &DataType| {
1303        schema.field_with_name(name).is_ok_and(|field| {
1304            matches!(
1305                field.data_type(),
1306                DataType::FixedSizeList(item, length)
1307                    if *length == i32::try_from(DEPTH10_LEN).expect("depth length fits i32")
1308                        && item.data_type() == expected
1309            )
1310        })
1311    };
1312    let widths = ["bid_price", "ask_price", "bid_size", "ask_size"]
1313        .iter()
1314        .filter_map(|name| {
1315            let field = schema.field_with_name(name).ok()?;
1316            let DataType::FixedSizeList(item, _) = field.data_type() else {
1317                return None;
1318            };
1319            let DataType::FixedSizeBinary(width @ (8 | 16)) = item.data_type() else {
1320                return None;
1321            };
1322            Some(*width)
1323        })
1324        .collect::<Vec<_>>();
1325    let order_id_fields = ["bid_order_id", "ask_order_id"]
1326        .iter()
1327        .filter(|name| schema.field_with_name(name).is_ok())
1328        .count();
1329
1330    widths.len() == 4
1331        && widths.iter().all(|width| *width == widths[0])
1332        && ["bid_count", "ask_count"]
1333            .iter()
1334            .all(|name| fixed_list_matches(name, &DataType::UInt32))
1335        && matches!(order_id_fields, 0 | 2)
1336        && (order_id_fields == 0
1337            || ["bid_order_id", "ask_order_id"]
1338                .iter()
1339                .all(|name| fixed_list_matches(name, &DataType::UInt64)))
1340}
1341
1342/// Returns whether `name` identifies a legacy catalog timestamp field.
1343#[must_use]
1344pub fn is_timestamp_field(name: &str) -> bool {
1345    name.starts_with("ts_")
1346        || matches!(
1347            name,
1348            "activation_ns"
1349                | "expiration_ns"
1350                | "next_funding_ns"
1351                | "expire_time"
1352                | "event_open_date"
1353                | "market_start_time"
1354        )
1355}
1356
1357/// Returns whether `name` identifies a legacy catalog enum field.
1358#[must_use]
1359pub fn is_legacy_enum_field(name: &str) -> bool {
1360    matches!(name, "action" | "side" | "aggressor_side" | "close_type")
1361}
1362
1363fn legacy_enum_name(name: &str, value: u8) -> Result<String, ArrowError> {
1364    let name_value = match name {
1365        "action" => BookAction::from_u8(value).map(|value| value.to_string()),
1366        "side" => match value {
1367            0 => Some("NO_ORDER_SIDE".to_string()),
1368            1 => Some(OrderSide::Buy.to_string()),
1369            2 => Some(OrderSide::Sell.to_string()),
1370            _ => None,
1371        },
1372        "aggressor_side" => AggressorSide::from_u8(value).map(|value| value.to_string()),
1373        "close_type" => InstrumentCloseType::from_u8(value).map(|value| value.to_string()),
1374        _ => None,
1375    };
1376    name_value.ok_or_else(|| {
1377        ArrowError::CastError(format!(
1378            "Invalid legacy enum value {value} for column '{name}'"
1379        ))
1380    })
1381}
1382
1383// Exact field inventory emitted by the legacy fixed-point schemas.
1384const LEGACY_QUANTITY_FIELDS: &[&str] = &[
1385    "size",
1386    "quantity",
1387    "qty",
1388    "volume",
1389    "bid_size",
1390    "ask_size",
1391    "bid_size_0",
1392    "bid_size_1",
1393    "bid_size_2",
1394    "bid_size_3",
1395    "bid_size_4",
1396    "bid_size_5",
1397    "bid_size_6",
1398    "bid_size_7",
1399    "bid_size_8",
1400    "bid_size_9",
1401    "ask_size_0",
1402    "ask_size_1",
1403    "ask_size_2",
1404    "ask_size_3",
1405    "ask_size_4",
1406    "ask_size_5",
1407    "ask_size_6",
1408    "ask_size_7",
1409    "ask_size_8",
1410    "ask_size_9",
1411];
1412
1413fn is_legacy_quantity_field(name: &str) -> bool {
1414    LEGACY_QUANTITY_FIELDS.contains(&name)
1415}
1416
1417#[cfg(test)]
1418mod tests {
1419    use std::collections::HashMap;
1420
1421    use arrow::{
1422        array::{BinaryArray, BooleanArray},
1423        datatypes::Schema,
1424    };
1425    use nautilus_model::{
1426        data::{FundingRateUpdate, InstrumentClose},
1427        enums::{InstrumentCloseType, MarketStatusAction},
1428    };
1429    use rstest::rstest;
1430
1431    use super::*;
1432    use crate::arrow::{
1433        DecodeFromRecordBatch, DecodeTypedFromRecordBatch, EncodeToRecordBatch, StringColumnRef,
1434    };
1435
1436    #[rstest]
1437    #[case("size", true)]
1438    #[case("volume", true)]
1439    #[case("bid_size_0", true)]
1440    #[case("ask_size_9", true)]
1441    #[case("price", false)]
1442    #[case("quantity_hint", false)]
1443    #[case("bid_size_10", false)]
1444    fn legacy_quantity_fields_use_exact_schema_names(#[case] name: &str, #[case] expected: bool) {
1445        assert_eq!(is_legacy_quantity_field(name), expected);
1446    }
1447
1448    #[rstest]
1449    fn legacy_market_columns_normalize_enums_without_changing_flags() {
1450        let price = 1_i64.to_le_bytes();
1451        let size = 2_u64.to_le_bytes();
1452        let batch = RecordBatch::try_new(
1453            Arc::new(Schema::new(vec![
1454                Field::new("action", DataType::UInt8, false),
1455                Field::new("side", DataType::UInt8, false),
1456                Field::new("price", DataType::FixedSizeBinary(8), false),
1457                Field::new("size", DataType::FixedSizeBinary(8), false),
1458                Field::new("order_id", DataType::UInt64, false),
1459                Field::new("flags", DataType::UInt8, false),
1460                Field::new("sequence", DataType::UInt64, false),
1461                Field::new("ts_event", DataType::UInt64, false),
1462                Field::new("ts_init", DataType::UInt64, false),
1463            ])),
1464            vec![
1465                Arc::new(UInt8Array::from(vec![BookAction::Add as u8])),
1466                Arc::new(UInt8Array::from(vec![OrderSide::Buy as u8])),
1467                Arc::new(
1468                    FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1469                        [Some(price.as_slice())].into_iter(),
1470                        8,
1471                    )
1472                    .unwrap(),
1473                ),
1474                Arc::new(
1475                    FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1476                        [Some(size.as_slice())].into_iter(),
1477                        8,
1478                    )
1479                    .unwrap(),
1480                ),
1481                Arc::new(UInt64Array::from(vec![11])),
1482                Arc::new(UInt8Array::from(vec![7])),
1483                Arc::new(UInt64Array::from(vec![12])),
1484                Arc::new(UInt64Array::from(vec![13])),
1485                Arc::new(UInt64Array::from(vec![1])),
1486            ],
1487        )
1488        .unwrap();
1489
1490        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1491        let action =
1492            StringColumnRef::try_from_array(normalized.column_by_name("action").unwrap().as_ref())
1493                .unwrap();
1494        let side =
1495            StringColumnRef::try_from_array(normalized.column_by_name("side").unwrap().as_ref())
1496                .unwrap();
1497
1498        assert_eq!(action.value(0), "ADD");
1499        assert_eq!(side.value(0), "BUY");
1500        assert_eq!(
1501            normalized.column_by_name("price").unwrap().data_type(),
1502            &fixed_decimal_data_type(),
1503        );
1504        assert_eq!(
1505            normalized.column_by_name("size").unwrap().data_type(),
1506            &fixed_decimal_data_type(),
1507        );
1508        assert_eq!(
1509            normalized.column_by_name("flags").unwrap().data_type(),
1510            &DataType::UInt8,
1511        );
1512        assert_eq!(
1513            normalized.column_by_name("ts_init").unwrap().data_type(),
1514            &timestamp_data_type(),
1515        );
1516    }
1517
1518    #[rstest]
1519    fn unrelated_named_columns_are_not_retyped() {
1520        let batch = RecordBatch::try_new(
1521            Arc::new(Schema::new(vec![
1522                Field::new("side", DataType::UInt8, false),
1523                Field::new("ts_recv", DataType::UInt64, false),
1524            ])),
1525            vec![
1526                Arc::new(UInt8Array::from(vec![127])),
1527                Arc::new(UInt64Array::from(vec![u64::MAX])),
1528            ],
1529        )
1530        .unwrap();
1531
1532        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1533
1534        assert_eq!(normalized, batch);
1535    }
1536
1537    #[rstest]
1538    fn unrelated_fixed_binary_column_is_not_retyped() {
1539        let value = i64::MIN.to_le_bytes();
1540        let batch = RecordBatch::try_new(
1541            Arc::new(Schema::new(vec![Field::new(
1542                "price",
1543                DataType::FixedSizeBinary(8),
1544                false,
1545            )])),
1546            vec![Arc::new(
1547                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1548                    [Some(value.as_slice())].into_iter(),
1549                    8,
1550                )
1551                .unwrap(),
1552            )],
1553        )
1554        .unwrap();
1555
1556        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1557
1558        assert_eq!(normalized, batch);
1559    }
1560
1561    #[rstest]
1562    fn legacy_family_fingerprints_cover_unpinned_schemas() {
1563        let fixed = DataType::FixedSizeBinary(8);
1564        let schemas = [
1565            Schema::new(vec![
1566                Field::new("close_price", fixed.clone(), false),
1567                Field::new("close_type", DataType::UInt8, false),
1568                Field::new("ts_event", DataType::UInt64, false),
1569                Field::new("ts_init", DataType::UInt64, false),
1570            ]),
1571            Schema::new(vec![
1572                Field::new("instrument_id", DataType::Utf8, true),
1573                Field::new("close_type", DataType::Utf8, true),
1574                Field::new("close_price", DataType::Utf8, true),
1575                Field::new("ts_event", DataType::UInt64, true),
1576                Field::new("ts_init", DataType::UInt64, true),
1577            ]),
1578            Schema::new(vec![
1579                Field::new("value", fixed, false),
1580                Field::new("ts_event", DataType::UInt64, false),
1581                Field::new("ts_init", DataType::UInt64, false),
1582            ]),
1583            Schema::new(vec![
1584                Field::new("rate", DataType::Binary, false),
1585                Field::new("interval", DataType::UInt16, true),
1586                Field::new("next_funding_ns", DataType::UInt64, true),
1587                Field::new("ts_event", DataType::UInt64, false),
1588                Field::new("ts_init", DataType::UInt64, false),
1589            ]),
1590            Schema::new(vec![
1591                Field::new("instrument_id", DataType::Utf8, true),
1592                Field::new("action", DataType::Utf8, true),
1593                Field::new("reason", DataType::Utf8, true),
1594                Field::new("trading_event", DataType::Utf8, true),
1595                Field::new("is_trading", DataType::Boolean, true),
1596                Field::new("is_quoting", DataType::Boolean, true),
1597                Field::new("is_short_sell_restricted", DataType::Boolean, true),
1598                Field::new("ts_event", DataType::UInt64, true),
1599                Field::new("ts_init", DataType::UInt64, true),
1600            ]),
1601        ];
1602
1603        assert!(schemas.iter().all(is_nautilus_legacy_schema));
1604    }
1605
1606    #[rstest]
1607    fn legacy_family_fingerprint_accepts_dictionary_string_fields() {
1608        let schema = Schema::new(vec![
1609            Field::new("price", DataType::FixedSizeBinary(8), false),
1610            Field::new("size", DataType::FixedSizeBinary(8), false),
1611            Field::new("aggressor_side", DataType::UInt8, false),
1612            Field::new(
1613                "trade_id",
1614                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
1615                false,
1616            ),
1617            Field::new("ts_event", DataType::UInt64, false),
1618            Field::new("ts_init", DataType::UInt64, false),
1619            Field::new(
1620                KEY_IDENTIFIER,
1621                DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)),
1622                false,
1623            ),
1624        ]);
1625
1626        assert!(is_nautilus_legacy_schema(&schema));
1627    }
1628
1629    #[rstest]
1630    fn legacy_flat_depth_fingerprint_accepts_all_nullable_fields() {
1631        let fixed = DataType::FixedSizeBinary(8);
1632        let mut fields = Vec::new();
1633
1634        for side in ["bid", "ask"] {
1635            for level in 0..DEPTH10_LEN {
1636                fields.extend([
1637                    Field::new(format!("{side}_price_{level}"), fixed.clone(), true),
1638                    Field::new(format!("{side}_size_{level}"), fixed.clone(), true),
1639                    Field::new(format!("{side}_count_{level}"), DataType::UInt32, true),
1640                ]);
1641            }
1642        }
1643        fields.extend([
1644            Field::new("flags", DataType::UInt8, true),
1645            Field::new("sequence", DataType::UInt64, true),
1646            Field::new("ts_event", DataType::UInt64, true),
1647            Field::new("ts_init", DataType::UInt64, true),
1648        ]);
1649        let schema = Schema::new(fields);
1650
1651        assert!(super::legacy_flat_depth_fingerprint_matches(&schema));
1652    }
1653
1654    #[rstest]
1655    fn legacy_fixed_list_depth_fingerprint_accepts_missing_order_ids() {
1656        let fixed = Arc::new(Field::new("item", DataType::FixedSizeBinary(8), false));
1657        let count = Arc::new(Field::new("item", DataType::UInt32, false));
1658        let schema = Schema::new(vec![
1659            Field::new(
1660                "bid_price",
1661                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1662                false,
1663            ),
1664            Field::new(
1665                "ask_price",
1666                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1667                false,
1668            ),
1669            Field::new(
1670                "bid_size",
1671                DataType::FixedSizeList(Arc::clone(&fixed), 10),
1672                false,
1673            ),
1674            Field::new("ask_size", DataType::FixedSizeList(fixed, 10), false),
1675            Field::new(
1676                "bid_count",
1677                DataType::FixedSizeList(Arc::clone(&count), 10),
1678                false,
1679            ),
1680            Field::new("ask_count", DataType::FixedSizeList(count, 10), false),
1681        ]);
1682
1683        assert!(is_nautilus_legacy_schema(&schema));
1684    }
1685
1686    #[rstest]
1687    fn metadata_only_schema_is_not_a_legacy_family() {
1688        let schema = Schema::new_with_metadata(
1689            vec![
1690                Field::new("side", DataType::UInt8, false),
1691                Field::new("ts_recv", DataType::UInt64, false),
1692            ],
1693            [("type_name".to_string(), "CustomData".to_string())].into(),
1694        );
1695
1696        assert!(!is_nautilus_legacy_schema(&schema));
1697    }
1698
1699    #[rstest]
1700    fn known_metadata_families_require_and_normalize_their_legacy_shape() {
1701        let schemas = [
1702            Schema::new_with_metadata(
1703                vec![
1704                    Field::new("rate", DataType::Binary, false),
1705                    Field::new("interval", DataType::UInt16, true),
1706                    Field::new("next_funding_ns", DataType::UInt64, true),
1707                    Field::new("ts_event", DataType::UInt64, false),
1708                    Field::new("ts_init", DataType::UInt64, false),
1709                ],
1710                [("type".to_string(), "FundingRateUpdate".to_string())].into(),
1711            ),
1712            Schema::new_with_metadata(
1713                vec![
1714                    Field::new("instrument_id", DataType::Utf8, true),
1715                    Field::new("action", DataType::Utf8, true),
1716                    Field::new("reason", DataType::Utf8, true),
1717                    Field::new("trading_event", DataType::Utf8, true),
1718                    Field::new("is_trading", DataType::Boolean, true),
1719                    Field::new("is_quoting", DataType::Boolean, true),
1720                    Field::new("is_short_sell_restricted", DataType::Boolean, true),
1721                    Field::new("ts_event", DataType::UInt64, true),
1722                    Field::new("ts_init", DataType::UInt64, true),
1723                ],
1724                [("type".to_string(), "InstrumentStatus".to_string())].into(),
1725            ),
1726            Schema::new_with_metadata(
1727                vec![
1728                    Field::new("instrument_id", DataType::Utf8, false),
1729                    Field::new("delta", DataType::Float64, false),
1730                    Field::new("gamma", DataType::Float64, false),
1731                    Field::new("vega", DataType::Float64, false),
1732                    Field::new("theta", DataType::Float64, false),
1733                    Field::new("rho", DataType::Float64, false),
1734                    Field::new("mark_iv", DataType::Float64, true),
1735                    Field::new("bid_iv", DataType::Float64, true),
1736                    Field::new("ask_iv", DataType::Float64, true),
1737                    Field::new("underlying_price", DataType::Float64, true),
1738                    Field::new("open_interest", DataType::Float64, true),
1739                    Field::new("ts_event", DataType::UInt64, false),
1740                    Field::new("ts_init", DataType::UInt64, false),
1741                    Field::new("convention", DataType::Utf8, false),
1742                ],
1743                [("type".to_string(), "OptionGreeks".to_string())].into(),
1744            ),
1745        ];
1746
1747        for schema in schemas {
1748            assert!(is_nautilus_legacy_schema(&schema));
1749            let normalized =
1750                normalize_legacy_fixed_columns(&RecordBatch::new_empty(Arc::new(schema))).unwrap();
1751
1752            assert_eq!(
1753                normalized
1754                    .schema()
1755                    .field_with_name("ts_init")
1756                    .unwrap()
1757                    .data_type(),
1758                &timestamp_data_type(),
1759            );
1760        }
1761    }
1762
1763    #[rstest]
1764    fn near_match_with_foreign_field_type_is_not_retyped() {
1765        let schema = Arc::new(Schema::new(vec![
1766            Field::new("action", DataType::UInt8, false),
1767            Field::new("side", DataType::UInt8, false),
1768            Field::new("price", DataType::FixedSizeBinary(8), false),
1769            Field::new("size", DataType::FixedSizeBinary(8), false),
1770            Field::new("order_id", DataType::UInt64, false),
1771            Field::new("flags", DataType::UInt8, false),
1772            Field::new("sequence", DataType::Int64, false),
1773            Field::new("ts_event", DataType::UInt64, false),
1774            Field::new("ts_init", DataType::UInt64, false),
1775        ]));
1776        let batch = RecordBatch::new_empty(schema);
1777
1778        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1779
1780        assert_eq!(normalized, batch);
1781    }
1782
1783    #[rstest]
1784    fn legacy_price_error_fails_with_field_and_row() {
1785        let fixed = |raw: [u8; 8]| -> ArrayRef {
1786            Arc::new(
1787                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1788                    [Some(raw.as_slice())].into_iter(),
1789                    8,
1790                )
1791                .unwrap(),
1792            )
1793        };
1794        let batch = RecordBatch::try_new(
1795            Arc::new(Schema::new_with_metadata(
1796                vec![
1797                    Field::new("bid_price", DataType::FixedSizeBinary(8), false),
1798                    Field::new("ask_price", DataType::FixedSizeBinary(8), false),
1799                    Field::new("bid_size", DataType::FixedSizeBinary(8), false),
1800                    Field::new("ask_size", DataType::FixedSizeBinary(8), false),
1801                    Field::new("ts_event", DataType::UInt64, false),
1802                    Field::new("ts_init", DataType::UInt64, false),
1803                ],
1804                [("type".to_string(), "QuoteTick".to_string())].into(),
1805            )),
1806            vec![
1807                fixed(i64::MIN.to_le_bytes()),
1808                fixed(2_i64.to_le_bytes()),
1809                fixed(3_u64.to_le_bytes()),
1810                fixed(4_u64.to_le_bytes()),
1811                Arc::new(UInt64Array::from(vec![5])),
1812                Arc::new(UInt64Array::from(vec![6])),
1813            ],
1814        )
1815        .unwrap();
1816
1817        let error = normalize_legacy_fixed_columns(&batch).unwrap_err();
1818
1819        assert_eq!(
1820            error.to_string(),
1821            format!(
1822                "Cast error: Legacy price column 'bid_price' contains PRICE_ERROR raw value {} at row 0",
1823                i64::MIN,
1824            ),
1825        );
1826    }
1827
1828    #[rstest]
1829    fn metadata_family_near_match_passes_through_unchanged() {
1830        let raw = i64::MIN.to_le_bytes();
1831        let batch = RecordBatch::try_new(
1832            Arc::new(Schema::new_with_metadata(
1833                vec![Field::new("bid_price", DataType::FixedSizeBinary(8), false)],
1834                [("type".to_string(), "QuoteTick".to_string())].into(),
1835            )),
1836            vec![Arc::new(
1837                FixedSizeBinaryArray::try_from_sparse_iter_with_size(
1838                    [Some(raw.as_slice())].into_iter(),
1839                    8,
1840                )
1841                .unwrap(),
1842            )],
1843        )
1844        .unwrap();
1845
1846        let normalized = normalize_legacy_fixed_columns(&batch).unwrap();
1847
1848        assert_eq!(normalized, batch);
1849    }
1850
1851    #[rstest]
1852    fn schema_fingerprint_ignores_metadata_and_includes_nullability() {
1853        let fields = vec![Field::new("value", DataType::Utf8, false)];
1854        let schema1 = Schema::new_with_metadata(
1855            fields.clone(),
1856            HashMap::from([("a".to_string(), "1".to_string())]),
1857        );
1858        let schema2 =
1859            Schema::new_with_metadata(fields, HashMap::from([("b".to_string(), "2".to_string())]));
1860        let nullable = Schema::new(vec![Field::new("value", DataType::Utf8, true)]);
1861
1862        assert_eq!(schema_fingerprint(&schema1), schema_fingerprint(&schema2));
1863        assert_ne!(schema_fingerprint(&schema1), schema_fingerprint(&nullable));
1864    }
1865
1866    #[rstest]
1867    fn current_schema_passes_through() {
1868        let batch = RecordBatch::new_empty(Arc::new(QuoteTick::get_schema(None)));
1869        let result = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap();
1870
1871        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1872        assert_eq!(result.batches.len(), 1);
1873    }
1874
1875    #[rstest]
1876    fn parquet_schema_with_plain_dictionary_strings_passes_through() {
1877        let fields = TradeTick::get_schema(None)
1878            .fields()
1879            .iter()
1880            .filter(|field| field.name() != KEY_IDENTIFIER)
1881            .map(|field| match field.data_type() {
1882                DataType::Dictionary(..) => field.as_ref().clone().with_data_type(DataType::Utf8),
1883                _ => field.as_ref().clone(),
1884            })
1885            .collect::<Vec<_>>();
1886
1887        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1888        let result = transcode_legacy_record_batch("trades", "trades.parquet", batch).unwrap();
1889
1890        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1891        assert_eq!(result.batches.len(), 1);
1892    }
1893
1894    #[rstest]
1895    fn reordered_nullable_schema_with_identity_column_passes_through() {
1896        let mut fields = QuoteTick::get_schema(None)
1897            .fields()
1898            .iter()
1899            .rev()
1900            .map(|field| field.as_ref().clone().with_nullable(true))
1901            .collect::<Vec<_>>();
1902        fields.push(Field::new(KEY_INSTRUMENT_ID, DataType::Utf8, false));
1903        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1904        let result = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap();
1905
1906        assert_eq!(result.kind, LegacyTranscodeKind::PassThrough);
1907        assert_eq!(result.batches.len(), 1);
1908    }
1909
1910    #[rstest]
1911    #[case::extra_column(Some(Field::new("venue", DataType::Utf8, true)), None)]
1912    #[case::missing_column(None, Some("ask_size"))]
1913    fn schema_with_unexpected_columns_is_rejected(
1914        #[case] extra: Option<Field>,
1915        #[case] missing: Option<&str>,
1916    ) {
1917        let mut fields = QuoteTick::get_schema(None)
1918            .fields()
1919            .iter()
1920            .filter(|field| Some(field.name().as_str()) != missing)
1921            .map(|field| field.as_ref().clone())
1922            .collect::<Vec<_>>();
1923        fields.extend(extra);
1924        let batch = RecordBatch::new_empty(Arc::new(Schema::new(fields)));
1925        let error = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap_err();
1926
1927        assert!(matches!(error, LegacyArrowError::UnknownSchema { .. }));
1928    }
1929
1930    #[rstest]
1931    fn unknown_registered_schema_is_rejected() {
1932        let batch = RecordBatch::new_empty(Arc::new(Schema::new(vec![Field::new(
1933            "unexpected",
1934            DataType::UInt64,
1935            false,
1936        )])));
1937        let error = transcode_legacy_record_batch("quotes", "quotes.parquet", batch).unwrap_err();
1938
1939        assert!(matches!(error, LegacyArrowError::UnknownSchema { .. }));
1940        assert!(error.to_string().contains("quotes.parquet"));
1941    }
1942
1943    #[rstest]
1944    #[case(false)]
1945    #[case(true)]
1946    fn instrument_status_fields_are_reordered(#[case] normalized: bool) {
1947        let schema = Arc::new(legacy_instrument_status_schema());
1948        let batch = RecordBatch::try_new(
1949            schema,
1950            vec![
1951                Arc::new(StringArray::from(vec!["AAPL.XNAS"])),
1952                Arc::new(StringArray::from(vec!["TRADING"])),
1953                Arc::new(StringArray::from(vec![Some("Normal")])),
1954                Arc::new(StringArray::from(vec![Some("OPEN")])),
1955                Arc::new(BooleanArray::from(vec![Some(true)])),
1956                Arc::new(BooleanArray::from(vec![Some(true)])),
1957                Arc::new(BooleanArray::from(vec![Some(false)])),
1958                Arc::new(UInt64Array::from(vec![1])),
1959                Arc::new(UInt64Array::from(vec![2])),
1960            ],
1961        )
1962        .unwrap();
1963
1964        // Migration normalizes legacy timestamps before resolving the schema
1965        let batch = if normalized {
1966            normalize_legacy_fixed_columns(&batch).unwrap()
1967        } else {
1968            batch
1969        };
1970
1971        let result =
1972            transcode_legacy_record_batch("instrument_status", "status.parquet", batch).unwrap();
1973        let output = &result.batches[0];
1974
1975        assert_eq!(result.kind, LegacyTranscodeKind::InstrumentStatusV1);
1976        assert_eq!(
1977            output
1978                .schema()
1979                .fields()
1980                .iter()
1981                .map(|field| field.name().as_str())
1982                .collect::<Vec<_>>(),
1983            vec![
1984                "action",
1985                "ts_event",
1986                "ts_init",
1987                "reason",
1988                "trading_event",
1989                "is_trading",
1990                "is_quoting",
1991                "is_short_sell_restricted",
1992                "identifier",
1993            ]
1994        );
1995        assert_eq!(
1996            output
1997                .schema()
1998                .field_with_name("action")
1999                .unwrap()
2000                .data_type(),
2001            &enum_dictionary_data_type()
2002        );
2003        let identifiers = output
2004            .column_by_name(KEY_IDENTIFIER)
2005            .unwrap()
2006            .as_any()
2007            .downcast_ref::<StringArray>()
2008            .unwrap();
2009        assert_eq!(identifiers.value(0), "AAPL.XNAS");
2010
2011        // Migration attaches the row identifier as instrument metadata before decoding
2012        let mut metadata = output.schema().metadata().clone();
2013        metadata.insert(
2014            KEY_INSTRUMENT_ID.to_string(),
2015            identifiers.value(0).to_string(),
2016        );
2017        let decoded = InstrumentStatus::decode_typed_batch(&metadata, output.clone()).unwrap();
2018        assert_eq!(decoded[0].action, MarketStatusAction::Trading);
2019        assert_eq!(decoded[0].instrument_id.to_string(), "AAPL.XNAS");
2020        assert_eq!(decoded[0].ts_event.as_u64(), 1);
2021        assert_eq!(decoded[0].ts_init.as_u64(), 2);
2022    }
2023
2024    #[rstest]
2025    fn funding_rate_is_decoded_and_interval_is_widened() {
2026        let schema = Arc::new(Schema::new_with_metadata(
2027            legacy_funding_rate_schema()
2028                .fields()
2029                .iter()
2030                .cloned()
2031                .collect::<Vec<_>>(),
2032            HashMap::from([(
2033                KEY_INSTRUMENT_ID.to_string(),
2034                "BTCUSDT-PERP.BINANCE".to_string(),
2035            )]),
2036        ));
2037        let batch = RecordBatch::try_new(
2038            schema,
2039            vec![
2040                Arc::new(BinaryArray::from(vec![b"\"0.0001\"".as_slice()])),
2041                Arc::new(UInt16Array::from(vec![Some(480)])),
2042                Arc::new(UInt64Array::from(vec![Some(9)])),
2043                Arc::new(UInt64Array::from(vec![1])),
2044                Arc::new(UInt64Array::from(vec![2])),
2045            ],
2046        )
2047        .unwrap();
2048
2049        let result =
2050            transcode_legacy_record_batch("funding_rates", "funding.parquet", batch).unwrap();
2051        let output = &result.batches[0];
2052        let decoded =
2053            FundingRateUpdate::decode_typed_batch(output.schema().metadata(), output.clone())
2054                .unwrap();
2055
2056        assert_eq!(result.kind, LegacyTranscodeKind::FundingRateUpdateV1);
2057        assert_eq!(decoded[0].instrument_id.to_string(), "BTCUSDT-PERP.BINANCE");
2058        assert_eq!(decoded[0].rate.to_string(), "0.0001");
2059        assert_eq!(decoded[0].interval, Some(480));
2060        assert_eq!(
2061            decoded[0].next_funding_ns.map(|value| value.as_u64()),
2062            Some(9)
2063        );
2064    }
2065
2066    #[rstest]
2067    #[case(false)]
2068    #[case(true)]
2069    fn instrument_close_is_split_and_decoded(#[case] normalized: bool) {
2070        let schema = Arc::new(legacy_instrument_close_schema());
2071        let batch = RecordBatch::try_new(
2072            schema,
2073            vec![
2074                Arc::new(StringArray::from(vec![
2075                    "AUD/USD.SIM",
2076                    "GBP/USD.SIM",
2077                    "AUD/USD.SIM",
2078                ])),
2079                Arc::new(StringArray::from(vec![
2080                    "END_OF_SESSION",
2081                    "CONTRACT_EXPIRED",
2082                    "END_OF_SESSION",
2083                ])),
2084                Arc::new(StringArray::from(vec!["1.0500", "2.1000", "1.0600"])),
2085                Arc::new(UInt64Array::from(vec![1, 2, 3])),
2086                Arc::new(UInt64Array::from(vec![4, 5, 6])),
2087            ],
2088        )
2089        .unwrap();
2090
2091        let batch = if normalized {
2092            normalize_legacy_fixed_columns(&batch).unwrap()
2093        } else {
2094            batch
2095        };
2096
2097        let result =
2098            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap();
2099        let decoded = result
2100            .batches
2101            .iter()
2102            .flat_map(|batch| {
2103                InstrumentClose::decode_batch(batch.schema().metadata(), batch.clone()).unwrap()
2104            })
2105            .collect::<Vec<_>>();
2106
2107        assert_eq!(
2108            decoded
2109                .iter()
2110                .map(|close| (close.ts_event.as_u64(), close.ts_init.as_u64()))
2111                .collect::<Vec<_>>(),
2112            vec![(1, 4), (3, 6), (2, 5)]
2113        );
2114        assert_eq!(result.kind, LegacyTranscodeKind::InstrumentCloseV1);
2115        assert_eq!(result.batches.len(), 2);
2116        assert_eq!(
2117            decoded
2118                .iter()
2119                .map(|close| close.instrument_id.to_string())
2120                .collect::<Vec<_>>(),
2121            vec!["AUD/USD.SIM", "AUD/USD.SIM", "GBP/USD.SIM"]
2122        );
2123        assert_eq!(
2124            decoded
2125                .iter()
2126                .map(|close| close.close_price.to_string())
2127                .collect::<Vec<_>>(),
2128            vec!["1.0500", "1.0600", "2.1000"]
2129        );
2130        assert_eq!(
2131            decoded
2132                .iter()
2133                .map(|close| close.close_type)
2134                .collect::<Vec<_>>(),
2135            vec![
2136                InstrumentCloseType::EndOfSession,
2137                InstrumentCloseType::EndOfSession,
2138                InstrumentCloseType::ContractExpired,
2139            ]
2140        );
2141
2142        // Transcoded files must consolidate with files written by the current encoder
2143        for batch in &result.batches {
2144            let schema = batch.schema();
2145            let closes = InstrumentClose::decode_batch(schema.metadata(), batch.clone()).unwrap();
2146            let fresh = InstrumentClose::encode_batch(&closes[0].metadata(), &closes).unwrap();
2147            assert_eq!(fresh.schema(), schema);
2148        }
2149    }
2150
2151    #[rstest]
2152    fn instrument_close_rejects_mixed_precision() {
2153        let batch = RecordBatch::try_new(
2154            Arc::new(legacy_instrument_close_schema()),
2155            vec![
2156                Arc::new(StringArray::from(vec!["AUD/USD.SIM", "AUD/USD.SIM"])),
2157                Arc::new(StringArray::from(vec!["END_OF_SESSION", "END_OF_SESSION"])),
2158                Arc::new(StringArray::from(vec!["1.0500", "1.060"])),
2159                Arc::new(UInt64Array::from(vec![1, 2])),
2160                Arc::new(UInt64Array::from(vec![3, 4])),
2161            ],
2162        )
2163        .unwrap();
2164
2165        let error =
2166            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap_err();
2167
2168        assert!(error.to_string().contains("close.parquet"));
2169        assert!(error.to_string().contains("precision conflict at row 1"));
2170    }
2171
2172    #[rstest]
2173    fn instrument_close_rejects_unknown_close_type() {
2174        let batch = RecordBatch::try_new(
2175            Arc::new(legacy_instrument_close_schema()),
2176            vec![
2177                Arc::new(StringArray::from(vec!["AUD/USD.SIM"])),
2178                Arc::new(StringArray::from(vec!["UNKNOWN"])),
2179                Arc::new(StringArray::from(vec!["1.0500"])),
2180                Arc::new(UInt64Array::from(vec![1])),
2181                Arc::new(UInt64Array::from(vec![2])),
2182            ],
2183        )
2184        .unwrap();
2185
2186        let error =
2187            transcode_legacy_record_batch("instrument_closes", "close.parquet", batch).unwrap_err();
2188
2189        assert!(error.to_string().contains("close.parquet"));
2190        assert!(
2191            error
2192                .to_string()
2193                .contains("unknown close_type \"UNKNOWN\" at row 0")
2194        );
2195    }
2196}