Struct KTable 

pub struct KTable<K, V, KS = <K as DefaultSerde>::Serde, VS = <V as DefaultSerde>::Serde> { /* private fields */ }

A changelog-backed table handle.

store_name is the materialized store this table reads and writes. The DSL uses it to derive changelog topics and to reuse the store in downstream materialized ops. source_topic is the Kafka topic this table was sourced from. It is set for builder.table_explicit() KTables and is None for derived KTables. The join DSL reads it to declare copartition groups.

Implementations

impl<K, V, KS, VS> KTable<K, V, KS, VS>

pub fn with_key_serde<NewKS>(self, serde: NewKS) -> KTable<K, V, NewKS, VS>

pub fn with_value_serde<NewVS>(self, serde: NewVS) -> KTable<K, V, KS, NewVS>

impl<K, V, KS, VS> KTable<K, V, KS, VS>

where K: Any + Send + Sync + Clone, V: Any + Send + Clone,

pub fn to_stream(&self) -> KStream<K, V, KS, VS>

where KS: Clone, VS: Clone,

toStream: view the table’s change-stream as a KStream, forwarding every record unchanged. Not key-changing.

pub fn map_values<V2, F>( &self, f: F, ) -> KTable<K, V2, KS, <V2 as DefaultSerde>::Serde>

where V2: DefaultSerde + Any + Send + Clone, F: Fn(&V) -> V2 + Clone + Send + Sync + 'static, KS: Clone,

mapValues: transform each value and forward the rewritten table view.

This op does not materialize a store. It matches the JVM’s non-materialized mapValues. The key stays unchanged, and the op emits no changelog topic. Use map_values_materialized for the store-backed form.

pub fn map_values_materialized<V2, NKS, NVS, F>( &self, f: F, materialized: impl Into<Materialized<NKS, NVS>>, ) -> KTable<K, V2, NKS, NVS>

where V2: Any + Send + Clone, NKS: Serde<K> + Clone + 'static, NVS: Serde<V2> + Clone + 'static, F: Fn(&V) -> V2 + Clone + Send + Sync + 'static,

mapValues: transform each value, materializing the rewritten table into a new store. Key unchanged.

pub fn filter<NKS, NVS, P>( &self, predicate: P, materialized: impl Into<Materialized<NKS, NVS>>, ) -> KTable<K, V, NKS, NVS>

where NKS: Serde<K> + Clone + 'static, NVS: Serde<V> + Clone + 'static, P: Fn(&K, &V) -> bool + Clone + Send + Sync + 'static,

filter: keep the rows that match predicate and materialize the view.

When a row that matched before stops matching, this op removes it from the store and forwards it as a Change<V> tombstone, so downstream views drop it. See the processor module doc.

pub fn group_by<KR, VR, M>(&self, mapper: M) -> KGroupedTable<KR, VR>

where KR: DefaultSerde + Any + Send + Sync + Clone + PartialEq, VR: DefaultSerde + Any + Send + Sync + Clone, <KR as DefaultSerde>::Serde: Serde<KR> + Clone, <VR as DefaultSerde>::Serde: Serde<VR> + Clone, M: Fn(&K, &V) -> (KR, VR) + Clone + Send + Sync + 'static, K: Any + Send + Sync + Clone, V: Any + Send + Clone,

groupBy: re-group the table by a new (KR, VR) derived from each entry.

After the re-group, aggregate with count, reduce, or aggregate. This op always repartitions: the JVM KTable.groupBy inserts a repartition-map, a sink, and a source.

pub fn group_by_explicit<KR, VR, GKS, GVS, M>( &self, mapper: M, grouped: impl Into<Grouped<GKS, GVS>>, ) -> KGroupedTable<KR, VR>

where KR: Any + Send + Sync + Clone + PartialEq, VR: Any + Send + Sync + Clone, GKS: Serde<KR> + Clone + 'static, GVS: Serde<VR> + Clone + 'static, M: Fn(&K, &V) -> (KR, VR) + Clone + Send + Sync + 'static, K: Any + Send + Sync + Clone, V: Any + Send + Clone,

groupBy with explicit repartition serdes.

pub fn join<VB, VR, F, VBS>( &self, other: &KTable<K, VB, KS, VBS>, joiner: F, ) -> KTable<K, VR, KS, <VR as DefaultSerde>::Serde>

where V: Sync, VB: Any + Send + Sync + Clone, VR: DefaultSerde + Any + Send + Clone, F: Fn(&V, &VB) -> VR + Clone + Send + Sync + 'static, KS: Clone, VBS: Clone,

join (inner KTable-KTable join): join rows that both tables hold.

