hydro_lang/live_collections/
mod.rs

1//! Definitions for live collections, which offer the core APIs for writing distributed applications.
2//!
3//! Traditional programs (like those in Rust) typically manipulate **collections** of data elements,
4//! such as those stored in a `Vec` or `HashMap`. These collections are **fixed** in the sense that
5//! any transformations applied to them such as `map` are immediately executed on a snapshot of the
6//! collection. This means that the output will not be updated when the input collection is modified.
7//!
8//! In Hydro, programs instead work with **live collections** which are expected to dynamically
9//! change over time as new elements are added or removed (in response to API requests, streaming
10//! ingestion, etc). Applying a transformation like `map` to a live collection results in another live
11//! collection that will dynamically change over time. All network inputs and outputs in Hydro are
12//! handled via live collections, so the majority of application logic written with Hydro will involve
13//! manipulating live collections.
14//!
15//! See the [Hydro docs](https://hydro.run/docs/hydro/reference/introduction/live-collections) for more.
16
17pub mod boundedness;
18
19/// The context type for quoted closures (`q!(...)`) passed to operators on live collections.
20///
21/// This bundles the [`crate::location::Location`] where the collection is materialized
22/// with a marker for the [`boundedness::Boundedness`] of the collection the closure operates on.
23/// Free variables captured inside such closures can constrain both components. For example,
24/// reference handles created via `by_ref()` / `by_mut()` (see [`crate::handoff_ref`]) require
25/// that both the location *and* the boundedness of the referenced collection match those of the
26/// collection whose operator captures the reference. This prevents, e.g., a reference to a
27/// [`boundedness::Bounded`] singleton from being accessed inside a `map` over an
28/// [`boundedness::Unbounded`] stream: the singleton is only materialized on the first tick,
29/// while the closure keeps running on later ticks, where accessing the reference would crash.
30pub struct OperatorContext<L, B>(L, std::marker::PhantomData<B>);
31
32impl<L: Clone, B> OperatorContext<L, B> {
33    /// Constructs an [`OperatorContext`] value for splicing quoted closures for an operator on
34    /// a collection materialized at `location` with boundedness `B`.
35    pub(crate) fn new(location: &L) -> Self {
36        OperatorContext(location.clone(), std::marker::PhantomData)
37    }
38}
39
40/// A context type for quoted snippets (`q!(...)`) from which a [`crate::location::Location`]
41/// can be extracted.
42///
43/// This is implemented both for bare locations (used when splicing quoted *values*, such as
44/// the argument to [`crate::location::Location::source_iter`]) and for [`OperatorContext`]
45/// (used when splicing quoted *closures* passed to operators on live collections). Free
46/// variables that only depend on the location of the splice site (such as
47/// [`crate::location::cluster::CLUSTER_SELF_ID`]) are generic over this trait so that they can
48/// be captured in both kinds of quoted snippets.
49pub trait ContextWithLocation<'a> {
50    /// The location type of this context.
51    type Location: crate::location::Location<'a>;
52
53    /// Extracts the location of the splice site from this context.
54    fn context_location(&self) -> &Self::Location;
55}
56
57impl<'a, L: crate::location::Location<'a>> ContextWithLocation<'a> for L {
58    type Location = L;
59
60    fn context_location(&self) -> &L {
61        self
62    }
63}
64
65impl<'a, L: crate::location::Location<'a>, B> ContextWithLocation<'a> for OperatorContext<L, B> {
66    type Location = L;
67
68    fn context_location(&self) -> &L {
69        &self.0
70    }
71}
72
73pub mod keyed_singleton;
74#[doc(inline)]
75pub use keyed_singleton::KeyedSingleton;
76
77pub mod keyed_stream;
78#[doc(inline)]
79pub use keyed_stream::KeyedStream;
80
81pub mod optional;
82#[doc(inline)]
83pub use optional::Optional;
84
85pub mod singleton;
86#[doc(inline)]
87pub use singleton::Singleton;
88
89pub mod stream;
90#[doc(inline)]
91pub use stream::Stream;
92
93pub mod sliced;
94
95#[doc(hidden)]
96pub mod batch_atomic;
97
98/// Wraps a freshly-created live collection IR node in an `Rc<RefCell<...>>` and registers it
99/// with the flow state, so that the [`crate::compile::builder::FlowBuilder`] can yank the IR
100/// of collections that are still alive when the flow is finalized (see hydro-project/hydro#3051).
101pub(crate) fn tracked_ir_node(
102    flow_state: &crate::compile::builder::FlowState,
103    ir_node: crate::compile::ir::HydroNode,
104) -> std::rc::Rc<std::cell::RefCell<crate::compile::ir::HydroNode>> {
105    let cell = std::rc::Rc::new(std::cell::RefCell::new(ir_node));
106    flow_state
107        .borrow_mut()
108        .live_collection_nodes
109        .push(std::rc::Rc::downgrade(&cell));
110    cell
111}
hydro_lang/live_collections/mod.rs

hydro_lang/live_collections/
mod.rs
