linkrs_query/executor/streaming/plan/
properties.rs

1//! Physical properties for physical plan nodes.
2//!
3//! Describes the output characteristics of each physical operator:
4//! distribution, ordering, pipeline kind, parallelism, and memory policy.
5//! Used in the cost model, optimizer, and parallel execution planning.
6
7use super::super::slot::SlotId;
8
9/// Data distribution strategy for the output of a physical node.
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub enum Distribution {
12    Single,
13    Broadcast,
14    HashPartitioned(Vec<SlotId>),
15}
16
17/// Output ordering guarantee.
18#[derive(Debug, Clone)]
19pub enum Ordering {
20    None,
21    Sorted(Vec<SortOrder>),
22}
23
24#[derive(Debug, Clone)]
25pub struct SortOrder {
26    pub slot: SlotId,
27    pub ascending: bool,
28}
29
30/// Whether the operator is streaming (produces incremental results)
31/// or blocking (must consume all input before producing output).
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub enum PipelineKind {
34    Streaming,
35    Blocking,
36}
37
38/// Parallelism constraints for this node.
39#[derive(Debug, Clone)]
40pub struct Parallelism {
41    pub min_workers: usize,
42    pub max_workers: usize,
43}
44
45impl Default for Parallelism {
46    fn default() -> Self {
47        Self {
48            min_workers: 1,
49            max_workers: 1,
50        }
51    }
52}
53
54/// Memory policy for blocking/spill-capable operators.
55///
56/// - `None`: no memory tracking (streaming operators that don't accumulate).
57/// - `RequiresBudget`: operator tracks memory via `MemoryTracker` but does NOT
58///   spill. When budget is exceeded, returns `ResourceExhausted`.
59/// - `Spillable`: operator can spill to disk when budget is exceeded.
60///   `threshold` is a plan-level trigger hint in bytes.
61#[derive(Debug, Clone, Default)]
62pub enum MemoryPolicy {
63    #[default]
64    None,
65    RequiresBudget,
66    Spillable {
67        threshold: u64,
68    },
69}
70
71/// Default spill threshold for operators with full external spill support.
72pub const SPILL_DEFAULT_THRESHOLD: u64 = 64 * 1024 * 1024; // 64 MB
73
74/// Physical plan properties attached to each node's output.
75#[derive(Debug, Clone)]
76pub struct PhysicalProperties {
77    pub distribution: Distribution,
78    pub ordering: Ordering,
79    pub pipeline_kind: PipelineKind,
80    pub parallelism: Parallelism,
81    pub memory_policy: MemoryPolicy,
82}
83
84impl PhysicalProperties {
85    pub fn new(
86        distribution: Distribution,
87        ordering: Ordering,
88        pipeline_kind: PipelineKind,
89        parallelism: Parallelism,
90        memory_policy: MemoryPolicy,
91    ) -> Self {
92        Self {
93            distribution,
94            ordering,
95            pipeline_kind,
96            parallelism,
97            memory_policy,
98        }
99    }
100
101    pub fn single_streaming() -> Self {
102        Self::new(
103            Distribution::Single,
104            Ordering::None,
105            PipelineKind::Streaming,
106            Parallelism::default(),
107            MemoryPolicy::None,
108        )
109    }
110
111    pub fn single_blocking() -> Self {
112        Self::new(
113            Distribution::Single,
114            Ordering::None,
115            PipelineKind::Blocking,
116            Parallelism::default(),
117            MemoryPolicy::None,
118        )
119    }
120
121    /// Blocking operator that tracks memory but does not spill.
122    pub fn single_blocking_with_budget() -> Self {
123        Self::new(
124            Distribution::Single,
125            Ordering::None,
126            PipelineKind::Blocking,
127            Parallelism::default(),
128            MemoryPolicy::RequiresBudget,
129        )
130    }
131
132    /// Blocking operator with full external spill support.
133    pub fn single_blocking_spillable(threshold: u64) -> Self {
134        Self::new(
135            Distribution::Single,
136            Ordering::None,
137            PipelineKind::Blocking,
138            Parallelism::default(),
139            MemoryPolicy::Spillable { threshold },
140        )
141    }
142
143    pub fn sorted_blocking(ordering: Ordering) -> Self {
144        Self::new(
145            Distribution::Single,
146            ordering,
147            PipelineKind::Blocking,
148            Parallelism::default(),
149            MemoryPolicy::None,
150        )
151    }
152}
linkrs_query/executor/streaming/plan/properties.rs

linkrs_query/executor/streaming/plan/
properties.rs
