Fix merge batcher flat

Signed-off-by: Moritz Hoffmann <[email protected]>
TimelyDataflow · Jun 11, 2024 · c7c4aa8 · c7c4aa8
1 parent 264c267
commit c7c4aa8
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/src/trace/implementations/merge_batcher_flat.rs b/src/trace/implementations/merge_batcher_flat.rs
@@ -99,6 +99,7 @@ where
         + Push<((FR::Key<'a>, FR::Val<'a>), FR::Time<'a>, FR::Diff<'a>)>,
     for<'a> FR::Time<'a>: PartialOrder<T> + Copy + IntoOwned<'a, Owned=T>,
     for<'a> FR::Diff<'a>: IntoOwned<'a, Owned=R>,
+    for<'a> FR::ReadItem<'a>: std::fmt::Debug,
 {
     type Time = T;
     type Chunk = FlatStack<FR>;
@@ -171,6 +172,10 @@ where
             head1.advance(advance);
             result = self.empty(stash);
         }
+        if !result.is_empty() {
+            output.push(result);
+            result = self.empty(stash);
+        }
         output.extend(list1);
         self.recycle(head1.done(), stash);
 
@@ -245,6 +250,8 @@ where
                 for (key, val, time, _diff) in buffer.iter().map(FR::into_parts) {
                     if !upper.less_equal(&time) {
                         if let Some((p_key, p_val)) = prev_keyval {
+                            debug_assert!(p_key <= key);
+                            debug_assert!(p_key != key || p_val <= val);
                             if p_key != key {
                                 keys += 1;
                                 vals += 1;