feat: physical planner — plan() builds PhysicalNode tree with Placement + Exchange (Layer 5)

[zzylol]

Files changed

controller/src/algebra/physical.rs (+471 lines)

New function: plan(expr: &QueryExpr, config: &PhysicalPlannerConfig) → PhysicalNode

Walks the optimized QueryExpr tree bottom-up and produces a PhysicalNode tree where each node has:

• op: PhysicalOp — concrete operator (OtelSketchBuild, SketchMerge, TopK, DbQuery, Exchange, etc.)
• placement: Placement — which pipeline stage runs it (AgentCollector, BackendCollector, QueryEngine, Database)
• cost: PhysicalCost — estimated bandwidth, memory, CPU
• children: Vec<PhysicalNode> — child nodes

New function: decide_sketch_placement(resolved, budgets) → Placement

Assigns sketch operations to Agent by default. Defers to BackendCollector when agent memory budget is exceeded, and to QueryEngine when both are exceeded.

New function: insert_exchange_if_needed(node)

When a parent node's placement differs from a child's, wraps the child in an Exchange node with the appropriate format:

• Agent → Backend: ExchangeFormat::Otlp
• Agent → QueryEngine: ExchangeFormat::Otlp
• Backend → QueryEngine: ExchangeFormat::SketchBinary
• Agent → Database: ExchangeFormat::RawSamples

New struct: PhysicalPlannerConfig

Holds StageResourceBudgets for budget-based deferral decisions.

New methods on PhysicalNode:

• node_count() — total nodes in tree
• placements() — distinct placements in tree
• exchange_count() — number of Exchange boundaries

Placement rules implemented in plan_node():

QueryExpr node	Placement
Source	AgentCollector
Filter	Same as child
SketchAgg, WindowedAgg	AgentCollector (or deferred by budget)
Partition, Merge, Dedup	BackendCollector
TopK, HistogramQuantile, BinaryOp, PromQLSubquery	QueryEngine
Aggregate (non-sketch exact)	Database
Sort, Limit, Project, Window, WindowFunc	Inherit child
Join, JoinSketch, SetOp	QueryEngine

controller/docs/query-to-sketch-translation.md (updated)

Layer 5 sections for all four concrete examples rewritten to show actual plan() output as PhysicalNode trees with Placement and Exchange annotations:

Example 4 (PromQL quantile): single-stage, all AgentCollector, 0 Exchanges

OtelSketchBuild { DDSketch, OtelTumblingFlush(5m) }  [AgentCollector]
  └── Filter                                         [AgentCollector]
        └── OtlpScan                                 [AgentCollector]

Example 5 (PromQL topk): 3-stage, 2 Exchanges

TopK                                [QueryEngine]
  └── Exchange(SketchBinary)        [QueryEngine]
        └── HashAggregate           [BackendCollector]
              └── Exchange(Otlp)    [BackendCollector]
                    └── OtelSketchBuild(CountSketch)  [AgentCollector]

Example 6 (SQL AVG): Agent→Database passthrough

DbQuery { GROUP BY }     [Database]
  └── Exchange(RawSamples)  [Database]
        └── OtlpScan      [AgentCollector]

Example 7 (SQL TUMBLE COUNT DISTINCT): 3-stage with HLL

TopK                                [QueryEngine]
  └── Exchange(SketchBinary)        [QueryEngine]
        └── HashAggregate           [BackendCollector]
              └── Exchange(Otlp)    [BackendCollector]
                    └── OtelSketchBuild(HLL)  [AgentCollector]

Tests

8 new tests in physical.rs:

• plan_simple_sketch_at_agent — SketchAgg placed at Agent
• plan_windowed_agg_has_window — WindowedAgg resolves physical window
• plan_topk_at_query_engine — TopK placed at QueryEngine
• plan_topk_inserts_exchange — Exchange inserted between Agent and QueryEngine
• plan_partition_at_backend — Partition placed at Backend
• plan_aggregate_at_database — non-sketch Aggregate placed at Database
• plan_full_pipeline_has_multiple_stages — verifies 3 stages + ≥2 exchanges
• plan_budget_deferral — tiny agent budget defers sketch to Backend

323 total tests pass.

🤖 Generated with Claude Code