arrow_mlake.ipc

The C Data Interface hands buffers to another library in the same process. IPC is for everything else: a file, a socket, a shared mapping another process reads. pyarrow's ipc.open_stream reads what IpcWriter writes, and read_ipc_stream reads what pyarrow's ipc.new_stream writes.

var w = IpcWriter(batch)      # the schema comes from the first batch
w.write(batch)
var bytes = w^.finish()

var s = read_ipc_stream(Span(bytes))
print(len(s.batches), "batches of", s.schema.num_columns(), "columns")

The framing

Each message in a stream is:

ffffffff                  continuation marker
<int32 metadata_size>     little-endian, a multiple of 8
<flatbuffer Message>      padded out to metadata_size
<body>                    the buffers, each padded to a multiple of 8

and the stream ends with a continuation marker and a zero length. The first message is the Schema; every later one is a RecordBatch whose nodes and buffers list each array in pre-order, parent before children. The reader also accepts the pre-0.15 framing without the continuation marker.

What is supported

Every type ArrayData holds: null, bool, the eight integers, the three floats, utf8 and binary and their large forms, fixed-size binary, decimal128, date32, time32/64, timestamps with a unit and zone, list, large list, struct and map, nested to any depth. Fields carry their nullability and two metadata keys, ARROW:extension:name and PARQUET:field_id, which is where ArrowType.extension and ArrayData.field_id travel.

Dictionary encoding, compression, unions, run-end encoding, the view types and the IPC file format are not implemented; a stream that uses them is rejected with an error that names the feature.

A validity bitmap is only written when an array has nulls; otherwise the buffer is empty, which the format allows. Reading copies each buffer into the ArrayData's own lists.

struct IpcWriter

def __init__(out self, schema: RecordBatch)

def write(mut self, batch: RecordBatch)

def finish(deinit self) -> List[UInt8]

struct IpcStream

def __init__(out self, var schema: RecordBatch, var batches: List[RecordBatch])

def encode_schema_message(batch: RecordBatch) -> List[UInt8]

def write_ipc_stream(batches: List[RecordBatch]) -> List[UInt8]

def read_ipc_stream(data: Span[UInt8]) -> IpcStream