Syncs the state of this operator to match that of the persist shard until the provided frontier
(
&mut self,
target_upper: AntichainRef<'_, IntoTime>,
)
| 86 | /// Syncs the state of this operator to match that of the persist shard until the provided |
| 87 | /// frontier |
| 88 | async fn sync( |
| 89 | &mut self, |
| 90 | target_upper: AntichainRef<'_, IntoTime>, |
| 91 | ) -> ReclockBatch<FromTime, IntoTime> { |
| 92 | let mut updates: Vec<(FromTime, IntoTime, Diff)> = Vec::new(); |
| 93 | |
| 94 | // Tail the remap collection until we reach the target upper frontier. Note that, in the |
| 95 | // common case, we are also the writer, so we are waiting to read-back what we wrote |
| 96 | while PartialOrder::less_than(&self.upper.borrow(), &target_upper) { |
| 97 | let (mut batch, upper) = self |
| 98 | .remap_handle |
| 99 | .next() |
| 100 | .await |
| 101 | .expect("requested data after empty antichain"); |
| 102 | self.upper = upper; |
| 103 | updates.append(&mut batch); |
| 104 | } |
| 105 | |
| 106 | self.source_upper.update_iter( |
| 107 | updates |
| 108 | .iter() |
| 109 | .map(|(src_ts, _dest_ts, diff)| (src_ts.clone(), diff.into_inner())), |
| 110 | ); |
| 111 | |
| 112 | ReclockBatch { |
| 113 | updates, |
| 114 | upper: self.upper.clone(), |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | pub async fn mint( |
| 119 | &mut self, |