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).
• Remote execution — RemoteFragmentTransport and RemoteFragmentEndpoint use a versioned, bounded Arrow IPC pull protocol. One batch per pull provides backpressure; query-scoped cancellation and adaptive top-k refill cross the same authenticated node-internal RPC carrier.

Integration point with DataFusion

DataFusionDistributedPlanner::lower converts a real datafusion::logical_expr::LogicalPlan (produced by MongrelSession after catalog/schema resolution) onto LogicalPlanLite and hands the tree to distribute. Supported operators: TableScan, Projection (column pushdown), Filter (pushdown onto scans as typed predicate text for worker evaluation), Aggregate, Sort, Limit, equi-Join, and Union. Every other DataFusion node is rejected with DistributedError::Unsupported.

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.

CoreFragmentTableSource

Core MVCC-backed tablet source used by remote SQL workers.

DataFusionDistributedPlanner

Lowers a DataFusion datafusion::logical_expr::LogicalPlan onto tablets.

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.

FragmentLifecycleMetrics

Lifecycle counters for remote fragment workers (P0.4-T6).

FragmentLifecycleSnapshot

Point-in-time fragment lifecycle view (P0.4-T6).

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

LoopbackFragmentRpcClient

In-process carrier for protocol and endpoint tests.

PlanDescription

Everything distribute needs to build a DistributedPlan.

PlanFragment

One executable unit of a DistributedPlan.

PlannerOptions

Planner tuning knobs.

PlanningTableCatalog

Arrow schemas for tables that DataFusion must resolve while planning SQL for distributed placement. Schemas are planning-only (no row data).

RemoteFragmentEndpoint

Worker-side endpoint for the internal fragment protocol.

RemoteFragmentTransport

Coordinator-side Arrow IPC transport for tablet fragments.

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.

DEFAULT_REMOTE_FRAGMENT_EXECUTIONS

Default maximum number of fragment streams held by one worker.

DEFAULT_REMOTE_FRAGMENT_MESSAGE_BYTES

Default maximum encoded request or response body for one fragment RPC.

MAX_FRAGMENT_AUTHORIZATION_CONTEXT_BYTES

Maximum server-issued authorization envelope forwarded to a worker.

REMOTE_FRAGMENT_PROTOCOL_VERSION

Version of the private cluster fragment protocol.

REMOTE_FRAGMENT_SERVICE_ID

Stable service discriminator inside the cluster’s authenticated internal RPC multiplexer.

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”).

FragmentAuthorizationResolver

Validates the forwarded server-issued authorization envelope.

FragmentDatabaseProvider

Resolves the local storage owner for one hosted tablet.

FragmentExecutor

Executes one fragment on its worker. Server/engine bindings install an implementation behind RemoteFragmentEndpoint; InMemoryFragmentExecutor remains the deterministic reference executor.

FragmentRpcClient

One authenticated request/response carrier for remote fragment bodies.

FragmentTableSource

Batch source consumed by the reference fragment operator engine.

FragmentTransport

Moves fragments, refill, and cancellation between the coordinator and workers (spec section 12.10). Implementations include the deterministic InMemoryTransport and the bounded Arrow IPC RemoteFragmentTransport.

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”).

plan_logical_distributed

Lowers an already-parsed DataFusion logical plan onto tablets.

plan_logical_distributed_with_id

Same as plan_logical_distributed but stamps a caller-chosen query id.

plan_sql_distributed

Public gateway entry: parse sql with DataFusion against catalog, then lower via DataFusionDistributedPlanner onto locator/metadata.

plan_sql_distributed_with_id

Same as plan_sql_distributed but stamps a caller-chosen query id.

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.