Enum QueryOp 

pub enum QueryOp {
    Exchange(ExchangeOp),
    ProviderScan {
        provider: Option<String>,
        rows: Vec<u8>,
        filters: Vec<u8>,
        projection: Vec<String>,
        sort_keys: Vec<(String, bool)>,
        limit: Option<usize>,
        offset: usize,
        distinct: bool,
    },
    Aggregate {
        collection: String,
        input: Option<Box<PhysicalPlan>>,
        group_by: Vec<GroupKeySpec>,
        aggregates: Vec<AggregateSpec>,
        filters: Vec<u8>,
        having: Vec<u8>,
        limit: usize,
        sub_group_by: Vec<String>,
        sub_aggregates: Vec<AggregateSpec>,
        grouping_sets: Vec<Vec<u32>>,
        sort_keys: Vec<(String, bool)>,
    },
    PartialAggregate {
        collection: String,
        group_by: Vec<String>,
        aggregates: Vec<AggregateSpec>,
        filters: Vec<u8>,
    },
    PartialAggregateState {
        collection: String,
        input: Option<Box<PhysicalPlan>>,
        group_by: Vec<GroupKeySpec>,
        aggregates: Vec<AggregateSpec>,
        filters: Vec<u8>,
    },
    HashJoin {

        left_collection: String,
        right_collection: String,
        left_alias: Option<String>,
        right_alias: Option<String>,
        on: Vec<(String, String)>,
        join_type: String,
        limit: usize,
        post_group_by: Vec<String>,
        post_aggregates: Vec<(String, String)>,
        projection: Vec<JoinProjection>,
        computed_projection: Vec<u8>,
        join_filters: Vec<u8>,
        post_filters: Vec<u8>,
        left_input: Option<Box<PhysicalPlan>>,
        right_input: Option<Box<PhysicalPlan>>,
        left_bitmap: Option<Box<PhysicalPlan>>,
        right_bitmap: Option<Box<PhysicalPlan>>,
    
},
    ShuffleJoinConsume {
        build_path: String,
        probe_path: String,
        on: Vec<(String, String)>,
        join_type: String,
        limit: usize,
        probe_qualifier: String,
        index_qualifier: String,
    },
    ShuffleAggregateConsume {
        state_path: String,
        group_by: Vec<String>,
        aggregates: Vec<AggregateSpec>,
        having: Vec<u8>,
        limit: usize,
        sort_keys: Vec<(String, bool)>,
    },
    NestedLoopJoin {
        left_collection: String,
        right_collection: String,
        condition: Vec<u8>,
        join_type: String,
        limit: usize,
    },
    SortMergeJoin {
        left_collection: String,
        right_collection: String,
        on: Vec<(String, String)>,
        join_type: String,
        limit: usize,
        pre_sorted: bool,
    },
    FacetCounts {
        collection: String,
        filters: Vec<u8>,
        fields: Vec<String>,
        limit_per_facet: usize,
    },
    RecursiveScan {
        collection: String,
        base_filters: Vec<u8>,
        recursive_filters: Vec<u8>,
        join_link: Option<(String, String)>,
        max_iterations: usize,
        distinct: bool,
        limit: usize,
    },
    RecursiveValue {
        cte_name: String,
        columns: Vec<String>,
        init_exprs: Vec<String>,
        step_exprs: Vec<String>,
        condition: Option<String>,
        max_depth: usize,
        distinct: bool,
    },
    LateralTopK {
        outer_plan: Box<PhysicalPlan>,
        outer_alias: String,
        inner_collection: String,
        inner_filters: Vec<u8>,
        inner_order_by: Vec<(String, bool)>,
        inner_limit: usize,
        correlation_keys: Vec<(String, String)>,
        lateral_alias: String,
        projection: Vec<JoinProjection>,
        left_join: bool,
    },
    LateralLoop {
        outer_plan: Box<PhysicalPlan>,
        outer_alias: String,
        inner_collection: String,
        inner_filters: Vec<u8>,
        correlation_predicates: Vec<(String, String)>,
        lateral_alias: String,
        projection: Vec<JoinProjection>,
        left_join: bool,
        outer_row_cap: usize,
    },
}

Query-level physical operations (joins, aggregates).

Variants

Exchange(ExchangeOp)

Coordinator-resolved data-movement wrapper. Defensive error if it reaches a core.

ProviderScan

Pre-materialized rows. provider: Some(name) => filled per-request (identity-scoped catalog table); None => rows is final (constant SELECT or gathered broadcast child). rows is a canonical msgpack array of map values (the encode_binary_rows shape).

The executor applies predicate, offset, sort, distinct, projection, and limit in that order over the rows before emitting the response. Empty filters/projection and None limit are no-ops (all rows emitted unchanged through those stages).

