dau_scheduler

Submodules

• dau_scheduler.errors
• dau_scheduler.merge
• dau_scheduler.posture
• dau_scheduler.profile
• dau_scheduler.rows
• dau_scheduler.split

Attributes

__version__
ADDITIVE
AGGREGATE_REFUSALS
BOUNDARY_STATS
DEFERRED_MEAN
POSTURES
Posture
SplitArm

Exceptions

PostureError	A posture was requested that cannot be resolved into a share.
SplitError	A split was asked for that cannot be executed or merged as asked.

Classes

MergeRule	What one query's partials are, and whether they can be put together.
HostCost	Anything that can price a row count on the engine the caller is on.
WorkSplit	How one query's rows divide between the two engines.
LinkRate	The measured host-to-device transfer rate of one link: the term the
NodeCost	One plan node's measured cost, and whether it scales with rows.
ProfileCache	Profiles on disk, keyed by plan and host, and link rates keyed by
QueryProfile	A measured host cost for one plan on one host.
RowBatch	Rows the caller has already materialized: the offsets AND the values.
RowRange	Rows named by their offsets in the table, and nothing else.
RowSource	The rows one arm owns, named by offset and optionally by value.
SplitExecution	One executed split: what each engine ran, how long, and the answer.

Functions

check_split() → None)	Refuse a query whose result cannot be reassembled, by name.
merge_keyed_groups(→ list[Any])	Merge per-key records that carry no ordering, over the whole key set.
merge_ordered_groups(→ list[Any])	Concatenate per-group records, folding the group that straddles the
merge_sorted_runs(→ list[Any])	Merge the arms' sorted runs into one sorted run.
merge_top_k(→ list[Any])	Re-rank the arms' k-lists into one.
merge_totals(→ Any)	Add the named aggregates, then RE-DERIVE anything computed from them.
batch_count(→ int)	How many batches one engine's share should be delivered in.
plan_split(→ WorkSplit)	Resolve a posture into a row split.
host_identity(→ str)	A name for the machine a profile was measured on.
profile_from_spans(→ QueryProfile)	Build a profile from an engine's per-node timings.
as_row_source(→ RowSource)	The seam's argument, from whatever the caller is holding.
execute_split() → SplitExecution)	Run split's share of the rows on one engine and the rest on the

Package Contents

dau_scheduler.__version__ = '0.1.0'

exception dau_scheduler.PostureError

Bases: ValueError

A posture was requested that cannot be resolved into a share.

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

exception dau_scheduler.SplitError

Bases: ValueError

A split was asked for that cannot be executed or merged as asked.

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

dau_scheduler.ADDITIVE: frozenset[str]

dau_scheduler.AGGREGATE_REFUSALS: dict[str, str]

dau_scheduler.BOUNDARY_STATS: frozenset[str]

dau_scheduler.DEFERRED_MEAN: frozenset[str]

class dau_scheduler.MergeRule

What one query’s partials are, and whether they can be put together.

terminal names the result shape, for messages. combine takes the arms’ partials IN ARM ORDER – the leading rows first – and returns the answer the unsplit query would have produced. mergeable is the set of aggregate kinds whose partials this shape can combine, consulted by check_split() and default-deny.

refusal, when set, says this shape cannot be split at all, and no combine is reached.

failed is the guard for an arm that could not complete. An engine that invalidated its own work returns a partial that says so, and merging that into the other arm’s would produce a plausible SHORT answer – so if any partial reports failure, the merged result is that partial, failure and all, rather than a total nobody can tell is missing rows.

terminal: str

combine: collections.abc.Callable[[collections.abc.Sequence[Any]], Any]

mergeable: frozenset[str]

failed: collections.abc.Callable[[Any], bool] | None = None

refusal: str | None = None

merge(partials: collections.abc.Sequence[Any]) → Any

The merged answer for these partials.

dau_scheduler.check_split(rule: MergeRule, aggregates: collections.abc.Iterable[tuple[str, str]] = ()) → None

Refuse a query whose result cannot be reassembled, by name.

aggregates are (output name, aggregate kind) pairs, so the refusal names the output the caller has to change rather than only the kind.

Kept separate from execution so a planner can ask whether a query is splittable at all before committing anything to either engine.

dau_scheduler.merge_keyed_groups(partials: collections.abc.Sequence[collections.abc.Sequence[Any]], *, key: collections.abc.Callable[[Any], Any], fold: collections.abc.Callable[[Any, Any], Any]) → list[Any]

Merge per-key records that carry no ordering, over the whole key set.

