dau_core.stream

Attributes

LOGICAL_TYPE_BYTE_WIDTHS
DEFAULT_RECORD_BATCH_LIMITS

Exceptions

InvalidStreamError

Inappropriate argument value (of correct type).

Classes

LogicalType	Enum where members are also (and must be) ints
OperationCode	Enum where members are also (and must be) ints
StreamAggregationErrorCode	Enum where members are also (and must be) ints
GroupRole	Enum where members are also (and must be) ints
PolicyFlag	Support for integer-based Flags
StreamFlag	Support for integer-based Flags
BatchHeader
ColumnDescriptor
OperationDescriptor
RecordBatchLimits	Caps for multi-column record batches (see dau-docs record-batch contract).

Functions

output_type_for(opcode) → LogicalType	The wire type an aggregation opcode produces (single source of truth
validate_record_batch(header, columns, *[, limits]) → None	Validate the multi-column batch shape shared by host codecs and HDL models.
_validate_u8(name, value) → None
_validate_u16(name, value) → None
_validate_u32(name, value) → None
_validate_u64(name, value) → None

Module Contents

exception dau_core.stream.InvalidStreamError

Bases: ValueError

Inappropriate argument value (of correct type).

Initialize self. See help(type(self)) for accurate signature.

class dau_core.stream.LogicalType

Bases: enum.IntEnum

Enum where members are also (and must be) ints

Initialize self. See help(type(self)) for accurate signature.

BOOL = 1

INT8 = 10

INT16 = 11

INT32 = 12

INT64 = 13

UINT8 = 20

UINT16 = 21

UINT32 = 22

UINT64 = 23

FLOAT32 = 30

FLOAT64 = 31

dau_core.stream.LOGICAL_TYPE_BYTE_WIDTHS

class dau_core.stream.OperationCode

Bases: enum.IntEnum

Enum where members are also (and must be) ints

Initialize self. See help(type(self)) for accurate signature.

MIN = 1

MAX = 2

SUM = 3

COUNT = 4

AVERAGE = 5

classmethod from_name(name: str) → OperationCode

dau_core.stream.output_type_for(opcode: OperationCode) → LogicalType

The wire type an aggregation opcode produces (single source of truth for every backend, API layer, and example).

class dau_core.stream.StreamAggregationErrorCode

Bases: enum.IntEnum

Enum where members are also (and must be) ints

Initialize self. See help(type(self)) for accurate signature.

DESCRIPTOR = 1

OPCODE = 2

TILE = 3

class dau_core.stream.GroupRole

Bases: enum.IntEnum

Enum where members are also (and must be) ints

Initialize self. See help(type(self)) for accurate signature.

VALUE = 0

KEY = 1

OUTPUT = 2

class dau_core.stream.PolicyFlag

Bases: enum.IntFlag

Support for integer-based Flags

Initialize self. See help(type(self)) for accurate signature.

NONE = 0

SATURATING_SUM = 1

IEEE_FLOAT_NAN = 2

FLOAT64_AVERAGE = 4

class dau_core.stream.StreamFlag

Bases: enum.IntFlag

Support for integer-based Flags

Initialize self. See help(type(self)) for accurate signature.

NONE = 0

HAS_GROUP_FIELDS = 1

class dau_core.stream.BatchHeader

schema_hash: int

batch_id: int

row_count: int

column_count: int

operation_count: int

group_key_count: int

descriptor_table_offset: int

payload_offset: int

payload_length: int

version_major: int = 0

version_minor: int = 1

header_length: int = 64

flags: StreamFlag

SIZE_BYTES = 64

_STRUCT

__post_init__() → None

pack() → bytes

classmethod unpack(data: bytes) → BatchHeader

class dau_core.stream.ColumnDescriptor

field_id: int

logical_type: LogicalType

physical_width_bits: int

row_count: int

null_count: int

payload_offset: int

payload_length: int

validity_offset: int = 0

validity_length: int = 0

group_role: GroupRole

flags: int = 0

SIZE_BYTES = 64

_STRUCT

__post_init__() → None

pack() → bytes

classmethod unpack(data: bytes) → ColumnDescriptor

class dau_core.stream.OperationDescriptor

operation_id: int

opcode: OperationCode

input_column_id: int

output_field_id: int

output_type: LogicalType

accumulator_width_bits: int

policy_flags: PolicyFlag

group_key_set_id: int = 0

SIZE_BYTES = 64

_STRUCT

__post_init__() → None

pack() → bytes

classmethod unpack(data: bytes) → OperationDescriptor

class dau_core.stream.RecordBatchLimits(/, **data: Any)

Bases: dau_core.models.CheckedModel

Caps for multi-column record batches (see dau-docs record-batch contract).

Defaults are sized to the 1 MiB stream-job staging buffers: header + descriptor table + payload must fit one staging region.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

max_columns: int = 16

max_rows: int = 1048576

max_batch_bytes: int = 1048576

alignment_bytes: int = 64

model_post_init(context) → None

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

dau_core.stream.DEFAULT_RECORD_BATCH_LIMITS

dau_core.stream.validate_record_batch(header: BatchHeader, columns: tuple[ColumnDescriptor, ...], *, limits: RecordBatchLimits = DEFAULT_RECORD_BATCH_LIMITS) → None

Validate the multi-column batch shape shared by host codecs and HDL models.

dau_core.stream._validate_u8(name: str, value: int) → None

dau_core.stream._validate_u16(name: str, value: int) → None

dau_core.stream._validate_u32(name: str, value: int) → None

dau_core.stream._validate_u64(name: str, value: int) → None