Module distributed 

Distributed SQL plan groundwork (Stage 3J, spec section 12.10).

This module is the planning and coordination seam for tablet-distributed queries. It deliberately depends on small local metadata traits (TabletLocator, ClusterMetadata) instead of mongreldb-cluster, so the query crate stays free of the cluster dependency; the cluster wave adapts mongreldb_cluster::tablet routing metadata onto these traits.

What lives here

• DistributedPlan — the plan structure from spec section 12.10: pinned QueryId + MetadataVersion, PlanFragments with tablet assignments, and ExchangeDescriptor edges between them.
• distribute — the planner. It lowers a LogicalPlanLite tree (the simplified input IR) onto tablets: colocated joins when both sides share partitioning and layout, broadcast joins when the small side fits under a byte threshold, repartition joins otherwise. Every fragment carries estimated rows/bytes and a maximum spill allowance (spec section 12.10).
• Distributed top-k — merge_top_k / exact_top_k are pure functions implementing the deterministic coordinator merge (final score descending, tablet id ascending, RowId ascending) with adaptive refill when a tablet’s unseen-score bound shows it could still contribute winners.
• Execution skeleton — FragmentExecutor, FragmentTransport, and the Coordinator runtime: per-fragment resource reservation, cancellation fan-out wired to the existing SqlQueryRegistry, worker lease expiry that cleans abandoned fragments, and real in-memory merge operators (k-way MergeSort, FinalAggregate combine, distributed top-k) used until the remote Arrow IPC transport lands.

Integration point with DataFusion

LogicalPlanLite is intentionally minimal. The DataFusion integration wave lowers datafusion::logical_expr::LogicalPlan (produced by MongrelSession after catalog/schema resolution) onto LogicalPlanLite — scans of MongrelScanExec-backed tables become LogicalPlanLite::Scan, aggregates/joins/sorts/limits map one-to-one — and hands the tree to distribute. Full DataFusion plan conversion is explicitly out of scope for this wave.

Structs

AggregateExpr

One aggregate expression.

BatchFrame

One Arrow-ish record-batch frame on a fragment stream.

Coordinator

The query coordinator (spec section 12.10): registers the query with the existing SqlQueryRegistry (so registry.cancel(...) reaches every fragment through the ExecutionControl hierarchy), reserves resources per fragment, fans out cancellation to every fragment, sweeps expired worker leases to clean abandoned fragments, and merges producer streams per the exchange descriptors with the real in-memory operators.

DistributedPlan

A distributed query plan pinned to one control-plane metadata version (spec section 12.10).

ExchangeDescriptor

One exchange edge between two fragments (spec section 12.10).

FragmentControl

Cooperative per-fragment execution control: cancellation/deadline shared with the query’s ExecutionControl hierarchy plus the fragment’s spill allowance.

InMemoryFragmentExecutor

Reference FragmentExecutor over an InMemoryTableStore. Interprets scans, projections, partial aggregates, local merge sorts, bounded local top-ks (with exact tie information), and limits for real; exchange sources drain their input streams; join operators reject (their execution binding lands with the tablet wave — plan shape is fully tested).

InMemoryTableStore

In-memory per-(table, tablet) record-batch store backing InMemoryFragmentExecutor.

InMemoryTransport

In-memory FragmentTransport: routes fragments to per-tablet executors, records starts/cancellations/refills for test introspection, and keeps each fragment’s ExecutionControl so cancellations are observable.

JoinKey

One join-key equality pair (left.col = right.col).

LeaseLedger

Worker lease ledger (spec section 12.10: “worker lease expiry cleans abandoned fragments”). Workers are keyed by the tablet whose data they serve this wave; the node-level binding lands with the transport wave.

PlanDescription

Everything distribute needs to build a DistributedPlan.

PlanFragment

One executable unit of a DistributedPlan.

PlannerOptions

Planner tuning knobs.

ResourceLedger

Per-query fragment resource ledger. Reservations are RAII: dropping a ResourcePermit releases its share.

ResourcePermit

A held resource reservation; released on drop.

SortKey

One sort key.

TableStats

Planner statistics for one table.

TabletTopK

One tablet’s bounded local top-k plus tie information (spec section 12.10: “each tablet returns a bounded local top-k plus tie information”).

TopKCandidate

One ranked row in the top-k model: the score plus the exact tie-break identity (spec section 12.10: final score descending, tablet id ascending, RowId ascending).

TopKMerge

The outcome of one deterministic coordinator merge step.

TopKRefill

The next batch of a tablet’s local top-k (adaptive refill, spec section 12.10).

Enums

AggregateFunction

Supported aggregate functions.

BuildSide

Which join input is the broadcast (build) side.

DistributedError

Errors raised by distributed planning and coordination.

ExchangeKind

How rows move across one exchange edge (spec section 12.10).

FragmentAssignment

Where one fragment executes.

FragmentOperator

One physical operator inside a fragment (spec section 12.10).

LogicalPlanLite

The simplified logical IR distribute lowers onto tablets.

PartitionSpec

How one table’s rows map onto tablets — the planner’s dependency-free mirror of mongreldb_cluster::tablet::Partitioning (spec section 12.2). The cluster wave adapts the cluster type onto this enum one-to-one.

ScoreBound

Tie information for scored (top-k) streams, carried on a producer’s terminal frame (spec section 12.10: “bounded local top-k plus tie information”).

Constants

DEFAULT_BROADCAST_THRESHOLD_BYTES

Small-side byte threshold below which a join is planned as a broadcast join (spec section 12.10: “broadcast small side”).

DEFAULT_MAX_SPILL_BYTES_PER_FRAGMENT

Default per-fragment maximum spill allowance (spec section 12.10: “every plan includes … a maximum spill allowance”). Bound to the core mongreldb_core::SpillManager via FragmentControl::begin_spill.

TOPK_ROWID_COLUMN

Name of the unsigned row-id column every scored (top-k) stream carries. Tablet scans feeding a distributed top-k must project this column; it is stripped from the coordinator’s final top-k output.

Traits

ClusterMetadata

Control-plane metadata the planner pins against (spec section 12.10: “coordinator plans using metadata version”).

FragmentExecutor

Executes one fragment on its worker. The real binding runs tablet fragments against a ClusterReplica core in a later wave; this wave ships InMemoryFragmentExecutor as the reference implementation.

FragmentTransport

Moves fragments and cancellation between the coordinator and workers (spec section 12.10). The Arrow IPC transport with backpressure is the remote-transport wave’s job; this wave ships the in-memory implementation.

TabletLocator

Resolves which tablets serve one table. Defined here (not in mongreldb-cluster) so the query crate stays free of the cluster dependency; the cluster wave binds TabletLayout/routing metadata onto this trait.

Functions

distribute

Lowers a LogicalPlanLite tree onto tablets (spec section 12.10).

exact_top_k

Drives merge_top_k with adaptive refill until the result is provably exact (spec section 12.10: “for exact global top-k, use adaptive refill when local bounds show unseen rows could still win”).

merge_top_k

Deterministically merges bounded local top-ks (spec section 12.10: “coordinator merges deterministically”).

topk_cmp

The deterministic winner order: score descending, then tablet id ascending, then RowId ascending. Returns Ordering::Less when a ranks strictly better than b.

Type Aliases

DistributedResult

Result alias for distributed planning and coordination.

ExchangeId

Plan-local exchange identifier (index into DistributedPlan::exchanges).

FragmentId

Plan-local fragment identifier (index into DistributedPlan::fragments).

FragmentStream

A fragment’s output stream.