linkrs_query/executor/streaming/plan/
context.rs

1//! PhysicalPlanBuildContext: read-only context for building physical plans.
2//!
3//! Contains only schema catalog, statistics, capability information, and
4//! planning configuration — never runtime handles, storage clients,
5//! transaction handles, or per-query mutable state.
6//!
7//! This ensures the resulting [`PhysicalPlan`](super::types::PhysicalPlan)
8//! is immutable, cacheable, and safe to share across concurrent executions.
9
10use super::types::{FragmentIdAllocator, PhysicalOperatorIdAllocator};
11use crate::executor::base::ExecutionContext;
12use crate::executor::streaming::parameters::ParameterSchema;
13use crate::executor::streaming::slot::SlotLayout;
14use crate::planning::plan::PartitionSpec;
15use linkrs_core::error::QueryError;
16
17/// Schema identifier for catalog lookups during plan building.
18#[derive(Debug, Clone)]
19pub struct SchemaRef {
20    pub space_name: String,
21    pub layout_version: u64,
22}
23
24/// Snapshot of table/index statistics used for cardinality estimation.
25#[derive(Debug, Clone, Default)]
26pub struct StatisticsSnapshot {
27    pub row_count_estimates: Vec<(String, u64)>,
28    /// Per-logical-node output row estimates from the cost-based phase,
29    /// keyed by logical node id. Consumed by the `estimated_rows` metadata
30    /// pass to write `estimated_cardinality` onto physical operators.
31    pub per_node_row_estimates: std::collections::HashMap<i64, u64>,
32}
33
34/// Planning configuration flags and thresholds.
35#[derive(Debug, Clone)]
36pub struct PlanningConfig {
37    /// Maximum number of partitions to generate.
38    pub max_partitions: usize,
39    /// Whether to enable hash join.
40    pub enable_hash_join: bool,
41    /// Optimizer rule set version.
42    pub optimizer_version: u64,
43    /// Hash of this config for cache compatibility.
44    pub config_hash: u64,
45}
46
47impl Default for PlanningConfig {
48    fn default() -> Self {
49        Self {
50            max_partitions: 4,
51            enable_hash_join: true,
52            optimizer_version: 1,
53            config_hash: 0,
54        }
55    }
56}
57
58/// Read-only context used during [`PhysicalPlan`] construction.
59///
60/// Intentionally free of:
61/// - `QueryStorage`, transaction/session handles
62/// - Runtime, memory tracker, cursor, buffer, emitted state
63/// - Per-execution parameter values, auth context, current snapshot
64/// - Temporary space/storage references that belong in bindings
65#[derive(Debug, Clone)]
66pub struct PhysicalPlanBuildContext {
67    /// Schema catalog reference (space schema + layout version).
68    pub schema: Option<SchemaRef>,
69    /// Statistics snapshot for cardinality estimation.
70    pub statistics: StatisticsSnapshot,
71    /// Planning configuration.
72    pub config: PlanningConfig,
73    /// The output slot layout expected by the parent (if known at build time).
74    pub expected_output_layout: Option<SlotLayout>,
75
76    /// Parameter schema for prepared-statement parameters.
77    pub parameter_schema: ParameterSchema,
78
79    /// optional partition spec for partitioned execution.
80    /// When set, the builder produces a PhysicalPlan with multiple source
81    /// fragments and exchange/gather fragments instead of a single linear chain.
82    pub partition_spec: Option<PartitionSpec>,
83
84    /// Why parallel partitioning was not applied (empty = partitioning
85    /// active or not requested). Copied into the built [`PhysicalPlan`] so
86    /// EXPLAIN / PROFILE diagnostics can surface it.
87    pub parallel_fallback_reason: String,
88
89    /// Cost-based decision notes (unnest / join order) copied from the
90    /// optimized [`ExecutionPlan`] into the built [`PhysicalPlan`] so
91    /// EXPLAIN diagnostics can surface why the optimizer chose a shape.
92    pub cbo_notes: Vec<String>,
93
94    /// Effective statistics collection settings (sampling window and
95    /// refresh threshold) copied from the pipeline into the built
96    /// [`PhysicalPlan`] so EXPLAIN diagnostics can surface them.
97    pub statistics_summary: String,
98
99    // ── Allocators ──
100    pub(crate) operator_id_alloc: PhysicalOperatorIdAllocator,
101    pub(crate) fragment_id_alloc: FragmentIdAllocator,
102    // NOTE: capacity planning and spill config is not yet wired.
103}
104
105impl PhysicalPlanBuildContext {
106    /// Create a new build context from an [`ExecutionContext`], extracting
107    /// only the immutable, plan-relevant portions.
108    ///
109    /// Note: `ExecutionContext` still carries runtime-bound state that
110    /// should eventually be separated.  This constructor extracts what it
111    /// needs and leaves the rest behind.
112    pub fn from_execution_context(context: &ExecutionContext) -> Self {
113        Self {
114            schema: context.space_name.as_ref().map(|space_name| SchemaRef {
115                space_name: space_name.clone(),
116                layout_version: 0,
117            }),
118            statistics: StatisticsSnapshot::default(),
119            config: PlanningConfig {
120                max_partitions: context.max_workers,
121                ..PlanningConfig::default()
122            },
123            expected_output_layout: None,
124            parameter_schema: ParameterSchema::default(),
125            partition_spec: None,
126            parallel_fallback_reason: String::new(),
127            cbo_notes: Vec::new(),
128            statistics_summary: String::new(),
129            operator_id_alloc: PhysicalOperatorIdAllocator::new(),
130            fragment_id_alloc: FragmentIdAllocator::new(),
131        }
132    }
133
134    /// Create a minimal context for testing or simple scans.
135    pub fn new() -> Self {
136        Self {
137            schema: None,
138            statistics: StatisticsSnapshot::default(),
139            config: PlanningConfig::default(),
140            expected_output_layout: None,
141            parameter_schema: ParameterSchema::default(),
142            partition_spec: None,
143            parallel_fallback_reason: String::new(),
144            cbo_notes: Vec::new(),
145            statistics_summary: String::new(),
146            operator_id_alloc: PhysicalOperatorIdAllocator::new(),
147            fragment_id_alloc: FragmentIdAllocator::new(),
148        }
149    }
150
151    /// Allocate a new physical operator ID from the unified arena.
152    pub fn allocate_operator_id(
153        &mut self,
154    ) -> crate::executor::streaming::plan::types::PhysicalOperatorId {
155        self.operator_id_alloc.allocate()
156    }
157
158    /// Allocate a new fragment ID.
159    pub fn allocate_fragment_id(&mut self) -> crate::executor::streaming::plan::types::FragmentId {
160        self.fragment_id_alloc.allocate()
161    }
162
163    /// Peek at the next operator ID without consuming it.
164    pub fn peek_operator_id(&self) -> crate::executor::streaming::plan::types::PhysicalOperatorId {
165        self.operator_id_alloc.peek()
166    }
167
168    /// Check that this context has been configured correctly.
169    pub fn validate(&self) -> Result<(), QueryError> {
170        Ok(())
171    }
172}
173
174impl Default for PhysicalPlanBuildContext {
175    fn default() -> Self {
176        Self::new()
177    }
178}
linkrs_query/executor/streaming/plan/context.rs

linkrs_query/executor/streaming/plan/
context.rs
