Stream codec reference

dau_driver.stream_codec encodes typed columns as DAU stream batches and decodes them back. The wire format (BatchHeader, ColumnDescriptor, OperationDescriptor, LogicalType, StreamFlag, RecordBatchLimits, InvalidStreamError) and the per-type widths (LOGICAL_TYPE_BYTE_WIDTHS) are defined in dau_core.stream; this module marshals them.

Batches are 64-byte aligned throughout: the descriptor table starts at BatchHeader.SIZE_BYTES, each payload begins on a 64-byte boundary, and the encoded batch is padded up to one.

Supported logical types: INT8, INT16, INT32, INT64, UINT8, UINT16, UINT32, UINT64, FLOAT32, FLOAT64. Any other value raises InvalidStreamError.

Validity bitmaps are not supported. Columns must be fixed-width and null-free.

StreamEncoder

All methods are static.

encode_column

StreamEncoder.encode_column(
    logical_type: LogicalType,
    payload: bytes,
    *,
    row_count: int,
    field_id: int = 0,
    schema_hash: int = 0,
    batch_id: int = 0,
    operations: Iterable[OperationDescriptor] = (),
) -> bytes

Encode one column as a single-column batch. Raises InvalidStreamError when row_count is negative or len(payload) is not row_count * width.

encode_record_batch

StreamEncoder.encode_record_batch(
    columns: Sequence[RecordBatchColumn],
    *,
    row_count: int,
    schema_hash: int = 0,
    batch_id: int = 0,
    operations: Iterable[OperationDescriptor] = (),
    limits: RecordBatchLimits = DEFAULT_RECORD_BATCH_LIMITS,
) -> bytes

Encode N fixed-width, null-free columns as one multi-column batch. Raises InvalidStreamError when row_count is negative, columns is empty, or any column’s payload length does not match its row count and width. The assembled header and descriptors are checked against limits before serialization.

encode_arrow_batch

StreamEncoder.encode_arrow_batch(
    batch,
    *,
    schema_hash: int = 0,
    batch_id: int = 0,
    operations: Iterable[OperationDescriptor] = (),
    limits: RecordBatchLimits = DEFAULT_RECORD_BATCH_LIMITS,
) -> bytes

Encode a PyArrow RecordBatch or Table of fixed-width, null-free columns. Field ids are assigned by column position. Column data is taken as a zero-copy, offset-aware view of the primitive data buffer, so a sliced array encodes without a materializing copy; one copy happens at stream assembly. A column with nulls raises InvalidStreamError naming the column. Requires pyarrow.

StreamDecoder

All methods are static.

Method	Returns
decode_column(data)	DecodedStreamColumn: a single-column batch. Requires column_count == 1.
decode_record_batch(data, *, limits=DEFAULT_RECORD_BATCH_LIMITS)	DecodedRecordBatch: the N-column counterpart.
decode_record_batch_to_arrow(data, *, names=None, limits=DEFAULT_RECORD_BATCH_LIMITS)	A PyArrow RecordBatch. Column names default to f{field_id}.
decode_scalar_result(data)	DecodedStreamResult: a one-row batch with its value unpacked.

Decoding validates the stream version (0.1), the header length, 64-byte alignment of the descriptor and payload offsets, every payload window against the data length, each column’s logical type, physical width, row count, and payload length, and each operation descriptor’s input column id, output type, and accumulator width (32 or 64). Any mismatch raises InvalidStreamError.

decode_scalar_result additionally requires exactly one row in both the header and the column, and a payload length matching the result type’s width.

Data models

RecordBatchColumn

One fixed-width, null-free input column for record-batch encoding.

Field	Type
field_id	int
logical_type	LogicalType
payload	bytes

DecodedStreamColumn

Field	Type	Default
header	BatchHeader	—
column	ColumnDescriptor	—
payload	bytes	—
operations	tuple[OperationDescriptor, ...]	()

DecodedRecordBatch

Field	Type	Default
header	BatchHeader	—
columns	tuple[ColumnDescriptor, ...]	—
payloads	tuple[bytes, ...]	—
operations	tuple[OperationDescriptor, ...]	()

column_payload(field_id) returns the payload for one column by field id, not by position. Raises InvalidStreamError when no column carries that id.

DecodedStreamResult

Field	Type
header	BatchHeader
column	ColumnDescriptor
payload	bytes
value	int | float