datum/
attributes.rs

1//! Akka-style stream/stage planner attributes.
2//!
3//! [`Attributes`] is a small, immutable, cloneable bag of [`Attribute`]s attached to a stream
4//! blueprint or a graph stage. Attributes carry validated planning intent and introspection:
5//! names, input-buffer and dispatcher hints, batching/concurrency hints, fusion constraints,
6//! metrics level, and kernel backend preference. A1 activates the existing execution machinery for
7//! [`Fusion::TypedOnly`] and [`Fusion::ErasedOnly`]; [`Attribute::BatchSize`],
8//! [`Attribute::Parallelism`], [`Attribute::MetricsLevel`], and [`Attribute::KernelBackend`] are
9//! validated, report-visible plan inputs whose execution semantics are consumed by later M14
10//! metrics/kernel rungs.
11//!
12//! Combine two bags with [`Attributes::and`]: the **argument** takes precedence over the receiver,
13//! so the most-recently-applied (most specific) value wins the accessor lookups
14//! ([`Attributes::name`], [`Attributes::input_buffer_hint`], [`Attributes::dispatcher_hint`]).
15//!
16//! Mirrors `akka.stream.Attributes`; `Attribute`/`Attributes` are re-exported from the crate root.
17
18use std::sync::Arc;
19
20use thiserror::Error;
21
22/// Validation error returned by fallible [`Attributes`] constructors.
23#[derive(Clone, Debug, PartialEq, Eq, Error)]
24pub enum AttributeError {
25    #[error("input buffer initial and max must be nonzero")]
26    ZeroInputBuffer,
27    #[error("input buffer initial ({initial}) must be <= max ({max})")]
28    InputBufferMinExceedsMax { initial: usize, max: usize },
29    #[error("batch size must be greater than zero")]
30    ZeroBatchSize,
31    #[error("parallelism must be greater than zero")]
32    ZeroParallelism,
33}
34
35/// Fallible result for validated attribute constructors.
36pub type AttributeResult<T> = Result<T, AttributeError>;
37
38/// Fusion preference for a graph node.
39///
40/// `TypedOnly` and `ErasedOnly` are active in M14-A1. `Auto` leaves the executor free to choose the
41/// best available tier. Datum deliberately does not insert async boundaries from this attribute in
42/// A1; explicit [`crate::AsyncBoundary`] stages remain the boundary mechanism.
43#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
44pub enum Fusion {
45    /// Let the planner select the fastest supported tier.
46    #[default]
47    Auto,
48    /// Require a typed executor tier and fail materialization if the graph would fall to erased.
49    TypedOnly,
50    /// Force the erased executor tier and report the tier change.
51    ErasedOnly,
52}
53
54/// Requested metrics detail for a graph node.
55///
56/// Metrics collection is implemented by the M14 metrics rung. In A1 this is a validated,
57/// report-visible planner input only.
58#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
59pub enum MetricsLevel {
60    /// No per-node metrics.
61    #[default]
62    Off,
63    /// Element/batch/row counters.
64    Counts,
65    /// Counters plus processing-time measurements.
66    Timing,
67    /// Counters, timing, and backpressure/stall observations.
68    Stalls,
69}
70
71/// Kernel backend preference for lowerable batch/scalar nodes.
72///
73/// Backend selection is implemented by the M14 kernel rungs. In A1 this is a validated,
74/// report-visible planner input only.
75#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash)]
76pub enum KernelBackend {
77    /// Let the planner choose the backend for a lowerable node.
78    Auto,
79    /// Use Datum's Rust/Arrow backend.
80    #[default]
81    ArrowRust,
82    /// Use a Python/C++ Arrow backend when that backend is available and selected.
83    Python,
84}
85
86/// A single validated stream/stage planner attribute.
87#[derive(Clone, Debug, PartialEq, Eq)]
88pub enum Attribute {
89    /// Human-readable stage/stream name (used in diagnostics).
90    Name(Arc<str>),
91    /// Requested input-buffer sizing hint (`initial` ≤ `max`).
92    InputBuffer { initial: usize, max: usize },
93    /// Dispatcher name hint for where the stage's work should run.
94    Dispatcher(Arc<str>),
95    /// Preferred batch size for batch-producing or batch-coalescing stages.
96    BatchSize(usize),
97    /// Per-node concurrency limit where semantics allow parallel work.
98    Parallelism(usize),
99    /// Fusion constraint/preference.
100    Fusion(Fusion),
101    /// Requested metrics detail.
102    MetricsLevel(MetricsLevel),
103    /// Kernel backend preference.
104    KernelBackend(KernelBackend),
105}
106
107/// An ordered, immutable collection of [`Attribute`]s.
108///
109/// Lookups return the first matching attribute; [`Attributes::and`] prepends the argument's
110/// attributes so the more specific bag wins. Construction never starts any work.
111#[derive(Clone, Debug, Default, PartialEq, Eq)]
112pub struct Attributes {
113    attribute_list: Arc<[Attribute]>,
114}
115
116impl Attributes {
117    /// An empty attribute set (the `Default`).
118    #[must_use]
119    pub fn none() -> Self {
120        Self::default()
121    }
122
123    /// A set holding a single [`Attribute::Name`].
124    #[must_use]
125    pub fn named(name: impl Into<String>) -> Self {
126        Self {
127            attribute_list: Arc::from([Attribute::Name(Arc::from(name.into()))]),
128        }
129    }
130
131    /// A set holding a single [`Attribute::InputBuffer`] hint.
132    #[must_use]
133    pub fn input_buffer(initial: usize, max: usize) -> Self {
134        Self::try_input_buffer(initial, max).expect("invalid input buffer attribute")
135    }
136
137    /// Fallible constructor for a single [`Attribute::InputBuffer`] hint.
138    pub fn try_input_buffer(initial: usize, max: usize) -> AttributeResult<Self> {
139        validate_input_buffer(initial, max)?;
140        Ok(Self {
141            attribute_list: Arc::from([Attribute::InputBuffer { initial, max }]),
142        })
143    }
144
145    /// A set holding a single [`Attribute::Dispatcher`] hint.
146    #[must_use]
147    pub fn dispatcher(name: impl Into<String>) -> Self {
148        Self {
149            attribute_list: Arc::from([Attribute::Dispatcher(Arc::from(name.into()))]),
150        }
151    }
152
153    /// A set holding a single [`Attribute::BatchSize`] hint.
154    ///
155    /// A1 records this in execution reports; batch execution semantics are added by later M14
156    /// rungs.
157    #[must_use]
158    pub fn batch_size(size: usize) -> Self {
159        Self::try_batch_size(size).expect("invalid batch size attribute")
160    }
161
162    /// Fallible constructor for a single [`Attribute::BatchSize`] hint.
163    pub fn try_batch_size(size: usize) -> AttributeResult<Self> {
164        if size == 0 {
165            return Err(AttributeError::ZeroBatchSize);
166        }
167        Ok(Self {
168            attribute_list: Arc::from([Attribute::BatchSize(size)]),
169        })
170    }
171
172    /// A set holding a single [`Attribute::Parallelism`] hint.
173    ///
174    /// A1 records this in execution reports; per-node parallel execution semantics are added by
175    /// later M14 rungs.
176    #[must_use]
177    pub fn parallelism(parallelism: usize) -> Self {
178        Self::try_parallelism(parallelism).expect("invalid parallelism attribute")
179    }
180
181    /// Fallible constructor for a single [`Attribute::Parallelism`] hint.
182    pub fn try_parallelism(parallelism: usize) -> AttributeResult<Self> {
183        if parallelism == 0 {
184            return Err(AttributeError::ZeroParallelism);
185        }
186        Ok(Self {
187            attribute_list: Arc::from([Attribute::Parallelism(parallelism)]),
188        })
189    }
190
191    /// A set holding a single [`Attribute::Fusion`] preference.
192    #[must_use]
193    pub fn fusion(fusion: Fusion) -> Self {
194        Self {
195            attribute_list: Arc::from([Attribute::Fusion(fusion)]),
196        }
197    }
198
199    /// A set holding a single [`Attribute::MetricsLevel`] preference.
200    #[must_use]
201    pub fn metrics_level(level: MetricsLevel) -> Self {
202        Self {
203            attribute_list: Arc::from([Attribute::MetricsLevel(level)]),
204        }
205    }
206
207    /// A set holding a single [`Attribute::KernelBackend`] preference.
208    #[must_use]
209    pub fn kernel_backend(backend: KernelBackend) -> Self {
210        Self {
211            attribute_list: Arc::from([Attribute::KernelBackend(backend)]),
212        }
213    }
214
215    /// Combine two sets; the attributes of `other` take precedence on accessor lookups.
216    #[must_use]
217    pub fn and(mut self, other: Self) -> Self {
218        if other.attribute_list.is_empty() {
219            return self;
220        }
221        if self.attribute_list.is_empty() {
222            return other;
223        }
224        let mut combined =
225            Vec::with_capacity(other.attribute_list.len() + self.attribute_list.len());
226        combined.extend(other.attribute_list.iter().cloned());
227        combined.extend(self.attribute_list.iter().cloned());
228        self.attribute_list = Arc::from(combined.into_boxed_slice());
229        self
230    }
231
232    /// The underlying attributes, in lookup order (most specific first).
233    #[must_use]
234    pub fn attribute_list(&self) -> &[Attribute] {
235        &self.attribute_list
236    }
237
238    /// Validate every attribute in this bag.
239    pub fn validate(&self) -> AttributeResult<()> {
240        for attribute in self.attribute_list.iter() {
241            match attribute {
242                Attribute::InputBuffer { initial, max } => validate_input_buffer(*initial, *max)?,
243                Attribute::BatchSize(0) => return Err(AttributeError::ZeroBatchSize),
244                Attribute::Parallelism(0) => return Err(AttributeError::ZeroParallelism),
245                Attribute::Name(_)
246                | Attribute::Dispatcher(_)
247                | Attribute::BatchSize(_)
248                | Attribute::Parallelism(_)
249                | Attribute::Fusion(_)
250                | Attribute::MetricsLevel(_)
251                | Attribute::KernelBackend(_) => {}
252            }
253        }
254        Ok(())
255    }
256
257    /// The first [`Attribute::Name`], if any.
258    #[must_use]
259    pub fn name(&self) -> Option<&str> {
260        self.attribute_list
261            .iter()
262            .find_map(|attribute| match attribute {
263                Attribute::Name(name) => Some(name.as_ref()),
264                _ => None,
265            })
266    }
267
268    /// The first [`Attribute::InputBuffer`] hint as `(initial, max)`, if any.
269    #[must_use]
270    pub fn input_buffer_hint(&self) -> Option<(usize, usize)> {
271        self.attribute_list
272            .iter()
273            .find_map(|attribute| match attribute {
274                Attribute::InputBuffer { initial, max } => Some((*initial, *max)),
275                _ => None,
276            })
277    }
278
279    /// The first [`Attribute::Dispatcher`] hint, if any.
280    #[must_use]
281    pub fn dispatcher_hint(&self) -> Option<&str> {
282        self.attribute_list
283            .iter()
284            .find_map(|attribute| match attribute {
285                Attribute::Dispatcher(name) => Some(name.as_ref()),
286                _ => None,
287            })
288    }
289
290    /// The first [`Attribute::BatchSize`] hint, if any.
291    #[must_use]
292    pub fn batch_size_hint(&self) -> Option<usize> {
293        self.attribute_list
294            .iter()
295            .find_map(|attribute| match attribute {
296                Attribute::BatchSize(size) => Some(*size),
297                _ => None,
298            })
299    }
300
301    /// The first [`Attribute::Parallelism`] hint, if any.
302    #[must_use]
303    pub fn parallelism_hint(&self) -> Option<usize> {
304        self.attribute_list
305            .iter()
306            .find_map(|attribute| match attribute {
307                Attribute::Parallelism(parallelism) => Some(*parallelism),
308                _ => None,
309            })
310    }
311
312    /// The first [`Attribute::Fusion`] preference, if any.
313    #[must_use]
314    pub fn fusion_hint(&self) -> Option<Fusion> {
315        self.attribute_list
316            .iter()
317            .find_map(|attribute| match attribute {
318                Attribute::Fusion(fusion) => Some(*fusion),
319                _ => None,
320            })
321    }
322
323    /// The first [`Attribute::MetricsLevel`] preference, if any.
324    #[must_use]
325    pub fn metrics_level_hint(&self) -> Option<MetricsLevel> {
326        self.attribute_list
327            .iter()
328            .find_map(|attribute| match attribute {
329                Attribute::MetricsLevel(level) => Some(*level),
330                _ => None,
331            })
332    }
333
334    /// The first [`Attribute::KernelBackend`] preference, if any.
335    #[must_use]
336    pub fn kernel_backend_hint(&self) -> Option<KernelBackend> {
337        self.attribute_list
338            .iter()
339            .find_map(|attribute| match attribute {
340                Attribute::KernelBackend(backend) => Some(*backend),
341                _ => None,
342            })
343    }
344
345    /// Return a copy with a [`Attribute::Name`] prepended so it wins [`Attributes::name`].
346    #[must_use]
347    pub fn with_name(mut self, name: impl Into<String>) -> Self {
348        let mut attributes = Vec::with_capacity(self.attribute_list.len() + 1);
349        attributes.push(Attribute::Name(Arc::from(name.into())));
350        attributes.extend(self.attribute_list.iter().cloned());
351        self.attribute_list = Arc::from(attributes.into_boxed_slice());
352        self
353    }
354}
355
356fn validate_input_buffer(initial: usize, max: usize) -> AttributeResult<()> {
357    if initial == 0 || max == 0 {
358        return Err(AttributeError::ZeroInputBuffer);
359    }
360    if initial > max {
361        return Err(AttributeError::InputBufferMinExceedsMax { initial, max });
362    }
363    Ok(())
364}
datum/attributes.rs

datum/
attributes.rs
