dbsp/operator/
star_join_macros.rs

1#[macro_export]
2macro_rules! generate_all_join_orderings {
3    ($($f:tt)::+, [$($fixed:tt)*], $first:expr, $($rest:expr),+ $(,)?) => {
4        $crate::generate_all_join_orderings!(@collect $($f)::+, [$($fixed)*], $first, []; []; $($rest),+)
5    };
6    (@collect $($f:tt)::+, [$($fixed:tt)*], $first:expr,
7        [$($out:expr,)*]; [$($prefix:expr,)*];
8        $head:expr $(, $tail:expr)*
9    ) => {
10        $crate::generate_all_join_orderings!(
11            @collect
12            $($f)::+,
13            [$($fixed)*],
14            $first,
15            [$($out,)* $($f)::+!($($fixed)* $($prefix,)* $first, $head $(, $tail)*),];
16            [$($prefix,)* $head,];
17            $($tail),*
18        )
19    };
20    (@collect $($f:tt)::+, [$($fixed:tt)*], $first:expr,
21        [$($out:expr,)*]; [$($prefix:expr,)*];
22    ) => {
23        [$($out,)* $($f)::+!($($fixed)* $($prefix,)* $first)]
24    };
25}
26
27#[macro_export]
28macro_rules! count_tts {
29    () => {
30        0usize
31    };
32    ($head:tt $(, $tail:tt)*) => {
33        1usize + $crate::count_tts!($($tail),*)
34    };
35}
36
37#[macro_export]
38macro_rules! build_star_join_index_func {
39    ($join_func:ident, $prefix_cursor:ident, $trace_cursors:ident, $($vals:expr),+ $(,)?) => {{
40        use $crate::{dynamic::{Erase, DowncastTrait}, trace::Cursor};
41
42        let mut ok: Box<DynData> = Box::<OK>::default().erase_box();
43        let mut ov: Box<DynData> = Box::<OV>::default().erase_box();
44        let join_func = $join_func.clone();
45
46        Box::new(move |$prefix_cursor, $trace_cursors, cb| {
47            for (k, v) in join_func(
48                unsafe { $prefix_cursor.key().downcast() },
49                $($vals),+
50            ) {
51                *unsafe { ok.downcast_mut() } = k;
52                *unsafe { ov.downcast_mut() } = v;
53                cb(ok.as_mut(), ov.as_mut());
54            }
55        })
56    }};
57}
58
59#[macro_export]
60macro_rules! build_star_join_flatmap_func {
61    ($join_func:ident, $prefix_cursor:ident, $trace_cursors:ident, $($vals:expr),+ $(,)?) => {{
62        use $crate::dynamic::{Erase, DowncastTrait};
63        use $crate::trace::Cursor;
64
65        let mut ov: Box<DynData> = Box::<OV>::default().erase_box();
66        let join_func = $join_func.clone();
67
68        Box::new(move |$prefix_cursor, $trace_cursors, cb| {
69            for v in join_func(
70                unsafe { $prefix_cursor.key().downcast() },
71                $($vals),+
72            ) {
73                *unsafe { ov.downcast_mut() } = v;
74                cb(ov.as_mut(), ().erase_mut());
75            }
76        })
77    }};
78}
79
80#[macro_export]
81macro_rules! build_star_join_func {
82    ($join_func:ident, $prefix_cursor:ident, $trace_cursors:ident, $($vals:expr),+ $(,)?) => {{
83        use $crate::dynamic::{Erase, DowncastTrait};
84        use $crate::trace::Cursor;
85
86        let mut ov: Box<DynData> = Box::<OV>::default().erase_box();
87        let join_func = $join_func.clone();
88
89        Box::new(move |$prefix_cursor, $trace_cursors, cb| {
90            let v = join_func(
91                unsafe { $prefix_cursor.key().downcast() },
92                $($vals),+
93            );
94            *unsafe { ov.downcast_mut() } = v;
95            cb(ov.as_mut(), ().erase_mut());
96        })
97    }};
98}
99
100#[macro_export]
101macro_rules! star_join_index_funcs {
102    (
103        $join_func:ident,
104        $prefix_cursor:ident,
105        $trace_cursors:ident,
106        [$($trace_idx:tt),+ $(,)?]
107    ) => {
108        $crate::generate_all_join_orderings!(
109            $crate::build_star_join_index_func,
110            [$join_func, $prefix_cursor, $trace_cursors,],
111            unsafe { $prefix_cursor.val().downcast() },
112            $(unsafe { $trace_cursors[$trace_idx].val().downcast() }),+
113        )
114    };
115}
116
117#[macro_export]
118macro_rules! star_join_flatmap_funcs {
119    (
120        $join_func:ident,
121        $prefix_cursor:ident,
122        $trace_cursors:ident,
123        [$($trace_idx:tt),+ $(,)?]
124    ) => {
125        $crate::generate_all_join_orderings!(
126            $crate::build_star_join_flatmap_func,
127            [$join_func, $prefix_cursor, $trace_cursors,],
128            unsafe { $prefix_cursor.val().downcast() },
129            $(unsafe { $trace_cursors[$trace_idx].val().downcast() }),+
130        )
131    };
132}
133
134#[macro_export]
135macro_rules! star_join_funcs {
136    (
137        $join_func:ident,
138        $prefix_cursor:ident,
139        $trace_cursors:ident,
140        [$($trace_idx:tt),+ $(,)?]
141    ) => {
142        $crate::generate_all_join_orderings!(
143            $crate::build_star_join_func,
144            [$join_func, $prefix_cursor, $trace_cursors,],
145            unsafe { $prefix_cursor.val().downcast() },
146            $(unsafe { $trace_cursors[$trace_idx].val().downcast() }),+
147        )
148    };
149}
150
151#[macro_export]
152macro_rules! inner_star_join_index_body {
153    (
154        $stream1:expr,
155        [$($stream:expr),+ $(,)?],
156        $join_func:ident,
157        [$($val_ty:ty),+ $(,)?],
158        [$($trace_idx:tt),+ $(,)?],
159        $k_ty:ty,
160        $ok_ty:ty,
161        $ov_ty:ty
162    ) => {{
163        use $crate::{
164            NestedCircuit,
165            dynamic::DynData,
166            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
167            trace::BatchReaderFactories,
168        };
169
170        let join_funcs: [StarJoinFunc<NestedCircuit, MonoIndexedZSet, DynData, DynData>;
171            $crate::count_tts!($($trace_idx),+) + 1
172        ] = $crate::star_join_index_funcs!(
173            $join_func,
174            prefix_cursor,
175            trace_cursors,
176            [$($trace_idx),+]
177        );
178
179        let mut join_factories = StarJoinFactories::new::<$ok_ty, $ov_ty>();
180        $(
181            join_factories.add_input_factories(
182                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
183                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
184            );
185        )+
186
187        $stream1
188            .inner()
189            .dyn_inner_star_join_index_mono(
190                &join_factories,
191                &[$($stream.inner()),+],
192                &join_funcs,
193            )
194            .typed()
195    }};
196}
197
198#[macro_export]
199macro_rules! inner_star_join_index_root_body {
200    (
201        $stream1:expr,
202        [$($stream:expr),+ $(,)?],
203        $join_func:ident,
204        [$($val_ty:ty),+ $(,)?],
205        [$($trace_idx:tt),+ $(,)?],
206        $k_ty:ty,
207        $ok_ty:ty,
208        $ov_ty:ty
209    ) => {{
210        use $crate::{
211            RootCircuit,
212            dynamic::DynData,
213            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
214            trace::BatchReaderFactories,
215        };
216
217        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynData>;
218            $crate::count_tts!($($trace_idx),+) + 1
219        ] = $crate::star_join_index_funcs!(
220            $join_func,
221            prefix_cursor,
222            trace_cursors,
223            [$($trace_idx),+]
224        );
225
226        let mut join_factories = StarJoinFactories::new::<$ok_ty, $ov_ty>();
227        $(
228            join_factories.add_input_factories(
229                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
230                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
231            );
232        )+
233
234        $stream1
235            .inner()
236            .dyn_inner_star_join_index_mono(
237                &join_factories,
238                &[$($stream.inner()),+],
239                &join_funcs,
240            )
241            .typed()
242    }};
243}
244
245#[macro_export]
246macro_rules! star_join_index_body {
247    (
248        $stream1:expr,
249        [$(($stream:expr, $saturate:expr)),+ $(,)?],
250        $join_func:ident,
251        [$($val_ty:ty),+ $(,)?],
252        [$($trace_idx:tt),+ $(,)?],
253        $k_ty:ty,
254        $ok_ty:ty,
255        $ov_ty:ty
256    ) => {{
257        use $crate::{
258            RootCircuit,
259            dynamic::DynData,
260            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
261            trace::BatchReaderFactories,
262        };
263
264        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynData>;
265            $crate::count_tts!($($trace_idx),+) + 1
266        ] = $crate::star_join_index_funcs!(
267            $join_func,
268            prefix_cursor,
269            trace_cursors,
270            [$($trace_idx),+]
271        );
272
273        let mut join_factories = StarJoinFactories::new::<$ok_ty, $ov_ty>();
274        $(
275            join_factories.add_input_factories(
276                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
277                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
278            );
279        )+
280
281        $stream1
282            .inner()
283            .dyn_star_join_index_mono(
284                &join_factories,
285                &[$(($stream.inner(), $saturate)),+],
286                &join_funcs,
287            )
288            .typed()
289    }};
290}
291
292#[macro_export]
293macro_rules! inner_star_join_flatmap_body {
294    (
295        $stream1:expr,
296        [$($stream:expr),+ $(,)?],
297        $join_func:ident,
298        [$($val_ty:ty),+ $(,)?],
299        [$($trace_idx:tt),+ $(,)?],
300        $k_ty:ty,
301        $ov_ty:ty
302    ) => {{
303        use $crate::{
304            NestedCircuit,
305            dynamic::{DynData, DynUnit},
306            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
307            trace::BatchReaderFactories,
308        };
309
310        let join_funcs: [StarJoinFunc<NestedCircuit, MonoIndexedZSet, DynData, DynUnit>;
311            $crate::count_tts!($($trace_idx),+) + 1
312        ] = $crate::star_join_flatmap_funcs!(
313            $join_func,
314            prefix_cursor,
315            trace_cursors,
316            [$($trace_idx),+]
317        );
318
319        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
320        $(
321            join_factories.add_input_factories(
322                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
323                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
324            );
325        )+
326
327        $stream1
328            .inner()
329            .dyn_inner_star_join_mono(
330                &join_factories,
331                &[$($stream.inner()),+],
332                &join_funcs,
333            )
334            .typed()
335    }};
336}
337
338#[macro_export]
339macro_rules! inner_star_join_flatmap_root_body {
340    (
341        $stream1:expr,
342        [$($stream:expr),+ $(,)?],
343        $join_func:ident,
344        [$($val_ty:ty),+ $(,)?],
345        [$($trace_idx:tt),+ $(,)?],
346        $k_ty:ty,
347        $ov_ty:ty
348    ) => {{
349        use $crate::{
350            RootCircuit,
351            dynamic::{DynData, DynUnit},
352            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
353            trace::BatchReaderFactories,
354        };
355
356        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynUnit>;
357            $crate::count_tts!($($trace_idx),+) + 1
358        ] = $crate::star_join_flatmap_funcs!(
359            $join_func,
360            prefix_cursor,
361            trace_cursors,
362            [$($trace_idx),+]
363        );
364
365        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
366        $(
367            join_factories.add_input_factories(
368                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
369                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
370            );
371        )+
372
373        $stream1
374            .inner()
375            .dyn_inner_star_join_mono(
376                &join_factories,
377                &[$($stream.inner()),+],
378                &join_funcs,
379            )
380            .typed()
381    }};
382}
383
384#[macro_export]
385macro_rules! star_join_flatmap_body {
386    (
387        $stream1:expr,
388        [$(($stream:expr, $saturate:expr)),+ $(,)?],
389        $join_func:ident,
390        [$($val_ty:ty),+ $(,)?],
391        [$($trace_idx:tt),+ $(,)?],
392        $k_ty:ty,
393        $ov_ty:ty
394    ) => {{
395        use $crate::{
396            RootCircuit,
397            dynamic::{DynData, DynUnit},
398            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
399            trace::BatchReaderFactories,
400        };
401
402        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynUnit>;
403            $crate::count_tts!($($trace_idx),+) + 1
404        ] = $crate::star_join_flatmap_funcs!(
405            $join_func,
406            prefix_cursor,
407            trace_cursors,
408            [$($trace_idx),+]
409        );
410
411        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
412        $(
413            join_factories.add_input_factories(
414                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
415                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
416            );
417        )+
418
419        $stream1
420            .inner()
421            .dyn_star_join_mono(
422                &join_factories,
423                &[$(($stream.inner(), $saturate)),+],
424                &join_funcs,
425            )
426            .typed()
427    }};
428}
429
430#[macro_export]
431macro_rules! inner_star_join_root_body {
432    (
433        $stream1:expr,
434        [$($stream:expr),+ $(,)?],
435        $join_func:ident,
436        [$($val_ty:ty),+ $(,)?],
437        [$($trace_idx:tt),+ $(,)?],
438        $k_ty:ty,
439        $ov_ty:ty
440    ) => {{
441        use $crate::{
442            RootCircuit,
443            dynamic::{DynData, DynUnit},
444            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
445            trace::BatchReaderFactories,
446        };
447
448        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynUnit>;
449            $crate::count_tts!($($trace_idx),+) + 1
450        ] = $crate::star_join_funcs!(
451            $join_func,
452            prefix_cursor,
453            trace_cursors,
454            [$($trace_idx),+]
455        );
456
457        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
458        $(
459            join_factories.add_input_factories(
460                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
461                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
462            );
463        )+
464
465        $stream1
466            .inner()
467            .dyn_inner_star_join_mono(
468                &join_factories,
469                &[$($stream.inner()),+],
470                &join_funcs,
471            )
472            .typed()
473    }};
474}
475
476#[macro_export]
477macro_rules! inner_star_join_body {
478    (
479        $stream1:expr,
480        [$($stream:expr),+ $(,)?],
481        $join_func:ident,
482        [$($val_ty:ty),+ $(,)?],
483        [$($trace_idx:tt),+ $(,)?],
484        $k_ty:ty,
485        $ov_ty:ty
486    ) => {{
487        use $crate::{
488            NestedCircuit,
489            dynamic::{DynData, DynUnit},
490            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
491            trace::BatchReaderFactories,
492        };
493
494
495        let join_funcs: [StarJoinFunc<NestedCircuit, MonoIndexedZSet, DynData, DynUnit>;
496            $crate::count_tts!($($trace_idx),+) + 1
497        ] = $crate::star_join_funcs!(
498            $join_func,
499            prefix_cursor,
500            trace_cursors,
501            [$($trace_idx),+]
502        );
503
504        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
505        $(
506            join_factories.add_input_factories(
507                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
508                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
509            );
510        )+
511
512        $stream1
513            .inner()
514            .dyn_inner_star_join_mono(
515                &join_factories,
516                &[$($stream.inner()),+],
517                &join_funcs,
518            )
519            .typed()
520    }};
521}
522
523#[macro_export]
524macro_rules! star_join_body {
525    (
526        $stream1:expr,
527        [$(($stream:expr, $saturate:expr)),+ $(,)?],
528        $join_func:ident,
529        [$($val_ty:ty),+ $(,)?],
530        [$($trace_idx:tt),+ $(,)?],
531        $k_ty:ty,
532        $ov_ty:ty
533    ) => {{
534        use $crate::{
535            RootCircuit,
536            dynamic::{DynData, DynUnit},
537            operator::dynamic::{MonoIndexedZSet, multijoin::{StarJoinFunc, StarJoinFactories}},
538            trace::BatchReaderFactories,
539        };
540
541        let join_funcs: [StarJoinFunc<RootCircuit, MonoIndexedZSet, DynData, DynUnit>;
542            $crate::count_tts!($($trace_idx),+) + 1
543        ] = $crate::star_join_funcs!(
544            $join_func,
545            prefix_cursor,
546            trace_cursors,
547            [$($trace_idx),+]
548        );
549
550        let mut join_factories = StarJoinFactories::new::<$ov_ty, ()>();
551        $(
552            join_factories.add_input_factories(
553                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
554                BatchReaderFactories::new::<$k_ty, $val_ty, $crate::ZWeight>(),
555            );
556        )+
557
558        $stream1
559            .inner()
560            .dyn_star_join_mono(
561                &join_factories,
562                &[$(($stream.inner(), $saturate)),+],
563                &join_funcs,
564            )
565            .typed()
566    }};
567}
568
569/// Generate a `inner_star_joinN`/`inner_star_joinN_nested` function.
570///
571/// Call this macro for every number N in order to generate an N-way inner star join
572/// operator for both RootCircuit and NestedCircuit.
573///
574/// The operator computes an incremental join of multiple streams on the same key using a
575/// user-provided join function that returns exactly one output value for each input tuple.
576///
577/// Example generated function signature:
578///
579/// ```text
580///     pub fn inner_star_join4<K, V1, V2, V3, V4, OV>(
581///         stream1: &Stream<crate::RootCircuit, OrdIndexedZSet<K, V1>>,
582///         stream2: &Stream<crate::RootCircuit, OrdIndexedZSet<K, V2>>,
583///         stream3: &Stream<crate::RootCircuit, OrdIndexedZSet<K, V3>>,
584///         stream4: &Stream<crate::RootCircuit, OrdIndexedZSet<K, V4>>,
585///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> OV + Clone + 'static,
586///     ) -> Stream<RootCircuit, OrdZSet<OV>>
587///     where
588///         K: DBData,
589///         V1: DBData,
590///         V2: DBData,
591///         V3: DBData,
592///         V4: DBData,
593///         OV: DBData;
594///
595///     pub fn inner_star_join4_nested<K, V1, V2, V3, V4, OV>(
596///         stream1: &Stream<crate::NestedCircuit, OrdIndexedZSet<K, V1>>,
597///         stream2: &Stream<crate::NestedCircuit, OrdIndexedZSet<K, V2>>,
598///         stream3: &Stream<crate::NestedCircuit, OrdIndexedZSet<K, V3>>,
599///         stream4: &Stream<crate::NestedCircuit, OrdIndexedZSet<K, V4>>,
600///     ) -> Stream<NestedCircuit, OrdZSet<OV>>
601///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> OV + Clone + 'static,
602///     where
603///         K: DBData,
604///         V1: DBData,
605///         V2: DBData,
606///         V3: DBData,
607///         V4: DBData,
608///         OV: DBData;
609/// ```
610#[macro_export]
611macro_rules! define_inner_star_join {
612    ($n:literal) => {
613        seq_macro::seq!(I in 2..=$n {
614            paste::paste! {
615                #[allow(unused)]
616                pub fn [<inner_star_join $n>]<K, V1, #(V~I,)* OV>(
617                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
618                    #(
619                        stream~I: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
620                    )*
621                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> OV + Clone + 'static,
622                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdZSet<OV>>
623                where
624                    K: $crate::DBData,
625                    V1: $crate::DBData,
626                    #(V~I: $crate::DBData,)*
627                    OV: $crate::DBData,
628                {
629                    $crate::inner_star_join_root_body!(
630                        stream1,
631                        [#(stream~I,)*],
632                        join_func,
633                        [V1, #(V~I,)*],
634                        [#((I - 2),)*],
635                        K,
636                        OV
637                    )
638                }
639
640                #[allow(unused)]
641                pub fn [<inner_star_join $n _nested>]<K, V1, #(V~I,)* OV>(
642                    stream1: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V1>>,
643                    #(
644                        stream~I: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V~I>>,
645                    )*
646                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> OV + Clone + 'static,
647                ) -> $crate::Stream<$crate::NestedCircuit, $crate::OrdZSet<OV>>
648                where
649                    K: $crate::DBData,
650                    V1: $crate::DBData,
651                    #(V~I: $crate::DBData,)*
652                    OV: $crate::DBData,
653                {
654                    $crate::inner_star_join_body!(
655                        stream1,
656                        [#(stream~I,)*],
657                        join_func,
658                        [V1, #(V~I,)*],
659                        [#((I - 2),)*],
660                        K,
661                        OV
662                    )
663                }
664            }
665        });
666    };
667}
668
669/// Generate a `inner_star_join_indexN`/`inner_star_join_indexN_nested` function.
670///
671/// Call this macro for every number N in order to generate an N-way inner star join
672/// index operator for both RootCircuit and NestedCircuit.
673///
674/// The operator computes an incremental join of multiple streams on the same key using a
675/// user-provided join function that can return 0 or more output key-value pairs for each input tuple.
676///
677/// Example generated function signature:
678///
679/// ```text
680///     pub fn inner_star_join_index4<K, V1, V2, V3, V4, OK, OV, It>(
681///         stream1: &Stream<RootCircuit, OrdIndexedZSet<K, V1>>,
682///         stream2: &Stream<RootCircuit, OrdIndexedZSet<K, V2>>,
683///         stream3: &Stream<RootCircuit, OrdIndexedZSet<K, V3>>,
684///         stream4: &Stream<RootCircuit, OrdIndexedZSet<K, V4>>,
685///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
686///     ) -> Stream<RootCircuit, OrdIndexedZSet<OK, OV>>
687///     where
688///         K: DBData,
689///         V1: DBData,
690///         V2: DBData,
691///         V3: DBData,
692///         V4: DBData,
693///         OK: DBData,
694///         OV: DBData,
695///         It: IntoIterator<Item = (OK, OV)> + 'static;
696///
697///     pub fn inner_star_join_index4_nested<K, V1, V2, V3, V4, OK, OV, It>(
698///         stream1: &Stream<NestedCircuit, OrdIndexedZSet<K, V1>>,
699///         stream2: &Stream<NestedCircuit, OrdIndexedZSet<K, V2>>,
700///         stream3: &Stream<NestedCircuit, OrdIndexedZSet<K, V3>>,
701///         stream4: &Stream<NestedCircuit, OrdIndexedZSet<K, V4>>,
702///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
703///     ) -> Stream<NestedCircuit, OrdIndexedZSet<OK, OV>>
704///     where
705///         K: DBData,
706///         V1: DBData,
707///         V2: DBData,
708///         V3: DBData,
709///         V4: DBData,
710///         OK: DBData,
711///         OV: DBData,
712///         It: IntoIterator<Item = (OK, OV)> + 'static;
713/// ```
714#[macro_export]
715macro_rules! define_inner_star_join_index {
716    ($n:literal) => {
717        seq_macro::seq!(I in 2..=$n {
718            paste::paste! {
719                #[allow(unused)]
720                pub fn [<inner_star_join_index $n>]<K, V1, #(V~I,)* OK, OV, It>(
721                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
722                    #(
723                        stream~I: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
724                    )*
725                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
726                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<OK, OV>>
727                where
728                    K: $crate::DBData,
729                    V1: $crate::DBData,
730                    #(V~I: $crate::DBData,)*
731                    OK: $crate::DBData,
732                    OV: $crate::DBData,
733                    It: IntoIterator<Item = (OK, OV)> + 'static,
734                {
735                    $crate::inner_star_join_index_root_body!(
736                        stream1,
737                        [#(stream~I,)*],
738                        join_func,
739                        [V1, #(V~I,)*],
740                        [#((I - 2),)*],
741                        K,
742                        OK,
743                        OV
744                    )
745                }
746
747                #[allow(unused)]
748                pub fn [<inner_star_join_index $n _nested>]<K, V1, #(V~I,)* OK, OV, It>(
749                    stream1: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V1>>,
750                    #(
751                        stream~I: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V~I>>,
752                    )*
753                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
754                ) -> $crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<OK, OV>>
755                where
756                    K: $crate::DBData,
757                    V1: $crate::DBData,
758                    #(V~I: $crate::DBData,)*
759                    OK: $crate::DBData,
760                    OV: $crate::DBData,
761                    It: IntoIterator<Item = (OK, OV)> + 'static,
762                {
763                    $crate::inner_star_join_index_body!(
764                        stream1,
765                        [#(stream~I,)*],
766                        join_func,
767                        [V1, #(V~I,)*],
768                        [#((I - 2),)*],
769                        K,
770                        OK,
771                        OV
772                    )
773                }
774            }
775
776        });
777    };
778}
779
780/// Generate a `inner_star_join_flatmapN`/`inner_star_join_flatmapN_nested` function.
781///
782/// Call this macro for every number N in order to generate an N-way inner star join
783/// flatmap operator for both RootCircuit and NestedCircuit.
784///
785/// The operator computes an incremental join of multiple streams on the same key using a
786/// user-provided join function that can return 0 or more output values for each input tuple.
787///
788/// Example generated function signature:
789///
790/// ```text
791///     pub fn inner_star_join_flatmap4<K, V1, V2, V3, V4, OV, It>(
792///         stream1: &Stream<RootCircuit, OrdIndexedZSet<K, V1>>,
793///         stream2: &Stream<RootCircuit, OrdIndexedZSet<K, V2>>,
794///         stream3: &Stream<RootCircuit, OrdIndexedZSet<K, V3>>,
795///         stream4: &Stream<RootCircuit, OrdIndexedZSet<K, V4>>,
796///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
797///     ) -> Stream<RootCircuit, OrdZSet<OV>>
798///     where
799///         K: DBData,
800///         V1: DBData,
801///         V2: DBData,
802///         V3: DBData,
803///         V4: DBData,
804///         OV: DBData,
805///         It: IntoIterator<Item = OV> + 'static;
806///
807///     pub fn inner_star_join_flatmap4_nested<K, V1, V2, V3, V4, OV, It>(
808///         stream1: &Stream<NestedCircuit, OrdIndexedZSet<K, V1>>,
809///         stream2: &Stream<NestedCircuit, OrdIndexedZSet<K, V2>>,
810///         stream3: &Stream<NestedCircuit, OrdIndexedZSet<K, V3>>,
811///         stream4: &Stream<NestedCircuit, OrdIndexedZSet<K, V4>>,
812///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
813///     ) -> Stream<NestedCircuit, OrdZSet<OV>>
814///     where
815///         K: DBData,
816///         V1: DBData,
817///         V2: DBData,
818///         V3: DBData,
819///         V4: DBData,
820///         OV: DBData,
821///         It: IntoIterator<Item = OV> + 'static;
822/// ```
823#[macro_export]
824macro_rules! define_inner_star_join_flatmap {
825    ($n:literal) => {
826        seq_macro::seq!(I in 2..=$n {
827            paste::paste! {
828                #[allow(unused)]
829                pub fn [<inner_star_join_flatmap $n>]<K, V1, #(V~I,)* OV, It>(
830                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
831                    #(
832                        stream~I: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
833                    )*
834                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
835                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdZSet<OV>>
836                where
837                    K: $crate::DBData,
838                    V1: $crate::DBData,
839                    #(V~I: $crate::DBData,)*
840                    OV: $crate::DBData,
841                    It: IntoIterator<Item = OV> + 'static,
842                {
843                    $crate::inner_star_join_flatmap_root_body!(
844                        stream1,
845                        [#(stream~I,)*],
846                        join_func,
847                        [V1, #(V~I,)*],
848                        [#((I - 2),)*],
849                        K,
850                        OV
851                    )
852                }
853
854                #[allow(unused)]
855                pub fn [<inner_star_join_flatmap $n _nested>]<K, V1, #(V~I,)* OV, It>(
856                    stream1: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V1>>,
857                    #(
858                        stream~I: &$crate::Stream<$crate::NestedCircuit, $crate::OrdIndexedZSet<K, V~I>>,
859                    )*
860                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
861                ) -> $crate::Stream<$crate::NestedCircuit, $crate::OrdZSet<OV>>
862                where
863                    K: $crate::DBData,
864                    V1: $crate::DBData,
865                    #(V~I: $crate::DBData,)*
866                    OV: $crate::DBData,
867                    It: IntoIterator<Item = OV> + 'static,
868                {
869                    $crate::inner_star_join_flatmap_body!(
870                        stream1,
871                        [#(stream~I,)*],
872                        join_func,
873                        [V1, #(V~I,)*],
874                        [#((I - 2),)*],
875                        K,
876                        OV
877                    )
878                }
879            }
880        });
881    };
882}
883
884/// Generate a `star_joinN` function.
885///
886/// Call this macro for every number N in order to generate an N-way star-join operator.
887/// This operator is only available for RootCircuit. See `define_inner_star_join` for a
888/// star join operator that works for NestedCircuit.
889///
890/// The operator computes an incremental join of multiple streams on the same key using a
891/// user-provided join function that must return exactly one value per input tuple.  The
892/// `saturate` flag for each input stream (except the first one) controls whether to compute
893/// an outer or inner join for that stream.
894///
895/// Example generated function signature:
896///
897/// ```text
898///     pub fn star_join4<K, V1, V2, V3, V4, OV>(
899///         stream1: &Stream<RootCircuit, OrdIndexedZSet<K, V1>>,
900///         (stream2, saturate2): (&Stream<RootCircuit, OrdIndexedZSet<K, V2>>, bool),
901///         (stream3, saturate3): (&Stream<RootCircuit, OrdIndexedZSet<K, V3>>, bool),
902///         (stream4, saturate4): (&Stream<RootCircuit, OrdIndexedZSet<K, V4>>, bool),
903///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> OV + Clone + 'static,
904///     ) -> Stream<RootCircuit, OrdZSet<OV>>
905///     where
906///         K: DBData,
907///         V1: DBData,
908///         V2: DBData,
909///         V3: DBData,
910///         V4: DBData,
911///         OV: DBData;
912/// ```
913#[macro_export]
914macro_rules! define_star_join {
915    ($n:literal) => {
916        seq_macro::seq!(I in 2..=$n {
917            paste::paste! {
918                pub fn [<star_join $n>]<K, V1, #(V~I,)* OV>(
919                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
920                    #(
921                        (stream~I, saturate~I): (
922                            &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
923                            bool
924                        ),
925                    )*
926                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> OV + Clone + 'static,
927                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdZSet<OV>>
928                where
929                    K: $crate::DBData,
930                    V1: $crate::DBData,
931                    #(V~I: $crate::DBData,)*
932                    OV: $crate::DBData,
933                {
934                    $crate::star_join_body!(
935                        stream1,
936                        [#((stream~I, saturate~I),)*],
937                        join_func,
938                        [V1, #(V~I,)*],
939                        [#((I - 2),)*],
940                        K,
941                        OV
942                    )
943                }
944            }
945
946        });
947    };
948}
949
950/// Generate a `star_join_indexN` function.
951///
952/// Call this macro for every number N in order to generate an N-way star-join-index operator.
953/// This operator is only available for RootCircuit. See `define_inner_star_join_index` for a
954/// star-join-index operator that works for NestedCircuit.
955///
956/// The operator computes an incremental join of multiple streams on the same key using a
957/// user-provided join function that can return 0 or more output key-value pairs for each input tuple.
958/// The `saturate` flag for each input stream (except the first one) controls whether to compute an outer or inner join
959/// for that stream.
960///
961/// Example generated function signature:
962///
963/// ```text
964///     pub fn star_join_index4<K, V1, V2, V3, V4, OK, OV, It>(
965///         stream1: &Stream<RootCircuit, OrdIndexedZSet<K, V1>>,
966///         (stream2, saturate2): (
967///             &Stream<RootCircuit, OrdIndexedZSet<K, V2>>,
968///             bool,
969///         ),
970///         (stream3, saturate3): (
971///             &Stream<RootCircuit, OrdIndexedZSet<K, V3>>,
972///             bool,
973///         ),
974///         (stream4, saturate4): (
975///             &Stream<RootCircuit, OrdIndexedZSet<K, V4>>,
976///             bool,
977///         ),
978///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
979///     ) -> Stream<RootCircuit, OrdIndexedZSet<OK, OV>>
980///     where
981///         K: DBData,
982///         V1: DBData,
983///         V2: DBData,
984///         V3: DBData,
985///         V4: DBData,
986///         OK: DBData,
987///         OV: DBData,
988///         It: IntoIterator<Item = (OK, OV)> + 'static;
989/// ```
990#[macro_export]
991macro_rules! define_star_join_index {
992    ($n:literal) => {
993        seq_macro::seq!(I in 2..=$n {
994            paste::paste! {
995                pub fn [<star_join_index $n>]<K, V1, #(V~I,)* OK, OV, It>(
996                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
997                    #(
998                        (stream~I, saturate~I): (
999                            &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
1000                            bool
1001                        ),
1002                    )*
1003                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
1004                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<OK, OV>>
1005                where
1006                    K: $crate::DBData,
1007                    V1: $crate::DBData,
1008                    #(V~I: $crate::DBData,)*
1009                    OK: $crate::DBData,
1010                    OV: $crate::DBData,
1011                    It: IntoIterator<Item = (OK, OV)> + 'static,
1012                {
1013                    $crate::star_join_index_body!(
1014                        stream1,
1015                        [#((stream~I, saturate~I),)*],
1016                        join_func,
1017                        [V1, #(V~I,)*],
1018                        [#((I - 2),)*],
1019                        K,
1020                        OK,
1021                        OV
1022                    )
1023                }
1024            }
1025
1026        });
1027    };
1028}
1029
1030/// Generate a `star_join_flatmapN` function.
1031///
1032/// Call this macro for every number N in order to generate an N-way star join flatmap operator.
1033/// This operator is only available for RootCircuit. See `define_inner_star_join_flatmap` for a
1034/// star join flatmap operator that works for NestedCircuit.
1035///
1036/// The operator computes an incremental join of multiple streams on the same key using a
1037/// user-provided join function that can return 0 or more output values for each input tuple.
1038/// The `saturate` flag for each input stream (except the first one) controls whether to compute
1039/// an outer or inner join for that stream.
1040///
1041/// Example generated function signature:
1042///
1043/// ```text
1044///     pub fn star_join_flatmap4<K, V1, V2, V3, V4, OV, It>(
1045///         stream1: &Stream<RootCircuit, OrdIndexedZSet<K, V1>>,
1046///         (stream2, saturate2): (
1047///             &Stream<RootCircuit, OrdIndexedZSet<K, V2>>,
1048///             bool,
1049///         ),
1050///         (stream3, saturate3): (
1051///             &Stream<RootCircuit, OrdIndexedZSet<K, V3>>,
1052///             bool,
1053///         ),
1054///         (stream4, saturate4): (
1055///             &Stream<RootCircuit, OrdIndexedZSet<K, V4>>,
1056///             bool,
1057///         ),
1058///         join_func: impl Fn(&K, &V1, &V2, &V3, &V4) -> It + Clone + 'static,
1059///     ) -> Stream<RootCircuit, OrdZSet<OV>>;
1060/// ```
1061#[macro_export]
1062macro_rules! define_star_join_flatmap {
1063    ($n:literal) => {
1064        seq_macro::seq!(I in 2..=$n {
1065            paste::paste! {
1066                pub fn [<star_join_flatmap $n>]<K, V1, #(V~I,)* OV, It>(
1067                    stream1: &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V1>>,
1068                    #(
1069                        (stream~I, saturate~I): (
1070                            &$crate::Stream<$crate::RootCircuit, $crate::OrdIndexedZSet<K, V~I>>,
1071                            bool
1072                        ),
1073                    )*
1074                    join_func: impl Fn(&K, &V1, #(&V~I,)*) -> It + Clone + 'static,
1075                ) -> $crate::Stream<$crate::RootCircuit, $crate::OrdZSet<OV>>
1076                where
1077                    K: $crate::DBData,
1078                    V1: $crate::DBData,
1079                    #(V~I: $crate::DBData,)*
1080                    OV: $crate::DBData,
1081                    It: IntoIterator<Item = OV> + 'static,
1082                {
1083                    $crate::star_join_flatmap_body!(
1084                        stream1,
1085                        [#((stream~I, saturate~I),)*],
1086                        join_func,
1087                        [V1, #(V~I,)*],
1088                        [#((I - 2),)*],
1089                        K,
1090                        OV
1091                    )
1092                }
1093            }
1094        });
1095    };
1096}
dbsp/operator/star_join_macros.rs

dbsp/operator/
star_join_macros.rs