The contrast with merge_ordered_groups() is the point. An engine holding one accumulator per distinct key can meet a key anywhere in its rows, so a key may appear in both arms and there is no boundary pair to fold – every key has to be looked up. The result is key-ascending, which is the order such an engine drains in.

dau_scheduler.merge_ordered_groups(partials: collections.abc.Sequence[collections.abc.Sequence[Any]], *, key: collections.abc.Callable[[Any], Any], fold: collections.abc.Callable[[Any, Any], Any]) → list[Any]

Concatenate per-group records, folding the group that straddles the boundary.

Valid only when the rows are SORTED on the group key, because that is what makes the boundary pair the only group two arms can share. The precondition is verified rather than trusted: on unsorted rows an order-sensitive engine emits several records per key, and merging those by key would collapse them into an answer no run produced.

fold combines two partial records of the same group; key reads a record’s group key.

dau_scheduler.merge_sorted_runs(partials: collections.abc.Sequence[collections.abc.Sequence[Any]], *, key: collections.abc.Callable[[Any], Any], descending: bool = False) → list[Any]

Merge the arms’ sorted runs into one sorted run.

Equal keys take the earlier range first, which is what keeps a stable sort’s arrival-order ties true of the whole table rather than of each half.

dau_scheduler.merge_top_k(partials: collections.abc.Sequence[collections.abc.Sequence[Any]], *, limit: int, key: collections.abc.Callable[[Any], Any], descending: bool = True) → list[Any]

Re-rank the arms’ k-lists into one.

A row in the whole query’s top k ranks at least as high within the range it arrived in, so it is in that range’s k-list: the arms’ outputs are the complete candidate set. Each arm emits equal keys in arrival order and the arms are given in row order, so a STABLE re-rank keeps an earliest-arrival tie-break true of the whole table rather than of each half.

dau_scheduler.merge_totals(partials: collections.abc.Sequence[dict[str, Any]], names: collections.abc.Sequence[str], *, finalize: collections.abc.Callable[[dict[str, Any]], Any] | None = None) → Any

Add the named aggregates, then RE-DERIVE anything computed from them.

finalize is where an output that is an EXPRESSION over the aggregates – a ratio, a share, a rate – is recomputed from the merged totals. Adding the two arms’ ratios would produce a number that is not a ratio of anything, and it would look exactly like a total.

dau_scheduler.POSTURES: tuple[str, ...] = ('cpu', 'offload', 'collaborative', 'adaptive')

class dau_scheduler.HostCost

Bases: Protocol

Anything that can price a row count on the engine the caller is on.

dau_scheduler.profile.QueryProfile is one, measured rather than modelled; a caller with an analytic model of its own host can pass that instead.

estimate(rows: int) → float

dau_scheduler.Posture

class dau_scheduler.WorkSplit(/, **data: Any)

Bases: pydantic.BaseModel

How one query’s rows divide between the two engines.

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

rationale: str

predicted_speedup: float | None = None

check() → WorkSplit

Re-run the invariants and return self.

Construction validates, but model_copy(update=...) does not, so every consumer calls this before acting on a split rather than trusting that the instance it was handed went through __init__.

_share_is_a_fraction() → WorkSplit

dau_scheduler.batch_count(payload_bytes: int, *, resident: bool, knee_bytes: int) → int

How many batches one engine’s share should be delivered in.

Batching exists to hide the transfer behind the compute. Each batch’s transfer overlaps the previous batch’s work, so the pipelined cost falls toward the transfer time alone. With the data already RESIDENT on the engine there is no transfer to hide and every batch is pure per-call overhead, so the answer is one.

Measured on one link with a 45.8 MiB payload:

batches	streamed	resident
1	22.08 ms	5.27 ms
3	18.78	5.91 ms
12	22.59 ms	–

The optima are opposite, so a single tuned batch count is wrong for one of them – which is why this takes resident rather than a constant.

Streamed, the count targets the link’s transfer knee rather than a fitted model. knee_bytes is the payload below which a transfer loses rate to per-transfer setup; it is a property of the link and has to be measured on it, so there is no default here. Fitting T(B) = transfer + work/B + c*B to the measured points above gives c anywhere from 83 to 568 microseconds depending on which pair is fitted, so the model does not generalize and the measured knee is the honest thing to encode. At a 15 MiB knee, 45.8 MiB rounds to 3, which is what measured fastest.

dau_scheduler.plan_split(posture: Posture = 'collaborative', *, cpu_seconds: float | None = None, device_seconds: float | None = None, profile: HostCost | None = None, rows: int | None = None, device_fixed_seconds: float = 0.0) → WorkSplit

Resolve a posture into a row split.

