nautechsystems/nautilus_trader · error · anyhow::Error

Failed to concatenate stream batches: {e}

Error message

Failed to concatenate stream batches: {e}

What it means

This error wraps an Arrow `concat_batches` failure while `apply_stream_conversion_transforms` merges all per-message RecordBatches from a feather/nautilus stream file into one batch before writing it to the catalog as parquet. `concat_batches` fails when the input batches cannot be combined under a single schema — typically because a batch's schema differs from `batches[0].schema()` (column order, field types, or metadata-driven field changes). It is an upstream data-shape problem, not an I/O failure.

Source

Thrown at crates/persistence/src/backend/catalog.rs:4088

            for batch in &mut batches {
                let mut columns = batch.columns().to_vec();
                columns[ts_init_idx] = columns[ts_event_idx].clone();

                *batch = RecordBatch::try_new(schema.clone(), columns).map_err(|e| {
                    anyhow::anyhow!("Failed to create stream conversion batch: {e}")
                })?;
            }
        } else if metadata_changed {
            for batch in &mut batches {
                *batch = RecordBatch::try_new(schema.clone(), batch.columns().to_vec()).map_err(
                    |e| anyhow::anyhow!("Failed to create stream conversion batch: {e}"),
                )?;
            }
        }

        let mut batch = concat_batches(&schema, batches.iter())
            .map_err(|e| anyhow::anyhow!("Failed to concatenate stream batches: {e}"))?;

        if batch.num_rows() == 0 {
            return Ok(None);
        }

        if !Self::is_record_batch_monotonic_by_ts_init(&batch)? {
            let indices = sort_to_indices(
                Self::ts_init_array(&batch)?,
                Some(SortOptions {
                    descending: false,
                    nulls_first: false,
                }),
                None,
            )
            .map_err(|e| anyhow::anyhow!("Failed to sort stream conversion batch: {e}"))?;
            batch = take_record_batch(&batch, &indices)
                .map_err(|e| anyhow::anyhow!("Failed to reorder stream conversion batch: {e}"))?;
        }

View on GitHub (pinned to 18893faf8b)

Solutions

1. Re-extract or re-generate the source feather/stream files so all batches share one schema (re-run the stream session or re-export with the current nautilus version).
2. Check the Arrow error text in {e}: it names the exact schema mismatch (field order/type); if it is a column-order difference, normalize batch schemas before conversion with `RecordBatch::try_new(schema, ...)` per batch.
3. Convert one data-type directory at a time to isolate which file contains the mismatched batch, then exclude/fix that file.
4. Upgrade both the writer and reader sides of the data to matching nautilus/arrow versions so serialization formats agree.

Example fix

// before: concatenating raw heterogeneous batches
let batch = concat_batches(&schema, batches.iter())
    .map_err(|e| anyhow::anyhow!("Failed to concatenate stream batches: {e}"))?;

// after: normalize each batch to the first batch's schema first
let schema = batches[0].schema();
let batches: Vec<RecordBatch> = batches
    .into_iter()
    .map(|b| {
        if b.schema() != schema {
            RecordBatch::try_new(schema.clone(), b.columns().to_vec())
                .expect("batch normalized to common schema")
        } else {
            b
        }
    })
    .collect();
let batch = concat_batches(&schema, batches.iter())
    .map_err(|e| anyhow::anyhow!("Failed to concatenate stream batches: {e}"))?;

// Rust: verify all batches share an identical schema before conversion
fn schemas_match(batches: &[RecordBatch]) -> bool {
    if batches.is_empty() { return false; }
    let first = batches[0].schema();
    batches.iter().all(|b| b.schema() == first)
}
if !schemas_match(&batches) {
    return Err(anyhow::anyhow!("source batches have heterogeneous schemas; re-extract data"));
}

// Rust
match convert_stream_to_catalog(&feather_path) {
    Ok(()) => info!("converted {feather_path}"),
    Err(e) if e.to_string().contains("Failed to concatenate stream batches") => {
        warn!("skipping {feather_path}: schema mismatch ({e})");
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: Called from stream-to-catalog conversion (e.g. `convert_catalog` / feather import path) when the Vec<RecordBatch> read from a `.feather` stream file contains batches with heterogeneous schemas — e.g. mixed arrow versions wrote the file, or the ts_event→ts_init substitution (`use_ts_event_for_ts_init`) rebuilt batches against a schema that other batches no longer match, or the bar_type metadata rewrite produced a schema inconsistent with later batches.

Common situations: Converting stream/feather data written by an older nautilus version into a newer catalog where the Arrow schema for a data type changed; hand-edited or partially-written feather files; a stream directory mixing data files of different schema versions; corrupted feather files read as batches with mismatched fields.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• height must be positive, was {self.height}
• Failed to reorder stream conversion batch: {e}
• Failed to merge custom data type metadata: {e}
• pandas is required for report generation; install it with `p
• pandas is required for visualization; install it with `pip i

AI-assisted analysis of nautechsystems/nautilus_trader@18893faf8b (2026-09-08). Data as JSON: /api/errors/aad0f088636bb2b4. Report an issue: GitHub.