pola-rs/polars · error

not implemented

Error message

not implemented

What it means

FlightConsumer::consume (crates/polars-arrow/src/io/ipc/read/flight.rs) decodes one Arrow IPC stream message per call and only handles Schema (rejected as unexpected), DictionaryBatch and RecordBatch headers. Any other MessageHeaderRef variant — flatbuffer union tag NONE (0) or a message kind this reader predates — falls into _ => unimplemented!() at line 372 and panics mid-stream. FlightstreamConsumer is built on top of it, so async flight streams hit the same arm.

Source

Thrown at crates/polars-arrow/src/io/ipc/read/flight.rs:372

                    )
                    .map(Some)
                } else {
                    // Needed to memory map.
                    let arrow_data = Arc::new(msg.arrow_data);
                    unsafe {
                        mmap_record(
                            &self.md.schema,
                            &self.md.ipc_schema.fields,
                            arrow_data,
                            batch,
                            0,
                            &self.dictionaries,
                        )
                        .map(Some)
                    }
                }
            },
            _ => unimplemented!(),
        }
    }
}

pub struct FlightstreamConsumer<S: Stream<Item = PolarsResult<EncodedData>> + Unpin> {
    inner: FlightConsumer,
    stream: S,
}

impl<S: Stream<Item = PolarsResult<EncodedData>> + Unpin> FlightstreamConsumer<S> {
    pub async fn new(mut stream: S) -> PolarsResult<Self> {
        let Some(first) = stream.next().await else {
            polars_bail!(ComputeError: "expected the schema")
        };
        let first = first?;

        Ok(FlightstreamConsumer {
            inner: FlightConsumer::new(first)?,

View on GitHub (pinned to df599052da)

Solutions

  1. Validate the header kind before calling consume and skip or error on unknown messages
  2. Pin producer and consumer to compatible Arrow IPC versions
  3. Upgrade polars-arrow once the message kind is supported
  4. Treat unknown headers as a corrupt stream: surface an error and abort the session instead of unwinding

Example fix

// before
let batch = consumer.consume(msg)?; // panics on unknown MessageHeader

// after
let header = arrow_format::ipc::MessageRef::read_as_root(&msg.ipc_message)
    .map_err(|e| polars_err!(oos = OutOfSpecKind::InvalidFlatbufferMessage(e)))?
    .header()
    .map_err(|e| polars_err!(oos = OutOfSpecKind::InvalidFlatbufferHeader(e)))?;
match header {
    Some(MessageHeaderRef::DictionaryBatch(_)) | Some(MessageHeaderRef::RecordBatch(_)) =>
        consumer.consume(msg),
    _ => polars_bail!(ComputeError: "unsupported IPC message header in flight stream"),
}
Defensive patterns

Strategy: validation

Validate before calling

let header = arrow_format::ipc::MessageRef::read_as_root(&msg.ipc_message)
    .map_err(|e| polars_err!(oos = OutOfSpecKind::InvalidFlatbufferMessage(e)))?
    .header()
    .map_err(|e| polars_err!(oos = OutOfSpecKind::InvalidFlatbufferHeader(e)))?;
match header {
    Some(MessageHeaderRef::DictionaryBatch(_)) | Some(MessageHeaderRef::RecordBatch(_)) => consumer.consume(msg),
    _ => polars_bail!(ComputeError: "unsupported IPC message header in flight stream"),
}

Try / catch

let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| consumer.consume(msg)));
let batch = match res {
    Ok(v) => v?,
    Err(_) => polars_bail!(ComputeError: "flight stream contained an unsupported message kind; treat stream as corrupt"),
};

Prevention

When it happens

Trigger: consumer.consume(msg) where MessageRef::read_as_root(...).header() yields something other than the three known kinds: truncated/corrupt flatbuffers, or a producer emitting message types this reader was never written to process.

Common situations: Arrow Flight / Flight SQL streams from newer servers; custom EncodedData producers; network truncation or proxy corruption producing garbage message headers.

Related errors

AI-assisted analysis of pola-rs/polars@df599052da (2026-08-16). Data as JSON: /api/errors/b641df2e30eb38ba. Report an issue: GitHub.