From ba8b92c32add73893686c25b73602a5c286fa8e7 Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Tue, 16 Sep 2025 14:54:57 -0700 Subject: [PATCH 1/3] [dbsp] Make Debug output for VecIndexedWSet readable. I really cannot decipher the default printed form of VecIndexedWSet. This is better. Signed-off-by: Ben Pfaff --- .../src/trace/ord/vec/indexed_wset_batch.rs | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) 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(()) } } From 3601658fe2a24386258968dd2337b1fcc6f2e1b1 Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Tue, 16 Sep 2025 16:23:51 -0700 Subject: [PATCH 2/3] [dbsp] Avoid hashing at all in trivial case in CursorList. Signed-off-by: Ben Pfaff --- crates/dbsp/src/trace/cursor/cursor_list.rs | 4 ++++ 1 file changed, 4 insertions(+) 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(); From 64309d52119996918b5ef1b57b66c534e979bc1d Mon Sep 17 00:00:00 2001 From: Ben Pfaff Date: Wed, 12 Aug 2026 17:27:56 -0700 Subject: [PATCH 3/3] [dbsp] Avoid cloning keys in InputUpsert::eval cur_key held a factory-allocated Box>, cloned from the current key via from_ref() on every key transition. The clone is unnecessary: the key reference comes from `updates`, a parameter borrowed for the entire function call, so it stays valid across the whole loop (and the final flush after it). Builder::push_key also only ever takes a borrowed &Output::Key, never ownership. Hold a plain Option<&T::Key> instead, dropping one allocation and copy per unique key. We can't do the same thing for cur_val because it gets mutated in-place and even if we introduce some kind of Cow-like logic it ends up getting copied most of the time anyway. No measured macrobenchmark performance difference but it seems like an obvious improvement. Signed-off-by: Ben Pfaff --- .../dbsp/src/operator/dynamic/input_upsert.rs | 24 ++++++++----------- 1 file changed, 10 insertions(+), 14 deletions(-) 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();