For each key, the join row exists only when both tables hold a value. On any change to either side, the join re-reads the other side’s current value from its store and forwards a Change<VR>. It forwards a tombstone when the row stops existing.

Both tables must be materialized, because the join reads each side’s store. The join declares the two source topics as a copartition group (KIP-1071).

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join<VB, VR, F, VBS>( &self, other: &KTable<K, VB, KS, VBS>, joiner: F, ) -> KTable<K, VR, KS, <VR as DefaultSerde>::Serde>

where V: Sync, VB: Any + Send + Sync + Clone, VR: DefaultSerde + Any + Send + Clone, F: Fn(&V, Option<&VB>) -> VR + Clone + Send + Sync + 'static, KS: Clone, VBS: Clone,

leftJoin (left KTable-KTable join): emit a row for every left row.

This join emits a row whenever the left (this) side is present. The right side is optional, and the joiner receives None for it on a miss.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn outer_join<VB, VR, F, VBS>( &self, other: &KTable<K, VB, KS, VBS>, joiner: F, ) -> KTable<K, VR, KS, <VR as DefaultSerde>::Serde>

where V: Sync, VB: Any + Send + Sync + Clone, VR: DefaultSerde + Any + Send + Clone, F: Fn(Option<&V>, Option<&VB>) -> VR + Clone + Send + Sync + 'static, KS: Clone, VBS: Clone,

outerJoin (outer KTable-KTable join): emit a row for either side.

This join emits a row whenever either side is present. The joiner receives an Option for each side.

pub fn join_on_foreign_key<KO, VB, VR, FKE, J, KOS, KosOther, VbsOther>( &self, other: &KTable<KO, VB, KosOther, VbsOther>, fk_extractor: FKE, joiner: J, fk_serde: KOS, ) -> KTable<K, VR, KS, <VR as DefaultSerde>::Serde>

where KO: Any + Send + Sync + Clone, VB: Any + Send + Sync + Clone, VR: DefaultSerde + Any + Send + Sync + Clone, K: Send + Sync, V: Send + Sync, FKE: Fn(&V) -> KO + Clone + Send + Sync + 'static, J: Fn(&V, &VB) -> VR + Clone + Send + Sync + 'static, KOS: Serde<KO> + Clone + 'static, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, KosOther: Clone, VbsOther: Serde<VB> + Clone + 'static,

join on a foreign key (KIP-213 inner FK join).

For each left record, the foreign key fk_extractor(&VA) selects a row in other, which is a KTable<KO, VB>. joiner(&VA, &VB) produces the result whenever both are present.

Both tables must be materialized source tables built with builder.table, because the join reads sa and sb and needs their serdes. The join lowers to the two-subtopology KIP-213 graph: the subscription registration and response repartition topics, a subscription state store, and the five FK-join processors. See the module dsl::processors::fk for the per-processor semantics.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.

pub fn left_join_on_foreign_key<KO, VB, VR, FKE, J, KOS, KosOther, VbsOther>( &self, other: &KTable<KO, VB, KosOther, VbsOther>, fk_extractor: FKE, joiner: J, fk_serde: KOS, ) -> KTable<K, VR, KS, <VR as DefaultSerde>::Serde>

where KO: Any + Send + Sync + Clone, VB: Any + Send + Sync + Clone, VR: DefaultSerde + Any + Send + Sync + Clone, K: Send + Sync, V: Send + Sync, FKE: Fn(&V) -> KO + Clone + Send + Sync + 'static, J: Fn(&V, Option<&VB>) -> VR + Clone + Send + Sync + 'static, KOS: Serde<KO> + Clone + 'static, KS: Serde<K> + Clone + 'static, VS: Serde<V> + Clone + 'static, KosOther: Clone, VbsOther: Serde<VB> + Clone + 'static,

leftJoin on a foreign key (KIP-213 left FK join).

This join emits a row for every left record. The joiner receives None for the foreign value when the foreign key has no matching row.

impl<K, V, KS, VS> KTable<K, V, KS, VS>

where K: Any + Send + Sync + Clone, V: Any + Send + Clone,

pub fn suppress(&self, suppressed: Suppressed<K>) -> KTable<K, V, KS, VS>

where KS: Clone, VS: Clone,

suppress(Suppressed): buffer updates and emit on a delay.

On a windowed table, until_window_closes emits each window’s final value once the window closes. until_time_limit rate-limits any table to one update per key per wait.

The buffer is a registered SuppressBytesStore. It is durable, with a changelog and restore, when logging is on. The serdes come from the table-producing operation. A call to suppress on a table that changed its value type through map_values panics, because no serde factory is available.

Panics

Panics if synchronized client state is poisoned or a response violates an invariant established by protocol validation.