hydro_lang/live_collections/
boundedness.rs

1//! Type declarations for boundedness markers, which indicate whether a live collection is finite
2//! and immutable ([`Bounded`]) or asynchronously arriving over time ([`Unbounded`]).
3
4use sealed::sealed;
5
6use super::keyed_singleton::KeyedSingletonBound;
7use super::optional::{InitNone, OptionalBound};
8use crate::compile::ir::BoundKind;
9use crate::live_collections::singleton::SingletonBound;
10
11/// A marker trait indicating whether a stream's length is bounded (finite) or unbounded (potentially infinite).
12///
13/// Implementors of this trait use it to signal the boundedness property of a stream.
14#[sealed]
15pub trait Boundedness:
16    SingletonBound<UnderlyingBound = Self>
17    + KeyedSingletonBound<UnderlyingBound = Self>
18    + OptionalBound<UnderlyingBound = Self>
19{
20    /// `true` if the bound is [`Bounded`], `false` if it is [`Unbounded`].
21    const BOUNDED: bool;
22
23    /// The [`BoundKind`] corresponding to this type.
24    const BOUND_KIND: BoundKind = if Self::BOUNDED {
25        BoundKind::Bounded
26    } else {
27        BoundKind::Unbounded
28    };
29
30    /// The [`OptionalBound`] of an optional produced by aggregating a stream with this
31    /// boundedness (e.g. `reduce`/`max`/`min`/`first`/`last`).
32    ///
33    /// A [`Bounded`] stream yields a [`Bounded`] optional. An [`Unbounded`] stream yields an
34    /// [`InitNone`] optional: such an aggregation is materialized at a top-level location where
35    /// its state persists across ticks, so once the first element arrives the result becomes
36    /// non-null and stays non-null (its value may still change).
37    type AggregatedOptional: OptionalBound<UnderlyingBound = Self>;
38
39    /// Determines the output ordering of a join based on this (right/build) side's boundedness.
40    ///
41    /// When this side is [`Bounded`], the join accumulates this side first and then
42    /// streams the left side through, preserving the left side's ordering `InO`.
43    /// When this side is [`Unbounded`], a symmetric hash join is used and ordering is lost.
44    type PreserveOrderIfBounded<InO: crate::live_collections::stream::Ordering>: crate::live_collections::stream::Ordering;
45}
46
47/// Marks the stream as being unbounded, which means that it is not
48/// guaranteed to be complete in finite time.
49pub enum Unbounded {}
50
51#[sealed]
52impl Boundedness for Unbounded {
53    const BOUNDED: bool = false;
54    type AggregatedOptional = InitNone;
55    type PreserveOrderIfBounded<InO: crate::live_collections::stream::Ordering> =
56        crate::live_collections::stream::NoOrder;
57}
58
59/// Marks the stream as being bounded, which means that it is guaranteed
60/// to be complete in finite time.
61pub enum Bounded {}
62
63#[sealed]
64impl Boundedness for Bounded {
65    const BOUNDED: bool = true;
66    type AggregatedOptional = Bounded;
67    type PreserveOrderIfBounded<InO: crate::live_collections::stream::Ordering> = InO;
68}
69
70#[sealed]
71#[diagnostic::on_unimplemented(
72    message = "The input collection must be bounded (`Bounded`), but has bound `{Self}`. Strengthen the boundedness upstream or consider a different API.",
73    label = "required here",
74    note = "To intentionally process a non-deterministic snapshot or batch, you may want to use a `sliced!` region. This introduces non-determinism so avoid unless necessary."
75)]
76/// Marker trait that is implemented for the [`Bounded`] boundedness guarantee.
77pub trait IsBounded: Boundedness {}
78
79#[sealed]
80#[diagnostic::do_not_recommend]
81impl IsBounded for Bounded {}
hydro_lang/live_collections/boundedness.rs

hydro_lang/live_collections/
boundedness.rs
