mz_timely_util/columnar/
builder_input.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! `BuilderInput` impls for [`ColumnBody`] and [`Column`] so DD `Builder`s can
17//! drain a batcher's chunks without an extra container conversion.
18//!
19//! Mirrors the impl on [`ColumnationStack`](crate::columnation::ColumnationStack)
20//! at `columnation.rs`, but the `Item<'a>` here is a columnar `Ref` tuple
21//! rather than a borrowed owned tuple, so:
22//!
23//! - `Key<'a>` / `Val<'a>` are `Ref<'a, K>` / `Ref<'a, V>` — no `Owned`
24//!   round-trip on the read side.
25//! - `Time` / `Diff` materialize as owned on `into_parts` (the trait
26//!   contract requires owned for these).
27//!
28//! Distinct-counts (`key_val_upd_counts`) tally per chunk and sum, accepting
29//! at most `chain.len()` over-counts at chunk boundaries. The downstream
30//! consumer uses these as capacity hints, so a small over-estimate is
31//! cheaper than the alternative (snapshotting `K::Owned` / `V::Owned`
32//! across chunk boundaries).
33//!
34//! The body impl serves the chunk batchers, whose chains are bodies. The
35//! column impl serves the column pager's batcher, whose chains are still
36//! edge containers.
37
38use columnar::{BorrowedOf, Columnar, Index, Len};
39use differential_dataflow::difference::Semigroup;
40use differential_dataflow::lattice::Lattice;
41use differential_dataflow::trace::implementations::BatchContainer;
42use differential_dataflow::trace::implementations::ord_neu::BuilderInput;
43use timely::progress::Timestamp;
44
45use crate::columnar::Column;
46use crate::columnar::body::ColumnBody;
47
48/// Distinct key, distinct `(key, val)`, and update counts over a chain of
49/// views, deduplicated within each view and summed.
50fn key_val_upd_counts<'a, K, V, T, R>(
51    chain: impl Iterator<Item = BorrowedOf<'a, ((K, V), T, R)>>,
52) -> (usize, usize, usize)
53where
54    K: Columnar,
55    V: Columnar,
56    T: Columnar,
57    R: Columnar,
58    for<'b> columnar::Ref<'b, K>: Copy + Ord,
59    for<'b> columnar::Ref<'b, V>: Copy + Ord,
60{
61    // Per-chunk dedup, summed. Skips cross-chunk equality checks; the
62    // counts may over-count by up to `chain.len()` (one boundary per
63    // chunk). Capacity-hint consumers tolerate over-estimates.
64    let mut keys = 0;
65    let mut vals = 0;
66    let mut upds = 0;
67    for view in chain {
68        let len = view.len();
69        if len == 0 {
70            continue;
71        }
72        let mut prev: Option<(columnar::Ref<'_, K>, columnar::Ref<'_, V>)> = None;
73        for i in 0..len {
74            let ((k, v), _, _) = view.get(i);
75            match prev {
76                None => {
77                    keys += 1;
78                    vals += 1;
79                }
80                Some((pk, pv)) => {
81                    if pk != k {
82                        keys += 1;
83                        vals += 1;
84                    } else if pv != v {
85                        vals += 1;
86                    }
87                }
88            }
89            upds += 1;
90            prev = Some((k, v));
91        }
92    }
93    (keys, vals, upds)
94}
95
96macro_rules! impl_builder_input {
97    ($container:ident) => {
98        impl<KBC, VBC, K, V, T, R> BuilderInput<KBC, VBC> for $container<((K, V), T, R)>
99        where
100            K: Columnar,
101            V: Columnar,
102            T: Columnar + Timestamp + Lattice + Clone,
103            R: Columnar + Ord + Semigroup + Clone,
104            for<'a> columnar::Ref<'a, K>: Copy + Ord,
105            for<'a> columnar::Ref<'a, V>: Copy + Ord,
106            KBC: BatchContainer,
107            VBC: BatchContainer,
108            for<'a, 'b> KBC::ReadItem<'a>: PartialEq<columnar::Ref<'b, K>>,
109            for<'a, 'b> VBC::ReadItem<'a>: PartialEq<columnar::Ref<'b, V>>,
110        {
111            type Key<'a> = columnar::Ref<'a, K>;
112            type Val<'a> = columnar::Ref<'a, V>;
113            type Time = T;
114            type Diff = R;
115
116            fn into_parts<'a>(
117                item: Self::Item<'a>,
118            ) -> (Self::Key<'a>, Self::Val<'a>, Self::Time, Self::Diff) {
119                let ((key, val), time, diff) = item;
120                (key, val, T::into_owned(time), R::into_owned(diff))
121            }
122
123            fn key_eq(this: &columnar::Ref<'_, K>, other: KBC::ReadItem<'_>) -> bool {
124                KBC::reborrow(other) == *this
125            }
126
127            fn val_eq(this: &columnar::Ref<'_, V>, other: VBC::ReadItem<'_>) -> bool {
128                VBC::reborrow(other) == *this
129            }
130
131            fn key_val_upd_counts(chain: &[Self]) -> (usize, usize, usize) {
132                key_val_upd_counts::<K, V, T, R>(chain.iter().map(|chunk| chunk.borrow()))
133            }
134        }
135    };
136}
137
138impl_builder_input!(ColumnBody);
139impl_builder_input!(Column);
mz_timely_util/columnar/builder_input.rs

mz_timely_util/columnar/
builder_input.rs
