Pass description itself to builder (#551)

antiguru · web-flow · commit 63212f9afc47 · 2024-12-07T20:16:22.000-05:00
Pass the description itself to the builder instead of supplying the
parameters required to build a description. No functional changes.

Signed-off-by: Moritz Hoffmann &lt;antiguru@gmail.com&gt;
diff --git a/src/operators/arrange/upsert.rs b/src/operators/arrange/upsert.rs
@@ -109,7 +109,7 @@ use timely::progress::Antichain;
 use timely::dataflow::operators::Capability;
 
 use crate::operators::arrange::arrangement::Arranged;
-use crate::trace::Builder;
+use crate::trace::{Builder, Description};
 use crate::trace::{self, Trace, TraceReader, Batch, Cursor};
 use crate::trace::cursor::IntoOwned;
 use crate::{ExchangeData, Hashable};
@@ -281,7 +281,8 @@ where
                                     updates.sort();
                                     builder.push(&mut updates);
                                 }
-                                let batch = builder.done(prev_frontier.clone(), upper.clone(), Antichain::from_elem(G::Timestamp::minimum()));
+                                let description = Description::new(prev_frontier.clone(), upper.clone(), Antichain::from_elem(G::Timestamp::minimum()));
+                                let batch = builder.done(description);
                                 prev_frontier.clone_from(&upper);
 
                                 // Communicate `batch` to the arrangement and the stream.
diff --git a/src/operators/reduce.rs b/src/operators/reduce.rs
@@ -23,7 +23,7 @@ use crate::trace::cursor::IntoOwned;
 
 use crate::operators::arrange::{Arranged, ArrangeByKey, ArrangeBySelf, TraceAgent};
 use crate::lattice::Lattice;
-use crate::trace::{Batch, BatchReader, Cursor, Trace, Builder, ExertionLogic};
+use crate::trace::{Batch, BatchReader, Cursor, Trace, Builder, ExertionLogic, Description};
 use crate::trace::cursor::CursorList;
 use crate::trace::implementations::{KeySpine, KeyBuilder, ValSpine, ValBuilder};
 
@@ -565,7 +565,8 @@ where
 
                             if output_upper.borrow() != output_lower.borrow() {
 
-                                let batch = builder.done(output_lower.clone(), output_upper.clone(), Antichain::from_elem(G::Timestamp::minimum()));
+                                let description = Description::new(output_lower.clone(), output_upper.clone(), Antichain::from_elem(G::Timestamp::minimum()));
+                                let batch = builder.done(description);
 
                                 // ship batch to the output, and commit to the output trace.
                                 output.session(&capabilities[index]).give(batch.clone());
diff --git a/src/trace/implementations/merge_batcher.rs b/src/trace/implementations/merge_batcher.rs
@@ -12,7 +12,7 @@ use timely::container::{ContainerBuilder, PushInto};
 
 use crate::difference::Semigroup;
 use crate::logging::{BatcherEvent, DifferentialEvent};
-use crate::trace::{Batcher, Builder};
+use crate::trace::{Batcher, Builder, Description};
 use crate::Data;
 
 /// Creates batches from unordered tuples.
@@ -109,7 +109,8 @@ where
 
         self.stash.clear();
 
-        let seal = B::seal(&mut readied, self.lower.borrow(), upper.borrow(), Antichain::from_elem(M::Time::minimum()).borrow());
+        let description = Description::new(self.lower.clone(), upper.clone(), Antichain::from_elem(M::Time::minimum()));
+        let seal = B::seal(&mut readied, description);
         self.lower = upper;
         seal
     }
diff --git a/src/trace/implementations/ord_neu.rs b/src/trace/implementations/ord_neu.rs
@@ -682,7 +682,7 @@ mod val_batch {
         }
 
         #[inline(never)]
-        fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> OrdValBatch<L> {
+        fn done(mut self, description: Description<Self::Time>) -> OrdValBatch<L> {
             // Record the final offsets
             self.result.vals_offs.push(self.result.times.len());
             // Remove any pending singleton, and if it was set increment our count.
@@ -691,23 +691,18 @@ mod val_batch {
             OrdValBatch {
                 updates: self.result.times.len() + self.singletons,
                 storage: self.result,
-                description: Description::new(lower, upper, since),
+                description,
             }
         }
 
-        fn seal(
-            chain: &mut Vec<Self::Input>,
-            lower: AntichainRef<Self::Time>,
-            upper: AntichainRef<Self::Time>,
-            since: AntichainRef<Self::Time>,
-        ) -> Self::Output {
+        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
             let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
             let mut builder = Self::with_capacity(keys, vals, upds);
             for mut chunk in chain.drain(..) {
                 builder.push(&mut chunk);
             }
     
-            builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())
+            builder.done(description)
         }
     }
 }
@@ -1170,31 +1165,26 @@ mod key_batch {
         }
 
         #[inline(never)]
