1#![allow(dead_code, missing_docs)]
13
14use columnar::{Container, Ref};
15use differential_dataflow::operators::arrange::Arranged;
16use differential_dataflow::operators::arrange::TraceAgent;
17use differential_dataflow::trace::implementations::chunker::ColumnationChunker;
18use differential_dataflow::trace::implementations::merge_batcher::{ColMerger, MergeBatcher};
19use differential_dataflow::trace::wrappers::enter::TraceEnter;
20use differential_dataflow::trace::wrappers::frontier::TraceFrontier;
21use mz_repr::Diff;
22use mz_storage_types::errors::DataflowError;
23use timely::dataflow::ScopeParent;
24
25use crate::row_spine::RowValBuilder;
26use crate::typedefs::spines::{ColKeyBatcher, ColKeyBuilder, ColValBatcher, ColValBuilder};
27
28pub use crate::row_spine::{RowRowSpine, RowSpine, RowValBatcher, RowValSpine};
29pub use crate::typedefs::spines::{ColKeySpine, ColValSpine};
30
31pub(crate) mod spines {
32 use std::rc::Rc;
33
34 use differential_dataflow::containers::{Columnation, TimelyStack};
35 use differential_dataflow::trace::implementations::ord_neu::{
36 OrdKeyBatch, OrdKeyBuilder, OrdValBatch, OrdValBuilder,
37 };
38 use differential_dataflow::trace::implementations::spine_fueled::Spine;
39 use differential_dataflow::trace::implementations::{Layout, Update};
40 use differential_dataflow::trace::rc_blanket_impls::RcBuilder;
41
42 use crate::row_spine::OffsetOptimized;
43 use crate::typedefs::{KeyBatcher, KeyValBatcher};
44
45 pub type ColValSpine<K, V, T, R> = Spine<Rc<OrdValBatch<MzStack<((K, V), T, R)>>>>;
47 pub type ColValBatcher<K, V, T, R> = KeyValBatcher<K, V, T, R>;
48 pub type ColValBuilder<K, V, T, R> =
49 RcBuilder<OrdValBuilder<MzStack<((K, V), T, R)>, TimelyStack<((K, V), T, R)>>>;
50
51 pub type ColKeySpine<K, T, R> = Spine<Rc<OrdKeyBatch<MzStack<((K, ()), T, R)>>>>;
53 pub type ColKeyBatcher<K, T, R> = KeyBatcher<K, T, R>;
54 pub type ColKeyBuilder<K, T, R> =
55 RcBuilder<OrdKeyBuilder<MzStack<((K, ()), T, R)>, TimelyStack<((K, ()), T, R)>>>;
56
57 pub struct MzStack<U: Update> {
59 phantom: std::marker::PhantomData<U>,
60 }
61
62 impl<U: Update> Layout for MzStack<U>
63 where
64 U::Key: Columnation + 'static,
65 U::Val: Columnation + 'static,
66 U::Time: Columnation,
67 U::Diff: Columnation,
68 {
69 type KeyContainer = TimelyStack<U::Key>;
70 type ValContainer = TimelyStack<U::Val>;
71 type TimeContainer = TimelyStack<U::Time>;
72 type DiffContainer = TimelyStack<U::Diff>;
73 type OffsetContainer = OffsetOptimized;
74 }
75}
76
77pub type KeyValSpine<K, V, T, R> = ColValSpine<K, V, T, R>;
82pub type KeyValAgent<K, V, T, R> = TraceAgent<KeyValSpine<K, V, T, R>>;
83pub type KeyValEnter<K, V, T, R, TEnter> =
84 TraceEnter<TraceFrontier<KeyValAgent<K, V, T, R>>, TEnter>;
85
86pub type KeySpine<K, T, R> = ColKeySpine<K, T, R>;
88pub type KeyAgent<K, T, R> = TraceAgent<KeySpine<K, T, R>>;
89pub type KeyEnter<K, T, R, TEnter> = TraceEnter<TraceFrontier<KeyAgent<K, T, R>>, TEnter>;
90
91pub type RowValAgent<V, T, R> = TraceAgent<RowValSpine<V, T, R>>;
93pub type RowValArrangement<S, V> = Arranged<S, RowValAgent<V, <S as ScopeParent>::Timestamp, Diff>>;
94pub type RowValEnter<V, T, R, TEnter> = TraceEnter<TraceFrontier<RowValAgent<V, T, R>>, TEnter>;
95pub type RowRowAgent<T, R> = TraceAgent<RowRowSpine<T, R>>;
97pub type RowRowArrangement<S> = Arranged<S, RowRowAgent<<S as ScopeParent>::Timestamp, Diff>>;
98pub type RowRowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowRowAgent<T, R>>, TEnter>;
99pub type RowAgent<T, R> = TraceAgent<RowSpine<T, R>>;
101pub type RowArrangement<S> = Arranged<S, RowAgent<<S as ScopeParent>::Timestamp, Diff>>;
102pub type RowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowAgent<T, R>>, TEnter>;
103
104pub type ErrSpine<T, R> = ColKeySpine<DataflowError, T, R>;
106pub type ErrBatcher<T, R> = ColKeyBatcher<DataflowError, T, R>;
107pub type ErrBuilder<T, R> = ColKeyBuilder<DataflowError, T, R>;
108
109pub type ErrAgent<T, R> = TraceAgent<ErrSpine<T, R>>;
110pub type ErrEnter<T, TEnter> = TraceEnter<TraceFrontier<ErrAgent<T, Diff>>, TEnter>;
111
112pub type KeyErrSpine<K, T, R> = ColValSpine<K, DataflowError, T, R>;
113pub type KeyErrBatcher<K, T, R> = ColValBatcher<K, DataflowError, T, R>;
114pub type KeyErrBuilder<K, T, R> = ColValBuilder<K, DataflowError, T, R>;
115
116pub type RowErrSpine<T, R> = RowValSpine<DataflowError, T, R>;
117pub type RowErrBatcher<T, R> = RowValBatcher<DataflowError, T, R>;
118pub type RowErrBuilder<T, R> = RowValBuilder<DataflowError, T, R>;
119
120pub type KeyBatcher<K, T, D> = KeyValBatcher<K, (), T, D>;
122pub type KeyValBatcher<K, V, T, D> =
123 MergeBatcher<Vec<((K, V), T, D)>, ColumnationChunker<((K, V), T, D)>, ColMerger<(K, V), T, D>>;
124
125pub trait MzTimestamp:
127 MzData + timely::progress::Timestamp + differential_dataflow::lattice::Lattice
128{
129}
130
131impl<T> MzTimestamp for T
132where
133 T: MzData,
134 T: timely::progress::Timestamp,
135 T: differential_dataflow::lattice::Lattice,
136{
137}
138
139pub trait MzData:
142 differential_dataflow::containers::Columnation
143 + for<'a> columnar::Columnar<Container: Container<Ref<'a>: Copy + Ord> + Clone + Send>
144{
145}
146
147impl<T> MzData for T
148where
149 T: differential_dataflow::containers::Columnation,
150 T: for<'a> columnar::Columnar<Container: Clone + Send>,
151 for<'a> Ref<'a, T>: Copy + Ord,
152{
153}
154
155pub trait MzArrangeData: differential_dataflow::containers::Columnation {}
156impl<T> MzArrangeData for T where T: differential_dataflow::containers::Columnation {}