Implementation:ArroyoSystems Arroyo Format Deserializer

Overview

ArrowDeserializer is the central deserialization engine that converts raw bytes from various wire formats (JSON, Avro, Protobuf, raw strings, raw bytes) into Arrow RecordBatch instances. It handles framing, bad data policies, metadata fields, and buffered batch construction.

Description

The module defines:

• ArrowDeserializer: The main struct that manages deserialization state including format configuration, schema, framing, bad data policy, schema resolver, and an internal ContextBuffer for accumulating rows before flushing.

• FramingIterator: Splits raw byte messages according to a framing configuration (e.g., newline-delimited), yielding individual message slices.

• ContextBuffer: An internal buffer that holds Arrow array builders for each column. It tracks buffer size and creation time to decide when to flush (based on row count or age).

• FieldValueType: An enum representing typed metadata field values (Int64, UInt64, Int32, String, Bytes) that can be injected into deserialized records.

Key features:

• Supports multiple formats via the Format enum dispatch
• Handles Confluent Schema Registry integration for Avro
• Supports protobuf deserialization via descriptor pools
• Configurable bad data handling (fail, drop, or dead letter)
• Metadata field injection from connector sources
• Timestamp assignment from event time or system time

Usage

ArrowDeserializer is instantiated per source operator and called on each incoming message. After accumulating enough rows, the caller invokes flush_buffer() to obtain a complete RecordBatch.

Code Reference

Source Location

crates/arroyo-formats/src/de.rs

Signature

pub struct ArrowDeserializer {
    // format, schema, framing, bad_data, schema_resolver, buffer, etc.
}

impl ArrowDeserializer {
    pub fn new(
        format: Format,
        framing: Option<Arc<Framing>>,
        schema: Arc<ArroyoSchema>,
        metadata_fields: &[MetadataField],
        bad_data: BadData,
    ) -> Self

    pub fn with_schema_resolver(
        format: Format,
        framing: Option<Arc<Framing>>,
        schema: Arc<ArroyoSchema>,
        metadata_fields: &[MetadataField],
        bad_data: BadData,
        resolver: Arc<dyn SchemaResolver + Sync>,
    ) -> Self

    pub async fn deserialize_slice(
        &mut self,
        msg: &[u8],
        timestamp: SystemTime,
        metadata: Option<&HashMap<String, FieldValueType<'_>>>,
    ) -> Vec<DataflowError>

    pub fn flush_buffer(&mut self) -> (Option<RecordBatch>, Vec<DataflowError>)
}

pub struct FramingIterator<'a> { /* ... */ }

pub enum FieldValueType<'a> {
    Int64(Option<i64>),
    UInt64(Option<u64>),
    Int32(Option<i32>),
    String(Option<&'a str>),
    Bytes(Option<&'a [u8]>),
}

Import

use arroyo_formats::de::{ArrowDeserializer, FramingIterator, FieldValueType};

I/O Contract

Inputs

Name	Type	Description
msg	&[u8]	Raw message bytes from a source connector
timestamp	SystemTime	System time for the message (used if no event time field)
metadata	Option<&HashMap<String, FieldValueType>>	Optional metadata fields from the connector

Outputs

Name	Type	Description
batch	Option<RecordBatch>	Flushed Arrow record batch (None if buffer is empty)
errors	Vec<DataflowError>	Deserialization errors encountered during processing

Usage Examples

let mut deserializer = ArrowDeserializer::new(
    Format::Json(JsonFormat::default()),
    Some(Arc::new(Framing::Newline(NewlineDelimitedFraming {}))),
    schema.clone(),
    &[],
    BadData::Drop {},
);

// Process incoming messages
let errors = deserializer.deserialize_slice(raw_bytes, SystemTime::now(), None).await;

// Flush when ready
if let (Some(batch), flush_errors) = deserializer.flush_buffer() {
    // process batch
}

Related Pages

• ArroyoSystems_Arroyo_Format_Serializer - Complementary serialization from Arrow to wire formats
• ArroyoSystems_Arroyo_Avro_Deserializer - Avro-specific deserialization used internally
• ArroyoSystems_Arroyo_Json_Schema_Module - JSON schema handling