ct: don't clone every row when producing ct_times
#30123
Conversation
In the previous impl, the split in the dataflow graph resulted in every row being cloned, just to be immediately thrown away. This is wasteful and potentially doubles the memory requirements. Instead use a custom 1-input 2-output operator (passing through the input). While I'm in here, also resolve a TODO to switch the operators to extension traits on `Collection`.
There was a problem hiding this comment.
Thank, this looks good. I left a comment around using the ConsolidatingContainerBuilder to opportunistically consolidate the times output, which would simplify your implementation (and would permit consolidating accross input batches. As a rule of thumb, use the consolidating container builder whenever an operator changes the key/value cardinality and we can expect reasonable amounts of consolidation. The cost of the builder is not free, but low enough that it doesn't matter too much.)
| move |_frontiers| { | ||
| let mut passthrough = passthrough.activate(); | ||
| let mut times = times.activate(); | ||
| while let Some((cap, data)) = input.next() { | ||
| data.swap(&mut buf); | ||
| for ((), ts, diff) in buf.drain(..) { | ||
| notificator.notify_at(cap.delayed(&ts)); | ||
| if let Some(sum) = stash.get_mut(&ts) { | ||
| sum.plus_equals(&diff); | ||
| data.swap(&mut passthrough_buf); | ||
| for (_data, ts, diff) in &passthrough_buf { | ||
| if let Some(d) = times_hash.get_mut(ts) { | ||
| d.plus_equals(diff); | ||
| } else { | ||
| stash.insert(ts, diff); | ||
| } | ||
| times_hash.insert(*ts, diff.clone()); | ||
| }; | ||
| } | ||
| let times_iter = times_hash | ||
| .drain() | ||
| // Silly to emit zero diffs. | ||
| .filter(|(_, diff)| !diff.is_zero()) | ||
| .map(|(ts, diff)| ((), ts, diff)); | ||
| times_buf.extend(times_iter); | ||
| passthrough | ||
| .session(&cap) | ||
| .give_container(&mut passthrough_buf); | ||
| times.session(&cap).give_container(&mut times_buf); | ||
| } |
There was a problem hiding this comment.
I think this could use the ConsolidatingContainerBuilder instead of maintaining its own hashmap, but it might change semantics. For an example, see flat_map. The consolidating container builder sits on a buffer of fixed length, appends, and whenever full or time ticks, it'll consolidate the buffer. This only ensures opportunistically consolidated outputs, so duplicate elements might still occur.
In this case, the stream seems to be ((), time, diff), with time being totally ordered. It seems this would be the best case example for the consolidating builder, unless you need potentially zero or more timestamps per input batch. (Note that the implementation emits to times on every input.next() iteration.)
In the previous impl, the split in the dataflow graph resulted in every row being cloned, just to be immediately thrown away. This is wasteful and potentially doubles the memory requirements. Instead use a custom 1-input 2-output operator (passing through the input).
While I'm in here, also resolve a TODO to switch the operators to extension traits on
Collection.Touches https://github.com/MaterializeInc/database-issues/issues/8427
Motivation
Tips for reviewer
Definitely review with hide whitespace changes on
Checklist
$T ⇔ Proto$Tmapping (possibly in a backwards-incompatible way), then it is tagged with aT-protolabel.