-        fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> OrdKeyBatch<L> {
+        fn done(mut self, description: Description<Self::Time>) -> OrdKeyBatch<L> {
             // Record the final offsets
             self.result.keys_offs.push(self.result.times.len());
             // Remove any pending singleton, and if it was set increment our count.
             if self.singleton.take().is_some() { self.singletons += 1; }
             OrdKeyBatch {
                 updates: self.result.times.len() + self.singletons,
                 storage: self.result,
-                description: Description::new(lower, upper, since),
+                description,
             }
         }
 
-        fn seal(
-            chain: &mut Vec<Self::Input>,
-            lower: AntichainRef<Self::Time>,
-            upper: AntichainRef<Self::Time>,
-            since: AntichainRef<Self::Time>,
-        ) -> Self::Output {
+        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
             let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
             let mut builder = Self::with_capacity(keys, vals, upds);
             for mut chunk in chain.drain(..) {
                 builder.push(&mut chunk);
             }
     
-            builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())
+            builder.done(description)
         }
     }
 
diff --git a/src/trace/implementations/rhh.rs b/src/trace/implementations/rhh.rs
@@ -855,7 +855,7 @@ mod val_batch {
         }
 
         #[inline(never)]
-        fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> RhhValBatch<L> {
+        fn done(mut self, description: Description<Self::Time>) -> RhhValBatch<L> {
             // Record the final offsets
             self.result.vals_offs.push(self.result.times.len());
             // Remove any pending singleton, and if it was set increment our count.
@@ -864,23 +864,18 @@ mod val_batch {
             RhhValBatch {
                 updates: self.result.times.len() + self.singletons,
                 storage: self.result,
-                description: Description::new(lower, upper, since),
+                description,
             }
         }
 
-        fn seal(
-            chain: &mut Vec<Self::Input>,
-            lower: AntichainRef<Self::Time>,
-            upper: AntichainRef<Self::Time>,
-            since: AntichainRef<Self::Time>,
-        ) -> Self::Output {
+        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
             let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);
             let mut builder = Self::with_capacity(keys, vals, upds);
             for mut chunk in chain.drain(..) {
                 builder.push(&mut chunk);
             }
-    
-            builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())
+
+            builder.done(description)
         }
     }
 
diff --git a/src/trace/mod.rs b/src/trace/mod.rs
@@ -342,18 +342,14 @@ pub trait Builder: Sized {
     /// Adds all elements from `chunk` to the builder and leaves `chunk` in an undefined state.
     fn push(&mut self, chunk: &mut Self::Input);
     /// Completes building and returns the batch.
-    fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Self::Output;
+    fn done(self, description: Description<Self::Time>) -> Self::Output;
 
     /// Builds a batch from a chain of updates corresponding to the indicated lower and upper bounds.
     ///
     /// This method relies on the chain only containing updates greater or equal to the lower frontier,
-    /// and not greater or equal to the upper frontier. Chains must also be sorted and consolidated.
-    fn seal(
-        chain: &mut Vec<Self::Input>,
-        lower: AntichainRef<Self::Time>,
-        upper: AntichainRef<Self::Time>,
-        since: AntichainRef<Self::Time>,
-    ) -> Self::Output;
+    /// and not greater or equal to the upper frontier, as encoded in the description. Chains must also
+    /// be sorted and consolidated.
+    fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output;
 }
 
 /// Represents a merge in progress.
