Construct a new [ReclockOperator] from the given collection metadata
(remap_handle: Handle)
| 62 | { |
| 63 | /// Construct a new [ReclockOperator] from the given collection metadata |
| 64 | pub async fn new(remap_handle: Handle) -> (Self, ReclockBatch<FromTime, IntoTime>) { |
| 65 | let upper = remap_handle.upper().clone(); |
| 66 | |
| 67 | let mut operator = Self { |
| 68 | upper: Antichain::from_elem(IntoTime::minimum()), |
| 69 | source_upper: MutableAntichain::new(), |
| 70 | remap_handle, |
| 71 | }; |
| 72 | |
| 73 | // Load the initial state that might exist in the shard |
| 74 | let trace_batch = if upper.elements() != [IntoTime::minimum()] { |
| 75 | operator.sync(upper.borrow()).await |
| 76 | } else { |
| 77 | ReclockBatch { |
| 78 | updates: vec![], |
| 79 | upper: Antichain::from_elem(IntoTime::minimum()), |
| 80 | } |
| 81 | }; |
| 82 | |
| 83 | (operator, trace_batch) |
| 84 | } |
| 85 | |
| 86 | /// Syncs the state of this operator to match that of the persist shard until the provided |
| 87 | /// frontier |