Fields

provider: Option<String>

Catalog provider name for deferred materialization by the coordinator. Some(name) is replaced with None after materialize_providers fills rows. Data-Plane cores must never see Some.

rows: Vec<u8>

Msgpack array of plain map rows. Filled by the coordinator from the catalog producer or from a gathered broadcast child.

filters: Vec<u8>

Serialized Vec<ScanFilter> (MessagePack). Empty = no predicate.

projection: Vec<String>

Output column names to keep. Empty = emit all columns.

sort_keys: Vec<(String, bool)>

Sort keys: (column_name, ascending). Empty = unordered.

limit: Option<usize>

Maximum rows to emit after offset/sort/distinct/projection. None = unlimited.

offset: usize

Number of rows to skip before applying limit.

distinct: bool

SQL DISTINCT: deduplicate on the projected row.

Aggregate

Aggregate: GROUP BY + aggregate functions.

Fields

collection: String

input: Option<Box<PhysicalPlan>>

Optional sub-plan whose decoded rows are aggregated instead of scanning collection per-shard. Some currently means EXACTLY a catalog source (a ProviderScan lowered by the converter): the aggregate runs over the coordinator-materialized catalog rows and is therefore coordinator-local (never broadcast — see is_sharded_source). None = legacy path: scan the named collection on every shard. collection stays populated in both cases so downstream RLS / permission / classification continue to read it; the executor simply prefers input when present.

group_by: Vec<GroupKeySpec>

aggregates: Vec<AggregateSpec>

filters: Vec<u8>

having: Vec<u8>

HAVING predicates applied post-aggregation.

limit: usize

sub_group_by: Vec<String>

sub_aggregates: Vec<AggregateSpec>

grouping_sets: Vec<Vec<u32>>

ROLLUP / CUBE / GROUPING SETS expansion. Each inner Vec<u32> is one grouping set — the indices into group_by that are present (non-NULL) for rows in that set. Empty outer vec = plain single-set GROUP BY (no null-filling needed).

sort_keys: Vec<(String, bool)>

Post-aggregation sort keys: (column_name, ascending). Empty = preserve executor’s natural order (hash-map iteration for plain GROUP BY). The executor applies the sort after all groups are finalized and HAVING is filtered.

PartialAggregate

Partial aggregate: each core computes locally, Control Plane merges.

Fields

collection: String

group_by: Vec<String>

aggregates: Vec<AggregateSpec>

filters: Vec<u8>

PartialAggregateState

Partial aggregate STATE producer (distributed GROUP BY shuffle, map side).

Accumulates exactly like QueryOp::PartialAggregate — scan the named collection, apply filters, build per-group GroupState accumulators keyed on group_by — but DOES NOT finalize. Instead it emits one row PER GROUP of the flat shape:

{ <group_by[0]>: value_0, ..., "__agg_state": <bytes> }

where __agg_state is the serialized partial GroupState for that group. A downstream QueryOp::ShuffleAggregateConsume merges these partial states (from every producer shard/node) and finalizes them.

This op carries NO node-local paths — it reads only the named collection (or its input sub-plan) — so it is wire-shippable: a coordinator embeds it in a producer plan and dispatches it to a remote node’s Data Plane.

Fields

collection: String

input: Option<Box<PhysicalPlan>>

Optional sub-plan whose decoded rows are aggregated instead of scanning collection per-shard. Mirrors QueryOp::Aggregate’s input: Some runs the producer over the sub-plan’s rows (e.g. a coordinator-materialized ProviderScan), None scans the named collection. collection stays populated either way.

group_by: Vec<GroupKeySpec>

aggregates: Vec<AggregateSpec>

filters: Vec<u8>

HashJoin

Hash join: build hash map on right, probe with left.

left_input/right_input carry a resolved child plan (an Exchange child during planning, or an embedded ProviderScan after coordinator resolution). When None the side is scanned locally by collection name.

Fields

left_collection: String

right_collection: String

left_alias: Option<String>

right_alias: Option<String>

on: Vec<(String, String)>

join_type: String

limit: usize

post_group_by: Vec<String>

Post-join GROUP BY columns (empty = no aggregation).

post_aggregates: Vec<(String, String)>

Post-join aggregates: (op, field) pairs (empty = no aggregation).

projection: Vec<JoinProjection>

Post-join projection: column names to keep (empty = all).

computed_projection: Vec<u8>

MessagePack-encoded computed projection expressions. When present, these represent the complete SELECT list in output order.

join_filters: Vec<u8>

Residual ON predicates evaluated for each equi-key candidate before outer-join match accounting (MessagePack).

