diff --git a/_workdir/progress-2026-01-22-013.md b/_workdir/progress-2026-01-22-013.md new file mode 100644 index 0000000..3c8bd66 --- /dev/null +++ b/_workdir/progress-2026-01-22-013.md @@ -0,0 +1,111 @@ +--- +date: 2026-01-22 +session: engine-runtime-separation-spec +objective: Create RFC-0102 and update implementation guide for engine-runtime separation +status: completed +--- + +## Objective + +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 + - 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/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 + +``` +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 + +### RFC-0102 Structure + +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 + +### Implementation Guide Updates + +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 Guarantees + +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 + +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 diff --git a/specs/8_bridge_distributed_impl.md b/specs/8_bridge_distributed_impl.md index 30579b2..3f97d64 100644 --- a/specs/8_bridge_distributed_impl.md +++ b/specs/8_bridge_distributed_impl.md @@ -1,419 +1,740 @@ -Below is a **Week-8 bridge design** that cleanly extends your Week-7 local engine into **distributed execution**, without changing semantics or operator contracts. +# Implementation: Engine-Runtime Separation -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)**. +*(Implementation guide for RFC-0102: Execution Engine Architecture)* --- -# Design: Exchange Operator & Stage Splitting +## 1. Overview -*(Week-8 bridge: Local → Distributed Execution)* +This document provides implementation-specific details for the engine architecture defined in **RFC-0102**. It covers: ---- - -## 0. Design Objective (Why this exists) - -Week-7 gave us: - -* A **fully executable local physical plan** -* Streaming, Arrow-native operators -* Blocking operators clearly marked - -Week-8 must add: +* Current code structure and status +* Migration path to target architecture +* Implementation considerations and technical decisions +* Concrete code examples and patterns -* **Distributed execution** -* **Data repartitioning** -* **Parallel execution** -* **Failure isolation** - -Without: - -* Changing logical semantics -* Changing physical operator contracts -* Introducing implicit magic - -> **Key principle** -> Distribution is achieved by **cutting** the physical plan into **stages**, connected by an explicit **Exchange** operator. +**For architectural design, see [RFC-0102: Execution Engine Architecture](rfc-0102.md).** --- -## 1. Core Concepts +## 2. Current Code Status -### 1.1 Exchange is a Physical Operator +### 2.1 Existing Crate Structure -**Exchange is NOT a runtime trick.** -It is a **first-class physical operator** that: - -* Repartitions data -* Introduces a synchronization boundary -* Separates execution stages +``` +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 +``` -```text -Local Plan: -Scan → Filter → Expand → Aggregate → Collect +### 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 -Distributed Plan: -Scan → Filter → Expand → Exchange → Aggregate → Collect +``` +grism-engine: 97 unit tests + 33 integration tests (all passing) +grism-distributed: 0 tests (skeleton only) ``` --- -### 1.2 Stage = Executable Physical Subplan - -A **Stage** is: +## 3. Migration Plan -* A connected sub-DAG of physical operators -* Executed as a unit (locally or remotely) -* Has no internal Exchange operators +### 3.1 Phase 1: Extract Common Traits -```rust -struct ExecutionStage { - stage_id: StageId, - plan: PhysicalSubPlan, - input_partitioning: Option, - output_partitioning: Option, -} -``` - ---- +**Goal**: Define runtime-agnostic interfaces in `grism-engine`. -## 2. Exchange Operator Design +**Tasks**: -### 2.1 ExchangeExec (Physical Operator) +1. Define `ExecutionContextTrait` in `grism-engine`: ```rust -pub struct ExchangeExec { - pub partitioning: PartitioningSpec, - pub mode: ExchangeMode, +// src/grism-engine/src/executor/traits.rs + +/// 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; } ``` -### 2.2 ExchangeMode +2. Define `PhysicalPlanner` trait in `grism-engine`: ```rust -pub enum ExchangeMode { - Local, // no-op (Week 7 compatibility) - Shuffle, // repartition across workers - Broadcast, // replicate to all workers - Gather, // many → one +// 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; } ``` -Only `Local` is used in Week-7. -All others activate in Week-8. - ---- - -### 2.3 PartitioningSpec (Normative) +3. Update `PhysicalOperator` trait to use trait objects: ```rust -pub enum PartitioningSpec { - Hash { - keys: Vec, - partitions: usize, - }, - Range { - key: PhysicalExpr, - ranges: Vec, - }, - Adjacency { - entity: EntityType, // Node | Hyperedge - }, - Single, +// Change from concrete ExecutionContext to trait +#[async_trait] +pub trait PhysicalOperator: Send + Sync { + async fn execute( + &self, + ctx: &dyn ExecutionContextTrait, // Changed from &ExecutionContext + ) -> Result, GrismError>; + + // ... rest unchanged } ``` -**Important**: -Partitioning is **explicit**, never inferred at runtime. +**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 ---- +### 3.2 Phase 2: Create grism-local -### 2.4 Exchange Semantics (Non-Negotiable) +**Goal**: Move local runtime components to dedicated crate. -| Property | Guarantee | -| ------------ | ---------------------------------------- | -| Completeness | All rows forwarded | -| Disjointness | Each row appears once (except Broadcast) | -| Determinism | Same input → same partition | -| Barrier | Downstream waits for upstream readiness | +**Tasks**: -Exchange is a **semantic boundary**, not an optimization hint. +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) +``` -## 3. Why Exchange Is Explicit (Architectural Justification) +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"] } +``` -**Why not implicit shuffle?** +3. Implement `LocalExecutionContext`: -Because Grism guarantees: +```rust +// src/grism-local/src/context.rs + +pub struct LocalExecutionContext { + storage: Arc, + snapshot_id: SnapshotId, + memory_manager: Arc, + metrics_sink: Option>, + cancellation: CancellationHandle, +} -> *Parallelism must not change meaning.* +impl ExecutionContextTrait for LocalExecutionContext { + 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() + } +} +``` -Explicit Exchange ensures: +4. Add re-exports for backward compatibility in grism-engine: -* Explainability (`EXPLAIN DISTRIBUTED`) -* Cost modeling -* Debuggability -* Operator capability checks -* Hypergraph adjacency correctness +```rust +// 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. Stage Splitting Algorithm +### 3.3 Phase 3: Rename grism-distributed to grism-ray -### 4.1 Inputs +**Goal**: Rename and restructure for Ray-specific implementation. -* Fully planned **PhysicalPlan** -* Operator metadata: +**Tasks**: - * `blocking` - * `requires_global_view` - * `parallelizable` -* Execution target: `Local | Distributed` +1. Rename directory: ---- +```bash +git mv src/grism-distributed src/grism-ray +``` -### 4.2 Stage Boundary Rules (Normative) +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 +``` -A **new stage MUST start** at: +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", + # ... +] +``` -1. Any `ExchangeExec` -2. Any **blocking operator** if distribution is enabled -3. Any operator that: +4. Update all imports across the codebase. - * requires global state - * cannot be parallelized +### 3.4 Phase 4: Implement Exchange Operator -```text -[Stage A] → Exchange → [Stage B] -``` +**Goal**: Add Exchange operator for data repartitioning. ---- +**Tasks**: -### 4.3 Stage Splitting Pseudocode +1. Add Exchange types to grism-ray: ```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(); - } - current.add(node); - } +// src/grism-ray/src/operators/exchange.rs - stages.push(current.finish()); - stages +/// 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, } -``` - ---- - -## 5. Execution Model After Splitting -### 5.1 Distributed Execution Flow +/// Exchange operator for distributed data movement +pub struct ExchangeExec { + pub child: Arc, + pub partitioning: PartitioningSpec, + pub mode: ExchangeMode, +} -```text -Stage 0 (parallel) - ├─ Worker 1 - ├─ Worker 2 - └─ Worker 3 - ↓ - Exchange - ↓ -Stage 1 (parallel or single) +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()] + } +} ``` -Each stage: - -* Executes independently -* Produces Arrow batches -* Hands off via Exchange - ---- - -## 6. Exchange Execution (Ray Backend) - -### 6.1 Control Plane vs Data Plane +2. Add partitioning specification: -| Plane | Responsibility | -| ------- | ----------------------- | -| Control | Stage graph, scheduling | -| Data | Arrow batch movement | +```rust +// src/grism-ray/src/planner/partitioning.rs -Ray owns **control**, Rust owns **execution**. +/// Specification for how data is partitioned +#[derive(Debug, Clone)] +pub enum PartitioningSpec { + /// Hash partitioning by key columns + Hash { + keys: Vec, + num_partitions: usize, + }, + /// Range partitioning by key + Range { + key: String, + ranges: Vec<(Value, Value)>, + }, + /// Partitioning by graph adjacency (keeps neighbors together) + Adjacency { + entity_type: EntityType, + }, + /// Single partition (no distribution) + Single, + /// Round-robin distribution + RoundRobin { + num_partitions: usize, + }, +} ---- +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 + } + } +} +``` -### 6.2 Exchange Execution Steps +### 3.5 Phase 5: Implement Stage Splitting -**Shuffle Exchange example**: +**Goal**: Implement algorithm to split physical plans into stages. -1. Upstream stage emits `RecordBatch` -2. Batch partitioned using `PartitioningSpec` -3. Sub-batches sent to Ray object store -4. Downstream tasks pull assigned partitions +**Tasks**: -No operator knows **where** data comes from. +1. Implement stage builder: ---- +```rust +// src/grism-ray/src/planner/stage.rs + +/// Unique identifier for an execution stage +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct StageId(pub usize); + +/// 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. Expand-Aware Partitioning (Critical) +/// Builder for constructing stages +pub struct StageBuilder { + operators: Vec>, + input_partitioning: Option, +} -### 7.1 Binary Expand (Adjacency-Aware) +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 + } + } +} +``` -**Preferred strategy**: +2. Implement stage splitting algorithm: ```rust -PartitioningSpec::Adjacency { - entity: 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 + } } ``` -Guarantee: +### 3.6 Phase 6: Implement Distributed Planner -* Most expands are local -* Cross-partition traversal is explicit +**Goal**: Create DistributedPlanner that inserts Exchange operators. ---- +```rust +// src/grism-ray/src/planner/distributed.rs -### 7.2 N-ary Expand (RoleExpand) +pub struct DistributedPlanner { + pub config: DistributedPlannerConfig, +} -* Often forces shuffle -* Exchange inserted **before** Expand -* Planner decides based on cost +pub struct DistributedPlannerConfig { + pub default_parallelism: usize, + pub prefer_adjacency_partitioning: bool, +} -This preserves correctness for hyperedges. +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)) + } +} +``` --- -## 8. Blocking Operators & Global Semantics +## 4. Implementation Considerations -### 8.1 Aggregate +### 4.1 Backward Compatibility -Two-phase aggregation pattern: - -```text -Stage 0: PartialAggregate (parallel) - ↓ Exchange(Hash by group key) -Stage 1: FinalAggregate -``` +To maintain backward compatibility during migration: -This is **planned explicitly**. +1. **Feature Flags**: Use Cargo features to control crate dependencies ---- - -### 8.2 Sort - -```text -Stage 0: LocalSort - ↓ Exchange(Range) -Stage 1: MergeSort +```toml +# grism-engine/Cargo.toml +[features] +default = ["local-runtime"] +local-runtime = ["grism-local"] ``` -No hidden coordination. +2. **Re-exports**: Re-export moved types from original locations with deprecation warnings ---- +3. **Version Pinning**: Maintain API compatibility within minor versions -## 9. Operator Capability Contract (Extended) +### 4.2 Testing Strategy -Operators must declare: +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 -struct OperatorCaps { - streaming: bool, - blocking: bool, - parallel_safe: bool, - requires_partitioning: Option, +#[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); } ``` -Planner uses this to: - -* Insert Exchange -* Split stages -* Reject illegal plans - ---- - -## 10. Failure & Retry Semantics (Week-8 Scope) - -Stage boundaries define **failure domains**. +### 4.3 Performance Considerations -| Failure | Recovery | -| ---------------------- | --------------------- | -| Worker crash | Retry stage | -| Exchange failure | Re-run upstream stage | -| Blocking stage failure | Restart stage | +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 -No partial stage result is reused unless idempotent. +### 4.4 Error Handling ---- - -## 11. Explainability Guarantees - -`EXPLAIN DISTRIBUTED` must show: - -```text -Stage 0: - NodeScan → Filter → Expand - Parallelism: 8 - Output: HashPartition(node_id) - -Stage 1: - Aggregate → Collect - Parallelism: 1 +```rust +/// 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), +} ``` -This is **non-optional**. - --- -## 12. Integration with Week-7 Local Engine - -### 12.1 Local Compatibility - -When execution mode = `Local`: - -* ExchangeExec becomes **no-op** -* Single stage only -* No scheduling overhead - -Same code path, different backend. +## 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` | --- -## 13. Minimal Week-8 Implementation Checklist - -✔ ExchangeExec struct -✔ PartitioningSpec implemented -✔ Stage splitter implemented -✔ Ray stage executor wrapper -✔ Partial/final aggregate variants -✔ Expand-aware partition rules -✔ Distributed EXPLAIN output +## 6. Implementation Checklist + +### Phase 1: Extract Common Traits +- [ ] Define `ExecutionContextTrait` in grism-engine +- [ ] Define `PhysicalPlanner` trait in grism-engine +- [ ] Update `PhysicalOperator` trait to use `dyn ExecutionContextTrait` +- [ ] Update all operator implementations +- [ ] Verify all tests pass + +### Phase 2: Create grism-local +- [ ] 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 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 +- [ ] 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 --- -## 14. Architectural Payoff - -This design ensures: - -* **Zero semantic drift** -* **Local engine remains reference** -* **Distributed execution is explainable** -* **Hypergraph Expand remains correct** -* **Future extensions (GPU, spill, adaptive) slot in cleanly** - ---- - -## Final Summary - -> **Exchange is the only way data moves across stages.** -> **Stages are the only unit of distribution.** -> **Operators never know they are distributed.** +## 7. References -This is the exact architectural separation that allows Grism to scale **without ever compromising its hypergraph semantics**. \ No newline at end of file +* **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)