Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ rust-version = "1.88"
differential-dataflow = { path = "differential-dataflow", default-features = false, version = "0.25.1" }
#timely = { version = "0.31", default-features = false }
columnar = { version = "0.13.3", default-features = false }
timely = { git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
timely = { version = "0.32", default-features = false }
#timely = { path = "../timely-dataflow/timely/", default-features = false }

[workspace.lints.clippy]
Expand Down
2 changes: 1 addition & 1 deletion differential-dataflow/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ lz4_flex = "0.11"
[dependencies]
columnar = { workspace = true }
columnation = "0.1.1"
foldhash = "=0.2.0"
foldhash = "0.2"
paste = "1.0"
serde = { version = "1.0", features = ["derive"] }
smallvec = { version = "1.15.1", features = ["const_generics", "union"] }
Expand Down
3 changes: 1 addition & 2 deletions differential-dataflow/examples/scc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,8 +2,7 @@
//!
//! Usage: scc [timely args] [nodes [edges [batch [rounds]]]]
//!
//! Mirrors `diagnostics/examples/scc-bench.rs` minus the logging dataflow,
//! for a clean baseline to compare against `interactive/examples/ddir_col`.
//! A baseline with no logging dataflow, for comparing other SCC implementations against.

use std::hash::{Hash, Hasher};
use std::collections::hash_map::DefaultHasher;
Expand Down
2 changes: 1 addition & 1 deletion differential-dataflow/src/collection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1110,7 +1110,7 @@ pub mod vec {
{
/// Matches pairs `(key,val1)` and `(key,val2)` based on `key` and yields pairs `(key, (val1, val2))`.
///
/// The [`join_map`](Join::join_map) method may be more convenient for non-trivial processing pipelines.
/// The [`join_map`](Collection::join_map) method may be more convenient for non-trivial processing pipelines.
///
/// # Examples
///
Expand Down
2 changes: 1 addition & 1 deletion differential-dataflow/src/columnar/collection/operators.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
//! Dataflow operators over the columnar [`RecordedUpdates`](super::RecordedUpdates)
//! Dataflow operators over the columnar [`RecordedUpdates`]
//! container — the collection-level surface of the columnar chunk.
//!
//! [`join_function`] is a columnar `flat_map` (subsuming map/filter/negate/
Expand Down
10 changes: 5 additions & 5 deletions differential-dataflow/src/columnar/trace/chunk.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,9 @@
//! A [`ColChunk`] is either [`Resident`](ColChunk::Resident) (the trie in memory)
//! or [`Paged`](ColChunk::Paged) (resident bounds + a byte handle). [`Chunk::settle`]
//! is the spill point: it pages committed chunks out via
//! [`spill`](self::spill) when a worker has installed a spiller. Reads
//! [`spill`] when a worker has installed a spiller. Reads
//! fetch a paged chunk's trie back, caching it in a [`OnceCell`] so repeated
//! cursor access pays the fetch once; [`len`](Chunk::len) and [`bounds`](Chunk::bounds)
//! cursor access pays the fetch once; [`len`](Chunk::len) and [`bounds`](crate::trace::chunk::NavigableChunk::bounds)
//! read the resident metadata and never fetch.
//!
//! [`Chunk::advance`] is trie-native: it melds its input in place and rewrites
Expand Down Expand Up @@ -59,7 +59,7 @@ use crate::trace::chunk::Chunk;
const TARGET: usize = 8192;

/// Resident bounds for a paged chunk: the first and last `(key, val, time)` as
/// single-element columnar containers (so [`Chunk::bounds`] returns refs without
/// single-element columnar containers (so [`crate::trace::chunk::NavigableChunk::bounds`] returns refs without
/// fetching), plus the record count.
pub struct ChunkMeta<U: ColumnarUpdate> {
fk: ContainerOf<U::Key>, fv: ContainerOf<U::Val>, ft: ContainerOf<U::Time>,
Expand Down Expand Up @@ -382,7 +382,7 @@ where U::Time: 'static {
/// withheld trailing `(key, val)` group (the `carry`) — it never pulls the whole
/// chain into core. Cross-chunk consolidation happens *only* where a `(key, val)`
/// straddles a boundary: the (un-advanced) carry is melded onto the next chunk's
/// head, a sorted-chain meld that just stitches that one group. [`advance_trie`]
/// head, a sorted-chain meld that just stitches that one group. `advance_trie`
/// then advances and consolidates the time runs *within* each chunk (keys / vals /
/// diffs stay columnar refs, no global re-sort). Grading the emitted chunks is
/// left to [`Chunk::settle`].
Expand Down Expand Up @@ -438,7 +438,7 @@ where U::Time: 'static {
/// Maximal packing via the harness [`pack`](crate::trace::chunk::pack): coalesce by melding
/// each trie of the run onto the first (adjacent chunks of a sorted, consolidated
/// chain, so meld's "strictly greater first triple" precondition holds), split
/// with [`trie_merger::split_at`], and seal through [`seal_chunk`] (the spill
/// with [`trie_merger::split_at`], and seal through `seal_chunk` (the spill
/// point — pages a committed chunk when a spiller is installed).
fn settle(input: &mut VecDeque<Self>, done: bool, out: &mut VecDeque<Self>) {
crate::trace::chunk::pack(
Expand Down
4 changes: 2 additions & 2 deletions differential-dataflow/src/columnar/trace/chunker.rs
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
//! The container→batch chunker for the columnar trace.
//!
//! [`TrieChunker`] is the `ContainerBuilder` that turns a stream of
//! [`RecordedUpdates`](crate::columnar::collection::RecordedUpdates) into
//! [`ColChunk`](crate::columnar::trace::ColChunk) batches for the `Chunk`
//! [`RecordedUpdates`] into
//! [`ColChunk`] batches for the `Chunk`
//! harness's batcher. It accumulates small inputs, consolidates, and ships
//! chunks sized within 1-2x [`LINK_TARGET`](crate::columnar::LINK_TARGET).

Expand Down
2 changes: 1 addition & 1 deletion differential-dataflow/src/columnar/trace/spill.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
//! [`ColChunk`](super::ColChunk) is either resident (a trie in
//! memory) or paged (resident bounds + a byte handle to fetch the trie back).
//! `Chunk::settle` is where committed chunks may be **paged out**: it calls
//! [`try_page`], which consults a per-worker [`SpillState`] and, when over budget,
//! `try_page`, which consults a per-worker `SpillState` and, when over budget,
//! serializes the trie and hands the bytes to a pluggable [`BytesStore`]. Reads
//! ([`Chunk::merge`](crate::trace::chunk::Chunk::merge) etc., and the cursor) fetch paged chunks back on demand.
//!
Expand Down
4 changes: 2 additions & 2 deletions differential-dataflow/src/operators/int_proxy/vec_backend.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
//!
//! Per window, the backend presents the input (the novel batches and the prior history, merged), the
//! times of the novel records as seeds, and the prior output.
//! Identifiers are ordinal positions minted by a single [`MergedWalk`] over the novel batches and the
//! Identifiers are ordinal positions minted by a single `MergedWalk` over the novel batches and the
//! prior ones together, so a value carries ONE identifier whichever run it appears in and a novel
//! retraction cancels against its own history. They are minted after the walk consolidates, so a value
//! whose records all cancel never reaches the tactic and never spends an identifier.
Expand Down Expand Up @@ -41,7 +41,7 @@ use super::{KeyPosition, ProxyReduceBackend, ReduceInstance, ReduceWindow};
/// `D` is `(K, V)` on the input side and `(K, W)` on the output side.
type VBatch<D, T, R> = Rc<ChunkBatch<VecChunk<u64, D, T, R>>>;

/// A reference [`ProxyReduceBackend`] over [`VBatch`] storage.
/// A reference [`ProxyReduceBackend`] over `VBatch` storage.
pub struct VecReduceBackend<K, V, W, T, R, L> {
/// User supplied reduce closure.
logic: L,
Expand Down
Loading