@@ -467,14 +463,9 @@ pub mod rc_blanket_impls {
         type Output = Rc<B::Output>;
         fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { RcBuilder { builder: B::with_capacity(keys, vals, upds) } }
         fn push(&mut self, input: &mut Self::Input) { self.builder.push(input) }
-        fn done(self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> Rc<B::Output> { Rc::new(self.builder.done(lower, upper, since)) }
-        fn seal(
-            chain: &mut Vec<Self::Input>,
-            lower: AntichainRef<Self::Time>,
-            upper: AntichainRef<Self::Time>,
-            since: AntichainRef<Self::Time>,
-        ) -> Self::Output {
-            Rc::new(B::seal(chain, lower, upper, since))
+        fn done(self, description: Description<Self::Time>) -> Rc<B::Output> { Rc::new(self.builder.done(description)) }
+        fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {
+            Rc::new(B::seal(chain, description))
         }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -682,7 +682,7 @@ mod val_batch {`
`682`	`682`	`}`
`683`	`683`
`684`	`684`	`#[inline(never)]`
`685`		`- fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> OrdValBatch<L> {`
	`685`	`+ fn done(mut self, description: Description<Self::Time>) -> OrdValBatch<L> {`
`686`	`686`	`// Record the final offsets`
`687`	`687`	`self.result.vals_offs.push(self.result.times.len());`
`688`	`688`	`// Remove any pending singleton, and if it was set increment our count.`
`@@ -691,23 +691,18 @@ mod val_batch {`
`691`	`691`	`OrdValBatch {`
`692`	`692`	`updates: self.result.times.len() + self.singletons,`
`693`	`693`	`storage: self.result,`
`694`		`- description: Description::new(lower, upper, since),`
	`694`	`+ description,`
`695`	`695`	`}`
`696`	`696`	`}`
`697`	`697`
`698`		`- fn seal(`
`699`		`- chain: &mut Vec<Self::Input>,`
`700`		`- lower: AntichainRef<Self::Time>,`
`701`		`- upper: AntichainRef<Self::Time>,`
`702`		`- since: AntichainRef<Self::Time>,`
`703`		`- ) -> Self::Output {`
	`698`	`+ fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {`
`704`	`699`	`let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);`
`705`	`700`	`let mut builder = Self::with_capacity(keys, vals, upds);`
`706`	`701`	`for mut chunk in chain.drain(..) {`
`707`	`702`	`builder.push(&mut chunk);`
`708`	`703`	`}`
`709`	`704`
`710`		`- builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())`
	`705`	`+ builder.done(description)`
`711`	`706`	`}`
`712`	`707`	`}`
`713`	`708`	`}`
`@@ -1170,31 +1165,26 @@ mod key_batch {`
`1170`	`1165`	`}`
`1171`	`1166`
`1172`	`1167`	`#[inline(never)]`
`1173`		`- fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> OrdKeyBatch<L> {`
	`1168`	`+ fn done(mut self, description: Description<Self::Time>) -> OrdKeyBatch<L> {`
`1174`	`1169`	`// Record the final offsets`
`1175`	`1170`	`self.result.keys_offs.push(self.result.times.len());`
`1176`	`1171`	`// Remove any pending singleton, and if it was set increment our count.`
`1177`	`1172`	`if self.singleton.take().is_some() { self.singletons += 1; }`
`1178`	`1173`	`OrdKeyBatch {`
`1179`	`1174`	`updates: self.result.times.len() + self.singletons,`
`1180`	`1175`	`storage: self.result,`
`1181`		`- description: Description::new(lower, upper, since),`
	`1176`	`+ description,`
`1182`	`1177`	`}`
`1183`	`1178`	`}`
`1184`	`1179`
`1185`		`- fn seal(`
`1186`		`- chain: &mut Vec<Self::Input>,`
`1187`		`- lower: AntichainRef<Self::Time>,`
`1188`		`- upper: AntichainRef<Self::Time>,`
`1189`		`- since: AntichainRef<Self::Time>,`
`1190`		`- ) -> Self::Output {`
	`1180`	`+ fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {`
`1191`	`1181`	`let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);`
`1192`	`1182`	`let mut builder = Self::with_capacity(keys, vals, upds);`
`1193`	`1183`	`for mut chunk in chain.drain(..) {`
`1194`	`1184`	`builder.push(&mut chunk);`
`1195`	`1185`	`}`
`1196`	`1186`
`1197`		`- builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())`
	`1187`	`+ builder.done(description)`
`1198`	`1188`	`}`
`1199`	`1189`	`}`
`1200`	`1190`
Original file line number	Diff line number	Diff line change
`@@ -855,7 +855,7 @@ mod val_batch {`
`855`	`855`	`}`
`856`	`856`
`857`	`857`	`#[inline(never)]`
`858`		`- fn done(mut self, lower: Antichain<Self::Time>, upper: Antichain<Self::Time>, since: Antichain<Self::Time>) -> RhhValBatch<L> {`
	`858`	`+ fn done(mut self, description: Description<Self::Time>) -> RhhValBatch<L> {`
`859`	`859`	`// Record the final offsets`
`860`	`860`	`self.result.vals_offs.push(self.result.times.len());`
`861`	`861`	`// Remove any pending singleton, and if it was set increment our count.`
`@@ -864,23 +864,18 @@ mod val_batch {`
`864`	`864`	`RhhValBatch {`
`865`	`865`	`updates: self.result.times.len() + self.singletons,`
`866`	`866`	`storage: self.result,`
`867`		`- description: Description::new(lower, upper, since),`
	`867`	`+ description,`
`868`	`868`	`}`
`869`	`869`	`}`
`870`	`870`
`871`		`- fn seal(`
`872`		`- chain: &mut Vec<Self::Input>,`
`873`		`- lower: AntichainRef<Self::Time>,`
`874`		`- upper: AntichainRef<Self::Time>,`
`875`		`- since: AntichainRef<Self::Time>,`
`876`		`- ) -> Self::Output {`
	`871`	`+ fn seal(chain: &mut Vec<Self::Input>, description: Description<Self::Time>) -> Self::Output {`
`877`	`872`	`let (keys, vals, upds) = Self::Input::key_val_upd_counts(&chain[..]);`
`878`	`873`	`let mut builder = Self::with_capacity(keys, vals, upds);`
`879`	`874`	`for mut chunk in chain.drain(..) {`
`880`	`875`	`builder.push(&mut chunk);`
`881`	`876`	`}`
`882`		`-`
`883`		`- builder.done(lower.to_owned(), upper.to_owned(), since.to_owned())`
	`877`	`+`
	`878`	`+ builder.done(description)`
`884`	`879`	`}`
`885`	`880`	`}`
`886`	`881`