timely-util: ColumnChunk, a pool-spillable Chunk implementation

src/timely-util/proptest-regressions/columnar/chunk.txt

@@ -0,0 +1,8 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc 79f9e0f3cf01ba590155d536a36d67864533c44d17ad3c38da58886159837681 # shrinks to input = [((0, 0), 0, 1)], cuts = [], probe_keys = {}, spill = false
+cc 6a70b97870c5a545462247056ceac3dd0c69f0ff2fb855502d3ed8d87201c8ea # shrinks to inputs = [[((0, 3), 3, -2), ((0, 4), 0, 3), ((3, 3), 1, 2)]], cuts = [4, 0]

src/timely-util/src/columnar.rs

@@ -20,6 +20,7 @@
 pub mod batcher;
 pub mod builder;
 pub mod builder_input;
+pub mod chunk;
 pub mod consolidate;
 pub mod merge_batcher;
 pub mod unload;
@@ -175,26 +176,29 @@ where
 /// merger and chunks shipped from the builder are sized comparably.
 const SHIP_WORDS: usize = 1 << 18;
 
-/// Returns true once the serialized size of `borrow` is within 10% of the next
-/// `SHIP_WORDS` boundary.
+/// Returns true once the serialized size of `borrow` reaches 10% under
+/// `SHIP_WORDS`.
 ///
-/// Same heuristic as `ColumnBuilder::push_into`; lifted out so the merger and
-/// the `SizableContainer` impl agree on the ship signal.
+/// Monotone in size, deliberately not a window below the boundary. A single
+/// record wider than a window steps clear over it, and a ship signal that
+/// un-fires past the boundary lets a chunk grow until it exceeds the buffer
+/// pool's largest size class, past which a spilled body degrades to
+/// permanently resident. The same heuristic as [`builder::ColumnBuilder`]'s
+/// ship point, lifted out so the builder, the merger, and the
+/// `SizableContainer` impl agree on the signal.
 #[inline]
 pub(crate) fn at_serialized_capacity<'a, A>(borrow: &A) -> bool
 where
     A: columnar::AsBytes<'a>,
 {
-    let words = indexed::length_in_words(borrow);
-    let round = (words + (SHIP_WORDS - 1)) & !(SHIP_WORDS - 1);
-    round - words < round / 10
+    indexed::length_in_words(borrow) >= SHIP_WORDS - SHIP_WORDS / 10
 }
 
 impl<C: Columnar> SizableContainer for Column<C> {
     fn at_capacity(&self) -> bool {
-        // Match `ColumnBuilder`'s ship heuristic: serialized size within 10%
-        // of the next 2 MiB. Aligns chunk-size choices across the two paths
-        // and keeps recipients dealing with a single granularity.
+        // Match `ColumnBuilder`'s ship heuristic: serialized size at the
+        // 2 MiB ship threshold. Aligns chunk-size choices across the two
+        // paths and keeps recipients dealing with a single granularity.
         //
         // Serialized chunks (`Bytes` / `Align`) have no typed builder to push
         // into, so they're trivially "at capacity" — there's no further work
@@ -365,4 +369,24 @@ mod tests {
             vec![&1, &2, &3]
         );
     }
+
+    /// The ship signal is monotone: once it fires it stays fired, even when
+    /// a single wide record steps far past the 2 MiB boundary in one push.
+    #[mz_ore::test]
+    fn ship_threshold_monotone() {
+        use columnar::Push;
+        let mut container = <Vec<u64> as Columnar>::Container::default();
+        // Wider than 10% of any boundary a 25 MiB run can reach.
+        let wide: Vec<u64> = vec![0u64; 50_000];
+        let mut fired = false;
+        for pushes in 1..=64 {
+            container.push(&wide);
+            let now = at_serialized_capacity(&container.borrow());
+            if fired {
+                assert!(now, "ship signal un-fired at {pushes} records");
+            }
+            fired = fired || now;
+        }
+        assert!(fired, "ship signal never fired");
+    }
 }

src/timely-util/src/columnar/batcher.rs

@@ -259,7 +259,7 @@ where
 /// Compared to a linear scan, this is `O(log K)` for a run of length `K`
 /// satisfying `cmp` — useful when one side of a sorted merge has long runs
 /// dominated by the other side.
-fn gallop(upper: usize, lower: &mut usize, mut cmp: impl FnMut(usize) -> bool) {
+pub(crate) fn gallop(upper: usize, lower: &mut usize, mut cmp: impl FnMut(usize) -> bool) {
     // If `cmp` is already false at `*lower`, the run is empty — nothing to do.
     if *lower < upper && cmp(*lower) {
         // Phase 1 (overshoot): advance by exponentially growing steps as long

src/timely-util/src/columnar/builder.rs

@@ -44,28 +44,25 @@ where
     #[inline]
     fn push_into(&mut self, item: T) {
         self.current.push(item);
-        // If there is less than 10% slop with 2MB backing allocations, mint a container.
+        // Mint a container once the serialized size reaches the ship threshold.
         use columnar::Borrow;
-        let words = indexed::length_in_words(&self.current.borrow());
-        let round = (words + ((1 << 18) - 1)) & !((1 << 18) - 1);
-        if round - words < round / 10 {
+        if crate::columnar::at_serialized_capacity(&self.current.borrow()) {
             /// Move the contents from `current` to a `Vec<u64>` allocation built via
             /// `indexed::encode` (so no zero-init pre-pass), and push it to `pending`.
             #[cold]
-            fn outlined_align<C>(
-                current: &mut C::Container,
-                words: usize,
-                pending: &mut VecDeque<Column<C>>,
-            ) where
+            fn outlined_align<C>(current: &mut C::Container, pending: &mut VecDeque<Column<C>>)
+            where
                 C: Columnar,
             {
+                use columnar::Borrow;
+                let words = indexed::length_in_words(&current.borrow());
                 let mut alloc: Vec<u64> = Vec::with_capacity(words);
                 indexed::encode(&mut alloc, &current.borrow());
                 pending.push_back(Column::Align(alloc));
                 current.clear();
             }
 
-            outlined_align(&mut self.current, words, &mut self.pending);
+            outlined_align(&mut self.current, &mut self.pending);
         }
     }
 }