From 42748abbc4e1ba5afb2e9dba5ec5009c10f1f6e5 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Jan 2026 14:42:01 +0000 Subject: [PATCH 1/4] docs: Update bridge design for engine-runtime separation - Redesigned architecture with three crates: - grism-engine: Common physical layer (operators, expressions, schemas) - grism-local: Local runtime (in-memory and Lance storage) - grism-ray: Distributed runtime on Ray (renamed from grism-distributed) - Added detailed crate responsibilities and architecture diagram - Defined ExecutionContextTrait for runtime abstraction - Documented Exchange operator and stage splitting for Ray - Added migration path from current structure - Added implementation checklist - Maintained backward compatibility for local execution API Co-authored-by: chenxm35 --- specs/8_bridge_distributed_impl.md | 728 +++++++++++++++++++---------- 1 file changed, 489 insertions(+), 239 deletions(-) diff --git a/specs/8_bridge_distributed_impl.md b/specs/8_bridge_distributed_impl.md index 30579b2..a336364 100644 --- a/specs/8_bridge_distributed_impl.md +++ b/specs/8_bridge_distributed_impl.md @@ -1,110 +1,359 @@ -Below is a **Week-8 bridge design** that cleanly extends your Week-7 local engine into **distributed execution**, without changing semantics or operator contracts. +# Design: Engine-Runtime Separation & Distributed Execution -This is written as a **continuation of the Week-7 design**, not a rewrite. -Everything here is **additive** and explicitly aligned with **RFC-0008 (physical operators)** and **RFC-0010 (distributed execution)**. +*(Bridge from Local Engine to Ray Distributed Execution)* --- -# Design: Exchange Operator & Stage Splitting +## 0. Design Objective -*(Week-8 bridge: Local → Distributed Execution)* +Week-7 delivered a **fully executable local physical plan** with streaming, Arrow-native operators. This document extends that foundation to support **distributed execution on Ray** while maintaining a clean architecture separation. ---- +### Goals + +1. **Runtime Separation**: Extract common engine components that work across local and distributed execution +2. **Local Runtime**: Dedicated crate for single-machine execution with in-memory and Lance storage +3. **Ray Runtime**: Dedicated crate for distributed execution using Ray as orchestration layer +4. **Zero Semantic Drift**: Distribution MUST NOT change query results -## 0. Design Objective (Why this exists) +### Key Principle -Week-7 gave us: +> **The engine defines what to compute. The runtime defines how to execute.** -* A **fully executable local physical plan** -* Streaming, Arrow-native operators -* Blocking operators clearly marked +--- -Week-8 must add: +## 1. Crate Architecture -* **Distributed execution** -* **Data repartitioning** -* **Parallel execution** -* **Failure isolation** +### 1.1 Architecture Overview -Without: +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ Application Layer │ +│ (Python API, gRPC, CLI) │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ grism-engine (Common) │ +│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ ┌────────────────┐ │ +│ │ Physical │ │ Operators │ │ Expression │ │ Physical │ │ +│ │ Plan Model │ │ (Exec) │ │ Evaluator │ │ Schema │ │ +│ └─────────────┘ └──────────────┘ └───────────────┘ └────────────────┘ │ +│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │ +│ │ Operator │ │ Schema │ │ Memory & │ │ +│ │ Traits │ │ Inference │ │ Metrics │ │ +│ └─────────────┘ └──────────────┘ └───────────────┘ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ │ + ▼ ▼ +┌───────────────────────────────┐ ┌───────────────────────────────────────┐ +│ grism-local (Runtime) │ │ grism-ray (Runtime) │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ LocalExecutor │ │ │ │ RayExecutor │ │ +│ │ - Pull-based │ │ │ │ - Stage-based │ │ +│ │ - Single-threaded │ │ │ │ - Parallel workers │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ LocalPhysicalPlanner │ │ │ │ DistributedPlanner │ │ +│ │ - No Exchange ops │ │ │ │ - Exchange insertion │ │ +│ │ - Direct execution │ │ │ │ - Stage splitting │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ ExecutionContext │ │ │ │ StageExecutor │ │ +│ │ - InMemoryStorage │ │ │ │ - Ray task management │ │ +│ │ - LanceStorage │ │ │ │ - Arrow IPC transport │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +└───────────────────────────────┘ └───────────────────────────────────────┘ + │ │ + └────────────────┬───────────────────┘ + ▼ +┌─────────────────────────────────────────────────────────────────────────────┐ +│ grism-storage │ +│ (InMemoryStorage, LanceStorage, Catalog) │ +└─────────────────────────────────────────────────────────────────────────────┘ +``` -* Changing logical semantics -* Changing physical operator contracts -* Introducing implicit magic +### 1.2 Crate Responsibilities -> **Key principle** -> Distribution is achieved by **cutting** the physical plan into **stages**, connected by an explicit **Exchange** operator. +| Crate | Responsibility | Dependencies | +|-------|----------------|--------------| +| **grism-engine** | Physical plan model, operators, expressions, schema inference | grism-core, grism-logical | +| **grism-local** | Local single-machine execution runtime | grism-engine, grism-storage | +| **grism-ray** | Distributed Ray-based execution runtime | grism-engine, grism-storage, ray | --- -## 1. Core Concepts +## 2. grism-engine (Common Components) + +The `grism-engine` crate contains all **runtime-agnostic** components that both local and distributed execution share. -### 1.1 Exchange is a Physical Operator +### 2.1 Physical Plan Model -**Exchange is NOT a runtime trick.** -It is a **first-class physical operator** that: +```rust +// Core plan structures (runtime-agnostic) +pub struct PhysicalPlan { + pub root: Box, + pub properties: PlanProperties, +} -* Repartitions data -* Introduces a synchronization boundary -* Separates execution stages +pub struct PlanProperties { + pub execution_mode: ExecutionMode, + pub partitioning: PartitioningSpec, + pub blocking: bool, +} +``` -```text -Local Plan: -Scan → Filter → Expand → Aggregate → Collect +### 2.2 Physical Operators + +All physical operators implement the `PhysicalOperator` trait: + +```rust +#[async_trait] +pub trait PhysicalOperator: Send + Sync { + /// Execute this operator, pulling from children + async fn execute( + &self, + ctx: &dyn ExecutionContextTrait, + ) -> Result, GrismError>; + + /// Operator's output schema + fn schema(&self) -> &PhysicalSchema; + + /// Operator capabilities + fn capabilities(&self) -> OperatorCaps; + + /// Child operators + fn children(&self) -> Vec<&dyn PhysicalOperator>; +} +``` + +**Operators in grism-engine:** + +| Operator | Description | Blocking | +|----------|-------------|----------| +| `NodeScanExec` | Scan nodes by label | No | +| `HyperedgeScanExec` | Scan hyperedges by label | No | +| `FilterExec` | Filter rows by predicate | No | +| `ProjectExec` | Project/compute columns | No | +| `LimitExec` | Limit rows with offset | No | +| `SortExec` | Multi-key sorting | **Yes** | +| `HashAggregateExec` | Aggregation with GROUP BY | **Yes** | +| `AdjacencyExpandExec` | Binary edge traversal | No | +| `RoleExpandExec` | N-ary hyperedge traversal | No | +| `UnionExec` | Union of two inputs | No | +| `RenameExec` | Rename columns | No | +| `EmptyExec` | Empty result | No | + +### 2.3 Expression Evaluator + +The `ExprEvaluator` converts `LogicalExpr` to Arrow arrays: + +```rust +pub struct ExprEvaluator; -Distributed Plan: -Scan → Filter → Expand → Exchange → Aggregate → Collect +impl ExprEvaluator { + pub fn evaluate( + expr: &LogicalExpr, + batch: &RecordBatch, + ) -> Result; +} +``` + +**Supported expressions:** +- Literals: Int64, Float64, String, Bool, Null +- Column references: Direct and qualified +- Binary operations: +, -, *, /, %, =, <>, <, <=, >, >=, AND, OR +- Unary operations: NOT, IS NULL, IS NOT NULL, NEG +- CASE expressions, IN lists, BETWEEN + +### 2.4 Schema Inference + +```rust +pub mod schema_inference { + pub fn infer_expr_type(expr: &LogicalExpr, input_schema: &Schema) -> DataType; + pub fn build_project_schema(exprs: &[LogicalExpr], input: &Schema) -> Schema; + pub fn build_aggregate_schema(aggs: &[AggExpr], groups: &[LogicalExpr]) -> Schema; +} +``` + +### 2.5 Operator Capabilities + +```rust +pub struct OperatorCaps { + pub streaming: bool, + pub blocking: bool, + pub parallel_safe: bool, + pub requires_partitioning: Option, +} +``` + +### 2.6 Memory & Metrics + +```rust +// Memory management trait +pub trait MemoryManager: Send + Sync { + fn try_reserve(&self, bytes: usize) -> Result; + fn current_usage(&self) -> usize; +} + +// Metrics collection trait +pub trait MetricsSink: Send + Sync { + fn record_rows(&self, operator: &str, rows: usize); + fn record_batches(&self, operator: &str, count: usize); + fn record_time(&self, operator: &str, duration: Duration); +} +``` + +### 2.7 Execution Context Trait + +```rust +/// Trait for execution context - implemented by both local and ray runtimes +#[async_trait] +pub trait ExecutionContextTrait: Send + Sync { + fn storage(&self) -> &dyn Storage; + fn snapshot_id(&self) -> SnapshotId; + fn memory_manager(&self) -> &dyn MemoryManager; + fn metrics_sink(&self) -> Option<&dyn MetricsSink>; + fn is_cancelled(&self) -> bool; +} ``` --- -### 1.2 Stage = Executable Physical Subplan +## 3. grism-local (Local Runtime) + +The `grism-local` crate provides single-machine execution with support for both in-memory and persistent storage. + +### 3.1 LocalExecutor + +```rust +pub struct LocalExecutor { + config: RuntimeConfig, +} + +impl LocalExecutor { + pub fn new() -> Self; + pub fn with_config(config: RuntimeConfig) -> Self; -A **Stage** is: + pub async fn execute( + &self, + plan: PhysicalPlan, + storage: Arc, + snapshot: SnapshotId, + ) -> Result; +} +``` -* A connected sub-DAG of physical operators -* Executed as a unit (locally or remotely) -* Has no internal Exchange operators +### 3.2 LocalPhysicalPlanner ```rust -struct ExecutionStage { - stage_id: StageId, - plan: PhysicalSubPlan, - input_partitioning: Option, - output_partitioning: Option, +pub struct LocalPhysicalPlanner { + config: PlannerConfig, } + +impl PhysicalPlanner for LocalPhysicalPlanner { + fn plan(&self, logical: &LogicalPlan) -> Result; +} +``` + +**Key characteristics:** +- No Exchange operators inserted +- Direct operator tree execution +- Pull-based batch streaming + +### 3.3 LocalExecutionContext + +```rust +pub struct LocalExecutionContext { + storage: Arc, + snapshot_id: SnapshotId, + memory_manager: Arc, + metrics_sink: Option>, + cancellation: CancellationHandle, +} + +impl ExecutionContextTrait for LocalExecutionContext { + // Implement all trait methods +} +``` + +### 3.4 Storage Support + +| Storage Backend | Description | Use Case | +|-----------------|-------------|----------| +| `InMemoryStorage` | Hash-map based storage | Testing, small datasets | +| `LanceStorage` | Lance format file storage | Production, large datasets | + +```rust +// Example usage with InMemoryStorage +let storage = InMemoryStorage::new(); +let executor = LocalExecutor::new(); +let result = executor.execute(plan, Arc::new(storage), SnapshotId::default()).await?; + +// Example usage with LanceStorage +let storage = LanceStorage::open("/data/graph.lance").await?; +let executor = LocalExecutor::new(); +let result = executor.execute(plan, Arc::new(storage), SnapshotId::default()).await?; ``` --- -## 2. Exchange Operator Design +## 4. grism-ray (Distributed Runtime) + +The `grism-ray` crate provides distributed execution using Ray as the orchestration layer. + +### 4.1 Core Concepts -### 2.1 ExchangeExec (Physical Operator) +#### Exchange Operator + +Exchange is a **first-class physical operator** that: +- Repartitions data +- Introduces a synchronization boundary +- Separates execution stages ```rust pub struct ExchangeExec { pub partitioning: PartitioningSpec, pub mode: ExchangeMode, + pub child: Box, +} + +pub enum ExchangeMode { + Shuffle, // repartition across workers + Broadcast, // replicate to all workers + Gather, // many → one } ``` -### 2.2 ExchangeMode +#### Execution Stage + +A **Stage** is a connected sub-DAG of physical operators executed as a unit: ```rust -pub enum ExchangeMode { - Local, // no-op (Week 7 compatibility) - Shuffle, // repartition across workers - Broadcast, // replicate to all workers - Gather, // many → one +pub struct ExecutionStage { + pub stage_id: StageId, + pub plan: PhysicalSubPlan, + pub input_partitioning: Option, + pub output_partitioning: Option, } ``` -Only `Local` is used in Week-7. -All others activate in Week-8. +### 4.2 DistributedPlanner ---- +```rust +pub struct DistributedPlanner { + config: DistributedPlannerConfig, +} -### 2.3 PartitioningSpec (Normative) +impl DistributedPlanner { + /// Plan a logical plan into distributed stages + pub fn plan(&self, logical: &LogicalPlan) -> Result; + + /// Split physical plan into stages + pub fn split_into_stages(&self, plan: PhysicalPlan) -> Vec; +} +``` + +### 4.3 Partitioning Specification ```rust pub enum PartitioningSpec { @@ -120,77 +369,64 @@ pub enum PartitioningSpec { entity: EntityType, // Node | Hyperedge }, Single, + RoundRobin { + partitions: usize, + }, } ``` -**Important**: -Partitioning is **explicit**, never inferred at runtime. - ---- - -### 2.4 Exchange Semantics (Non-Negotiable) - -| Property | Guarantee | -| ------------ | ---------------------------------------- | -| Completeness | All rows forwarded | -| Disjointness | Each row appears once (except Broadcast) | -| Determinism | Same input → same partition | -| Barrier | Downstream waits for upstream readiness | - -Exchange is a **semantic boundary**, not an optimization hint. - ---- - -## 3. Why Exchange Is Explicit (Architectural Justification) - -**Why not implicit shuffle?** +### 4.4 RayExecutor -Because Grism guarantees: +```rust +pub struct RayExecutor { + config: RayExecutorConfig, +} -> *Parallelism must not change meaning.* +impl RayExecutor { + pub async fn execute( + &self, + stages: Vec, + storage: Arc, + ) -> Result; +} +``` -Explicit Exchange ensures: +### 4.5 Stage Execution Flow -* Explainability (`EXPLAIN DISTRIBUTED`) -* Cost modeling -* Debuggability -* Operator capability checks -* Hypergraph adjacency correctness +```text +┌──────────────────────────────────────────────────────────────────────┐ +│ DistributedPlan │ +├──────────────────────────────────────────────────────────────────────┤ +│ │ +│ Stage 0 (parallel) Stage 1 (parallel) Stage 2 │ +│ ┌─────────────────┐ ┌─────────────────┐ ┌──────────┐ │ +│ │ Scan → Filter │───────▶ │ Expand → Agg │───▶│ Collect │ │ +│ │ → Project │ Exchange│ (partial) │ │ (final) │ │ +│ └─────────────────┘ (Hash) └─────────────────┘ └──────────┘ │ +│ │ │ │ │ +│ ▼ ▼ ▼ │ +│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │ +│ │ Worker 1 │ │ Worker 1 │ │ Driver │ │ +│ │ Worker 2 │ │ Worker 2 │ │ │ │ +│ │ Worker 3 │ │ Worker 3 │ │ │ │ +│ └─────────────┘ └─────────────┘ └──────────┘ │ +│ │ +└──────────────────────────────────────────────────────────────────────┘ +``` --- -## 4. Stage Splitting Algorithm - -### 4.1 Inputs +## 5. Stage Splitting Algorithm -* Fully planned **PhysicalPlan** -* Operator metadata: - - * `blocking` - * `requires_global_view` - * `parallelizable` -* Execution target: `Local | Distributed` - ---- - -### 4.2 Stage Boundary Rules (Normative) +### 5.1 Stage Boundary Rules A **new stage MUST start** at: -1. Any `ExchangeExec` -2. Any **blocking operator** if distribution is enabled -3. Any operator that: +1. Any `ExchangeExec` operator +2. Any **blocking operator** when distribution is enabled +3. Any operator that requires global state - * requires global state - * cannot be parallelized - -```text -[Stage A] → Exchange → [Stage B] -``` - ---- - -### 4.3 Stage Splitting Pseudocode +### 5.2 Splitting Algorithm ```rust fn split_into_stages(plan: PhysicalPlan) -> Vec { @@ -212,208 +448,222 @@ fn split_into_stages(plan: PhysicalPlan) -> Vec { --- -## 5. Execution Model After Splitting +## 6. Exchange Semantics -### 5.1 Distributed Execution Flow +### 6.1 Guarantees -```text -Stage 0 (parallel) - ├─ Worker 1 - ├─ Worker 2 - └─ Worker 3 - ↓ - Exchange - ↓ -Stage 1 (parallel or single) -``` +| Property | Guarantee | +|----------|-----------| +| Completeness | All rows forwarded | +| Disjointness | Each row appears once (except Broadcast) | +| Determinism | Same input → same partition | +| Barrier | Downstream waits for upstream readiness | -Each stage: +### 6.2 Exchange Execution Steps (Shuffle) -* Executes independently -* Produces Arrow batches -* Hands off via Exchange +1. Upstream stage emits `RecordBatch` +2. Batch partitioned using `PartitioningSpec` +3. Sub-batches sent to Ray object store via Arrow IPC +4. Downstream tasks pull assigned partitions --- -## 6. Exchange Execution (Ray Backend) - -### 6.1 Control Plane vs Data Plane +## 7. Blocking Operators in Distributed Context -| Plane | Responsibility | -| ------- | ----------------------- | -| Control | Stage graph, scheduling | -| Data | Arrow batch movement | +### 7.1 Aggregate (Two-Phase) -Ray owns **control**, Rust owns **execution**. - ---- - -### 6.2 Exchange Execution Steps - -**Shuffle Exchange example**: +```text +Stage 0: PartialAggregate (parallel) + ↓ Exchange(Hash by group key) +Stage 1: FinalAggregate +``` -1. Upstream stage emits `RecordBatch` -2. Batch partitioned using `PartitioningSpec` -3. Sub-batches sent to Ray object store -4. Downstream tasks pull assigned partitions +### 7.2 Sort (Two-Phase) -No operator knows **where** data comes from. +```text +Stage 0: LocalSort (parallel) + ↓ Exchange(Range) +Stage 1: MergeSort +``` --- -## 7. Expand-Aware Partitioning (Critical) +## 8. Expand-Aware Partitioning -### 7.1 Binary Expand (Adjacency-Aware) +### 8.1 Adjacency Partitioning -**Preferred strategy**: +For graph traversal workloads, adjacency-based partitioning keeps most expansions local: ```rust PartitioningSpec::Adjacency { - entity: Node, + entity: EntityType::Node, } ``` -Guarantee: +### 8.2 N-ary Expand + +N-ary hyperedge expansion often requires shuffle: -* Most expands are local -* Cross-partition traversal is explicit +```text +Stage 0: Scan nodes + ↓ Exchange(Adjacency) +Stage 1: RoleExpand (hyperedges co-located with nodes) +``` --- -### 7.2 N-ary Expand (RoleExpand) +## 9. Control Plane vs Data Plane -* Often forces shuffle -* Exchange inserted **before** Expand -* Planner decides based on cost +| Plane | Responsibility | Owner | +|-------|----------------|-------| +| Control | Stage graph, scheduling, progress | Ray | +| Data | Arrow batch movement, execution | Rust | -This preserves correctness for hyperedges. +**Ray orchestrates, Rust executes.** --- -## 8. Blocking Operators & Global Semantics - -### 8.1 Aggregate +## 10. Failure & Retry Semantics -Two-phase aggregation pattern: +Stage boundaries define **failure domains**: -```text -Stage 0: PartialAggregate (parallel) - ↓ Exchange(Hash by group key) -Stage 1: FinalAggregate -``` +| Failure | Recovery | +|---------|----------| +| Worker crash | Retry stage | +| Exchange failure | Re-run upstream stage | +| Blocking stage failure | Restart stage | -This is **planned explicitly**. +No partial stage result is reused unless idempotent. --- -### 8.2 Sort +## 11. Explainability -```text -Stage 0: LocalSort - ↓ Exchange(Range) -Stage 1: MergeSort -``` +### 11.1 EXPLAIN DISTRIBUTED -No hidden coordination. +```text +Stage 0: + NodeScan(Person) → Filter(age > 21) → Project(name, age) + Parallelism: 8 + Output: HashPartition(node_id) ---- + ↓ Exchange(Shuffle, Hash(node_id)) -## 9. Operator Capability Contract (Extended) +Stage 1: + Aggregate(COUNT(*), GROUP BY city) + Parallelism: 4 + Output: HashPartition(city) -Operators must declare: + ↓ Exchange(Gather) -```rust -struct OperatorCaps { - streaming: bool, - blocking: bool, - parallel_safe: bool, - requires_partitioning: Option, -} +Stage 2: + Collect + Parallelism: 1 ``` -Planner uses this to: +--- + +## 12. Migration Path -* Insert Exchange -* Split stages -* Reject illegal plans +### 12.1 Current State ---- +``` +grism-engine/ # Contains both engine AND local runtime +grism-distributed/ # Partial Ray integration +``` -## 10. Failure & Retry Semantics (Week-8 Scope) +### 12.2 Target State -Stage boundaries define **failure domains**. +``` +grism-engine/ # Common: operators, expressions, schemas, traits +grism-local/ # Runtime: LocalExecutor, LocalPhysicalPlanner +grism-ray/ # Runtime: RayExecutor, DistributedPlanner, Exchange +``` -| Failure | Recovery | -| ---------------------- | --------------------- | -| Worker crash | Retry stage | -| Exchange failure | Re-run upstream stage | -| Blocking stage failure | Restart stage | +### 12.3 Migration Steps -No partial stage result is reused unless idempotent. +1. **Extract common traits** from `grism-engine` into stable interfaces +2. **Create `grism-local`** by moving `LocalExecutor`, `LocalPhysicalPlanner`, `ExecutionContext` +3. **Rename `grism-distributed`** to `grism-ray` +4. **Implement Exchange operator** in `grism-ray` +5. **Implement stage splitting** in `grism-ray` +6. **Keep current local engine functionality unchanged** - users should see no difference --- -## 11. Explainability Guarantees +## 13. API Compatibility -`EXPLAIN DISTRIBUTED` must show: +### 13.1 Local Execution (Unchanged) -```text -Stage 0: - NodeScan → Filter → Expand - Parallelism: 8 - Output: HashPartition(node_id) +```rust +// Before and after: identical API +use grism_local::{LocalExecutor, LocalPhysicalPlanner}; -Stage 1: - Aggregate → Collect - Parallelism: 1 +let planner = LocalPhysicalPlanner::new(); +let plan = planner.plan(&logical_plan)?; + +let executor = LocalExecutor::new(); +let result = executor.execute(plan, storage, snapshot).await?; ``` -This is **non-optional**. +### 13.2 Distributed Execution (New) ---- +```rust +use grism_ray::{RayExecutor, DistributedPlanner}; -## 12. Integration with Week-7 Local Engine +let planner = DistributedPlanner::new(); +let stages = planner.plan(&logical_plan)?; -### 12.1 Local Compatibility +let executor = RayExecutor::connect("ray://cluster:10001")?; +let result = executor.execute(stages, storage).await?; +``` -When execution mode = `Local`: +--- -* ExchangeExec becomes **no-op** -* Single stage only -* No scheduling overhead +## 14. Implementation Checklist -Same code path, different backend. +### Phase 1: Engine Separation +- [ ] Define `ExecutionContextTrait` in grism-engine +- [ ] Define `PhysicalPlanner` trait in grism-engine +- [ ] Ensure all operators depend only on traits, not concrete types ---- +### Phase 2: Create grism-local +- [ ] Move `LocalExecutor` to grism-local +- [ ] Move `LocalPhysicalPlanner` to grism-local +- [ ] Move `ExecutionContext` implementation to grism-local +- [ ] Add re-exports for backward compatibility -## 13. Minimal Week-8 Implementation Checklist +### Phase 3: Rename to grism-ray +- [ ] Rename `grism-distributed` to `grism-ray` +- [ ] Update all imports and references +- [ ] Implement `ExchangeExec` operator +- [ ] Implement `DistributedPlanner` with stage splitting +- [ ] Implement Ray task submission -✔ ExchangeExec struct -✔ PartitioningSpec implemented -✔ Stage splitter implemented -✔ Ray stage executor wrapper -✔ Partial/final aggregate variants -✔ Expand-aware partition rules -✔ Distributed EXPLAIN output +### Phase 4: Integration +- [ ] Integrate with Ray object store for Arrow IPC +- [ ] Implement two-phase aggregation +- [ ] Implement range-partitioned sort +- [ ] Add distributed EXPLAIN support --- -## 14. Architectural Payoff +## 15. Summary -This design ensures: +| Crate | What It Contains | Key Types | +|-------|------------------|-----------| +| **grism-engine** | Runtime-agnostic physical layer | `PhysicalPlan`, `PhysicalOperator`, `ExprEvaluator`, `OperatorCaps` | +| **grism-local** | Single-machine execution | `LocalExecutor`, `LocalPhysicalPlanner`, `LocalExecutionContext` | +| **grism-ray** | Distributed Ray execution | `RayExecutor`, `DistributedPlanner`, `ExchangeExec`, `ExecutionStage` | -* **Zero semantic drift** -* **Local engine remains reference** -* **Distributed execution is explainable** -* **Hypergraph Expand remains correct** -* **Future extensions (GPU, spill, adaptive) slot in cleanly** +### Architectural Guarantees ---- - -## Final Summary +1. **Zero semantic drift**: Local and distributed execution produce identical results +2. **Local engine unchanged**: Current local execution API remains stable +3. **Explicit distribution**: Exchange operators are visible in plans +4. **Hypergraph-correct**: Expand operators preserve adjacency semantics +5. **Explainable**: Distributed plans can be inspected and debugged > **Exchange is the only way data moves across stages.** > **Stages are the only unit of distribution.** > **Operators never know they are distributed.** - -This is the exact architectural separation that allows Grism to scale **without ever compromising its hypergraph semantics**. \ No newline at end of file From add615a5dd0f6029b748ef5c37df60be1afab634 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Jan 2026 14:47:24 +0000 Subject: [PATCH 2/4] docs: Add progress file for engine-runtime separation spec update Co-authored-by: chenxm35 --- _workdir/progress-2026-01-22-013.md | 89 +++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 _workdir/progress-2026-01-22-013.md diff --git a/_workdir/progress-2026-01-22-013.md b/_workdir/progress-2026-01-22-013.md new file mode 100644 index 0000000..c56779c --- /dev/null +++ b/_workdir/progress-2026-01-22-013.md @@ -0,0 +1,89 @@ +--- +date: 2026-01-22 +session: engine-runtime-separation-spec +objective: Update bridge design spec for engine-runtime separation architecture +status: completed +--- + +## Objective + +Update `specs/8_bridge_distributed_impl.md` to define the architecture for separating common engine components from runtime-specific implementations, enabling both local and Ray-based distributed execution. + +## Completed + +1. **Reviewed existing specifications** + - Read `specs/7_local_engine_milestone.md` for current local engine architecture + - Read `specs/8_bridge_distributed_impl.md` for existing distributed bridge design + - Reviewed `specs/rfc-0010.md` for distributed execution model + - Reviewed `specs/rfc-namings.md` for naming conventions + +2. **Updated `specs/8_bridge_distributed_impl.md`** with: + - New three-crate architecture diagram + - `grism-engine`: Common physical layer (operators, expressions, schemas, traits) + - `grism-local`: Local runtime (in-memory and Lance storage support) + - `grism-ray`: Distributed runtime on Ray (renamed from `grism-distributed`) + - Defined `ExecutionContextTrait` for runtime abstraction + - Detailed Exchange operator and stage splitting for Ray + - Added migration path from current structure + - Implementation checklist for phased migration + - API compatibility examples showing unchanged local execution API + +## Files Changed + +| File | Description | +|------|-------------| +| `specs/8_bridge_distributed_impl.md` | Complete rewrite with engine-runtime separation architecture | + +## Tests + +``` +make test +All 335 tests passed (90 grism-core, 97 grism-engine, 91 grism-logical, 46 grism-optimizer, 5 grism-storage, plus doc tests) +``` + +## Lint + +``` +make lint +0 warnings, 0 errors - PASSED +``` + +## Notes + +### Architecture Summary + +The new design separates concerns into three crates: + +1. **grism-engine (Common)** + - `PhysicalPlan`, `PhysicalOperator` trait + - All operator implementations (`*Exec`) + - `ExprEvaluator` for expression evaluation + - Schema inference + - `OperatorCaps` for capability declarations + - Memory and metrics traits + +2. **grism-local (Local Runtime)** + - `LocalExecutor` - pull-based single-machine execution + - `LocalPhysicalPlanner` - no Exchange operators + - `LocalExecutionContext` - implements `ExecutionContextTrait` + - Support for `InMemoryStorage` and `LanceStorage` + +3. **grism-ray (Distributed Runtime)** + - `RayExecutor` - stage-based distributed execution + - `DistributedPlanner` - Exchange insertion, stage splitting + - `ExchangeExec` operator for data repartitioning + - Ray task management and Arrow IPC transport + +### Key Design Decisions + +- **Zero semantic drift**: Local and distributed execution must produce identical results +- **Explicit distribution**: Exchange operators are visible in plans (no hidden shuffles) +- **Backward compatibility**: Current local execution API remains unchanged +- **Stage-based execution**: Stages are the only unit of distribution + +## Next Steps + +1. Extract common traits (`ExecutionContextTrait`, `PhysicalPlanner`) from grism-engine +2. Create `grism-local` crate with moved components +3. Rename `grism-distributed` to `grism-ray` +4. Implement Exchange operator and stage splitting in grism-ray From 286c943f299435d32521585d10333cc9ed10b05f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Jan 2026 14:58:54 +0000 Subject: [PATCH 3/4] docs: Add RFC-0102 for execution engine architecture - Created RFC-0102 defining the finalized engine architecture: - Common engine layer (grism-engine): operators, expressions, schemas - Local runtime (grism-local): single-machine execution - Ray runtime (grism-ray): distributed execution on Ray - Updated 8_bridge_distributed_impl.md as implementation guide: - Current code status and what works today - Detailed migration plan with 6 phases - Concrete code examples and patterns - Implementation checklist - References RFC-0102 for authoritative design - Updated rfc-index.md to include RFC-0102 Co-authored-by: chenxm35 --- specs/8_bridge_distributed_impl.md | 1081 +++++++++++++++------------- specs/rfc-0102.md | 751 +++++++++++++++++++ specs/rfc-index.md | 8 + 3 files changed, 1335 insertions(+), 505 deletions(-) create mode 100644 specs/rfc-0102.md diff --git a/specs/8_bridge_distributed_impl.md b/specs/8_bridge_distributed_impl.md index a336364..3f97d64 100644 --- a/specs/8_bridge_distributed_impl.md +++ b/specs/8_bridge_distributed_impl.md @@ -1,268 +1,178 @@ -# Design: Engine-Runtime Separation & Distributed Execution +# Implementation: Engine-Runtime Separation -*(Bridge from Local Engine to Ray Distributed Execution)* +*(Implementation guide for RFC-0102: Execution Engine Architecture)* --- -## 0. Design Objective +## 1. Overview -Week-7 delivered a **fully executable local physical plan** with streaming, Arrow-native operators. This document extends that foundation to support **distributed execution on Ray** while maintaining a clean architecture separation. +This document provides implementation-specific details for the engine architecture defined in **RFC-0102**. It covers: -### Goals +* Current code structure and status +* Migration path to target architecture +* Implementation considerations and technical decisions +* Concrete code examples and patterns -1. **Runtime Separation**: Extract common engine components that work across local and distributed execution -2. **Local Runtime**: Dedicated crate for single-machine execution with in-memory and Lance storage -3. **Ray Runtime**: Dedicated crate for distributed execution using Ray as orchestration layer -4. **Zero Semantic Drift**: Distribution MUST NOT change query results - -### Key Principle - -> **The engine defines what to compute. The runtime defines how to execute.** +**For architectural design, see [RFC-0102: Execution Engine Architecture](rfc-0102.md).** --- -## 1. Crate Architecture +## 2. Current Code Status -### 1.1 Architecture Overview +### 2.1 Existing Crate Structure ``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ Application Layer │ -│ (Python API, gRPC, CLI) │ -└─────────────────────────────────────────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────────────┐ -│ grism-engine (Common) │ -│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ ┌────────────────┐ │ -│ │ Physical │ │ Operators │ │ Expression │ │ Physical │ │ -│ │ Plan Model │ │ (Exec) │ │ Evaluator │ │ Schema │ │ -│ └─────────────┘ └──────────────┘ └───────────────┘ └────────────────┘ │ -│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │ -│ │ Operator │ │ Schema │ │ Memory & │ │ -│ │ Traits │ │ Inference │ │ Metrics │ │ -│ └─────────────┘ └──────────────┘ └───────────────┘ │ -└─────────────────────────────────────────────────────────────────────────────┘ - │ │ - ▼ ▼ -┌───────────────────────────────┐ ┌───────────────────────────────────────┐ -│ grism-local (Runtime) │ │ grism-ray (Runtime) │ -│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ -│ │ LocalExecutor │ │ │ │ RayExecutor │ │ -│ │ - Pull-based │ │ │ │ - Stage-based │ │ -│ │ - Single-threaded │ │ │ │ - Parallel workers │ │ -│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ -│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ -│ │ LocalPhysicalPlanner │ │ │ │ DistributedPlanner │ │ -│ │ - No Exchange ops │ │ │ │ - Exchange insertion │ │ -│ │ - Direct execution │ │ │ │ - Stage splitting │ │ -│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ -│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ -│ │ ExecutionContext │ │ │ │ StageExecutor │ │ -│ │ - InMemoryStorage │ │ │ │ - Ray task management │ │ -│ │ - LanceStorage │ │ │ │ - Arrow IPC transport │ │ -│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ -└───────────────────────────────┘ └───────────────────────────────────────┘ - │ │ - └────────────────┬───────────────────┘ - ▼ -┌─────────────────────────────────────────────────────────────────────────────┐ -│ grism-storage │ -│ (InMemoryStorage, LanceStorage, Catalog) │ -└─────────────────────────────────────────────────────────────────────────────┘ +src/ +├── grism-engine/ # Currently contains BOTH engine AND local runtime +│ ├── executor/ # LocalExecutor, ExecutionContext +│ ├── expr/ # ExprEvaluator +│ ├── memory/ # MemoryManager implementations +│ ├── metrics/ # MetricsSink +│ ├── operators/ # All physical operators (*Exec) +│ ├── physical/ # PhysicalPlan, PhysicalSchema, PlanProperties +│ └── planner/ # LocalPhysicalPlanner, schema_inference +│ +├── grism-distributed/ # Partial Ray integration (to be renamed grism-ray) +│ ├── planner/ # RayPlanner, Stage definitions +│ ├── transport/ # Arrow IPC transport +│ └── worker/ # Worker task definitions ``` -### 1.2 Crate Responsibilities - -| Crate | Responsibility | Dependencies | -|-------|----------------|--------------| -| **grism-engine** | Physical plan model, operators, expressions, schema inference | grism-core, grism-logical | -| **grism-local** | Local single-machine execution runtime | grism-engine, grism-storage | -| **grism-ray** | Distributed Ray-based execution runtime | grism-engine, grism-storage, ray | - ---- - -## 2. grism-engine (Common Components) - -The `grism-engine` crate contains all **runtime-agnostic** components that both local and distributed execution share. +### 2.2 What Works Today + +| Component | Status | Notes | +|-----------|--------|-------| +| Physical Plan Model | ✅ Complete | `PhysicalPlan`, `PhysicalSchema`, `PlanProperties` | +| All Physical Operators | ✅ Complete | 12 operators fully implemented and tested | +| Expression Evaluator | ✅ Complete | Full support for all expression types | +| Schema Inference | ✅ Complete | Type inference for all operators | +| LocalExecutor | ✅ Complete | Pull-based pipeline execution | +| LocalPhysicalPlanner | ✅ Complete | Logical to physical conversion | +| Memory Management | ✅ Complete | `MemoryManager` trait with implementations | +| Metrics Collection | ✅ Complete | `MetricsSink` trait | +| InMemoryStorage | ✅ Complete | For testing | +| Ray Stage Definitions | ⚠️ Partial | Basic Stage and StageId types | +| Exchange Operator | ❌ Not Started | Required for distributed execution | +| DistributedPlanner | ❌ Not Started | Stage splitting not implemented | +| RayExecutor | ❌ Not Started | Ray task submission not implemented | + +### 2.3 Test Coverage -### 2.1 Physical Plan Model - -```rust -// Core plan structures (runtime-agnostic) -pub struct PhysicalPlan { - pub root: Box, - pub properties: PlanProperties, -} - -pub struct PlanProperties { - pub execution_mode: ExecutionMode, - pub partitioning: PartitioningSpec, - pub blocking: bool, -} ``` - -### 2.2 Physical Operators - -All physical operators implement the `PhysicalOperator` trait: - -```rust -#[async_trait] -pub trait PhysicalOperator: Send + Sync { - /// Execute this operator, pulling from children - async fn execute( - &self, - ctx: &dyn ExecutionContextTrait, - ) -> Result, GrismError>; - - /// Operator's output schema - fn schema(&self) -> &PhysicalSchema; - - /// Operator capabilities - fn capabilities(&self) -> OperatorCaps; - - /// Child operators - fn children(&self) -> Vec<&dyn PhysicalOperator>; -} +grism-engine: 97 unit tests + 33 integration tests (all passing) +grism-distributed: 0 tests (skeleton only) ``` -**Operators in grism-engine:** - -| Operator | Description | Blocking | -|----------|-------------|----------| -| `NodeScanExec` | Scan nodes by label | No | -| `HyperedgeScanExec` | Scan hyperedges by label | No | -| `FilterExec` | Filter rows by predicate | No | -| `ProjectExec` | Project/compute columns | No | -| `LimitExec` | Limit rows with offset | No | -| `SortExec` | Multi-key sorting | **Yes** | -| `HashAggregateExec` | Aggregation with GROUP BY | **Yes** | -| `AdjacencyExpandExec` | Binary edge traversal | No | -| `RoleExpandExec` | N-ary hyperedge traversal | No | -| `UnionExec` | Union of two inputs | No | -| `RenameExec` | Rename columns | No | -| `EmptyExec` | Empty result | No | +--- -### 2.3 Expression Evaluator +## 3. Migration Plan -The `ExprEvaluator` converts `LogicalExpr` to Arrow arrays: +### 3.1 Phase 1: Extract Common Traits -```rust -pub struct ExprEvaluator; - -impl ExprEvaluator { - pub fn evaluate( - expr: &LogicalExpr, - batch: &RecordBatch, - ) -> Result; -} -``` +**Goal**: Define runtime-agnostic interfaces in `grism-engine`. -**Supported expressions:** -- Literals: Int64, Float64, String, Bool, Null -- Column references: Direct and qualified -- Binary operations: +, -, *, /, %, =, <>, <, <=, >, >=, AND, OR -- Unary operations: NOT, IS NULL, IS NOT NULL, NEG -- CASE expressions, IN lists, BETWEEN +**Tasks**: -### 2.4 Schema Inference +1. Define `ExecutionContextTrait` in `grism-engine`: ```rust -pub mod schema_inference { - pub fn infer_expr_type(expr: &LogicalExpr, input_schema: &Schema) -> DataType; - pub fn build_project_schema(exprs: &[LogicalExpr], input: &Schema) -> Schema; - pub fn build_aggregate_schema(aggs: &[AggExpr], groups: &[LogicalExpr]) -> Schema; -} -``` - -### 2.5 Operator Capabilities +// src/grism-engine/src/executor/traits.rs -```rust -pub struct OperatorCaps { - pub streaming: bool, - pub blocking: bool, - pub parallel_safe: bool, - pub requires_partitioning: Option, -} -``` - -### 2.6 Memory & Metrics - -```rust -// Memory management trait -pub trait MemoryManager: Send + Sync { - fn try_reserve(&self, bytes: usize) -> Result; - fn current_usage(&self) -> usize; -} - -// Metrics collection trait -pub trait MetricsSink: Send + Sync { - fn record_rows(&self, operator: &str, rows: usize); - fn record_batches(&self, operator: &str, count: usize); - fn record_time(&self, operator: &str, duration: Duration); -} -``` - -### 2.7 Execution Context Trait - -```rust -/// Trait for execution context - implemented by both local and ray runtimes +/// Runtime-agnostic execution context trait #[async_trait] pub trait ExecutionContextTrait: Send + Sync { + /// Access to storage layer fn storage(&self) -> &dyn Storage; + + /// Current snapshot for consistent reads fn snapshot_id(&self) -> SnapshotId; + + /// Memory management interface fn memory_manager(&self) -> &dyn MemoryManager; + + /// Optional metrics collection fn metrics_sink(&self) -> Option<&dyn MetricsSink>; + + /// Check if execution has been cancelled fn is_cancelled(&self) -> bool; } ``` ---- - -## 3. grism-local (Local Runtime) - -The `grism-local` crate provides single-machine execution with support for both in-memory and persistent storage. - -### 3.1 LocalExecutor +2. Define `PhysicalPlanner` trait in `grism-engine`: ```rust -pub struct LocalExecutor { - config: RuntimeConfig, +// src/grism-engine/src/planner/traits.rs + +/// Runtime-agnostic physical planner trait +pub trait PhysicalPlanner { + type Output; + + /// Convert logical plan to physical representation + fn plan(&self, logical: &LogicalPlan) -> Result; } +``` -impl LocalExecutor { - pub fn new() -> Self; - pub fn with_config(config: RuntimeConfig) -> Self; +3. Update `PhysicalOperator` trait to use trait objects: - pub async fn execute( +```rust +// Change from concrete ExecutionContext to trait +#[async_trait] +pub trait PhysicalOperator: Send + Sync { + async fn execute( &self, - plan: PhysicalPlan, - storage: Arc, - snapshot: SnapshotId, - ) -> Result; + ctx: &dyn ExecutionContextTrait, // Changed from &ExecutionContext + ) -> Result, GrismError>; + + // ... rest unchanged } ``` -### 3.2 LocalPhysicalPlanner +**Files to Modify**: +- `src/grism-engine/src/executor/mod.rs` - Add traits module +- `src/grism-engine/src/operators/traits.rs` - Update operator trait +- All operator files - Update execute() signature -```rust -pub struct LocalPhysicalPlanner { - config: PlannerConfig, -} +### 3.2 Phase 2: Create grism-local -impl PhysicalPlanner for LocalPhysicalPlanner { - fn plan(&self, logical: &LogicalPlan) -> Result; -} +**Goal**: Move local runtime components to dedicated crate. + +**Tasks**: + +1. Create new crate structure: + +``` +src/grism-local/ +├── Cargo.toml +└── src/ + ├── lib.rs + ├── executor.rs # LocalExecutor (moved from grism-engine) + ├── context.rs # LocalExecutionContext (moved from grism-engine) + ├── planner.rs # LocalPhysicalPlanner (moved from grism-engine) + └── result.rs # ExecutionResult (moved from grism-engine) ``` -**Key characteristics:** -- No Exchange operators inserted -- Direct operator tree execution -- Pull-based batch streaming +2. Update `Cargo.toml` for grism-local: + +```toml +[package] +name = "grism-local" +version = "0.1.0" +edition = "2021" + +[dependencies] +grism-engine = { path = "../grism-engine" } +grism-storage = { path = "../grism-storage" } +grism-logical = { path = "../grism-logical" } +arrow = "53" +async-trait = "0.1" +tokio = { version = "1", features = ["rt"] } +``` -### 3.3 LocalExecutionContext +3. Implement `LocalExecutionContext`: ```rust +// src/grism-local/src/context.rs + pub struct LocalExecutionContext { storage: Arc, snapshot_id: SnapshotId, @@ -272,398 +182,559 @@ pub struct LocalExecutionContext { } impl ExecutionContextTrait for LocalExecutionContext { - // Implement all trait methods + fn storage(&self) -> &dyn Storage { + self.storage.as_ref() + } + + fn snapshot_id(&self) -> SnapshotId { + self.snapshot_id + } + + fn memory_manager(&self) -> &dyn MemoryManager { + self.memory_manager.as_ref() + } + + fn metrics_sink(&self) -> Option<&dyn MetricsSink> { + self.metrics_sink.as_ref().map(|s| s.as_ref()) + } + + fn is_cancelled(&self) -> bool { + self.cancellation.is_cancelled() + } } ``` -### 3.4 Storage Support - -| Storage Backend | Description | Use Case | -|-----------------|-------------|----------| -| `InMemoryStorage` | Hash-map based storage | Testing, small datasets | -| `LanceStorage` | Lance format file storage | Production, large datasets | +4. Add re-exports for backward compatibility in grism-engine: ```rust -// Example usage with InMemoryStorage -let storage = InMemoryStorage::new(); -let executor = LocalExecutor::new(); -let result = executor.execute(plan, Arc::new(storage), SnapshotId::default()).await?; - -// Example usage with LanceStorage -let storage = LanceStorage::open("/data/graph.lance").await?; -let executor = LocalExecutor::new(); -let result = executor.execute(plan, Arc::new(storage), SnapshotId::default()).await?; -``` +// src/grism-engine/src/lib.rs ---- +// Re-export from grism-local for backward compatibility +// TODO: Add deprecation warnings in future version +#[cfg(feature = "local-runtime")] +pub use grism_local::{LocalExecutor, LocalPhysicalPlanner, ExecutionContext}; +``` -## 4. grism-ray (Distributed Runtime) +### 3.3 Phase 3: Rename grism-distributed to grism-ray -The `grism-ray` crate provides distributed execution using Ray as the orchestration layer. +**Goal**: Rename and restructure for Ray-specific implementation. -### 4.1 Core Concepts +**Tasks**: -#### Exchange Operator +1. Rename directory: -Exchange is a **first-class physical operator** that: -- Repartitions data -- Introduces a synchronization boundary -- Separates execution stages +```bash +git mv src/grism-distributed src/grism-ray +``` -```rust -pub struct ExchangeExec { - pub partitioning: PartitioningSpec, - pub mode: ExchangeMode, - pub child: Box, -} +2. Update `Cargo.toml`: + +```toml +[package] +name = "grism-ray" +version = "0.1.0" +edition = "2021" + +[dependencies] +grism-engine = { path = "../grism-engine" } +grism-storage = { path = "../grism-storage" } +grism-logical = { path = "../grism-logical" } +arrow = "53" +arrow-ipc = "53" +async-trait = "0.1" +tokio = { version = "1", features = ["rt", "net"] } +# ray = "..." # Ray Rust bindings when available +``` -pub enum ExchangeMode { - Shuffle, // repartition across workers - Broadcast, // replicate to all workers - Gather, // many → one -} +3. Update workspace `Cargo.toml`: + +```toml +[workspace] +members = [ + "src/grism-core", + "src/grism-logical", + "src/grism-optimizer", + "src/grism-engine", + "src/grism-local", # New + "src/grism-ray", # Renamed from grism-distributed + "src/grism-storage", + # ... +] ``` -#### Execution Stage +4. Update all imports across the codebase. -A **Stage** is a connected sub-DAG of physical operators executed as a unit: +### 3.4 Phase 4: Implement Exchange Operator -```rust -pub struct ExecutionStage { - pub stage_id: StageId, - pub plan: PhysicalSubPlan, - pub input_partitioning: Option, - pub output_partitioning: Option, -} -``` +**Goal**: Add Exchange operator for data repartitioning. -### 4.2 DistributedPlanner +**Tasks**: + +1. Add Exchange types to grism-ray: ```rust -pub struct DistributedPlanner { - config: DistributedPlannerConfig, +// src/grism-ray/src/operators/exchange.rs + +/// Exchange modes for data repartitioning +#[derive(Debug, Clone)] +pub enum ExchangeMode { + /// Repartition data by hash of keys + Shuffle, + /// Replicate data to all workers + Broadcast, + /// Collect all data to coordinator + Gather, } -impl DistributedPlanner { - /// Plan a logical plan into distributed stages - pub fn plan(&self, logical: &LogicalPlan) -> Result; +/// Exchange operator for distributed data movement +pub struct ExchangeExec { + pub child: Arc, + pub partitioning: PartitioningSpec, + pub mode: ExchangeMode, +} - /// Split physical plan into stages - pub fn split_into_stages(&self, plan: PhysicalPlan) -> Vec; +impl PhysicalOperator for ExchangeExec { + async fn execute( + &self, + ctx: &dyn ExecutionContextTrait, + ) -> Result, GrismError> { + // In local execution, Exchange is a no-op passthrough + // In distributed execution, this is handled by the stage executor + self.child.execute(ctx).await + } + + fn schema(&self) -> &PhysicalSchema { + self.child.schema() + } + + fn capabilities(&self) -> OperatorCaps { + OperatorCaps { + streaming: false, // Exchange is a barrier + blocking: true, + parallel_safe: true, + requires_partitioning: Some(self.partitioning.clone()), + } + } + + fn children(&self) -> Vec<&dyn PhysicalOperator> { + vec![self.child.as_ref()] + } } ``` -### 4.3 Partitioning Specification +2. Add partitioning specification: ```rust +// src/grism-ray/src/planner/partitioning.rs + +/// Specification for how data is partitioned +#[derive(Debug, Clone)] pub enum PartitioningSpec { + /// Hash partitioning by key columns Hash { - keys: Vec, - partitions: usize, + keys: Vec, + num_partitions: usize, }, + /// Range partitioning by key Range { - key: PhysicalExpr, - ranges: Vec, + key: String, + ranges: Vec<(Value, Value)>, }, + /// Partitioning by graph adjacency (keeps neighbors together) Adjacency { - entity: EntityType, // Node | Hyperedge + entity_type: EntityType, }, + /// Single partition (no distribution) Single, + /// Round-robin distribution RoundRobin { - partitions: usize, + num_partitions: usize, }, } -``` - -### 4.4 RayExecutor - -```rust -pub struct RayExecutor { - config: RayExecutorConfig, -} - -impl RayExecutor { - pub async fn execute( - &self, - stages: Vec, - storage: Arc, - ) -> Result; -} -``` - -### 4.5 Stage Execution Flow - -```text -┌──────────────────────────────────────────────────────────────────────┐ -│ DistributedPlan │ -├──────────────────────────────────────────────────────────────────────┤ -│ │ -│ Stage 0 (parallel) Stage 1 (parallel) Stage 2 │ -│ ┌─────────────────┐ ┌─────────────────┐ ┌──────────┐ │ -│ │ Scan → Filter │───────▶ │ Expand → Agg │───▶│ Collect │ │ -│ │ → Project │ Exchange│ (partial) │ │ (final) │ │ -│ └─────────────────┘ (Hash) └─────────────────┘ └──────────┘ │ -│ │ │ │ │ -│ ▼ ▼ ▼ │ -│ ┌─────────────┐ ┌─────────────┐ ┌──────────┐ │ -│ │ Worker 1 │ │ Worker 1 │ │ Driver │ │ -│ │ Worker 2 │ │ Worker 2 │ │ │ │ -│ │ Worker 3 │ │ Worker 3 │ │ │ │ -│ └─────────────┘ └─────────────┘ └──────────┘ │ -│ │ -└──────────────────────────────────────────────────────────────────────┘ -``` - ---- - -## 5. Stage Splitting Algorithm - -### 5.1 Stage Boundary Rules - -A **new stage MUST start** at: - -1. Any `ExchangeExec` operator -2. Any **blocking operator** when distribution is enabled -3. Any operator that requires global state -### 5.2 Splitting Algorithm - -```rust -fn split_into_stages(plan: PhysicalPlan) -> Vec { - let mut stages = vec![]; - let mut current = StageBuilder::new(); - - for node in plan.topo_order() { - if node.is_exchange() || node.is_blocking() { - stages.push(current.finish()); - current = StageBuilder::new(); +impl PartitioningSpec { + /// Calculate partition for a given row + pub fn partition_for(&self, batch: &RecordBatch, row: usize) -> usize { + match self { + Self::Hash { keys, num_partitions } => { + let mut hasher = DefaultHasher::new(); + for key in keys { + // Hash the value at this column + let col = batch.column_by_name(key).unwrap(); + hash_array_value(col, row, &mut hasher); + } + (hasher.finish() as usize) % num_partitions + } + Self::Single => 0, + Self::RoundRobin { num_partitions } => row % num_partitions, + // ... other cases } - current.add(node); } - - stages.push(current.finish()); - stages } ``` ---- +### 3.5 Phase 5: Implement Stage Splitting -## 6. Exchange Semantics +**Goal**: Implement algorithm to split physical plans into stages. -### 6.1 Guarantees +**Tasks**: -| Property | Guarantee | -|----------|-----------| -| Completeness | All rows forwarded | -| Disjointness | Each row appears once (except Broadcast) | -| Determinism | Same input → same partition | -| Barrier | Downstream waits for upstream readiness | +1. Implement stage builder: -### 6.2 Exchange Execution Steps (Shuffle) - -1. Upstream stage emits `RecordBatch` -2. Batch partitioned using `PartitioningSpec` -3. Sub-batches sent to Ray object store via Arrow IPC -4. Downstream tasks pull assigned partitions - ---- - -## 7. Blocking Operators in Distributed Context +```rust +// src/grism-ray/src/planner/stage.rs -### 7.1 Aggregate (Two-Phase) +/// Unique identifier for an execution stage +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct StageId(pub usize); -```text -Stage 0: PartialAggregate (parallel) - ↓ Exchange(Hash by group key) -Stage 1: FinalAggregate -``` +/// An execution stage containing a sub-plan +pub struct ExecutionStage { + pub stage_id: StageId, + pub root: Arc, + pub input_partitioning: Option, + pub output_partitioning: Option, +} -### 7.2 Sort (Two-Phase) +/// Builder for constructing stages +pub struct StageBuilder { + operators: Vec>, + input_partitioning: Option, +} -```text -Stage 0: LocalSort (parallel) - ↓ Exchange(Range) -Stage 1: MergeSort +impl StageBuilder { + pub fn new() -> Self { + Self { + operators: vec![], + input_partitioning: None, + } + } + + pub fn with_input_partitioning(mut self, spec: PartitioningSpec) -> Self { + self.input_partitioning = Some(spec); + self + } + + pub fn add(&mut self, op: Arc) { + self.operators.push(op); + } + + pub fn finish(self, stage_id: StageId) -> ExecutionStage { + // Build operator tree from collected operators + let root = self.build_tree(); + ExecutionStage { + stage_id, + root, + input_partitioning: self.input_partitioning, + output_partitioning: None, // Set by planner + } + } +} ``` ---- - -## 8. Expand-Aware Partitioning - -### 8.1 Adjacency Partitioning - -For graph traversal workloads, adjacency-based partitioning keeps most expansions local: +2. Implement stage splitting algorithm: ```rust -PartitioningSpec::Adjacency { - entity: EntityType::Node, +// src/grism-ray/src/planner/splitter.rs + +pub struct StageSplitter; + +impl StageSplitter { + /// Split a physical plan into execution stages + pub fn split(plan: &PhysicalPlan) -> Vec { + let mut stages = vec![]; + let mut current = StageBuilder::new(); + let mut stage_id = 0; + + // Topological traversal of operator tree + for node in Self::topo_order(&plan.root) { + if Self::is_stage_boundary(node.as_ref()) { + // Finish current stage + if !current.is_empty() { + stages.push(current.finish(StageId(stage_id))); + stage_id += 1; + } + + // Handle Exchange specially + if let Some(exchange) = node.as_any().downcast_ref::() { + // Exchange defines output partitioning of previous stage + // and input partitioning of next stage + if let Some(last) = stages.last_mut() { + last.output_partitioning = Some(exchange.partitioning.clone()); + } + current = StageBuilder::new() + .with_input_partitioning(exchange.partitioning.clone()); + } else { + current = StageBuilder::new(); + } + } + + current.add(node.clone()); + } + + // Final stage + if !current.is_empty() { + stages.push(current.finish(StageId(stage_id))); + } + + stages + } + + fn is_stage_boundary(op: &dyn PhysicalOperator) -> bool { + // Exchange always creates a boundary + if op.as_any().is::() { + return true; + } + + // Blocking operators create boundaries in distributed mode + let caps = op.capabilities(); + caps.blocking + } } ``` -### 8.2 N-ary Expand - -N-ary hyperedge expansion often requires shuffle: - -```text -Stage 0: Scan nodes - ↓ Exchange(Adjacency) -Stage 1: RoleExpand (hyperedges co-located with nodes) -``` - ---- - -## 9. Control Plane vs Data Plane +### 3.6 Phase 6: Implement Distributed Planner -| Plane | Responsibility | Owner | -|-------|----------------|-------| -| Control | Stage graph, scheduling, progress | Ray | -| Data | Arrow batch movement, execution | Rust | +**Goal**: Create DistributedPlanner that inserts Exchange operators. -**Ray orchestrates, Rust executes.** - ---- - -## 10. Failure & Retry Semantics - -Stage boundaries define **failure domains**: - -| Failure | Recovery | -|---------|----------| -| Worker crash | Retry stage | -| Exchange failure | Re-run upstream stage | -| Blocking stage failure | Restart stage | - -No partial stage result is reused unless idempotent. - ---- - -## 11. Explainability - -### 11.1 EXPLAIN DISTRIBUTED - -```text -Stage 0: - NodeScan(Person) → Filter(age > 21) → Project(name, age) - Parallelism: 8 - Output: HashPartition(node_id) - - ↓ Exchange(Shuffle, Hash(node_id)) +```rust +// src/grism-ray/src/planner/distributed.rs -Stage 1: - Aggregate(COUNT(*), GROUP BY city) - Parallelism: 4 - Output: HashPartition(city) +pub struct DistributedPlanner { + pub config: DistributedPlannerConfig, +} - ↓ Exchange(Gather) +pub struct DistributedPlannerConfig { + pub default_parallelism: usize, + pub prefer_adjacency_partitioning: bool, +} -Stage 2: - Collect - Parallelism: 1 +impl DistributedPlanner { + /// Plan a logical plan for distributed execution + pub fn plan(&self, logical: &LogicalPlan) -> Result { + // First, create a basic physical plan + let physical = self.create_physical_plan(logical)?; + + // Insert Exchange operators where needed + let with_exchanges = self.insert_exchanges(physical)?; + + // Split into stages + let stages = StageSplitter::split(&with_exchanges); + + Ok(DistributedPlan { stages }) + } + + fn insert_exchanges(&self, plan: PhysicalPlan) -> Result { + // Walk the plan and insert Exchange operators at: + // 1. Before blocking operators (Aggregate, Sort) + // 2. Before Expand operators that need repartitioning + // 3. Before the final Collect + + self.transform_plan(&plan.root, None) + } + + fn transform_plan( + &self, + op: &Arc, + required_partitioning: Option<&PartitioningSpec>, + ) -> Result { + // Check if we need to insert an Exchange + let current_partitioning = self.infer_partitioning(op); + + let result = if let Some(required) = required_partitioning { + if !self.partitioning_satisfies(¤t_partitioning, required) { + // Need Exchange + Arc::new(ExchangeExec { + child: op.clone(), + partitioning: required.clone(), + mode: ExchangeMode::Shuffle, + }) + } else { + op.clone() + } + } else { + op.clone() + }; + + // ... recursively transform children + Ok(PhysicalPlan::new(result)) + } +} ``` --- -## 12. Migration Path +## 4. Implementation Considerations -### 12.1 Current State +### 4.1 Backward Compatibility -``` -grism-engine/ # Contains both engine AND local runtime -grism-distributed/ # Partial Ray integration -``` +To maintain backward compatibility during migration: -### 12.2 Target State +1. **Feature Flags**: Use Cargo features to control crate dependencies -``` -grism-engine/ # Common: operators, expressions, schemas, traits -grism-local/ # Runtime: LocalExecutor, LocalPhysicalPlanner -grism-ray/ # Runtime: RayExecutor, DistributedPlanner, Exchange +```toml +# grism-engine/Cargo.toml +[features] +default = ["local-runtime"] +local-runtime = ["grism-local"] ``` -### 12.3 Migration Steps +2. **Re-exports**: Re-export moved types from original locations with deprecation warnings -1. **Extract common traits** from `grism-engine` into stable interfaces -2. **Create `grism-local`** by moving `LocalExecutor`, `LocalPhysicalPlanner`, `ExecutionContext` -3. **Rename `grism-distributed`** to `grism-ray` -4. **Implement Exchange operator** in `grism-ray` -5. **Implement stage splitting** in `grism-ray` -6. **Keep current local engine functionality unchanged** - users should see no difference +3. **Version Pinning**: Maintain API compatibility within minor versions ---- +### 4.2 Testing Strategy -## 13. API Compatibility - -### 13.1 Local Execution (Unchanged) +1. **Unit Tests**: Each component has comprehensive unit tests +2. **Integration Tests**: End-to-end tests for both local and distributed execution +3. **Equivalence Tests**: Verify local and distributed produce identical results ```rust -// Before and after: identical API -use grism_local::{LocalExecutor, LocalPhysicalPlanner}; +#[test] +fn test_local_distributed_equivalence() { + let logical_plan = create_test_plan(); + + // Execute locally + let local_result = LocalExecutor::new() + .execute(local_plan, storage.clone()) + .await?; + + // Execute distributed (single worker for comparison) + let ray_result = RayExecutor::new_local() + .execute(distributed_plan, storage.clone()) + .await?; + + assert_batches_equal(&local_result.batches, &ray_result.batches); +} +``` -let planner = LocalPhysicalPlanner::new(); -let plan = planner.plan(&logical_plan)?; +### 4.3 Performance Considerations -let executor = LocalExecutor::new(); -let result = executor.execute(plan, storage, snapshot).await?; -``` +1. **Zero-Copy Where Possible**: Use Arrow's zero-copy semantics +2. **Batch Size Tuning**: Configurable batch sizes for different workloads +3. **Memory Budgets**: Respect memory limits in blocking operators +4. **Metrics Collection**: Optional to avoid overhead in production -### 13.2 Distributed Execution (New) +### 4.4 Error Handling ```rust -use grism_ray::{RayExecutor, DistributedPlanner}; +/// Execution-layer errors +#[derive(Debug, thiserror::Error)] +pub enum ExecutionError { + #[error("Memory limit exceeded: requested {requested}, limit {limit}")] + MemoryExceeded { requested: usize, limit: usize }, + + #[error("Execution cancelled")] + Cancelled, + + #[error("Stage {0} failed: {1}")] + StageFailed(StageId, String), + + #[error("Exchange failed: {0}")] + ExchangeFailed(String), + + #[error("Worker {0} unreachable")] + WorkerUnreachable(String), +} +``` -let planner = DistributedPlanner::new(); -let stages = planner.plan(&logical_plan)?; +--- -let executor = RayExecutor::connect("ray://cluster:10001")?; -let result = executor.execute(stages, storage).await?; -``` +## 5. File Changes Summary + +### New Files to Create + +| File | Description | +|------|-------------| +| `src/grism-local/Cargo.toml` | New crate manifest | +| `src/grism-local/src/lib.rs` | Crate entry point | +| `src/grism-local/src/executor.rs` | LocalExecutor | +| `src/grism-local/src/context.rs` | LocalExecutionContext | +| `src/grism-local/src/planner.rs` | LocalPhysicalPlanner | +| `src/grism-engine/src/executor/traits.rs` | ExecutionContextTrait | +| `src/grism-engine/src/planner/traits.rs` | PhysicalPlanner trait | +| `src/grism-ray/src/operators/exchange.rs` | ExchangeExec | +| `src/grism-ray/src/planner/distributed.rs` | DistributedPlanner | +| `src/grism-ray/src/planner/splitter.rs` | StageSplitter | + +### Files to Modify + +| File | Changes | +|------|---------| +| `Cargo.toml` (workspace) | Add new crate members | +| `src/grism-engine/src/lib.rs` | Export traits, add re-exports | +| `src/grism-engine/src/operators/*.rs` | Update execute() signatures | +| `src/grism-ray/Cargo.toml` | Rename from grism-distributed | +| `src/grism-ray/src/lib.rs` | Update exports | + +### Files to Move + +| From | To | +|------|-----| +| `src/grism-engine/src/executor/local.rs` | `src/grism-local/src/executor.rs` | +| `src/grism-engine/src/executor/context.rs` | `src/grism-local/src/context.rs` | +| `src/grism-engine/src/executor/result.rs` | `src/grism-local/src/result.rs` | +| `src/grism-engine/src/planner/local.rs` | `src/grism-local/src/planner.rs` | --- -## 14. Implementation Checklist +## 6. Implementation Checklist -### Phase 1: Engine Separation +### Phase 1: Extract Common Traits - [ ] Define `ExecutionContextTrait` in grism-engine - [ ] Define `PhysicalPlanner` trait in grism-engine -- [ ] Ensure all operators depend only on traits, not concrete types +- [ ] Update `PhysicalOperator` trait to use `dyn ExecutionContextTrait` +- [ ] Update all operator implementations +- [ ] Verify all tests pass ### Phase 2: Create grism-local -- [ ] Move `LocalExecutor` to grism-local -- [ ] Move `LocalPhysicalPlanner` to grism-local -- [ ] Move `ExecutionContext` implementation to grism-local -- [ ] Add re-exports for backward compatibility +- [ ] Create `src/grism-local/` directory structure +- [ ] Create `Cargo.toml` with dependencies +- [ ] Move `LocalExecutor` from grism-engine +- [ ] Move `LocalPhysicalPlanner` from grism-engine +- [ ] Move `ExecutionContext` as `LocalExecutionContext` +- [ ] Implement `ExecutionContextTrait` for `LocalExecutionContext` +- [ ] Add re-exports to grism-engine for compatibility +- [ ] Update workspace Cargo.toml +- [ ] Verify all tests pass ### Phase 3: Rename to grism-ray -- [ ] Rename `grism-distributed` to `grism-ray` -- [ ] Update all imports and references +- [ ] Rename directory from grism-distributed to grism-ray +- [ ] Update package name in Cargo.toml +- [ ] Update workspace Cargo.toml +- [ ] Update all imports across codebase +- [ ] Verify compilation + +### Phase 4: Implement Exchange +- [ ] Add `ExchangeMode` enum +- [ ] Add `PartitioningSpec` enum - [ ] Implement `ExchangeExec` operator -- [ ] Implement `DistributedPlanner` with stage splitting -- [ ] Implement Ray task submission - -### Phase 4: Integration -- [ ] Integrate with Ray object store for Arrow IPC -- [ ] Implement two-phase aggregation -- [ ] Implement range-partitioned sort -- [ ] Add distributed EXPLAIN support +- [ ] Add hash partitioning logic +- [ ] Add unit tests for Exchange + +### Phase 5: Implement Stage Splitting +- [ ] Implement `StageBuilder` +- [ ] Implement `StageSplitter` +- [ ] Add unit tests for stage splitting +- [ ] Verify correct boundary detection + +### Phase 6: Implement Distributed Planner +- [ ] Implement `DistributedPlanner` +- [ ] Implement Exchange insertion logic +- [ ] Implement partitioning inference +- [ ] Add integration tests + +### Phase 7: Implement Ray Integration +- [ ] Add Ray task submission +- [ ] Implement Arrow IPC transport +- [ ] Implement stage execution on workers +- [ ] Add distributed execution tests --- -## 15. Summary - -| Crate | What It Contains | Key Types | -|-------|------------------|-----------| -| **grism-engine** | Runtime-agnostic physical layer | `PhysicalPlan`, `PhysicalOperator`, `ExprEvaluator`, `OperatorCaps` | -| **grism-local** | Single-machine execution | `LocalExecutor`, `LocalPhysicalPlanner`, `LocalExecutionContext` | -| **grism-ray** | Distributed Ray execution | `RayExecutor`, `DistributedPlanner`, `ExchangeExec`, `ExecutionStage` | - -### Architectural Guarantees - -1. **Zero semantic drift**: Local and distributed execution produce identical results -2. **Local engine unchanged**: Current local execution API remains stable -3. **Explicit distribution**: Exchange operators are visible in plans -4. **Hypergraph-correct**: Expand operators preserve adjacency semantics -5. **Explainable**: Distributed plans can be inspected and debugged +## 7. References -> **Exchange is the only way data moves across stages.** -> **Stages are the only unit of distribution.** -> **Operators never know they are distributed.** +* **RFC-0102**: Execution Engine Architecture (authoritative design) +* **RFC-0008**: Physical Plan & Operator Interfaces +* **RFC-0010**: Distributed & Parallel Execution +* **RFC-0100**: Architecture Design Document diff --git a/specs/rfc-0102.md b/specs/rfc-0102.md new file mode 100644 index 0000000..5d9529e --- /dev/null +++ b/specs/rfc-0102.md @@ -0,0 +1,751 @@ +# RFC-0102: Execution Engine Architecture + +**Status**: Draft +**Authors**: Grism Team +**Created**: 2026-01-22 +**Last Updated**: 2026-01-22 +**Depends on**: RFC-0002, RFC-0008, RFC-0010, RFC-0100 +**Supersedes**: — + +--- + +## 1. Abstract + +This RFC defines the **execution engine architecture** for Grism, specifying how logical plans are transformed into physical plans and executed across different runtime environments. + +The engine architecture is structured around three distinct concerns: + +1. **Common Engine Layer**: Runtime-agnostic physical planning, operators, and expression evaluation +2. **Local Runtime**: Single-machine execution with pull-based streaming +3. **Ray Runtime**: Distributed execution with stage-based parallelism + +This separation ensures that execution semantics remain identical regardless of runtime environment while allowing each runtime to optimize for its specific characteristics. + +--- + +## 2. Scope and Non-Goals + +### 2.1 Scope + +This RFC specifies: + +* Crate organization and responsibilities +* Physical operator model and contracts +* Expression evaluation architecture +* Local execution model +* Distributed execution model +* Runtime abstraction interfaces + +### 2.2 Non-Goals + +This RFC does **not** define: + +* Storage engine internals (see RFC-0012) +* Logical operator semantics (see RFC-0002) +* Cost model details (see RFC-0007) +* Network protocols or serialization formats +* Cluster management or deployment + +--- + +## 3. Design Principles + +### 3.1 Core Principles + +1. **Semantic Equivalence** + Local and distributed execution MUST produce identical results for the same logical plan. + +2. **Runtime Agnosticism** + Physical operators and expressions are defined independently of execution runtime. + +3. **Explicit Distribution** + Data movement in distributed execution is always explicit via Exchange operators. + +4. **Operator Transparency** + Operators do not know whether they execute locally or in a distributed context. + +### 3.2 Architectural Separation + +The engine separates **what to compute** from **how to execute**: + +| Concern | Responsibility | Layer | +|---------|----------------|-------| +| What to compute | Physical operators, expressions, schemas | Common Engine | +| How to execute locally | Pull-based pipeline, memory management | Local Runtime | +| How to execute distributed | Stage splitting, Exchange, task scheduling | Ray Runtime | + +--- + +## 4. Crate Architecture + +### 4.1 Overview + +``` +┌─────────────────────────────────────────────────────────────────────────────┐ +│ grism-engine (Common) │ +│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ ┌────────────────┐ │ +│ │ Physical │ │ Operators │ │ Expression │ │ Physical │ │ +│ │ Plan Model │ │ (Exec) │ │ Evaluator │ │ Schema │ │ +│ └─────────────┘ └──────────────┘ └───────────────┘ └────────────────┘ │ +│ ┌─────────────┐ ┌──────────────┐ ┌───────────────┐ │ +│ │ Operator │ │ Schema │ │ Memory & │ │ +│ │ Traits │ │ Inference │ │ Metrics │ │ +│ └─────────────┘ └──────────────┘ └───────────────┘ │ +└─────────────────────────────────────────────────────────────────────────────┘ + │ │ + ▼ ▼ +┌───────────────────────────────┐ ┌───────────────────────────────────────┐ +│ grism-local (Runtime) │ │ grism-ray (Runtime) │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ LocalExecutor │ │ │ │ RayExecutor │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ LocalPhysicalPlanner │ │ │ │ DistributedPlanner │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +│ ┌────────────────────────┐ │ │ ┌─────────────────────────────────┐ │ +│ │ ExecutionContext │ │ │ │ ExchangeExec / StageExecutor │ │ +│ └────────────────────────┘ │ │ └─────────────────────────────────┘ │ +└───────────────────────────────┘ └───────────────────────────────────────┘ +``` + +### 4.2 Crate Responsibilities + +| Crate | Responsibility | Key Types | +|-------|----------------|-----------| +| **grism-engine** | Runtime-agnostic physical layer | `PhysicalPlan`, `PhysicalOperator`, `ExprEvaluator`, `OperatorCaps` | +| **grism-local** | Single-machine execution | `LocalExecutor`, `LocalPhysicalPlanner`, `LocalExecutionContext` | +| **grism-ray** | Distributed Ray execution | `RayExecutor`, `DistributedPlanner`, `ExchangeExec`, `ExecutionStage` | + +--- + +## 5. Common Engine Layer (grism-engine) + +The common engine layer provides all runtime-agnostic components that both local and distributed execution share. + +### 5.1 Physical Plan Model + +A physical plan is a tree of physical operators with associated execution properties. + +``` +PhysicalPlan +├── root: PhysicalOperator +└── properties: PlanProperties + ├── execution_mode: ExecutionMode + ├── partitioning: PartitioningSpec + └── blocking: bool +``` + +**Invariants**: + +* Physical plans are semantically equivalent to their source logical plans +* All expressions are executable +* All operators are compatible with the execution mode + +### 5.2 Physical Operator Trait + +All physical operators implement a common trait defining execution behavior: + +``` +PhysicalOperator +├── execute(ctx) → RecordBatchStream +├── schema() → PhysicalSchema +├── capabilities() → OperatorCaps +└── children() → [PhysicalOperator] +``` + +**Operator Contract**: + +* Operators are stateless between executions +* Operators receive context via trait, not concrete type +* Operators produce Arrow RecordBatch streams +* Operators declare their capabilities explicitly + +### 5.3 Physical Operators + +The following operators are defined in the common engine layer: + +| Operator | Category | Blocking | Description | +|----------|----------|----------|-------------| +| `NodeScanExec` | Source | No | Scan nodes by label | +| `HyperedgeScanExec` | Source | No | Scan hyperedges by label | +| `FilterExec` | Unary | No | Apply predicate per batch | +| `ProjectExec` | Unary | No | Compute expressions per batch | +| `LimitExec` | Unary | No | Limit output rows | +| `RenameExec` | Unary | No | Rename columns | +| `SortExec` | Blocking | **Yes** | Multi-key sorting | +| `HashAggregateExec` | Blocking | **Yes** | Aggregation with GROUP BY | +| `AdjacencyExpandExec` | Graph | No | Binary edge traversal | +| `RoleExpandExec` | Graph | No | N-ary hyperedge traversal | +| `UnionExec` | Binary | No | Union of two inputs | +| `CollectExec` | Sink | **Yes** | Collect all results | +| `EmptyExec` | Source | No | Empty result | + +### 5.4 Operator Capabilities + +Each operator declares its capabilities: + +``` +OperatorCaps +├── streaming: bool // Can process input incrementally +├── blocking: bool // Must consume all input before output +├── parallel_safe: bool // Safe to execute in parallel +└── requires_partitioning: Option +``` + +These capabilities inform runtime decisions: + +* Local runtime uses blocking information for memory management +* Ray runtime uses capabilities for stage splitting and Exchange insertion + +### 5.5 Expression Evaluator + +The `ExprEvaluator` converts logical expressions to Arrow arrays: + +**Supported Expression Types**: + +| Category | Operations | +|----------|------------| +| Literals | Int64, Float64, String, Bool, Null | +| Column References | Direct, qualified (`entity.column`) | +| Binary Operations | `+`, `-`, `*`, `/`, `%`, `=`, `<>`, `<`, `<=`, `>`, `>=`, `AND`, `OR` | +| Unary Operations | `NOT`, `IS NULL`, `IS NOT NULL`, `NEG` | +| Special | `CASE`, `IN`, `BETWEEN` | + +**Evaluation Contract**: + +* Expressions are evaluated against a single RecordBatch +* Evaluation is side-effect free +* Type coercion follows RFC-0003 rules +* Null handling uses three-valued logic + +### 5.6 Schema Inference + +Schema inference provides type information for physical planning: + +| Function | Purpose | +|----------|---------| +| `infer_expr_type()` | Infer Arrow DataType from expression | +| `build_project_schema()` | Build schema for projections | +| `build_aggregate_schema()` | Build schema for aggregations | + +### 5.7 Execution Context Trait + +The execution context trait abstracts runtime-specific resources: + +``` +ExecutionContextTrait +├── storage() → Storage +├── snapshot_id() → SnapshotId +├── memory_manager() → MemoryManager +├── metrics_sink() → Option +└── is_cancelled() → bool +``` + +Both local and Ray runtimes implement this trait with their specific resource management. + +### 5.8 Memory and Metrics + +**Memory Management**: + +``` +MemoryManager +├── try_reserve(bytes) → Result +├── current_usage() → usize +└── limit() → Option +``` + +**Metrics Collection**: + +``` +MetricsSink +├── record_rows(operator, count) +├── record_batches(operator, count) +└── record_time(operator, duration) +``` + +--- + +## 6. Local Runtime (grism-local) + +The local runtime provides single-machine execution with pull-based streaming. + +### 6.1 Execution Model + +Local execution uses a **pull-based pipeline** model: + +``` +┌─────────────┐ ┌─────────────┐ ┌─────────────┐ +│ Scan │ ──▶ │ Filter │ ──▶ │ Project │ +│ (pull) │ │ (pull) │ │ (pull) │ +└─────────────┘ └─────────────┘ └─────────────┘ + ▲ ▲ ▲ + │ │ │ + RecordBatch RecordBatch RecordBatch +``` + +**Characteristics**: + +* Each operator pulls from its children on demand +* Streaming execution minimizes memory usage +* Blocking operators buffer internally +* Single-threaded by default (async-ready) + +### 6.2 LocalExecutor + +The `LocalExecutor` drives plan execution: + +``` +LocalExecutor +├── execute(plan, storage, snapshot) → ExecutionResult +└── config: RuntimeConfig + ├── batch_size: usize + └── memory_limit: Option +``` + +**Execution Flow**: + +1. Create execution context with storage and configuration +2. Initialize operator tree from physical plan +3. Pull batches from root operator until exhausted +4. Collect results into `ExecutionResult` + +### 6.3 LocalPhysicalPlanner + +The `LocalPhysicalPlanner` converts logical plans to physical plans: + +``` +LocalPhysicalPlanner +├── plan(logical) → PhysicalPlan +└── config: PlannerConfig +``` + +**Planning Rules**: + +* No Exchange operators are inserted +* All operators execute in a single stage +* Operator selection based on capabilities and cost + +### 6.4 LocalExecutionContext + +Implements `ExecutionContextTrait` for local execution: + +``` +LocalExecutionContext +├── storage: Arc +├── snapshot_id: SnapshotId +├── memory_manager: Arc +├── metrics_sink: Option> +└── cancellation: CancellationHandle +``` + +### 6.5 Storage Support + +The local runtime supports multiple storage backends: + +| Backend | Description | Use Case | +|---------|-------------|----------| +| `InMemoryStorage` | Hash-map based storage | Testing, small datasets | +| `LanceStorage` | Lance format file storage | Production, large datasets | + +Storage is accessed through the `Storage` trait defined in `grism-storage`. + +--- + +## 7. Ray Runtime (grism-ray) + +The Ray runtime provides distributed execution using Ray as the orchestration layer. + +### 7.1 Execution Model + +Distributed execution uses a **stage-based** model: + +``` +┌──────────────────────────────────────────────────────────────────────┐ +│ Distributed Plan │ +├──────────────────────────────────────────────────────────────────────┤ +│ │ +│ Stage 0 (parallel) Exchange Stage 1 (parallel) │ +│ ┌─────────────────┐ ┌─────────┐ ┌─────────────────┐ │ +│ │ Scan → Filter │───▶│ Shuffle │────▶│ Agg → Collect │ │ +│ │ → Project │ │ (Hash) │ │ │ │ +│ └─────────────────┘ └─────────┘ └─────────────────┘ │ +│ │ │ │ +│ ┌──────┴──────┐ ┌──────┴──────┐ │ +│ │ Worker 1-N │ │ Worker 1-M │ │ +│ └─────────────┘ └─────────────┘ │ +│ │ +└──────────────────────────────────────────────────────────────────────┘ +``` + +**Principle**: + +> **Ray orchestrates, Rust executes.** + +Ray handles task scheduling, data movement, and fault tolerance, while Rust workers perform actual query execution using the same operators as local execution. + +### 7.2 Exchange Operator + +`ExchangeExec` is a **first-class physical operator** that: + +* Repartitions data across workers +* Introduces a synchronization boundary +* Separates execution stages + +``` +ExchangeExec +├── partitioning: PartitioningSpec +├── mode: ExchangeMode +└── child: PhysicalOperator +``` + +**Exchange Modes**: + +| Mode | Description | +|------|-------------| +| `Shuffle` | Repartition across workers by key | +| `Broadcast` | Replicate to all workers | +| `Gather` | Collect to single coordinator | + +### 7.3 Partitioning Specification + +``` +PartitioningSpec +├── Hash { keys, partitions } +├── Range { key, ranges } +├── Adjacency { entity_type } +├── Single +└── RoundRobin { partitions } +``` + +**Partitioning Invariants**: + +* Partitions MUST cover entire input +* Partitions MUST be disjoint (except Broadcast) +* Same input always maps to same partition + +### 7.4 Execution Stage + +A **Stage** is a connected sub-DAG of physical operators executed as a unit: + +``` +ExecutionStage +├── stage_id: StageId +├── plan: PhysicalSubPlan +├── input_partitioning: Option +└── output_partitioning: Option +``` + +**Stage Properties**: + +* Contains no internal Exchange operators +* Executed as a unit on one or more workers +* Has explicit input and output partitioning + +### 7.5 Stage Splitting Algorithm + +Stage boundaries are determined by the following rules: + +**A new stage MUST start at**: + +1. Any `ExchangeExec` operator +2. Any **blocking operator** in distributed mode +3. Any operator requiring global state + +**Splitting Algorithm**: + +``` +for node in plan.topological_order(): + if node.is_exchange() or node.is_blocking(): + stages.push(current_stage.finish()) + current_stage = new_stage() + current_stage.add(node) +stages.push(current_stage.finish()) +``` + +### 7.6 DistributedPlanner + +The `DistributedPlanner` transforms logical plans for distributed execution: + +``` +DistributedPlanner +├── plan(logical) → DistributedPlan +├── split_into_stages(physical) → [ExecutionStage] +└── config: DistributedPlannerConfig +``` + +**Planning Responsibilities**: + +* Insert Exchange operators at appropriate points +* Choose partitioning strategies based on operators +* Split plan into execution stages +* Configure stage parallelism + +### 7.7 RayExecutor + +The `RayExecutor` orchestrates distributed execution: + +``` +RayExecutor +├── execute(stages, storage) → ExecutionResult +└── config: RayExecutorConfig + ├── default_parallelism: usize + └── ray_address: String +``` + +**Execution Flow**: + +1. Submit stage DAG to Ray +2. Ray schedules stage tasks on workers +3. Workers execute using Rust operators +4. Exchange moves data between stages via Arrow IPC +5. Final stage collects results + +### 7.8 Data Transport + +Data moves between stages via Arrow IPC: + +``` +Stage A Worker → Arrow IPC → Ray Object Store → Arrow IPC → Stage B Worker +``` + +**Transport Guarantees**: + +* Zero-copy when possible via Ray Plasma +* Backpressure support +* Deterministic routing + +--- + +## 8. Blocking Operators in Distributed Context + +### 8.1 Two-Phase Aggregation + +Aggregation in distributed execution uses a two-phase pattern: + +``` +Stage 0: PartialAggregate (parallel on each partition) + ↓ Exchange(Hash by group key) +Stage 1: FinalAggregate (parallel by group) +``` + +**Aggregate Functions**: + +| Function | Partial State | Merge Operation | +|----------|---------------|-----------------| +| `COUNT` | count | sum of counts | +| `SUM` | sum | sum of sums | +| `AVG` | (sum, count) | sum of sums / sum of counts | +| `MIN` | min | min of mins | +| `MAX` | max | max of maxs | + +### 8.2 Two-Phase Sort + +Sorting in distributed execution: + +``` +Stage 0: LocalSort (parallel per partition) + ↓ Exchange(Range) +Stage 1: MergeSort (parallel per range) +``` + +--- + +## 9. Graph Operators in Distributed Context + +### 9.1 Adjacency-Aware Partitioning + +For graph traversal workloads, adjacency-based partitioning keeps most expansions local: + +``` +PartitioningSpec::Adjacency { entity: Node } +``` + +**Benefits**: + +* Most binary Expand operations are partition-local +* Reduces shuffle volume for traversal queries +* Preserves locality for multi-hop patterns + +### 9.2 Expand Distribution + +| Expand Type | Distribution Strategy | +|-------------|----------------------| +| Binary (AdjacencyExpand) | Prefer adjacency partitioning | +| N-ary (RoleExpand) | May require Exchange for non-local roles | + +**Cross-Partition Expand**: + +When expansion crosses partition boundaries, explicit Exchange is inserted: + +``` +Stage 0: Scan nodes (partitioned by node_id) + ↓ Exchange(Adjacency) +Stage 1: RoleExpand (hyperedges co-located with participating nodes) +``` + +--- + +## 10. Failure and Recovery + +### 10.1 Failure Domains + +Stage boundaries define **failure domains**: + +| Failure | Recovery Action | +|---------|-----------------| +| Worker crash | Retry stage on different worker | +| Exchange failure | Re-run upstream stage | +| Coordinator crash | Query fails, client retries | + +### 10.2 Retry Semantics + +* Stateless operators may be safely retried +* Partial stage results are discarded on failure +* Exactly-once semantics are **not guaranteed** by default + +--- + +## 11. Explainability + +### 11.1 EXPLAIN Output + +Both runtimes support `EXPLAIN` output: + +**Local EXPLAIN**: + +``` +NodeScan(Person) + └── Filter(age > 21) + └── Project(name, age) + └── Collect +``` + +**Distributed EXPLAIN**: + +``` +Stage 0 (parallelism=8): + NodeScan(Person) → Filter(age > 21) → Project(name, age) + Output: HashPartition(node_id) + + ↓ Exchange(Shuffle, Hash(node_id)) + +Stage 1 (parallelism=4): + Aggregate(COUNT(*), GROUP BY city) + Output: HashPartition(city) + + ↓ Exchange(Gather) + +Stage 2 (parallelism=1): + Collect +``` + +### 11.2 Runtime Metrics + +Both runtimes collect per-operator metrics: + +| Metric | Description | +|--------|-------------| +| `rows_in` | Input row count | +| `rows_out` | Output row count | +| `batches` | Number of batches processed | +| `execution_time` | Wall-clock execution time | +| `memory_peak` | Peak memory usage | + +--- + +## 12. API Compatibility + +### 12.1 Local Execution API + +```python +from grism import LocalExecutor, LocalPhysicalPlanner + +planner = LocalPhysicalPlanner() +plan = planner.plan(logical_plan) + +executor = LocalExecutor() +result = executor.execute(plan, storage, snapshot) + +for batch in result.batches: + process(batch) +``` + +### 12.2 Distributed Execution API + +```python +from grism import RayExecutor, DistributedPlanner + +planner = DistributedPlanner() +stages = planner.plan(logical_plan) + +executor = RayExecutor.connect("ray://cluster:10001") +result = executor.execute(stages, storage) + +for batch in result.batches: + process(batch) +``` + +### 12.3 Unified Interface + +A unified interface can select runtime based on configuration: + +```python +from grism import Hypergraph + +hg = Hypergraph.connect("grism://local") # Local execution +# or +hg = Hypergraph.connect("grism://ray:10001") # Ray execution + +# Same query API regardless of runtime +result = ( + hg.nodes("Person") + .filter(col("age") > 21) + .select("name", "age") + .collect() +) +``` + +--- + +## 13. Relationship to Other RFCs + +| RFC | Relationship | +|-----|--------------| +| **RFC-0002** | Defines logical operator semantics that physical operators implement | +| **RFC-0003** | Defines expression semantics that ExprEvaluator implements | +| **RFC-0008** | Defines physical operator contracts that this RFC extends | +| **RFC-0010** | Defines distributed execution model that grism-ray implements | +| **RFC-0012** | Defines storage contracts that both runtimes use | +| **RFC-0100** | Defines overall architecture that this RFC refines for execution | + +--- + +## 14. Guarantees + +This RFC guarantees: + +1. **Semantic Equivalence**: Local and Ray execution produce identical results +2. **Operator Transparency**: Operators are unaware of their execution context +3. **Explicit Distribution**: All data movement is visible in distributed plans +4. **Hypergraph Correctness**: Expand operators preserve adjacency semantics +5. **Explainability**: Both runtimes support plan explanation and metrics + +--- + +## 15. Open Questions + +* Adaptive repartitioning based on runtime statistics +* Speculative execution for stragglers +* Hybrid local/distributed execution for mixed workloads +* GPU operator acceleration +* Spill-to-disk for memory-constrained execution + +--- + +## 16. Conclusion + +This RFC defines the execution engine architecture for Grism, establishing a clean separation between runtime-agnostic components and runtime-specific implementations. + +> **The engine defines what to compute.** +> **The runtime defines how to execute.** +> **RFC-0102 defines their boundary.** diff --git a/specs/rfc-index.md b/specs/rfc-index.md index 110458c..b337760 100644 --- a/specs/rfc-index.md +++ b/specs/rfc-index.md @@ -65,6 +65,7 @@ These RFCs are under active development and may be modified. | RFC | Title | Last Updated | Dependencies | Description | |-----|-------|--------------|--------------|-------------| | [RFC-0101](rfc-0101.md) | Python API Specification | 2026-01-22 | RFC-0001, RFC-0002, RFC-0003, RFC-0017, RFC-0100 | Canonical Python API for Grism. Authoritative user-facing interface with backward compatibility guarantees. | +| [RFC-0102](rfc-0102.md) | Execution Engine Architecture | 2026-01-22 | RFC-0002, RFC-0008, RFC-0010, RFC-0100 | Defines execution engine architecture with common engine layer, local runtime, and Ray distributed runtime. | --- @@ -101,6 +102,7 @@ graph TD RFC0017[RFC-0017: Transactions] RFC0100[RFC-0100: Architecture] RFC0101[RFC-0101: Python API] + RFC0102[RFC-0102: Execution Engine] RFC0001 --> RFC0002 RFC0002 --> RFC0003 @@ -162,6 +164,11 @@ graph TD RFC0017 --> RFC0101 RFC0100 --> RFC0101 + RFC0002 --> RFC0102 + RFC0008 --> RFC0102 + RFC0010 --> RFC0102 + RFC0100 --> RFC0102 + style RFC0001 fill:#e1f5ff style RFC0002 fill:#e1f5ff style RFC0003 fill:#e1f5ff @@ -188,6 +195,7 @@ graph TD - RFC-0009: Indexes & Access Paths (Draft) - RFC-0010: Distributed Execution (Draft) - RFC-0011: Runtime & Scheduling (Draft) +- RFC-0102: Execution Engine Architecture (Draft) ### Storage & Persistence - RFC-0012: Storage Layer (Draft) From 2fbba3d3da3486e978bd0cae2e902ffbb99d6318 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 22 Jan 2026 14:59:41 +0000 Subject: [PATCH 4/4] docs: Update progress file for RFC-0102 and implementation guide Co-authored-by: chenxm35 --- _workdir/progress-2026-01-22-013.md | 108 +++++++++++++++++----------- 1 file changed, 65 insertions(+), 43 deletions(-) diff --git a/_workdir/progress-2026-01-22-013.md b/_workdir/progress-2026-01-22-013.md index c56779c..3c8bd66 100644 --- a/_workdir/progress-2026-01-22-013.md +++ b/_workdir/progress-2026-01-22-013.md @@ -1,38 +1,59 @@ --- date: 2026-01-22 session: engine-runtime-separation-spec -objective: Update bridge design spec for engine-runtime separation architecture +objective: Create RFC-0102 and update implementation guide for engine-runtime separation status: completed --- ## Objective -Update `specs/8_bridge_distributed_impl.md` to define the architecture for separating common engine components from runtime-specific implementations, enabling both local and Ray-based distributed execution. +1. Create RFC-0102 as the authoritative specification for execution engine architecture +2. Update `specs/8_bridge_distributed_impl.md` as implementation-specific guide ## Completed +### Part 1: Initial Specification Update 1. **Reviewed existing specifications** - Read `specs/7_local_engine_milestone.md` for current local engine architecture - Read `specs/8_bridge_distributed_impl.md` for existing distributed bridge design - Reviewed `specs/rfc-0010.md` for distributed execution model - Reviewed `specs/rfc-namings.md` for naming conventions - -2. **Updated `specs/8_bridge_distributed_impl.md`** with: - - New three-crate architecture diagram - - `grism-engine`: Common physical layer (operators, expressions, schemas, traits) - - `grism-local`: Local runtime (in-memory and Lance storage support) - - `grism-ray`: Distributed runtime on Ray (renamed from `grism-distributed`) - - Defined `ExecutionContextTrait` for runtime abstraction - - Detailed Exchange operator and stage splitting for Ray - - Added migration path from current structure - - Implementation checklist for phased migration - - API compatibility examples showing unchanged local execution API + - Reviewed `specs/rfc-0100.md` and `specs/rfc-0008.md` for RFC format + +### Part 2: Created RFC-0102 +2. **Created `specs/rfc-0102.md`** - Execution Engine Architecture: + - Standalone authoritative RFC (no references to intermediate files) + - Defines three-crate architecture: grism-engine, grism-local, grism-ray + - Specifies physical operator model and contracts + - Defines expression evaluation architecture + - Documents local execution model (pull-based pipeline) + - Documents distributed execution model (stage-based with Exchange) + - Covers failure/recovery semantics + - Includes API compatibility section + - References other RFCs appropriately + +### Part 3: Updated Implementation Guide +3. **Rewrote `specs/8_bridge_distributed_impl.md`** as implementation guide: + - References RFC-0102 for authoritative design + - Documents current code status and what works today + - Provides detailed 6-phase migration plan + - Includes concrete code examples and patterns + - Lists specific files to create, modify, and move + - Contains comprehensive implementation checklist + +### Part 4: Updated RFC Index +4. **Updated `specs/rfc-index.md`**: + - Added RFC-0102 to API & Interfaces section + - Updated dependency graph + - Added RFC-0102 to "RFC by Layer" section ## Files Changed | File | Description | |------|-------------| -| `specs/8_bridge_distributed_impl.md` | Complete rewrite with engine-runtime separation architecture | +| `specs/rfc-0102.md` | NEW - Execution Engine Architecture RFC | +| `specs/8_bridge_distributed_impl.md` | Rewritten as implementation guide | +| `specs/rfc-index.md` | Updated to include RFC-0102 | ## Tests @@ -50,40 +71,41 @@ make lint ## Notes -### Architecture Summary - -The new design separates concerns into three crates: +### RFC-0102 Structure -1. **grism-engine (Common)** - - `PhysicalPlan`, `PhysicalOperator` trait - - All operator implementations (`*Exec`) - - `ExprEvaluator` for expression evaluation - - Schema inference - - `OperatorCaps` for capability declarations - - Memory and metrics traits +The new RFC covers: +- **Section 4**: Crate architecture with responsibilities +- **Section 5**: Common engine layer (operators, expressions, schemas) +- **Section 6**: Local runtime (LocalExecutor, pull-based model) +- **Section 7**: Ray runtime (ExchangeExec, stages, DistributedPlanner) +- **Section 8**: Blocking operators in distributed context +- **Section 9**: Graph operators in distributed context +- **Section 10-11**: Failure recovery and explainability -2. **grism-local (Local Runtime)** - - `LocalExecutor` - pull-based single-machine execution - - `LocalPhysicalPlanner` - no Exchange operators - - `LocalExecutionContext` - implements `ExecutionContextTrait` - - Support for `InMemoryStorage` and `LanceStorage` +### Implementation Guide Updates -3. **grism-ray (Distributed Runtime)** - - `RayExecutor` - stage-based distributed execution - - `DistributedPlanner` - Exchange insertion, stage splitting - - `ExchangeExec` operator for data repartitioning - - Ray task management and Arrow IPC transport +The updated implementation guide now includes: +- Current code status table (what works, what's partial, what's not started) +- 6-phase migration plan with concrete tasks +- Code snippets for new traits and types +- File changes summary (new, modified, moved) +- Implementation checklist for tracking progress -### Key Design Decisions +### Key Guarantees -- **Zero semantic drift**: Local and distributed execution must produce identical results -- **Explicit distribution**: Exchange operators are visible in plans (no hidden shuffles) -- **Backward compatibility**: Current local execution API remains unchanged -- **Stage-based execution**: Stages are the only unit of distribution +From RFC-0102: +1. Semantic Equivalence: Local and Ray execution produce identical results +2. Operator Transparency: Operators unaware of execution context +3. Explicit Distribution: All data movement visible in plans +4. Hypergraph Correctness: Expand operators preserve adjacency semantics +5. Explainability: Both runtimes support plan explanation and metrics ## Next Steps -1. Extract common traits (`ExecutionContextTrait`, `PhysicalPlanner`) from grism-engine -2. Create `grism-local` crate with moved components -3. Rename `grism-distributed` to `grism-ray` -4. Implement Exchange operator and stage splitting in grism-ray +Implementation phases as defined in `8_bridge_distributed_impl.md`: +1. Phase 1: Extract common traits from grism-engine +2. Phase 2: Create grism-local crate +3. Phase 3: Rename grism-distributed to grism-ray +4. Phase 4: Implement Exchange operator +5. Phase 5: Implement stage splitting +6. Phase 6: Implement DistributedPlanner