cpu_seconds and device_seconds are whole-query estimates for each engine alone. Only collaborative needs them; the other postures are definitional and take none, which is the point of offering them – a caller who does not want to pay for cost estimation can still express an intent.

The host figure may instead come from a measurement: pass a HostCost and the rows this run will process, and the estimate is taken from it. An explicit cpu_seconds still wins – a caller who measured the thing itself should not be overridden by a model of it.

What is refused is guessing. With neither a figure nor a profile, collaborative declines to invent one, because a fabricated number underneath a real decision is worse than no decision: the split changes which engine runs which rows, so an error in it is self-reinforcing.

device_fixed_seconds is what one device job costs before it has processed a row – launching it, polling it, reading its records – a per-invocation term the proportional form omitted and one measurement put at a few milliseconds beside a 12.3 ms host time (2026-09-27). With it the balance is (1 - s) * cpu = fixed + s * device, and a share that cannot beat the host alone (the fixed cost is already most of the host time) is a cpu split that says why, not a split predicted to win by a margin the fixed cost eats.

class dau_scheduler.LinkRate(/, **data: Any)

Bases: pydantic.BaseModel

The measured host-to-device transfer rate of one link: the term the adaptive cut model and a streamed split price every byte crossing with.

Measured, not declared: a platform states its bus width and lane count, but what a transfer actually sustains depends on the host, the slot, the driver and the transfer size, and the sweeps of record put the same card at different rates in different hosts. So the rate is keyed by (host, device), like a host profile, and measured where it will be used. transfer_bytes is part of the identity, not a note: a link’s rate at one transfer size says little about another (per-transfer setup makes small transfers materially slower), so a rate is measured at the size the caller will move and cached under it. samples keeps the individual rates so the spread is visible beside the figure the model uses. measured_at (seconds since the epoch) is what staleness is judged by: a driver change, a link renegotiation or a reboot cannot invalidate an entry, so an old one is re-measured rather than trusted.

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].

host: str

device: str

bytes_per_second: float

transfer_bytes: int

samples: tuple[float, ...] = ()

measured_at: float = 0.0

is_stale(*, now: float, max_age_seconds: float) → bool

Older than max_age_seconds at now, or never dated.

_rate_is_usable() → LinkRate

check() → LinkRate

Re-run the invariants and return self (see QueryProfile.check()).

class dau_scheduler.NodeCost(/, **data: Any)

Bases: pydantic.BaseModel

One plan node’s measured cost, and whether it scales with rows.

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].

node: str

seconds: float

scales_with_rows: bool

check() → NodeCost

Re-run the invariants and return self (see QueryProfile.check()).

_seconds_is_finite() → NodeCost

class dau_scheduler.ProfileCache(root: pathlib.Path | str)

Profiles on disk, keyed by plan and host, and link rates keyed by device and host.

Deliberately a plain directory of JSON rather than anything cleverer: a profile is small, human-readable, and worth being able to delete by hand when it goes stale.

The engine version is deliberately not part of the key. A host is a key because you cannot re-measure on a machine you are not on; an engine version is not, because you can always re-measure under the one you have. So it is staleness (QueryProfile.is_stale_for()), handled exactly as a row count that has moved too far, rather than a second slot on disk.

root

_under_root(name: str) → pathlib.Path

A file directly under the cache root. Every key component is percent-encoded with no safe characters, so a plan key or host that contains a slash, .. or a space becomes one flat, reversible file name and cannot name a path outside the root. The check is kept anyway: a cache that writes where it is told is the failure mode this guards.

static _component(text: str) → str

_path(plan_key: str, host: str) → pathlib.Path

get(plan_key: str, host: str | None = None) → QueryProfile | None

put(profile: QueryProfile) → pathlib.Path

entries() → collections.abc.Iterable[pathlib.Path]

_link_path(device: str, host: str, transfer_bytes: int) → pathlib.Path

get_link_rate(device: str, transfer_bytes: int, host: str | None = None) → LinkRate | None

put_link_rate(rate: LinkRate) → pathlib.Path

class dau_scheduler.QueryProfile(/, **data: Any)

Bases: pydantic.BaseModel

A measured host cost for one plan on one host.

selectivity is the fraction of rows surviving the query’s filters when the caller measured it. It is left None rather than inferred: nothing in a timing profile reveals row counts, and a guessed selectivity poisons everything downstream that prices bytes.

engine_version records which build of the engine was measured. It defaults to "" – unknown – and NOT to whatever is installed, which would let a hand-written profile claim to have been measured under an engine it never ran on.

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].

plan_key: str

host: str

rows: int

fixed_seconds: float

