dau_scheduler.split

Run a planned split: two engines, disjoint row ranges, one answer.

dau_scheduler.posture decides how much of a query the second engine should carry; this runs that decision. The second engine takes the LEADING row range and the first the trailing one, both run the WHOLE query over their own rows, and the partial results merge into the answer an unsplit run would have produced – the same object, so a split is a drop-in for whatever the caller ran before rather than a second result shape to handle.

Both engines are seams. An arm is a callable handed the query, its own RowSource, and the broadcast side of a join if there is one; what it does with them is entirely its own business, and nothing here assumes either arm is in this process.

The refusal comes before either arm starts. A result that cannot be reassembled is refused while it still costs nothing; a refusal that arrives after the work is a refusal that already paid for it.

The accounting covers the whole call. wall_seconds spans from entry to answer, and the two terms that are not an arm are named: materialize_seconds for everything done to the rows before an arm starts, and merge_seconds for putting the partials together. A timing field that excludes the dominant term is worse than no field – and on a large split where the caller passes an iterator, marshalling IS the dominant term.

The collector stays the caller’s. Staging tens of millions of row tuples leaves a heap whose collection pauses cost more than the query being split, and a library that quietly repartitions its caller’s heap is an imposition. Handing an arm a range rather than a fresh list removes this module’s own contribution to that; the rest belongs to whoever allocated the rows.

Attributes

SplitArm

Classes

SplitExecution

One executed split: what each engine ran, how long, and the answer.

Functions

execute_split() → SplitExecution)

Run split's share of the rows on one engine and the rest on the

Module Contents

dau_scheduler.split.SplitArm

class dau_scheduler.split.SplitExecution(/, **data: Any)

Bases: pydantic.BaseModel

One executed split: what each engine ran, how long, and the answer.

result is whatever an UNSPLIT run of this query returns, so the caller’s decode path is unchanged by having split the work.

overlap_seconds is the intersection of the two arms’ wall-clock intervals. It measures that the arms were IN FLIGHT together, which is the property a split needs and the one a serialized executor loses. It does not claim parallel progress inside a single interpreter, which nothing here could promise on the caller’s behalf.

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.

model_config: ClassVar

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

posture: str

device_share: float

terminal: str

device_rows: int

host_rows: int

device_seconds: float

host_seconds: float

wall_seconds: float

overlap_seconds: float

materialize_seconds: float

merge_seconds: float

device_granularity: int = 1

result: Any = None

_figures_are_sane() → SplitExecution

dau_scheduler.split.execute_split(query: Any, rows: dau_scheduler.rows.RowSource | collections.abc.Iterable[Any], split: dau_scheduler.posture.WorkSplit, merge: dau_scheduler.merge.MergeRule, *, device: SplitArm, host: SplitArm, build_rows: dau_scheduler.rows.RowSource | collections.abc.Iterable[Any] | None = None, aggregates: collections.abc.Iterable[tuple[str, str]] = ()) → SplitExecution

Run split’s share of the rows on one engine and the rest on the other, at the same time, and merge the partials into the query’s answer.

query is passed to both arms untouched; this module never inspects it.

The device arm takes rows [0, share*n) and the host arm the rest. That order is part of the contract, not an implementation detail: a grouped first/last and a top-k’s earliest-arrival tie-break are defined by it. An arm with no rows is not launched at all – a share of 0.0 or 1.0 runs exactly the unsplit path (given a table on the device’s granularity), which is what makes the degenerate ends of a share sweep meaningful.

rows is the whole table as a RowSource – a RowRange when both arms run from data they already hold, a RowBatch when one of them reads values – or any iterable of rows, which is wrapped and charged to materialize_seconds.

aggregates are (output name, aggregate kind) pairs checked against merge before anything runs (see dau_scheduler.merge.check_split()).

A device arm carrying row_granularity (see SplitArm) can only take a row count that divides it – an engine whose job length must sit on a fixed row grid. The boundary is snapped to the nearest such multiple and the count actually given is reported as device_rows. It is an execution fact about the engine, not part of the share the planner chose, which is why it rides the arm rather than the WorkSplit.