Skip to content

Commit febcc17

Browse files
antiguruantiguru
authored
Update Timely, Columnar versions (#569)
* Upgrade columnar, timely Signed-off-by: Moritz Hoffmann <[email protected]> * Correct columnar example Signed-off-by: Moritz Hoffmann <[email protected]> * Tack on some auxiliary changes Fix versions in mdbook to precise versions (=), make check against Timely master more robust. Signed-off-by: Moritz Hoffmann <[email protected]> --------- Signed-off-by: Moritz Hoffmann <[email protected]>
1 parent 4217a1c commit febcc17

File tree

4 files changed

+28
-36
lines changed

4 files changed

+28
-36
lines changed

‎.github/workflows/deploy.yml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,8 @@ jobs:
1515
run: |
1616
export DIFFERENTIAL_VERSION=$(cargo metadata --format-version 1 | jq -r '.packages[] | select(.name == "differential-dataflow") | .version')
1717
export TIMELY_VERSION=$(cargo metadata --format-version 1 | jq -r '.packages[] | select(.name == "timely") | .version')
18-
sed -i "s/^differential-dataflow = .*/differential = \"$DIFFERENTIAL_VERSION\"/" mdbook/src/chapter_0/chapter_0_0.md
19-
sed -i "s/^timely = .*/timely = \"$TIMELY_VERSION\"/" mdbook/src/chapter_0/chapter_0_0.md
18+
sed -i "s/^differential-dataflow = .*/differential = \"=$DIFFERENTIAL_VERSION\"/" mdbook/src/chapter_0/chapter_0_0.md
19+
sed -i "s/^timely = .*/timely = \"=$TIMELY_VERSION\"/" mdbook/src/chapter_0/chapter_0_0.md
2020
- run: cd mdbook && mdbook build
2121
- uses: JamesIves/github-pages-deploy-action@v4
2222
with:

‎.github/workflows/test-timely-master.yml‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,13 @@ jobs:
2525
[patch.crates-io]
2626
timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
2727
EOF
28+
- name: Cargo upgrade
29+
run: |
30+
cargo install cargo-edit
31+
cargo upgrade -p timely --incompatible
2832
- name: Cargo check against Timely master
2933
run: cargo check --all-targets
3034
- name: Cargo test against Timely master
31-
working-directory: timely_master
3235
run: cargo test
3336

3437
notify:

‎Cargo.toml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,10 +45,10 @@ bytemuck = "1.18.0"
4545
serde = { version = "1.0", features = ["derive"] }
4646
fnv="1.0.2"
4747
timely = {workspace = true}
48-
columnar = "0.2"
48+
columnar = "0.3"
4949

5050
[workspace.dependencies]
51-
timely = { version = "0.17", default-features = false }
51+
timely = { version = "0.18", default-features = false }
5252
#timely = { path = "../timely-dataflow/timely/", default-features = false }
5353

5454
[features]

‎examples/columnar.rs‎

Lines changed: 20 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -154,17 +154,17 @@ mod container {
154154
}
155155
}
156156

157-
use columnar::{Clear, Len, Index, AsBytes, FromBytes};
158-
use columnar::bytes::serialization::decode;
157+
use columnar::{Clear, Len, Index, FromBytes};
158+
use columnar::bytes::{EncodeDecode, Indexed};
159159
use columnar::common::IterOwn;
160160

161161
use timely::Container;
162162
impl<C: Columnar> Container for Column<C> {
163163
fn len(&self) -> usize {
164164
match self {
165165
Column::Typed(t) => t.len(),
166-
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'_> as FromBytes>::from_bytes(&mut decode(bytemuck::cast_slice(b))).len(),
167-
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'_> as FromBytes>::from_bytes(&mut decode(a)).len(),
166+
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'_> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).len(),
167+
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'_> as FromBytes>::from_bytes(&mut Indexed::decode(a)).len(),
168168
}
169169
}
170170
// This sets the `Bytes` variant to be an empty `Typed` variant, appropriate for pushing into.
@@ -181,8 +181,8 @@ mod container {
181181
fn iter<'a>(&'a self) -> Self::Iter<'a> {
182182
match self {
183183
Column::Typed(t) => t.borrow().into_iter(),
184-
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut decode(bytemuck::cast_slice(b))).into_iter(),
185-
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut decode(a)).into_iter(),
184+
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_iter(),
185+
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_iter(),
186186
}
187187
}
188188