per_row_seconds: float

nodes: tuple[NodeCost, ...] = ()

selectivity: float | None = None

engine_version: str = ''

check() → QueryProfile

Re-run the invariants and return self.

Construction validates, but model_copy(update=...) does not, so every use site (estimate, is_stale_for, the cache) calls this rather than trusting that the instance went through __init__.

_costs_are_sane() → QueryProfile

estimate(rows: int) → float

Host seconds for rows, as fixed + per_row * rows.

is_stale_for(rows: int, *, engine_version: str, factor: float = 10.0) → bool

Whether this measurement can still price rows on the engine here now.

Two things invalidate a measurement, and both are answered by this one predicate on purpose: a caller that has to remember to ask a second question is how a version stamp comes to be recorded on every profile and consulted by nothing.

Row count. Extrapolating one measurement across orders of magnitude is where a profile-guided split goes confidently wrong, and confidently wrong is worse than absent: the split decides which engine runs which rows, so the error is self-reinforcing.

Engine version. The argument for keying on the host – two machines measured 1.41x apart, so a profile keyed by plan alone steers the split on the wrong one – applies verbatim to the engine being measured, and this module exists BECAUSE an analytic model of that engine would be guesswork. A measurement taken under one build says nothing about another.

A profile with no recorded version is unknown, not current, and is stale for the same reason a guessed selectivity is refused: it may have been measured under anything.

dau_scheduler.host_identity() → str

A name for the machine a profile was measured on.

dau_scheduler.profile_from_spans(spans: collections.abc.Iterable[tuple[int, int, str]], *, rows: int, plan_key: str, host: str | None = None, is_fixed: collections.abc.Callable[[str], bool] = lambda _node: ..., selectivity: float | None = None, engine_version: str = '') → QueryProfile

Build a profile from an engine’s per-node timings.

spans are (start, end, node name) in MICROSECONDS, as a profiling query engine reports them. is_fixed names the plan-time nodes – the ones that must not be divided by the row count.

Nodes overlap (an engine runs some in parallel) and leave gaps, so summing ``end - start`` is NOT elapsed time. It double-counts the overlap and drops the gaps entirely: with optimization 0-1000, scan 1000-11000 and filter 5000-15000, the true elapsed cost is 15 ms and the sum reports 21 ms.

So the span is taken as elapsed wall time, the fixed nodes are measured as a UNION of their intervals, and everything else – including the gaps, which are real elapsed cost – is attributed to scaling work.

class dau_scheduler.RowBatch

Bases: RowSource

Rows the caller has already materialized: the offsets AND the values.

The rows are WRAPPED rather than copied. Whatever normalization an arm performs on its own input is not repeated here, because repeating it over the whole table was the cost this seam exists to remove.

rows: collections.abc.Sequence[Any]

start: int = 0

__post_init__() → None

property stop: int

__repr__() → str

slice(start: int, stop: int) → RowBatch

The sub-source covering [start, stop) of THIS source’s rows.

Relative indices, absolute offsets: the result carries where it sits in the whole table, which is the number an arm slices its own resident copy at.

values() → collections.abc.Sequence[Any]

class dau_scheduler.RowRange

Bases: RowSource

Rows named by their offsets in the table, and nothing else.

The source a split between two engines that each hold the data wants: it costs two integers no matter how many rows it names, so a sweep over millions of rows allocates nothing per call.

start: int

stop: int

__post_init__() → None

slice(start: int, stop: int) → RowRange

The sub-source covering [start, stop) of THIS source’s rows.

Relative indices, absolute offsets: the result carries where it sits in the whole table, which is the number an arm slices its own resident copy at.

class dau_scheduler.RowSource

The rows one arm owns, named by offset and optionally by value.

__slots__ = ()

start: int

stop: int

__len__() → int

abstractmethod slice(start: int, stop: int) → RowSource

The sub-source covering [start, stop) of THIS source’s rows.

Relative indices, absolute offsets: the result carries where it sits in the whole table, which is the number an arm slices its own resident copy at.

values() → collections.abc.Sequence[Any]

__iter__() → collections.abc.Iterator[Any]

dau_scheduler.as_row_source(rows: RowSource | collections.abc.Iterable[Any]) → RowSource

The seam’s argument, from whatever the caller is holding.

A source passes through untouched – that is the point, and the reason this is not list(rows) under a better name. A sequence is wrapped in place, so a caller who already has the rows pays nothing to hand them over. Only an ITERATOR is drained, and it has to be: both arms read from one source, and an iterator cannot serve two.

dau_scheduler.SplitArm

class dau_scheduler.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.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.