Struct KeyedSingleton 

pub struct KeyedSingleton<K, V, Loc, Bound: KeyedSingletonBound> { /* private fields */ }

Mapping from keys of type K to values of type V.

Keyed Singletons capture an asynchronously updated mapping from keys of the K to values of type V, where the order of keys is non-deterministic. In addition to the standard boundedness variants (Bounded for finite and immutable, Unbounded for asynchronously changing), keyed singletons can use BoundedValue to declare that new keys may be added over time, but keys cannot be removed and the value for each key is immutable.

Type Parameters:

• K: the type of the key for each entry
• V: the type of the value for each entry
• Loc: the Location where the keyed singleton is materialized
• Bound: tracks whether the entries are:
  • Bounded (local and finite)
  • Unbounded (asynchronous with entries added / removed / changed over time)
  • BoundedValue (asynchronous with immutable values for each key and no removals)

Implementations

impl<'a, K, V, L: Location<'a>, B: KeyedSingletonBound<ValueBound = Bounded>> KeyedSingleton<K, V, L, B>

pub fn entries( self, ) -> Stream<(K, V), L, B::UnderlyingBound, NoOrder, ExactlyOnce>

Flattens the keyed singleton into an unordered stream of key-value pairs.

The value for each key must be bounded, otherwise the resulting stream elements would be non-determinstic. As new entries are added to the keyed singleton, they will be streamed into the output.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton.entries()
// (1, 2), (2, 4) in any order

pub fn values(self) -> Stream<V, L, B::UnderlyingBound, NoOrder, ExactlyOnce>

Flattens the keyed singleton into an unordered stream of just the values.

The value for each key must be bounded, otherwise the resulting stream elements would be non-determinstic. As new entries are added to the keyed singleton, they will be streamed into the output.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton.values()
// 2, 4 in any order

pub fn keys(self) -> Stream<K, L, B::UnderlyingBound, NoOrder, ExactlyOnce>

Flattens the keyed singleton into an unordered stream of just the keys.

The value for each key must be bounded, otherwise the removal of keys would result in non-determinism. As new entries are added to the keyed singleton, they will be streamed into the output.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton.keys()
// 1, 2 in any order

pub fn filter_key_not_in<O2: Ordering, R2: Retries>( self, other: Stream<K, L, Bounded, O2, R2>, ) -> Self

where K: Hash + Eq,

Given a bounded stream of keys K, returns a new keyed singleton containing only the entries whose keys are not in the provided stream.

Example

let tick = process.tick();
let keyed_singleton = // { 1: 2, 2: 4 }
let keys_to_remove = process
    .source_iter(q!(vec![1]))
    .batch(&tick, nondet!(/** test */));
keyed_singleton.filter_key_not_in(keys_to_remove)
// { 2: 4 }