@@ -191,8 +191,8 @@ mod container {
191191
fn drain<'a>(&'a mut self) -> Self::DrainIter<'a> {
192192
match self {
193193
Column::Typed(t) => t.borrow().into_iter(),
194-
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut decode(bytemuck::cast_slice(b))).into_iter(),
195-
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut decode(a)).into_iter(),
194+
Column::Bytes(b) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(bytemuck::cast_slice(b))).into_iter(),
195+
Column::Align(a) => <<C::Container as columnar::Container<C>>::Borrowed<'a> as FromBytes>::from_bytes(&mut Indexed::decode(a)).into_iter(),
196196
}
197197
}
198198
}
@@ -202,7 +202,7 @@ mod container {
202202
fn at_capacity(&self) -> bool {
203203
match self {
204204
Self::Typed(t) => {
205-
let length_in_bytes = t.borrow().length_in_words() * 8;
205+
let length_in_bytes = Indexed::length_in_bytes(&t.borrow());
206206
length_in_bytes >= (1 << 20)
207207
},
208208
Self::Bytes(_) => true,
@@ -249,28 +249,15 @@ mod container {
249249
fn length_in_bytes(&self) -> usize {
250250
match self {
251251
// We'll need one u64 for the length, then the length rounded up to a multiple of 8.
252-
Column::Typed(t) => 8 * t.borrow().length_in_words(),
252+
Column::Typed(t) => Indexed::length_in_bytes(&t.borrow()),
253253
Column::Bytes(b) => b.len(),
254254
Column::Align(a) => 8 * a.len(),
255255
}
256256
}
257257

258258
fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
259259
match self {
260-
Column::Typed(t) => {
261-
use columnar::Container;
262-
// Columnar data is serialized as a sequence of `u64` values, with each `[u8]` slice
263-
// serialize as first its length in bytes, and then as many `u64` values as needed.
264-
// Padding should be added, but only for alignment; no specific values are required.
265-
for (align, bytes) in t.borrow().as_bytes() {
266-
assert!(align <= 8);
267-
let length: u64 = bytes.len().try_into().unwrap();
268-
writer.write_all(bytemuck::cast_slice(std::slice::from_ref(&length))).unwrap();
269-
writer.write_all(bytes).unwrap();
270-
let padding: usize = ((8 - (length % 8)) % 8).try_into().unwrap();
271-
writer.write_all(&[0; 8][..padding]).unwrap();
272-
}
273-
},
260+
Column::Typed(t) => Indexed::write(writer, &t.borrow()).unwrap(),
274261
Column::Bytes(b) => writer.write_all(b).unwrap(),
275262
Column::Align(a) => writer.write_all(bytemuck::cast_slice(a)).unwrap(),
276263
}
@@ -281,9 +268,11 @@ mod container {
281268

282269
use builder::ColumnBuilder;
283270
mod builder {
284-
285271
use std::collections::VecDeque;
286-
use columnar::{Columnar, Clear, Len, AsBytes, Push};
272+
273+
use columnar::{Columnar, Clear, Len, Push};
274+
use columnar::bytes::{EncodeDecode, Indexed};
275+
287276
use super::Column;
288277

289278
/// A container builder for `Column<C>`.
@@ -303,11 +292,11 @@ mod builder {
303292
self.current.push(item);
304293
// If there is less than 10% slop with 2MB backing allocations, mint a container.
305294
use columnar::Container;
306-
let words = self.current.borrow().length_in_words();
295+
let words = Indexed::length_in_words(&self.current.borrow());
307296
let round = (words + ((1 << 18) - 1)) & !((1 << 18) - 1);
308297
if round - words < round / 10 {
309298
let mut alloc = Vec::with_capacity(words);
310-
columnar::bytes::serialization::encode(&mut alloc, self.current.borrow().as_bytes());
299+
Indexed::encode(&mut alloc, &self.current.borrow());
311300
self.pending.push_back(Column::Align(alloc.into_boxed_slice()));
312301
self.current.clear();
313302
}
@@ -342,9 +331,9 @@ mod builder {
342331
fn finish(&mut self) -> Option<&mut Self::Container> {
343332
if !self.current.is_empty() {
344333
use columnar::Container;
345-
let words = self.current.borrow().length_in_words();
334+
let words = Indexed::length_in_words(&self.current.borrow());
346335
let mut alloc = Vec::with_capacity(words);
347-
columnar::bytes::serialization::encode(&mut alloc, self.current.borrow().as_bytes());
336+
Indexed::encode(&mut alloc, &self.current.borrow());
348337
self.pending.push_back(Column::Align(alloc.into_boxed_slice()));
349338
self.current.clear();
350339
}
@@ -405,7 +394,7 @@ pub mod batcher {
405394
}
406395
}
407396

408-
impl<'a, D, T, R, C2> PushInto<&'a mut Column<(D, T, R)>> for Chunker<C2>
397+
impl<'a, D, T, R, C2> PushInto<&'a mut Column<(D, T, R)>> for Chunker<C2>
409398
where
410399
D: Columnar,
411400
for<'b> D::Ref<'b>: Ord + Copy,

0 commit comments

Comments
 (0)