post_filters: Vec<u8>

Post-join WHERE filter predicates (MessagePack).

left_input: Option<Box<PhysicalPlan>>

Resolved child plan for the left side. An Exchange child during planning; a ProviderScan after coordinator resolution. None means the left side is scanned locally by left_collection.

right_input: Option<Box<PhysicalPlan>>

Resolved child plan for the right side. Same semantics as left_input but applied to right_collection.

left_bitmap: Option<Box<PhysicalPlan>>

Bitmap-producer sub-plan for the left side. When set, the executor executes this plan first, collects surrogates from all returned rows, and injects the resulting bitmap into the probe-side prefilter before scanning. None = no bitmap pushdown for the left side.

right_bitmap: Option<Box<PhysicalPlan>>

Bitmap-producer sub-plan for the right side. Same semantics as left_bitmap but applied to the right (probe) collection.

ShuffleJoinConsume

Cross-node shuffle-join CONSUMER (E4b): run the node-local grace-hash join over two already-staged shuffle frame files.

This variant is BUILT LOCALLY by the part-owner node’s consume hook with resolved local absolute paths and dispatched to the SAME node’s Data Plane. It carries node-local filesystem paths (as plain strings) and must NEVER be serialized cross-node — the wire encoder rejects it for exactly this reason (see physical_plan::wire::encode).

The owned fields mirror the borrowed JoinParams the Data-Plane handler (execute_shuffle_join) reconstructs: on (equi-join key pairs), join_type, limit, and the two column qualifiers. build_path / probe_path are the staged build (right) and probe (left) frame files.

Fields

build_path: String

Local absolute path to the staged BUILD (right) side frame file.

probe_path: String

Local absolute path to the staged PROBE (left) side frame file.

on: Vec<(String, String)>

Equi-join key pairs (left_key, right_key).

join_type: String

Join type (inner / left / right / full). cross/keyless is rejected by the handler.

limit: usize

Output row cap. usize::MAX = no explicit LIMIT (budget-bounded).

probe_qualifier: String

Column qualifier (prefix) for probe-side (left) columns.

index_qualifier: String

Column qualifier (prefix) for build-side (right/index) columns.

ShuffleAggregateConsume

Distributed GROUP BY shuffle CONSUMER (reduce side): merge the staged partial GroupStates produced by QueryOp::PartialAggregateState producers, then finalize, HAVING-filter, sort, and LIMIT.

state_path is a node-local frame file written by the shuffle receive path: a sequence of [u32 LE len][row-bytes] frames, one msgpack row per frame, each row the flat {<group_by cols>, "__agg_state": <bytes>} shape a PartialAggregateState producer emits. The handler rebuilds each group key (byte-identically to the accumulate path), decodes __agg_state, and merge_froms into a consolidated per-group map before running the same finalize tail as a normal aggregate.

Like QueryOp::ShuffleJoinConsume, this variant carries a node-local path and is BUILT LOCALLY by the consume hook for dispatch to the SAME node’s Data Plane — it must NEVER be serialized cross-node (the wire encoder rejects it; see physical_plan::wire::encode).

Fields

state_path: String

Local absolute path to the staged partial-state frame file.

group_by: Vec<String>

GROUP BY columns (the key the partial states were grouped on).

aggregates: Vec<AggregateSpec>

Aggregate specs (must match the producers’ specs, in order).

having: Vec<u8>

HAVING predicates applied post-merge (serialized Vec<ScanFilter>).

limit: usize

Output row cap after sort. usize::MAX = no explicit LIMIT.

sort_keys: Vec<(String, bool)>

Post-aggregation sort keys: (column_name, ascending).

NestedLoopJoin

Nested loop join: fallback for non-equi joins.

Fields

left_collection: String

right_collection: String

condition: Vec<u8>

Join condition as serialized Vec<ScanFilter>.

join_type: String

limit: usize

SortMergeJoin

Sort-merge join: both sides pre-sorted by join key. Optimal when both collections have index-ordered scans or when the planner sorts both sides before joining.

Fields

left_collection: String

right_collection: String

on: Vec<(String, String)>

join_type: String

limit: usize

pre_sorted: bool

If true, both sides are assumed pre-sorted by join key (skip sort phase).

FacetCounts

Multi-facet aggregation: compute facet counts for multiple fields in a single query, sharing the filter evaluation across all facets.

Fields

collection: String

filters: Vec<u8>

Serialized Vec<ScanFilter> predicates (MessagePack).

fields: Vec<String>

Field names to facet on (each produces a [{value, count}] array).

limit_per_facet: usize

Maximum number of values to return per facet field (0 = unlimited).

RecursiveScan

Recursive CTE: iterative fixed-point execution.