pub fn inspect<F>( self, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> KeyedSingleton<K, V, L, B>

where F: Fn(&V) + 'a,

An operator which allows you to “inspect” each value of a keyed singleton without modifying it. The closure f is called on a reference to each value. This is mainly useful for debugging, and should not be used to generate side-effects.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton
    .inspect(q!(|v| println!("{}", v)))
// { 1: 2, 2: 4 }

pub fn inspect_with_key<F>( self, f: impl IntoQuotedMut<'a, F, L>, ) -> KeyedSingleton<K, V, L, B>

where F: Fn(&(K, V)) + 'a,

An operator which allows you to “inspect” each entry of a keyed singleton without modifying it. The closure f is called on a reference to each key-value pair. This is mainly useful for debugging, and should not be used to generate side-effects.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton
    .inspect_with_key(q!(|(k, v)| println!("{}: {}", k, v)))
// { 1: 2, 2: 4 }

pub fn into_keyed_stream( self, ) -> KeyedStream<K, V, L, B::UnderlyingBound, TotalOrder, ExactlyOnce>

Converts this keyed singleton into a KeyedStream with each group having a single element, the value.

This is the equivalent of Singleton::into_stream but keyed.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton
    .clone()
    .into_keyed_stream()
    .interleave(
        keyed_singleton.into_keyed_stream()
    )
/// // { 1: [2, 2], 2: [4, 4] }

impl<'a, K, V, L: Location<'a>, B: KeyedSingletonBound> KeyedSingleton<K, V, L, B>

pub fn map<U, F>( self, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> KeyedSingleton<K, U, L, B>

where F: Fn(V) -> U + 'a,

Transforms each value by invoking f on each element, with keys staying the same after transformation. If you need access to the key, see KeyedStream::map_with_key.

If you do not want to modify the stream and instead only want to view each item use KeyedStream::inspect instead.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton.map(q!(|v| v + 1))
// { 1: 3, 2: 5 }

pub fn map_with_key<U, F>( self, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> KeyedSingleton<K, U, L, B>

where F: Fn((K, V)) -> U + 'a, K: Clone,

Transforms each value by invoking f on each key-value pair, with keys staying the same after transformation. Unlike KeyedSingleton::map, this gives access to both the key and value.

The closure f receives a tuple (K, V) containing both the key and value, and returns the new value U. The key remains unchanged in the output.

Example

let keyed_singleton = // { 1: 2, 2: 4 }
keyed_singleton.map_with_key(q!(|(k, v)| k + v))
// { 1: 3, 2: 6 }

pub fn filter<F>( self, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> KeyedSingleton<K, V, L, B>

where F: Fn(&V) -> bool + 'a,

Creates a keyed singleton containing only the key-value pairs where the value satisfies a predicate f.

The closure f receives a reference &V to each value and returns a boolean. If the predicate returns true, the key-value pair is included in the output. If it returns false, the pair is filtered out.

The closure f receives a reference &V rather than an owned value V because filtering does not modify or take ownership of the values. If you need to modify the values while filtering use KeyedSingleton::filter_map instead.

Example

let keyed_singleton = // { 1: 2, 2: 4, 3: 1 }
keyed_singleton.filter(q!(|&v| v > 1))
// { 1: 2, 2: 4 }

pub fn filter_map<F, U>( self, f: impl IntoQuotedMut<'a, F, L> + Copy, ) -> KeyedSingleton<K, U, L, B>

where F: Fn(V) -> Option<U> + 'a,

An operator that both filters and maps values. It yields only the key-value pairs where the supplied closure f returns Some(value).

The closure f receives each value V and returns Option<U>. If the closure returns Some(new_value), the key-value pair (key, new_value) is included in the output. If it returns None, the key-value pair is filtered out.

Example

let keyed_singleton = // { 1: "42", 2: "hello", 3: "100" }
keyed_singleton.filter_map(q!(|s| s.parse::<i32>().ok()))
// { 1: 42, 3: 100 }

pub fn key_count(self) -> Singleton<usize, L, B::UnderlyingBound>

Gets the number of keys in the keyed singleton.

The output singleton will be unbounded if the input is Unbounded or BoundedValue, since keys may be added / removed over time. When the set of keys changes, the count will be asynchronously updated.

Example

let keyed_singleton = // { 1: "a", 2: "b", 3: "c" }
keyed_singleton.key_count()
// 3

pub fn ir_node_named(self, name: &str) -> KeyedSingleton<K, V, L, B>

An operator which allows you to “name” a HydroNode. This is only used for testing, to correlate certain HydroNodes with IDs.

impl<'a, K: Hash + Eq, V, L: Location<'a>> KeyedSingleton<K, V, Tick<L>, Bounded>

pub fn get( self, key: Singleton<K, Tick<L>, Bounded>, ) -> Optional<V, Tick<L>, Bounded>

Gets the value associated with a specific key from the keyed singleton.

Example

let tick = process.tick();
let keyed_data = process
    .source_iter(q!(vec![(1, 2), (2, 3)]))
    .into_keyed()
    .batch(&tick, nondet!(/** test */))
    .first();
let key = tick.singleton(q!(1));
keyed_data.get(key).all_ticks()
// 2

pub fn get_many_if_present<O2: Ordering, R2: Retries, V2>( self, requests: KeyedStream<K, V2, Tick<L>, Bounded, O2, R2>, ) -> KeyedStream<K, (V, V2), Tick<L>, Bounded, NoOrder, R2>

Given a keyed stream of lookup requests, where the key is the lookup and the value is some additional metadata, emits a keyed stream of lookup results where the key is the same as before, but the value is a tuple of the lookup result and the metadata of the request. If the key is not found, no output will be produced.

Example

let tick = process.tick();
let keyed_data = process
    .source_iter(q!(vec![(1, 10), (2, 20)]))
    .into_keyed()
    .batch(&tick, nondet!(/** test */))
    .first();
let other_data = process
    .source_iter(q!(vec![(1, 100), (2, 200), (1, 101)]))
    .into_keyed()
    .batch(&tick, nondet!(/** test */));
keyed_data.get_many_if_present(other_data).entries().all_ticks()
// { 1: [(10, 100), (10, 101)], 2: [(20, 200)] } in any order

pub fn get_from<V2: Clone>( self, from: KeyedSingleton<V, V2, Tick<L>, Bounded>, ) -> KeyedSingleton<K, (V, Option<V2>), Tick<L>, Bounded>

where K: Clone, V: Hash + Eq + Clone,

For each entry in self, looks up the entry in the from with a key that matches the value of the entry in self. The output is a keyed singleton with tuple values containing the value from self and an option of the value from from. If the key is not present in from, the option will be None.

Example

let requests = // { 1: 10, 2: 20 }
let other_data = // { 10: 100, 11: 101 }
requests.get_from(other_data)
// { 1: (10, Some(100)), 2: (20, None) }

impl<'a, K, V, L, B: KeyedSingletonBound> KeyedSingleton<K, V, L, B>

where L: Location<'a> + NoTick + NoAtomic,

pub fn atomic(self, tick: &Tick<L>) -> KeyedSingleton<K, V, Atomic<L>, B>

Shifts this keyed singleton into an atomic context, which guarantees that any downstream logic will all be executed synchronously before any outputs are yielded (in KeyedSingleton::end_atomic).

This is useful to enforce local consistency constraints, such as ensuring that a write is processed before an acknowledgement is emitted. Entering an atomic section requires a Tick argument that declares where the keyed singleton will be atomically processed. Batching a keyed singleton into the same Tick will preserve the synchronous execution, while batching into a different Tick will introduce asynchrony.

pub fn snapshot( self, tick: &Tick<L>, nondet: NonDet, ) -> KeyedSingleton<K, V, Tick<L>, Bounded>

Returns a keyed singleton with a snapshot of each key-value entry at a non-deterministic point in time.

Non-Determinism

Because this picks a snapshot of each entry, which is continuously changing, each output has a non-deterministic set of entries since each snapshot can be at an arbitrary point in time.

impl<'a, K, V, L, B: KeyedSingletonBound> KeyedSingleton<K, V, Atomic<L>, B>

where L: Location<'a> + NoTick + NoAtomic,

pub fn snapshot(self, _nondet: NonDet) -> KeyedSingleton<K, V, Tick<L>, Bounded>

Returns a keyed singleton with a snapshot of each key-value entry, consistent with the state of the keyed singleton being atomically processed.

Non-Determinism

Because this picks a snapshot of each entry, which is continuously changing, each output has a non-deterministic set of entries since each snapshot can be at an arbitrary point in time.

pub fn end_atomic(self) -> KeyedSingleton<K, V, L, B>

Yields the elements of this keyed singleton back into a top-level, asynchronous execution context. See KeyedSingleton::atomic for more details.

impl<'a, K, V, L: Location<'a>> KeyedSingleton<K, V, Tick<L>, Bounded>

pub fn latest(self) -> KeyedSingleton<K, V, L, Unbounded>

Asynchronously yields this keyed singleton outside the tick, which will be asynchronously updated with the latest set of entries inside the tick.

This converts a bounded value inside a tick into an asynchronous value outside the tick that tracks the inner value. This is useful for getting the value as of the “most recent” tick, but note that updates are propagated asynchronously outside the tick.

The entire set of entries are propagated on each tick, which means that if a tick does not have a key “XYZ” that was present in the previous tick, the entry for “XYZ” will also be removed from the output.

Example

let tick = process.tick();
input_batch // first tick: { 1: 2, 2: 3 }, second tick: { 2: 4, 3: 5 }
    .latest()
// asynchronously changes from { 1: 2, 2: 3 } ~> { 2: 4, 3: 5 }

pub fn latest_atomic(self) -> KeyedSingleton<K, V, Atomic<L>, Unbounded>

Synchronously yields this keyed singleton outside the tick as an unbounded keyed singleton, which will be updated with the latest set of entries inside the tick.

Unlike KeyedSingleton::latest, this preserves synchronous execution, as the output keyed singleton is emitted in an Atomic context that will process elements synchronously with the input keyed singleton’s Tick context.

pub fn defer_tick(self) -> KeyedSingleton<K, V, Tick<L>, Bounded>

impl<'a, K, V, L, B: KeyedSingletonBound<ValueBound = Bounded>> KeyedSingleton<K, V, L, B>

where L: Location<'a> + NoTick + NoAtomic,

pub fn batch( self, tick: &Tick<L>, nondet: NonDet, ) -> KeyedSingleton<K, V, Tick<L>, Bounded>

Returns a keyed singleton with entries consisting of new key-value pairs that have arrived since the previous batch was released.

Currently, there is no all_ticks dual on KeyedSingleton, instead you may want to use KeyedSingleton::into_keyed_stream then yield with KeyedStream::all_ticks.

Non-Determinism

Because this picks a batch of asynchronously added entries, each output keyed singleton has a non-deterministic set of key-value pairs.

impl<'a, K, V, L, B: KeyedSingletonBound<ValueBound = Bounded>> KeyedSingleton<K, V, Atomic<L>, B>

where L: Location<'a> + NoTick + NoAtomic,

pub fn batch(self, nondet: NonDet) -> KeyedSingleton<K, V, Tick<L>, Bounded>

Returns a keyed singleton with entries consisting of new key-value pairs that are being atomically processed.

Currently, there is no dual to asynchronously yield back outside the tick, instead you should use KeyedSingleton::into_keyed_stream and yield a KeyedStream.

Non-Determinism

Because this picks a batch of asynchronously added entries, each output keyed singleton has a non-deterministic set of key-value pairs.

Trait Implementations

impl<'a, K: Clone, V: Clone, Loc: Location<'a>, Bound: KeyedSingletonBound> Clone for KeyedSingleton<K, V, Loc, Bound>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.