diff --git a/crates/dbsp/src/operator/dynamic/input_upsert.rs b/crates/dbsp/src/operator/dynamic/input_upsert.rs index 38e0c68e925..a923f4b532e 100644 --- a/crates/dbsp/src/operator/dynamic/input_upsert.rs +++ b/crates/dbsp/src/operator/dynamic/input_upsert.rs @@ -307,7 +307,6 @@ where .add_binary_operator( , U, B>>::new( factories.batch_factories.clone(), - factories.opt_key_factory, factories.opt_val_factory, patch_func, ), @@ -480,7 +479,6 @@ where U: DataTrait + ?Sized, { batch_factories: B::Factories, - opt_key_factory: &'static dyn Factory>, opt_val_factory: &'static dyn Factory>, patch_func: PatchFunc, @@ -501,13 +499,11 @@ where { pub fn new( batch_factories: B::Factories, - opt_key_factory: &'static dyn Factory>, opt_val_factory: &'static dyn Factory>, patch_func: PatchFunc, ) -> Self { Self { batch_factories, - opt_key_factory, opt_val_factory, patch_func, input_batch_stats: BatchSizeStats::new(), @@ -579,7 +575,7 @@ where B::Builder::with_capacity(&self.batch_factories, n_updates * 2, n_updates * 2); // Current key for which we are processing updates. - let mut cur_key: Box> = self.opt_key_factory.default_box(); + let mut cur_key = None; // Current value associated with the key after applying all processed updates // to it. @@ -603,9 +599,9 @@ where // We finished processing updates for the previous key. Push them to the // builder and generate a retraction for the new key. - if cur_key.get() != Some(key) { + if cur_key != Some(key) { // Push updates for the previous key to the builder. - if let Some(cur_key) = cur_key.get_mut() { + if let Some(cur_key) = cur_key { if let Some(val) = cur_val.get_mut() { key_updates.push_with(&mut |item| { let (v, w) = item.split_mut(); @@ -625,7 +621,7 @@ where key_updates.clear(); } - cur_key.from_ref(key); + cur_key = Some(key); cur_val.set_none(); // Generate retraction if `key` is present in the trace. @@ -670,7 +666,7 @@ where } // Push updates for the last key. - if let Some(cur_key) = cur_key.get_mut() { + if let Some(cur_key) = cur_key { if let Some(val) = cur_val.get_mut() { key_updates.push_with(&mut |item| { let (v, w) = item.split_mut(); @@ -839,7 +835,7 @@ where ); // Current key for which we are processing updates. - let mut cur_key: Box> = self.factories.opt_key_factory.default_box(); + let mut cur_key = None; // Current value associated with the key after applying all processed updates // to it. @@ -867,9 +863,9 @@ where // We finished processing updates for the previous key. Push them to the // builder and generate a retraction for the new key. - if cur_key.get() != Some(key) { + if cur_key != Some(key) { // Push updates for the previous key to the builder. - if let Some(cur_key) = cur_key.get_mut() { + if let Some(cur_key) = cur_key { if let Some(val) = cur_val.get_mut() { key_updates.push_with(&mut |item| { let (v, w) = item.split_mut(); @@ -890,7 +886,7 @@ where } skip_key = false; - cur_key.from_ref(key); + cur_key = Some(key); cur_val.set_none(); // Generate retraction if `key` is present in the trace. @@ -970,7 +966,7 @@ where } // Push updates for the last key. - if let Some(cur_key) = cur_key.get_mut() { + if let Some(cur_key) = cur_key { if let Some(val) = cur_val.get_mut() { key_updates.push_with(&mut |item| { let (v, w) = item.split_mut(); diff --git a/crates/dbsp/src/trace/cursor/cursor_list.rs b/crates/dbsp/src/trace/cursor/cursor_list.rs index d4082d82c04..2c43519a8ed 100644 --- a/crates/dbsp/src/trace/cursor/cursor_list.rs +++ b/crates/dbsp/src/trace/cursor/cursor_list.rs @@ -692,6 +692,10 @@ where fn seek_key_exact(&mut self, key: &K, hash: Option) -> bool { self.set_key_direction(None); + if self.cursors.is_empty() { + return false; + } + let hash = hash.unwrap_or_else(|| key.default_hash()); self.current_key.clear(); diff --git a/crates/dbsp/src/trace/ord/vec/indexed_wset_batch.rs b/crates/dbsp/src/trace/ord/vec/indexed_wset_batch.rs index 6a05f101bee..a5f08b9228f 100644 --- a/crates/dbsp/src/trace/ord/vec/indexed_wset_batch.rs +++ b/crates/dbsp/src/trace/ord/vec/indexed_wset_batch.rs @@ -245,9 +245,20 @@ where O: OrdOffset, { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("VecIndexedWSet") - .field("layer", &self.layer) - .finish() + writeln!(f, "VecIndexedWSet:")?; + let mut cursor = self.cursor(); + while let Some(key) = cursor.get_key() { + writeln!(f, " key {key:?}:")?; + while let Some(val) = cursor.get_val() { + writeln!(f, " value {val:?}:")?; + cursor.map_times(&mut |time, diff| { + let _ = writeln!(f, " ({time:?}, {diff:?})"); + }); + cursor.step_val(); + } + cursor.step_key(); + } + Ok(()) } }