Executes the base query once, then repeatedly executes the recursive query using the previous iteration’s results as the working table, until no new rows are produced (fixed point).

Fields

collection: String

Collection for the recursive scan.

base_filters: Vec<u8>

Base query filters (seeded once).

recursive_filters: Vec<u8>

Recursive step filters (applied to working table each iteration).

join_link: Option<(String, String)>

Equi-join link for tree-traversal recursion: (collection_field, working_table_field). Each iteration finds rows where collection_field value matches a working_table_field value from the previous iteration.

max_iterations: usize

Maximum iterations to prevent infinite loops. Default: 100.

distinct: bool

Whether to deduplicate results (UNION vs UNION ALL).

limit: usize

RecursiveValue

Value-generating recursive CTE: iterative expression evaluation.

No collection is needed. The executor evaluates the anchor expressions once to produce the first row, then repeatedly applies the step expressions to the previous row until the condition becomes false, a fixed point is reached, or max_depth is exceeded (typed error).

Fields

cte_name: String

CTE name (used in error messages).

columns: Vec<String>

Column names (length == init_exprs.len() == step_exprs.len()).

init_exprs: Vec<String>

Anchor SELECT expressions as raw SQL text.

step_exprs: Vec<String>

Recursive step SELECT expressions as raw SQL text.

condition: Option<String>

Optional WHERE condition as raw SQL text.

max_depth: usize

Maximum iterations before returning a depth-exceeded error.

distinct: bool

Whether to deduplicate (UNION vs UNION ALL).

LateralTopK

LATERAL equi-correlated top-K: scan inner_collection once per outer row, applying the equi-correlation as an equality filter, then return the top inner_limit rows ordered by inner_order_by.

The executor first runs outer_plan to materialise outer rows, then for each outer row injects the equi-correlation as an equality filter on inner_collection, applies inner_order_by, and keeps the top inner_limit rows. Output rows are (outer_row merged with inner_row).

correlation_keys are (outer_col, inner_col) equi-join pairs.

Fields

outer_plan: Box<PhysicalPlan>

Sub-plan that produces the outer (driving) rows.

outer_alias: String

Alias qualifying the outer columns in output rows.

inner_collection: String

Inner collection to scan per outer row.

inner_filters: Vec<u8>

Non-correlated filters applied to every inner scan (msgpack bytes).

inner_order_by: Vec<(String, bool)>

Sort keys for the inner per-outer-row result. Each entry is (field_name, ascending).

inner_limit: usize

Maximum inner rows per outer row.

correlation_keys: Vec<(String, String)>

Equi-join pairs (outer_col, inner_col).

lateral_alias: String

Alias qualifying inner columns in output rows.

projection: Vec<JoinProjection>

Output projection (empty = all columns).

left_join: bool

LEFT join semantics: preserve outer rows when inner is empty.

LateralLoop

LATERAL nested-loop: for each outer row, re-execute the inner plan with correlated values injected as additional equality filters.

The executor runs outer_plan, then for each outer row reads the outer_col values from correlation_predicates, appends equality filters on the corresponding inner_col fields, and scans inner_collection.

Fields

outer_plan: Box<PhysicalPlan>

Sub-plan that produces the outer (driving) rows.

outer_alias: String

Alias qualifying the outer columns in output rows.

inner_collection: String

Inner collection to scan per outer row.

inner_filters: Vec<u8>

Base inner filters (non-correlated, msgpack bytes).

correlation_predicates: Vec<(String, String)>

Correlated predicates: (inner_field, outer_field).

lateral_alias: String

Alias qualifying inner columns in output rows.

projection: Vec<JoinProjection>

Output projection (empty = all columns).

left_join: bool

LEFT join semantics.

outer_row_cap: usize

Hard cap on outer rows. Queries that exceed this cap return a typed LateralCapExceeded error instead of silently truncating.

Trait Implementations

impl Clone for QueryOp

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for QueryOp

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for QueryOp

fn deserialize<__D>(__deserializer: __D) -> Result<Self, __D::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl<'__msgpack_de> FromMessagePack<'__msgpack_de> for QueryOp

fn read<R: Read<'__msgpack_de>>(reader: &mut R) -> Result<Self, Error>

where Self: Sized,

Reads the MessagePack representation of this value from the provided reader.

impl PartialEq for QueryOp

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl Serialize for QueryOp

fn serialize<__S>(&self, __serializer: __S) -> Result<__S::Ok, __S::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.

impl StructuralPartialEq for QueryOp

impl ToMessagePack for QueryOp

fn write<W: Write>(&self, writer: &mut W) -> Result<(), Error>

Writes the MessagePack representation of this value into the provided writer.