krabka_client_streams/dsl/
kgrouped.rs

1//! `KGroupedStream<K,V>`: the handle between a group op and an aggregation.
2//!
3//! The group ops are `groupByKey` and `groupBy`. The terminal aggregations are
4//! `count`, `reduce`, and `aggregate`.
5//!
6//! `groupByKey` and `groupBy` record **no** graph node. They capture four things:
7//! whether the upstream was key-changing, the optional `Grouped` name, the value
8//! serdes for the repartition topic, and a typed repartition-lowering thunk. The
9//! terminal aggregation then does three steps:
10//!
11//! 1. It mints the **store name** at the JVM counter position. The name comes
12//!    from `Materialized`, or from a fresh `KSTREAM-AGGREGATE-STATE-STORE-` or
13//!    `KSTREAM-REDUCE-STATE-STORE-` counter.
14//! 2. If the upstream is key-changing, it records a `Repartition` node. That
15//!    node's thunk lowers `sink → add_repartition_topic → source`, so the runtime
16//!    re-groups records by key before the aggregation. This splits the topology
17//!    into 2 subtopologies.
18//! 3. It records an `Aggregate` node whose thunk adds the aggregation processor
19//!    and its state store.
20//!
21//! Repartition and changelog topic names follow the JVM naming rules, so the
22//! client can send the built topology through `StreamsGroupHeartbeat`.
23
24use std::{any::Any, cell::RefCell, marker::PhantomData, rc::Rc};
25
26use crate::{
27    dsl::{
28        builder::InternalStreamsBuilder,
29        config::Materialized,
30        graph::{GraphNodeKind, LowerState, NodeId},
31        ktable::KTable,
32        names,
33        processors::{
34            aggregate::{KStreamAggregateProcessor, KStreamReduceProcessor},
35            tuple_forwarder::TupleForwarder,
36        },
37    },
38    processor::serde::{DefaultSerde, I64Serde, Serde},
39    topology::NodeHandle,
40};
41
42/// A typed thunk that wires a repartition `sink → topic → source` triple into
43/// the Processor-API topology.
44///
45/// The thunk returns the source node's name. It is built from the `Grouped`
46/// serdes, so the serialization round-trip through the repartition topic is
47/// type-correct. It is type-erased here, so `KGroupedStream<K,V>` stays free of
48/// the serde type parameters.
49///
50/// Args: `(state, parent_name, sink_name, source_name, topic)`.
51pub(crate) type RepartitionLowerFn =
52    Box<dyn FnOnce(&mut LowerState, String, String, String, String) + Send>;
53
54/// Build a [`RepartitionLowerFn`] that captures the `Grouped` key and value
55/// serdes.
56pub(crate) fn repartition_lower<K, V, KS, VS>(key_serde: KS, value_serde: VS) -> RepartitionLowerFn
57where
58    K: Any + Send + Clone,
59    V: Any + Send + Clone,
60    KS: Serde<K> + Clone + 'static,
61    VS: Serde<V> + Clone + 'static,
62{
63    Box::new(
64        move |state: &mut LowerState,
65              parent_name: String,
66              sink_name: String,
67              source_name: String,
68              topic: String| {
69            let parent = NodeHandle::<K, V>::from_name(parent_name);
70            // sink: parent → repartition topic
71            state.topology.add_sink_explicit::<K, V, KS, VS, _, _>(
72                sink_name,
73                topic.clone(),
74                [parent],
75                crate::processor::serde::Produced::with(key_serde.clone(), value_serde.clone()),
76            );
77            // mark the topic internal so the test driver / wire treat it as a
78            // repartition topic (loop-back source)
79            state.topology.add_repartition_topic(topic.clone());
80            // source: repartition topic → new source node
81            state.topology.add_source_explicit::<K, V, KS, VS>(
82                source_name,
83                [topic],
84                crate::processor::serde::Consumed::with(key_serde, value_serde),
85            );
86        },
87    )
88}
89
90/// Handle produced by `groupByKey` and `groupBy`. A terminal aggregation
91/// consumes it.
92pub struct KGroupedStream<K, V> {
93    builder: Rc<RefCell<InternalStreamsBuilder>>,
94    /// Logical id of the node that feeds the aggregation. That node is the
95    /// source or the select-key.
96    parent: NodeId,
97    /// True when the upstream key was rewritten without a re-group. The
98    /// aggregation must then insert a repartition before the aggregate node.
99    key_changing_upstream: bool,
100    /// Explicit `Grouped` name used to derive the repartition topic name.
101    #[allow(dead_code)]
102    grouped_name: Option<String>,
103    /// Typed repartition-lowering thunk. The terminal op takes it once.
104    repartition_lower: Option<RepartitionLowerFn>,
105    /// The single source topic this grouped stream traces to, when one applies.
106    /// One applies for `groupByKey` on a single-source, key-unchanged stream.
107    /// Cogroup reads it to register a copartition group over all input source
108    /// topics.
109    pub(crate) source_topic: Option<String>,
110    _pd: PhantomData<fn() -> (K, V)>,
111}
112
113impl<K, V> KGroupedStream<K, V>
114where
115    K: Any + Send + Sync + Clone,
116    V: Any + Send + Clone,
117{
118    pub(crate) fn new(
119        builder: Rc<RefCell<InternalStreamsBuilder>>,
120        parent: NodeId,
121        key_changing_upstream: bool,
122        grouped_name: Option<String>,
123        repartition_lower: RepartitionLowerFn,
124    ) -> Self {
125        Self {
126            builder,
127            parent,
128            key_changing_upstream,
129            grouped_name,
130            repartition_lower: Some(repartition_lower),
131            source_topic: None,
132            _pd: PhantomData,
133        }
134    }
135
136    /// Set the single source-topic lineage for cogroup copartition registration.
137    pub(crate) fn with_source_topic(mut self, topic: Option<String>) -> Self {
138        self.source_topic = topic;
139        self
140    }
141
142    /// `count`: count records per key into a materialized `KTable<K, i64>`.
143    ///
144    /// `init = || 0`, `agg = |_k, _v, acc| acc + 1`.
145    pub fn count_explicit<KS, VS>(
146        self,
147        materialized: impl Into<Materialized<KS, VS>>,
148    ) -> KTable<K, i64, KS, VS>
149    where
150        KS: Serde<K> + Clone + 'static,
151        VS: Serde<i64> + Clone + 'static,
152    {
153        self.aggregate_inner(
154            materialized.into(),
155            names::AGGREGATE_STORE,
156            || 0i64,
157            |_k: &K, _v: &V, acc: i64| acc + 1,
158        )
159    }
160
161    /// `reduce`: combine values per key with `reducer`.
162    ///
163    /// The result is materialized as `KTable<K, V>`. The first value for a key
164    /// seeds the accumulator, because the JVM `Reducer` has no separate `init`.
165    /// Later values fold with `reducer(&acc, &value)`. The backing processor keeps
166    /// the public value type `V`, so no `Option` and no sentinel leaks into the
167    /// `KTable`.
168    pub fn reduce_explicit<KS, VS, R>(
169        self,
170        reducer: R,
171        materialized: impl Into<Materialized<KS, VS>>,
172    ) -> KTable<K, V, KS, VS>
173    where
174        KS: Serde<K> + Clone + 'static,
175        VS: Serde<V> + Clone + 'static,
176        R: Fn(&V, &V) -> V + Clone + Send + Sync + 'static,
177    {
178        let materialized = materialized.into();
179        let store_name = mint_store_name(&self.builder, &materialized, names::REDUCE_STORE);
180        self.lower_reduce::<KS, VS, R>(materialized, store_name, reducer)
181    }
182
183    /// `aggregate`: general aggregation with a caller-supplied `init` and `agg`.
184    ///
185    /// The result is materialized as `KTable<K, VA>`.
186    pub fn aggregate_explicit<KS, VS, VA, I, A>(
187        self,
188        init: I,
189        agg: A,
190        materialized: impl Into<Materialized<KS, VS>>,
191    ) -> KTable<K, VA, KS, VS>
192    where
193        VA: Any + Send + Clone,
194        KS: Serde<K> + Clone + 'static,
195        VS: Serde<VA> + Clone + 'static,
196        I: Fn() -> VA + Clone + Send + Sync + 'static,
197        A: Fn(&K, &V, VA) -> VA + Clone + Send + Sync + 'static,
198    {
199        self.aggregate_inner(materialized.into(), names::AGGREGATE_STORE, init, agg)
200    }
201
202    pub fn count(
203        self,
204        store_name: impl Into<String>,
205    ) -> KTable<K, i64, <K as DefaultSerde>::Serde, I64Serde>
206    where
207        K: DefaultSerde,
208        <K as DefaultSerde>::Serde: Serde<K> + Clone,
209    {
210        self.count_explicit(
211            Materialized::with(<K as DefaultSerde>::Serde::default(), I64Serde)
212                .as_store(store_name),
213        )
214    }
215
216    pub fn reduce<R>(
217        self,
218        reducer: R,
219        store_name: impl Into<String>,
220    ) -> KTable<K, V, <K as DefaultSerde>::Serde, <V as DefaultSerde>::Serde>
221    where
222        K: DefaultSerde,
223        V: DefaultSerde,
224        <K as DefaultSerde>::Serde: Serde<K> + Clone,
225        <V as DefaultSerde>::Serde: Serde<V> + Clone,
226        R: Fn(&V, &V) -> V + Clone + Send + Sync + 'static,
227    {
228        self.reduce_explicit(
229            reducer,
230            Materialized::with(
231                <K as DefaultSerde>::Serde::default(),
232                <V as DefaultSerde>::Serde::default(),
233            )
234            .as_store(store_name),
235        )
236    }
237
238    pub fn aggregate<VA, I, A>(
239        self,
240        init: I,
241        agg: A,
242        store_name: impl Into<String>,
243    ) -> KTable<K, VA, <K as DefaultSerde>::Serde, <VA as DefaultSerde>::Serde>
244    where
245        VA: DefaultSerde + Any + Send + Clone,
246        K: DefaultSerde,
247        <K as DefaultSerde>::Serde: Serde<K> + Clone,
248        <VA as DefaultSerde>::Serde: Serde<VA> + Clone,
249        I: Fn() -> VA + Clone + Send + Sync + 'static,
250        A: Fn(&K, &V, VA) -> VA + Clone + Send + Sync + 'static,
251    {
252        self.aggregate_explicit(
253            init,
254            agg,
255            Materialized::with(
256                <K as DefaultSerde>::Serde::default(),
257                <VA as DefaultSerde>::Serde::default(),
258            )
259            .as_store(store_name),
260        )
261    }
262
263    /// `windowedBy(TimeWindows)`: switch to a windowed aggregation.
264    ///
265    /// This method moves the grouped lineage into a
266    /// [`crate::dsl::TimeWindowedKGroupedStream`]. The lineage is the parent, the
267    /// key-changing flag, and the repartition thunk. That handle exposes the
268    /// windowed `count`, `reduce`, and `aggregate`, which produce a
269    /// `KTable<Windowed<K>, _>`.
270    #[must_use]
271    pub fn windowed_by(
272        mut self,
273        windows: crate::dsl::windows::TimeWindows,
274    ) -> crate::dsl::windowed_kgrouped::TimeWindowedKGroupedStream<K, V> {
275        crate::dsl::windowed_kgrouped::TimeWindowedKGroupedStream::new(
276            Rc::clone(&self.builder),
277            self.parent,
278            self.key_changing_upstream,
279            self.grouped_name.take(),
280            self.repartition_lower.take(),
281            windows,
282        )
283    }
284
285    /// `windowedBy(SessionWindows)`: switch to a session aggregation.
286    ///
287    /// This method moves the grouped lineage into a
288    /// [`crate::dsl::SessionWindowedKGroupedStream`]. That handle exposes the
289    /// session `count`, `reduce`, and `aggregate`, which produce a
290    /// `KTable<Windowed<K>, _>`. The method name is distinct, because Rust cannot
291    /// overload `windowed_by` by the window-spec argument type as the JVM does.
292    #[must_use]
293    pub fn windowed_by_session(
294        mut self,
295        windows: crate::dsl::windows::SessionWindows,
296    ) -> crate::dsl::session_windowed_kgrouped::SessionWindowedKGroupedStream<K, V>
297    where
298        V: Sync,
299    {
300        crate::dsl::session_windowed_kgrouped::SessionWindowedKGroupedStream::new(
301            Rc::clone(&self.builder),
302            self.parent,
303            self.key_changing_upstream,
304            self.grouped_name.take(),
305            self.repartition_lower.take(),
306            windows,
307        )
308    }
309
310    /// `windowedBy(SlidingWindows)`: switch to a sliding-window aggregation
311    /// (KIP-450).
312    ///
313    /// This method moves the grouped lineage into a
314    /// [`crate::dsl::SlidingWindowedKGroupedStream`]. The method name is
315    /// distinct, because Rust cannot overload `windowed_by` by the window-spec
316    /// argument type as the JVM does.
317    #[must_use]
318    pub fn windowed_by_sliding(
319        mut self,
320        windows: crate::dsl::windows::SlidingWindows,
321    ) -> crate::dsl::sliding_windowed_kgrouped::SlidingWindowedKGroupedStream<K, V>
322    where
323        V: Sync,
324    {
325        crate::dsl::sliding_windowed_kgrouped::SlidingWindowedKGroupedStream::new(
326            Rc::clone(&self.builder),
327            self.parent,
328            self.key_changing_upstream,
329            self.grouped_name.take(),
330            self.repartition_lower.take(),
331            windows,
332        )
333    }
334
335    /// `cogroup`: begin a KIP-150 cogroup with this stream as the first input.
336    ///
337    /// `agg` is this input's aggregator. Returns a
338    /// [`CogroupedKStream<K, VOut>`]. Chain more inputs onto it, then terminate
339    /// with `aggregate` or one of the `windowed_by*` methods.
340    ///
341    /// [`CogroupedKStream<K, VOut>`]: crate::dsl::cogrouped::CogroupedKStream
342    #[must_use]
343    pub fn cogroup<VOut, A>(self, agg: A) -> crate::dsl::cogrouped::CogroupedKStream<K, VOut>
344    where
345        V: Sync,
346        VOut: Any + Send + Sync + Clone,
347        A: Fn(&K, &V, VOut) -> VOut + Send + Sync + 'static,
348    {
349        let builder = Rc::clone(&self.builder);
350        let (parent, key_changing, rp_lower, source_topic) = self.into_cogroup_parts();
351        crate::dsl::cogrouped::CogroupedKStream::new(
352            builder,
353            vec![crate::dsl::cogrouped::CogroupInput {
354                parent,
355                key_changing_upstream: key_changing,
356                repartition_lower: rp_lower,
357                make_agg: crate::dsl::cogrouped::make_agg_for_input::<K, V, VOut, A>(agg),
358                source_topic,
359            }],
360        )
361    }
362
363    /// Decompose into the lineage parts a cogroup input needs. This method
364    /// consumes the handle. It returns
365    /// `(parent_id, key_changing, repartition_lower, source_topic)`.
366    pub(crate) fn into_cogroup_parts(
367        mut self,
368    ) -> (NodeId, bool, Option<RepartitionLowerFn>, Option<String>) {
369        (
370            self.parent,
371            self.key_changing_upstream,
372            self.repartition_lower.take(),
373            self.source_topic,
374        )
375    }
376
377    /// Shared body for `count` and `aggregate`. This method mints the store name
378    /// at the JVM counter position, then lowers the optional repartition node and
379    /// the aggregate node.
380    fn aggregate_inner<KS, VS, VA, I, A>(
381        self,
382        materialized: Materialized<KS, VS>,
383        store_prefix: &'static str,
384        init: I,
385        agg: A,
386    ) -> KTable<K, VA, KS, VS>
387    where
388        VA: Any + Send + Clone,
389        KS: Serde<K> + Clone + 'static,
390        VS: Serde<VA> + Clone + 'static,
391        I: Fn() -> VA + Clone + Send + Sync + 'static,
392        A: Fn(&K, &V, VA) -> VA + Clone + Send + Sync + 'static,
393    {
394        let store_name = mint_store_name(&self.builder, &materialized, store_prefix);
395        self.lower_aggregate::<KS, VS, VA, I, A>(materialized, store_name, init, agg)
396    }
397
398    /// Record the optional repartition node and a `KStreamAggregateProcessor`
399    /// aggregate node, then return the resulting `KTable`. The caller has already
400    /// minted the store name at the JVM counter position.
401    fn lower_aggregate<KS, VS, VA, I, A>(
402        mut self,
403        materialized: Materialized<KS, VS>,
404        store_name: String,
405        init: I,
406        agg: A,
407    ) -> KTable<K, VA, KS, VS>
408    where
409        VA: Any + Send + Clone,
410        KS: Serde<K> + Clone + 'static,
411        VS: Serde<VA> + Clone + 'static,
412        I: Fn() -> VA + Clone + Send + Sync + 'static,
413        A: Fn(&K, &V, VA) -> VA + Clone + Send + Sync + 'static,
414    {
415        let Materialized {
416            key_serde,
417            value_serde,
418            logging,
419            caching,
420            ..
421        } = materialized;
422        let suppress_factory = crate::dsl::ktable::kv_suppress_factory::<K, VA, KS, VS>(
423            key_serde.clone(),
424            value_serde.clone(),
425        );
426        let parent = self.parent;
427        let key_changing = self.key_changing_upstream;
428        let rp_lower = self.repartition_lower.take();
429        let mut g = self.builder.borrow_mut();
430        let agg_parent =
431            Self::record_repartition(&mut g, &store_name, parent, key_changing, rp_lower);
432
433        let agg_name = g.new_processor_name(names::AGGREGATE);
434        let agg_id = g.graph.add(
435            agg_name.clone(),
436            GraphNodeKind::Aggregate {
437                store_name: store_name.clone(),
438                changelog: logging,
439            },
440            vec![agg_parent],
441        );
442        let store_for_thunk = store_name.clone();
443        let key_serde_for_lower = key_serde.clone();
444        let value_serde_for_lower = value_serde.clone();
445        g.graph.nodes[agg_id].lower = Some(Box::new(move |state: &mut LowerState| {
446            let parent = NodeHandle::<K, V>::from_name(state.handle_name[&agg_parent].clone());
447            let store_for_proc = store_for_thunk.clone();
448            // The aggregate forwards Change<VA> (prior store value as old).
449            let h = state
450                .topology
451                .add_processor::<K, V, K, crate::dsl::processors::change::Change<VA>, _, _, _>(
452                    agg_name.clone(),
453                    move || KStreamAggregateProcessor {
454                        store_name: store_for_proc.clone(),
455                        init: init.clone(),
456                        agg: agg.clone(),
457                        forwarder: TupleForwarder::default(),
458                        _pd: PhantomData,
459                    },
460                    [parent],
461                );
462            // Honor `Materialized::with_logging(bool)`:
463            // - logging=true  → standard add_state_store (changelog topic emitted)
464            // - logging=false → add_state_store_no_changelog (store usable at
465            //   runtime, but NO state_changelog_topics entry in the wire topology)
466            if logging {
467                state.topology.add_state_store::<K, VA, KS, VS>(
468                    store_for_thunk.clone(),
469                    key_serde_for_lower.clone(),
470                    value_serde_for_lower.clone(),
471                    [h.name().to_string()],
472                );
473            } else {
474                state
475                    .topology
476                    .add_state_store_no_changelog::<K, VA, KS, VS>(
477                        store_for_thunk.clone(),
478                        key_serde_for_lower.clone(),
479                        value_serde_for_lower.clone(),
480                    );
481            }
482            state.topology.mark_store_caching(&store_for_thunk, caching);
483            state.handle_name.insert(agg_id, h.name().to_string());
484        }));
485
486        drop(g);
487        KTable::new(
488            Rc::clone(&self.builder),
489            agg_id,
490            Some(store_name),
491            None,
492            key_serde,
493            value_serde,
494        )
495        .with_suppress_factory(Some(suppress_factory))
496    }
497
498    /// Record the optional repartition node and a `KStreamReduceProcessor` node,
499    /// then return the `KTable<K, V>`. The first value seeds the accumulator and
500    /// later values fold into it.
501    fn lower_reduce<KS, VS, R>(
502        mut self,
503        materialized: Materialized<KS, VS>,
504        store_name: String,
505        reducer: R,
506    ) -> KTable<K, V, KS, VS>
507    where
508        KS: Serde<K> + Clone + 'static,
509        VS: Serde<V> + Clone + 'static,
510        R: Fn(&V, &V) -> V + Clone + Send + Sync + 'static,
511    {
512        let Materialized {
513            key_serde,
514            value_serde,
515            logging,
516            caching,
517            ..
518        } = materialized;
519        let suppress_factory = crate::dsl::ktable::kv_suppress_factory::<K, V, KS, VS>(
520            key_serde.clone(),
521            value_serde.clone(),
522        );
523        let parent = self.parent;
524        let key_changing = self.key_changing_upstream;
525        let rp_lower = self.repartition_lower.take();
526        let mut g = self.builder.borrow_mut();
527        let agg_parent =
528            Self::record_repartition(&mut g, &store_name, parent, key_changing, rp_lower);
529
530        let red_name = g.new_processor_name(names::REDUCE);
531        let red_id = g.graph.add(
532            red_name.clone(),
533            GraphNodeKind::Aggregate {
534                store_name: store_name.clone(),
535                changelog: logging,
536            },
537            vec![agg_parent],
538        );
539        let store_for_thunk = store_name.clone();
540        let key_serde_for_lower = key_serde.clone();
541        let value_serde_for_lower = value_serde.clone();
542        g.graph.nodes[red_id].lower = Some(Box::new(move |state: &mut LowerState| {
543            let parent = NodeHandle::<K, V>::from_name(state.handle_name[&agg_parent].clone());
544            let store_for_proc = store_for_thunk.clone();
545            let reducer = reducer.clone();
546            // The reduce forwards Change<V> (prior store value as old).
547            let h = state
548                .topology
549                .add_processor::<K, V, K, crate::dsl::processors::change::Change<V>, _, _, _>(
550                    red_name.clone(),
551                    move || KStreamReduceProcessor {
552                        store_name: store_for_proc.clone(),
553                        reducer: reducer.clone(),
554                        forwarder: TupleForwarder::default(),
555                        _pd: PhantomData,
556                    },
557                    [parent],
558                );
559            // Honor `Materialized::with_logging(bool)` for reduce as well.
560            if logging {
561                state.topology.add_state_store::<K, V, KS, VS>(
562                    store_for_thunk.clone(),
563                    key_serde_for_lower.clone(),
564                    value_serde_for_lower.clone(),
565                    [h.name().to_string()],
566                );
567            } else {
568                state.topology.add_state_store_no_changelog::<K, V, KS, VS>(
569                    store_for_thunk.clone(),
570                    key_serde_for_lower.clone(),
571                    value_serde_for_lower.clone(),
572                );
573            }
574            state.topology.mark_store_caching(&store_for_thunk, caching);
575            state.handle_name.insert(red_id, h.name().to_string());
576        }));
577
578        drop(g);
579        KTable::new(
580            Rc::clone(&self.builder),
581            red_id,
582            Some(store_name),
583            None,
584            key_serde,
585            value_serde,
586        )
587        .with_suppress_factory(Some(suppress_factory))
588    }
589
590    /// Record a `Repartition` node when the upstream is key-changing.
591    ///
592    /// The node lowers as `sink → repartition topic → source` and re-groups by
593    /// key. This method then returns its id, which is the aggregation's parent.
594    /// If the upstream is not key-changing, it returns the grouped parent id
595    /// unchanged. The thunk finalizes the repartition topic name
596    /// `<app>-<store>-repartition` with the same naming rule as the JVM client.
597    ///
598    /// This method takes the fields it needs by value and not through
599    /// `&mut self`, so the caller can hold the `builder.borrow_mut()` guard `g`
600    /// across the call.
601    pub(crate) fn record_repartition(
602        g: &mut InternalStreamsBuilder,
603        store_name: &str,
604        parent: NodeId,
605        key_changing_upstream: bool,
606        repartition_lower: Option<RepartitionLowerFn>,
607    ) -> NodeId {
608        if !key_changing_upstream {
609            return parent;
610        }
611        // The JVM's `KStreamImpl.createRepartitionedSource` first mints a
612        // null-key **filter** node (records with a null key can't be repartitioned
613        // by key), then the sink + source. We don't lower a filter node (the
614        // aggregate processors already reject null keys), but we must consume the
615        // counter index so downstream auto-names match the JVM byte-for-byte — e.g.
616        // a second aggregation's store lands at the same index as the JVM fixture.
617        let _filter_name = g.new_processor_name(names::FILTER);
618        let sink_name = g.new_processor_name(names::SINK);
619        let source_name = g.new_processor_name(names::SOURCE);
620        let upstream = parent;
621        let topic_store = store_name.to_string();
622        let lower = repartition_lower.expect("repartition_lower already consumed");
623        let rp_id = g.graph.add(
624            source_name.clone(),
625            GraphNodeKind::Repartition {
626                topic: format!("{topic_store}{}", names::REPARTITION_SUFFIX),
627                partitions: None,
628            },
629            vec![upstream],
630        );
631        g.graph.nodes[rp_id].lower = Some(Box::new(move |state: &mut LowerState| {
632            let parent_name = state.handle_name[&upstream].clone();
633            let topic = format!(
634                "{}-{topic_store}{}",
635                state.app_id,
636                names::REPARTITION_SUFFIX
637            );
638            lower(
639                state,
640                parent_name,
641                sink_name.clone(),
642                source_name.clone(),
643                topic,
644            );
645            // The aggregate wires to the repartition source node.
646            state.handle_name.insert(rp_id, source_name.clone());
647        }));
648        rp_id
649    }
650}
651
652/// Mint the store name.
653///
654/// Returns the `Materialized` name when it is present, otherwise a fresh counter
655/// at the JVM position. The caller calls this *before* it mints the aggregate
656/// processor name.
657pub(crate) fn mint_store_name<KS, VS>(
658    builder: &Rc<RefCell<InternalStreamsBuilder>>,
659    materialized: &Materialized<KS, VS>,
660    prefix: &str,
661) -> String {
662    match &materialized.store_name {
663        Some(name) => name.clone(),
664        None => builder.borrow_mut().new_processor_name(prefix),
665    }
666}
krabka_client_streams/dsl/kgrouped.rs

krabka_client_streams/dsl/
kgrouped.rs
