mz_compute/
typedefs.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Convience typedefs for differential types.
11
12#![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::merge_batcher::MergeBatcher;
18use differential_dataflow::trace::wrappers::enter::TraceEnter;
19use differential_dataflow::trace::wrappers::frontier::TraceFrontier;
20use mz_repr::{Diff, Row};
21use mz_row_spine::{RowRowBuilder, RowRowColPagedBuilder, RowValBuilder};
22use mz_timely_util::columnar::batcher::{Chunker, ColumnChunker};
23use mz_timely_util::columnar::chunk::{AccountedChunkBatcher, ComputeSpill, UnchunkBuilder};
24use mz_timely_util::columnar::{Col2ValBatcher, Col2ValColBatcher};
25use mz_timely_util::columnation::{ColInternalMerger, ColumnationChunker, ColumnationStack};
26use mz_timely_util::operator::ConsolidatingBatcher;
27
28use crate::render::errors::DataflowErrorSer;
29use crate::typedefs::spines::{ColKeyBatcher, ColKeyBuilder, ColValBatcher, ColValBuilder};
30
31pub use crate::typedefs::spines::{ColKeySpine, ColValSpine};
32pub use mz_row_spine::{RowRowSpine, RowSpine, RowValBatcher, RowValSpine};
33
34pub(crate) mod spines {
35    use columnation::Columnation;
36    use differential_dataflow::trace::implementations::ord_neu::{
37        OrdKeyBatch, OrdKeyBuilder, OrdValBatch, OrdValBuilder,
38    };
39    use differential_dataflow::trace::implementations::spine_fueled::Spine;
40    use differential_dataflow::trace::implementations::{Layout, Update};
41    use mz_timely_util::columnation::ColumnationStack;
42
43    use mz_row_spine::{ArcBatch, ArcBuilder, OffsetOptimized};
44
45    use crate::typedefs::{KeyBatcher, KeyValBatcher};
46
47    /// A spine for generic keys and values.
48    pub type ColValSpine<K, V, T, R> = Spine<ArcBatch<OrdValBatch<MzStack<((K, V), T, R)>>>>;
49    pub type ColValBatcher<K, V, T, R> = KeyValBatcher<K, V, T, R, ColValBuilder<K, V, T, R>>;
50    pub type ColValBuilder<K, V, T, R> =
51        ArcBuilder<OrdValBuilder<MzStack<((K, V), T, R)>, ColumnationStack<((K, V), T, R)>>>;
52
53    /// A spine for generic keys
54    pub type ColKeySpine<K, T, R> = Spine<ArcBatch<OrdKeyBatch<MzStack<((K, ()), T, R)>>>>;
55    pub type ColKeyBatcher<K, T, R> = KeyBatcher<K, T, R, ColKeyBuilder<K, T, R>>;
56    pub type ColKeyBuilder<K, T, R> =
57        ArcBuilder<OrdKeyBuilder<MzStack<((K, ()), T, R)>, ColumnationStack<((K, ()), T, R)>>>;
58
59    /// A layout based on chunked timely stacks
60    pub struct MzStack<U: Update> {
61        phantom: std::marker::PhantomData<U>,
62    }
63
64    impl<U: Update> Layout for MzStack<U>
65    where
66        U::Key: Columnation + 'static,
67        U::Val: Columnation + 'static,
68        U::Time: Columnation,
69        U::Diff: Columnation,
70    {
71        type KeyContainer = ColumnationStack<U::Key>;
72        type ValContainer = ColumnationStack<U::Val>;
73        type TimeContainer = ColumnationStack<U::Time>;
74        type DiffContainer = ColumnationStack<U::Diff>;
75        type OffsetContainer = OffsetOptimized;
76    }
77}
78
79// Spines are data structures that collect and maintain updates.
80// Agents are wrappers around spines that allow shared read access.
81
82// Fully generic spines and agents.
83pub type KeyValSpine<K, V, T, R> = ColValSpine<K, V, T, R>;
84pub type KeyValAgent<K, V, T, R> = TraceAgent<KeyValSpine<K, V, T, R>>;
85pub type KeyValEnter<K, V, T, R, TEnter> =
86    TraceEnter<TraceFrontier<KeyValAgent<K, V, T, R>>, TEnter>;
87
88// Fully generic key-only spines and agents
89pub type KeySpine<K, T, R> = ColKeySpine<K, T, R>;
90pub type KeyAgent<K, T, R> = TraceAgent<KeySpine<K, T, R>>;
91pub type KeyEnter<K, T, R, TEnter> = TraceEnter<TraceFrontier<KeyAgent<K, T, R>>, TEnter>;
92
93// Row specialized spines and agents.
94pub type RowValAgent<V, T, R> = TraceAgent<RowValSpine<V, T, R>>;
95pub type RowValArrangement<'scope, T, V> = Arranged<'scope, RowValAgent<V, T, Diff>>;
96pub type RowValEnter<V, T, R, TEnter> = TraceEnter<TraceFrontier<RowValAgent<V, T, R>>, TEnter>;
97// Row specialized spines and agents.
98pub type RowRowAgent<T, R> = TraceAgent<RowRowSpine<T, R>>;
99pub type RowRowArrangement<'scope, T> = Arranged<'scope, RowRowAgent<T, Diff>>;
100pub type RowRowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowRowAgent<T, R>>, TEnter>;
101// Row specialized spines and agents.
102pub type RowAgent<T, R, DC = ColumnationStack<R>> = TraceAgent<RowSpine<T, R, DC>>;
103pub type RowArrangement<'scope, T> = Arranged<'scope, RowAgent<T, Diff>>;
104pub type RowEnter<T, R, TEnter> = TraceEnter<TraceFrontier<RowAgent<T, R>>, TEnter>;
105
106// Error specialized spines and agents.
107pub type ErrSpine<T, R> = ColKeySpine<DataflowErrorSer, T, R>;
108pub type ErrBatcher<T, R> = ColKeyBatcher<DataflowErrorSer, T, R>;
109pub type ErrBuilder<T, R> = ColKeyBuilder<DataflowErrorSer, T, R>;
110
111pub type ErrAgent<T, R> = TraceAgent<ErrSpine<T, R>>;
112pub type ErrEnter<T, TEnter> = TraceEnter<TraceFrontier<ErrAgent<T, Diff>>, TEnter>;
113
114pub type KeyErrSpine<K, T, R> = ColValSpine<K, DataflowErrorSer, T, R>;
115pub type KeyErrBatcher<K, T, R> = ColValBatcher<K, DataflowErrorSer, T, R>;
116pub type KeyErrBuilder<K, T, R> = ColValBuilder<K, DataflowErrorSer, T, R>;
117
118pub type RowErrSpine<T, R> = RowValSpine<DataflowErrorSer, T, R>;
119pub type RowErrBatcher<T, R> = RowValBatcher<DataflowErrorSer, T, R>;
120pub type RowErrBuilder<T, R> = RowValBuilder<DataflowErrorSer, T, R>;
121
122// Batchers that chunk `Vec` input into columnation chains. `Se`, a spine builder, seals an
123// extracted chain into a batch. Consolidation that wants the chain itself uses
124// `mz_timely_util::operator::ConsolidatingBatcher`.
125pub type KeyBatcher<K, T, D, Se> = KeyValBatcher<K, (), T, D, Se>;
126pub type KeyValBatcher<K, V, T, D, Se> =
127    MergeBatcher<ColumnationChunker<((K, V), T, D)>, ColInternalMerger<(K, V), T, D>, Se>;
128
129// Row-to-row arrangement batchers, one per `ArrangementBatcher` flavor. `C` is the input
130// container the columnation chunker melds.
131pub type RowRowChunkedBatcher<T> = AccountedChunkBatcher<
132    (Row, Row),
133    T,
134    Diff,
135    ComputeSpill,
136    UnchunkBuilder<RowRowColPagedBuilder<T, Diff>, (Row, Row), T, Diff, ComputeSpill>,
137>;
138pub type RowRowColumnarBatcher<T> = Col2ValColBatcher<
139    Row,
140    Row,
141    T,
142    Diff,
143    ColumnChunker<((Row, Row), T, Diff)>,
144    RowRowColPagedBuilder<T, Diff>,
145>;
146pub type RowRowColumnationBatcher<C, T> =
147    Col2ValBatcher<Row, Row, T, Diff, Chunker<C>, RowRowBuilder<T, Diff>>;
148
149/// The batcher a consolidation wants: chunks `Vec` input, hands the chain back unsealed.
150pub type ConsolidateKeyValBatcher<K, V, T, D> =
151    ConsolidatingBatcher<ColumnationChunker<((K, V), T, D)>, ColInternalMerger<(K, V), T, D>>;
152/// [`ConsolidateKeyValBatcher`] with unit values.
153pub type ConsolidateBatcher<K, T, D> = ConsolidateKeyValBatcher<K, (), T, D>;
154/// [`ConsolidateKeyValBatcher`] over `Column` input rather than `Vec` input.
155pub type ConsolidateColumnBatcher<K, V, T, D> = ConsolidatingBatcher<
156    Chunker<ColumnationStack<((K, V), T, D)>>,
157    ColInternalMerger<(K, V), T, D>,
158>;
159
160/// Timestamp trait for rendering, constraint to support [`MzData`] and [timely::progress::Timestamp].
161pub trait MzTimestamp:
162    MzData + timely::progress::Timestamp + differential_dataflow::lattice::Lattice + std::hash::Hash
163{
164}
165
166impl<T> MzTimestamp for T
167where
168    T: MzData,
169    T: timely::progress::Timestamp,
170    T: differential_dataflow::lattice::Lattice + std::hash::Hash,
171{
172}
173
174/// Trait for data types that can be used in Materialize's dataflow, supporting both columnar and
175/// columnation.
176pub trait MzData:
177    columnation::Columnation
178    + mz_timely_util::columnar::chunk::PoolColumnar
179    + for<'a> columnar::Columnar<
180        Container: Container<Ref<'a>: Copy + Ord> + Clone + Send + columnar::Push<Self>,
181    >
182{
183}
184
185impl<T> MzData for T
186where
187    T: columnation::Columnation,
188    T: mz_timely_util::columnar::chunk::PoolColumnar,
189    T: columnar::Columnar<Container: Clone + Send + columnar::Push<T>>,
190    for<'a> Ref<'a, T>: Copy + Ord,
191{
192}
193
194pub trait MzArrangeData: columnation::Columnation {}
195impl<T> MzArrangeData for T where T: columnation::Columnation {}
mz_compute/typedefs.rs

mz_compute/
typedefs.rs
