diff --git a/differential-dataflow/src/columnar/trace/chunk.rs b/differential-dataflow/src/columnar/trace/chunk.rs index a8357ae28..393058920 100644 --- a/differential-dataflow/src/columnar/trace/chunk.rs +++ b/differential-dataflow/src/columnar/trace/chunk.rs @@ -678,7 +678,8 @@ mod test { // resumable merge -> advance -> settle pipeline end to end. #[test] fn batch_merger_resumable_matches_reference() { - use crate::trace::{Description, Merger}; + use crate::trace::Description; + use crate::trace::implementations::spine_fueled::Merger; use crate::trace::chunk::{ChunkBatch, ChunkBatchMerger, is_graded}; use crate::trace::cursor::Cursor; use crate::consolidation::consolidate_updates; diff --git a/differential-dataflow/src/trace/chunk/mod.rs b/differential-dataflow/src/trace/chunk/mod.rs index 93a4d28aa..6c894b85b 100644 --- a/differential-dataflow/src/trace/chunk/mod.rs +++ b/differential-dataflow/src/trace/chunk/mod.rs @@ -58,6 +58,7 @@ use timely::progress::Antichain; use timely::progress::frontier::AntichainRef; use crate::lattice::Lattice; use crate::trace::{Batch, BatchReader, Description, Navigable}; +use crate::trace::implementations::spine_fueled::SpineBatch; use crate::trace::cursor::Cursor; use crate::trace::implementations::BatchContainer; @@ -268,12 +269,17 @@ impl BatchReader for ChunkBatch { fn description(&self) -> &Description { &self.description } } -impl Batch for ChunkBatch +impl SpineBatch for ChunkBatch where C::Time: timely::progress::Timestamp + Lattice + Ord, { type Merger = ChunkBatchMerger; +} +impl Batch for ChunkBatch +where + C::Time: timely::progress::Timestamp + Lattice + Ord, +{ fn empty(lower: Antichain, upper: Antichain) -> Self { use timely::progress::Timestamp; let since = Antichain::from_elem(Self::Time::minimum()); @@ -560,7 +566,7 @@ where fn len(chunk: &C) -> usize { chunk.len() } } -/// The resumable [`Batch::Merger`] for [`ChunkBatch`]: merges two batches and advances +/// The resumable [`SpineBatch::Merger`] for [`ChunkBatch`]: merges two batches and advances /// their times to the compaction frontier, a fuel-bounded step at a time. /// /// Each step pipelines [`merge`](Chunk::merge) → [`advance`](Chunk::advance) → @@ -591,7 +597,7 @@ pub struct ChunkBatchMerger { complete: bool, } -impl crate::trace::Merger> for ChunkBatchMerger +impl crate::trace::implementations::spine_fueled::Merger> for ChunkBatchMerger where C: Chunk + Default + 'static, C::Time: timely::progress::Timestamp + Lattice + Ord + 'static, diff --git a/differential-dataflow/src/trace/chunk/vec.rs b/differential-dataflow/src/trace/chunk/vec.rs index 18dc05d78..d2e32e9c5 100644 --- a/differential-dataflow/src/trace/chunk/vec.rs +++ b/differential-dataflow/src/trace/chunk/vec.rs @@ -601,7 +601,8 @@ mod test { // resumable merge→advance→settle pipeline and the grade-at-yield invariant. #[test] fn batch_merger_resumable_matches_reference() { - use crate::trace::{Description, Merger}; + use crate::trace::Description; + use crate::trace::implementations::spine_fueled::Merger; use crate::trace::chunk::{ChunkBatch, ChunkBatchMerger, is_graded}; use crate::trace::cursor::Cursor; use crate::consolidation::consolidate_updates; diff --git a/differential-dataflow/src/trace/implementations/ord_neu.rs b/differential-dataflow/src/trace/implementations/ord_neu.rs index cf0c22a5f..e58271a60 100644 --- a/differential-dataflow/src/trace/implementations/ord_neu.rs +++ b/differential-dataflow/src/trace/implementations/ord_neu.rs @@ -248,7 +248,8 @@ pub mod val_batch { use timely::container::PushInto; use timely::progress::{Antichain, frontier::AntichainRef}; - use crate::trace::{Batch, BatchReader, Builder, Cursor, Description, Merger}; + use crate::trace::{Batch, BatchReader, Builder, Cursor, Description}; + use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; use crate::trace::implementations::{BatchContainer, BuilderInput}; use crate::trace::implementations::layout; @@ -323,13 +324,15 @@ pub mod val_batch { fn description(&self) -> &Description> { &self.description } } - impl Batch for OrdValBatch { + impl SpineBatch for OrdValBatch { type Merger = OrdValMerger; fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef>) -> Self::Merger { OrdValMerger::new(self, other, compaction_frontier) } + } + impl Batch for OrdValBatch { fn empty(lower: Antichain, upper: Antichain) -> Self { use timely::progress::Timestamp; Self { @@ -740,7 +743,8 @@ pub mod key_batch { use timely::container::PushInto; use timely::progress::{Antichain, frontier::AntichainRef}; - use crate::trace::{Batch, BatchReader, Builder, Cursor, Description, Merger}; + use crate::trace::{Batch, BatchReader, Builder, Cursor, Description}; + use crate::trace::implementations::spine_fueled::{SpineBatch, Merger}; use crate::trace::implementations::{BatchContainer, BuilderInput}; use crate::trace::implementations::layout; @@ -824,13 +828,15 @@ pub mod key_batch { fn description(&self) -> &Description> { &self.description } } - impl>> Batch for OrdKeyBatch { + impl>> SpineBatch for OrdKeyBatch { type Merger = OrdKeyMerger; fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef>) -> Self::Merger { OrdKeyMerger::new(self, other, compaction_frontier) } + } + impl>> Batch for OrdKeyBatch { fn empty(lower: Antichain, upper: Antichain) -> Self { use timely::progress::Timestamp; Self { diff --git a/differential-dataflow/src/trace/implementations/spine_fueled.rs b/differential-dataflow/src/trace/implementations/spine_fueled.rs index a6fc7b904..e3dec5693 100644 --- a/differential-dataflow/src/trace/implementations/spine_fueled.rs +++ b/differential-dataflow/src/trace/implementations/spine_fueled.rs @@ -2,7 +2,7 @@ //! //! The `Spine` is a general-purpose trace implementation based on collection and merging //! immutable batches of updates. It is generic with respect to the batch type, and can be -//! instantiated for any implementor of `trace::Batch`. +//! instantiated for any implementor of [`SpineBatch`]. //! //! ## Design //! @@ -69,19 +69,72 @@ //! have paid back any "debt" to higher layers by continuing to provide fuel as updates arrive. +use std::rc::Rc; + use crate::logging::Logger; -use crate::trace::{Batch, ExertionLogic, Merger, Trace, TraceReader}; +use crate::trace::{Batch, ExertionLogic, Trace, TraceReader}; use ::timely::dataflow::operators::generic::OperatorInfo; use ::timely::progress::{Antichain, frontier::AntichainRef}; use ::timely::order::PartialOrder; +/// The requirements this spine imposes on its batches, beyond [`Batch`]. +/// +/// This is an opinion of this spine, not a property of batches in general: batches must +/// support progressive (fuel-limited) merging through a [`Merger`]. Other trace +/// implementations may want to merge differently, or not at all, which is why this +/// requirement lives here rather than on the common [`Batch`] trait. +pub trait SpineBatch : Batch { + /// A type used to progressively merge batches. + type Merger: Merger; + + /// Initiates the merging of consecutive batches. + /// + /// The result of this method can be exercised to eventually produce the same result + /// that a call to `self.merge(other)` would produce, but it can be done in a measured + /// fashion. This can help to avoid latency spikes where a large merge needs to happen. + fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef) -> Self::Merger { + Self::Merger::new(self, other, compaction_frontier) + } +} + +/// Represents a merge in progress. +pub trait Merger { + /// Creates a new merger to merge the supplied batches, optionally compacting + /// up to the supplied frontier. + fn new(source1: &Output, source2: &Output, compaction_frontier: AntichainRef) -> Self; + /// Perform some amount of work, decrementing `fuel`. + /// + /// If `fuel` is non-zero after the call, the merging is complete and + /// one should call `done` to extract the merged results. + fn work(&mut self, source1: &Output, source2: &Output, fuel: &mut isize); + /// Extracts merged results. + /// + /// This method should only be called after `work` has been called and + /// has not brought `fuel` to zero. Otherwise, the merge is still in + /// progress. + fn done(self) -> Output; +} + +impl SpineBatch for Rc { + type Merger = RcMerger; +} + +/// Wrapper type for merging reference counted batches. +pub struct RcMerger { merger: B::Merger } + +impl Merger> for RcMerger { + fn new(source1: &Rc, source2: &Rc, compaction_frontier: AntichainRef) -> Self { RcMerger { merger: B::begin_merge(source1, source2, compaction_frontier) } } + fn work(&mut self, source1: &Rc, source2: &Rc, fuel: &mut isize) { self.merger.work(source1, source2, fuel) } + fn done(self) -> Rc { Rc::new(self.merger.done()) } +} + /// An append-only collection of update tuples. /// /// A spine maintains a small number of immutable collections of update tuples, merging the collections when /// two have similar sizes. In this way, it allows the addition of more tuples, which may then be merged with /// other immutable collections. -pub struct Spine { +pub struct Spine { operator: OperatorInfo, logger: Option, logical_frontier: Antichain, // Times after which the trace must accumulate correctly. @@ -97,7 +150,7 @@ pub struct Spine { exert_logic: Option, } -impl TraceReader for Spine { +impl TraceReader for Spine { type Time = B::Time; type Batch = B; @@ -225,7 +278,7 @@ impl TraceReader for Spine { // A trace implementation for any key type that can be borrowed from or converted into `Key`. // TODO: Almost all this implementation seems to be generic with respect to the trace and batch types. -impl Trace for Spine { +impl Trace for Spine { fn new( info: ::timely::dataflow::operators::generic::OperatorInfo, logging: Option, @@ -294,14 +347,14 @@ impl Trace for Spine { } // Drop implementation allows us to log batch drops, to zero out maintained totals. -impl Drop for Spine { +impl Drop for Spine { fn drop(&mut self) { self.drop_batches(); } } -impl Spine { +impl Spine { /// Drops and logs batches. Used in `set_logical_compaction` and drop. fn drop_batches(&mut self) { if let Some(logger) = &self.logger { @@ -342,7 +395,7 @@ impl Spine { } } -impl Spine { +impl Spine { /// Determine the amount of effort we should exert in the absence of updates. /// /// This method prepares an iterator over batches, including the level, count, and length of each layer. @@ -715,7 +768,7 @@ impl Spine { /// /// A layer can be empty, contain a single batch, or contain a pair of batches /// that are in the process of merging into a batch for the next layer. -enum MergeState { +enum MergeState { /// An empty layer, containing no updates. Vacant, /// A layer containing a single batch. @@ -727,7 +780,7 @@ enum MergeState { Double(MergeVariant), } -impl> MergeState { +impl> MergeState { /// The number of actual updates contained in the level. fn len(&self) -> usize { @@ -813,7 +866,7 @@ impl> MergeState { match (batch1, batch2) { (Some(batch1), Some(batch2)) => { assert!(batch1.upper() == batch2.lower()); - let begin_merge = ::begin_merge(&batch1, &batch2, compaction_frontier); + let begin_merge = ::begin_merge(&batch1, &batch2, compaction_frontier); MergeVariant::InProgress(batch1, batch2, begin_merge) } (None, Some(x)) => MergeVariant::Complete(Some((x, None))), @@ -825,14 +878,14 @@ impl> MergeState { } } -enum MergeVariant { +enum MergeVariant { /// Describes an actual in-progress merge between two non-trivial batches. - InProgress(B, B, ::Merger), + InProgress(B, B, ::Merger), /// A merge that requires no further work. May or may not represent a non-trivial batch. Complete(Option<(B, Option<(B, B)>)>), } -impl MergeVariant { +impl MergeVariant { /// Completes and extracts the batch, unless structurally empty. /// diff --git a/differential-dataflow/src/trace/mod.rs b/differential-dataflow/src/trace/mod.rs index 1b5439f87..7dea2a76b 100644 --- a/differential-dataflow/src/trace/mod.rs +++ b/differential-dataflow/src/trace/mod.rs @@ -203,19 +203,12 @@ pub trait BatchReader : Sized { } /// An immutable collection of updates. +/// +/// This trait asks only that batches can be minted empty over an indicated interval, +/// which trace maintenance uses to pad out otherwise empty intervals of time. Opinions +/// about how batches merge belong to individual trace implementations (for example +/// [`SpineBatch`](crate::trace::implementations::spine_fueled::SpineBatch)). pub trait Batch : BatchReader + Sized { - /// A type used to progressively merge batches. - type Merger: Merger; - - /// Initiates the merging of consecutive batches. - /// - /// The result of this method can be exercised to eventually produce the same result - /// that a call to `self.merge(other)` would produce, but it can be done in a measured - /// fashion. This can help to avoid latency spikes where a large merge needs to happen. - fn begin_merge(&self, other: &Self, compaction_frontier: AntichainRef) -> Self::Merger { - Self::Merger::new(self, other, compaction_frontier) - } - /// Produce an empty batch over the indicated interval. fn empty(lower: Antichain, upper: Antichain) -> Self; } @@ -274,32 +267,13 @@ pub trait Builder: Sized { fn seal(chain: &mut Vec, description: Description) -> Self::Output; } -/// Represents a merge in progress. -pub trait Merger { - /// Creates a new merger to merge the supplied batches, optionally compacting - /// up to the supplied frontier. - fn new(source1: &Output, source2: &Output, compaction_frontier: AntichainRef) -> Self; - /// Perform some amount of work, decrementing `fuel`. - /// - /// If `fuel` is non-zero after the call, the merging is complete and - /// one should call `done` to extract the merged results. - fn work(&mut self, source1: &Output, source2: &Output, fuel: &mut isize); - /// Extracts merged results. - /// - /// This method should only be called after `work` has been called and - /// has not brought `fuel` to zero. Otherwise, the merge is still in - /// progress. - fn done(self) -> Output; -} - - /// Blanket implementations for reference counted batches. pub mod rc_blanket_impls { use std::rc::Rc; - use timely::progress::{Antichain, frontier::AntichainRef}; - use super::{Batch, BatchReader, Builder, Merger, Navigable, Cursor, Description}; + use timely::progress::Antichain; + use super::{Batch, BatchReader, Builder, Navigable, Cursor, Description}; impl Navigable for Rc { /// The type used to enumerate the batch's contents. @@ -374,7 +348,6 @@ pub mod rc_blanket_impls { /// An immutable collection of updates. impl Batch for Rc { - type Merger = RcMerger; fn empty(lower: Antichain, upper: Antichain) -> Self { Rc::new(B::empty(lower, upper)) } @@ -396,13 +369,4 @@ pub mod rc_blanket_impls { } } - /// Wrapper type for merging reference counted batches. - pub struct RcMerger { merger: B::Merger } - - /// Represents a merge in progress. - impl Merger> for RcMerger { - fn new(source1: &Rc, source2: &Rc, compaction_frontier: AntichainRef) -> Self { RcMerger { merger: B::begin_merge(source1, source2, compaction_frontier) } } - fn work(&mut self, source1: &Rc, source2: &Rc, fuel: &mut isize) { self.merger.work(source1, source2, fuel) } - fn done(self) -> Rc { Rc::new(self.merger.done()) } - } } diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs index 35c617a6c..84440c987 100644 --- a/interactive/src/corgi/chunk.rs +++ b/interactive/src/corgi/chunk.rs @@ -618,7 +618,8 @@ where mod test { use super::*; use differential_dataflow::trace::chunk::{ChunkBatchMerger, is_graded}; - use differential_dataflow::trace::{Description, Merger}; + use differential_dataflow::trace::Description; + use differential_dataflow::trace::implementations::spine_fueled::Merger; use std::collections::BTreeMap; fn xorshift(s: &mut u64) -> u64 { *s ^= *s << 13; *s ^= *s >> 7; *s ^= *s << 17; *s }