krabka_client_streams/dsl/
global_table.rs

1//! `GlobalKTable<K,V>`: a fully-replicated lookup table.
2//!
3//! It is a join target only. It supports no aggregations and no `to_stream`.
4//! [`StreamsBuilder::global_table`] builds it, and
5//! [`KStream::join_global`] and [`KStream::left_join_global`] consume it.
6//!
7//! A `GlobalKTable` is invisible in the wire topology. It has no source topic, no
8//! changelog, and no subtopology of its own. Its source node only occupies a
9//! node-group index during grouping, so the other subtopology ids shift. The DSL
10//! registers the store with [`Topology::add_global_store`].
11//!
12//! [`StreamsBuilder::global_table`]: crate::dsl::builder::StreamsBuilder::global_table
13//! [`KStream::join_global`]: crate::dsl::kstream::KStream::join_global
14//! [`KStream::left_join_global`]: crate::dsl::kstream::KStream::left_join_global
15//! [`Topology::add_global_store`]: crate::topology::Topology::add_global_store
16use std::{cell::RefCell, marker::PhantomData, rc::Rc};
17
18use crate::{
19    dsl::{builder::InternalStreamsBuilder, graph::NodeId},
20    processor::serde::DefaultSerde,
21};
22
23/// A handle to a fully-replicated `GlobalKTable`. It works only as a join
24/// target.
25///
26/// `store_name` is the materialized store that `KStream::join_global` and
27/// `left_join_global` wire the join processor's lookup against. `source_topic`
28/// is the broker topic that fully replicates the global store.
29pub struct GlobalKTable<K, V, KS = <K as DefaultSerde>::Serde, VS = <V as DefaultSerde>::Serde> {
30    // Keeps the shared builder alive while this handle is live.
31    #[allow(dead_code)]
32    pub(crate) builder: Rc<RefCell<InternalStreamsBuilder>>,
33    // Logical graph node for the global source.
34    #[allow(dead_code)]
35    pub(crate) node: NodeId,
36    pub(crate) store_name: String,
37    // Source topic feeding the fully replicated store.
38    #[allow(dead_code)]
39    pub(crate) source_topic: String,
40    pub(crate) key_serde: KS,
41    pub(crate) value_serde: VS,
42    _pd: PhantomData<fn() -> (K, V)>,
43}
44
45impl<K, V, KS, VS> GlobalKTable<K, V, KS, VS> {
46    pub(crate) fn new(
47        builder: Rc<RefCell<InternalStreamsBuilder>>,
48        node: NodeId,
49        store_name: String,
50        source_topic: String,
51        key_serde: KS,
52        value_serde: VS,
53    ) -> Self {
54        Self {
55            builder,
56            node,
57            store_name,
58            source_topic,
59            key_serde,
60            value_serde,
61            _pd: PhantomData,
62        }
63    }
64
65    /// The name of the materialized state store that backs this global table.
66    ///
67    /// [`KStream::join_global`](crate::dsl::kstream::KStream::join_global) and
68    /// `left_join_global` read it to wire the join processor's global-store
69    /// lookup.
70    pub(crate) fn store_name(&self) -> &str {
71        &self.store_name
72    }
73
74    pub fn with_key_serde<NewKS>(self, serde: NewKS) -> GlobalKTable<K, V, NewKS, VS> {
75        GlobalKTable {
76            builder: self.builder,
77            node: self.node,
78            store_name: self.store_name,
79            source_topic: self.source_topic,
80            key_serde: serde,
81            value_serde: self.value_serde,
82            _pd: PhantomData,
83        }
84    }
85
86    pub fn with_value_serde<NewVS>(self, serde: NewVS) -> GlobalKTable<K, V, KS, NewVS> {
87        GlobalKTable {
88            builder: self.builder,
89            node: self.node,
90            store_name: self.store_name,
91            source_topic: self.source_topic,
92            key_serde: self.key_serde,
93            value_serde: serde,
94            _pd: PhantomData,
95        }
96    }
97}
krabka_client_streams/dsl/global_table.rs

krabka_client_streams/dsl/
global_table.rs
