fn internal_restore(
        &self,
        read: StreamReadHandle<SourceTimestamp, AssignedTimestamp>,
        upper_bindings_seal_ts: u64,
        upper_data_seal_ts: u64,
    ) -> Result<
        (
            HashMap<PartitionId, Option<MzOffset>>,
            Vec<(SourceTimestamp, AssignedTimestamp)>,
        ),
        persist::error::Error,
    > {
        assert!(upper_bindings_seal_ts >= upper_data_seal_ts);

        let mut bindings: HashMap<_, isize> = HashMap::new();
        let mut starting_offsets = HashMap::new();

        let snapshot = read.snapshot()?;

        let buf = snapshot.into_iter().collect::<Result<Vec<_>, _>>()?;

        for ((source_timestamp, assigned_timestamp), ts, diff) in buf.into_iter() {
            // Only restore starting offsets that are not beyond the uppser_data_seal_ts. This is the
            // timestamp up to which we have sealed the persistent collection storing the actual source
            // data/updates.
            if ts < upper_data_seal_ts {
                starting_offsets
                    .entry(source_timestamp.partition.clone())
                    .and_modify(|current_offset| {
                        match current_offset {
                            Some(current_offset) if source_timestamp.offset > *current_offset => {
                                *current_offset = source_timestamp.offset;
                            }
                            _ => (), // ignore "older" starting offsets
                        }
                    })
                    .or_insert_with(|| Some(source_timestamp.offset));
            } else {
                // If the binding is beyond the upper seal timestamp, we don't want to use its
                // offset as a starting offset. We still want to record that we know about the
                // partition so that we can add bindings to `timestamp_histories`. We do this,
                // because we restore all bindings, ignoring whether they are before or after the
                // seal timestamp.
                starting_offsets
                    .entry(source_timestamp.partition.clone())
                    .or_insert_with(|| None);
            }

            // Collect all the bindings. The bindings are potentially beyond the starting offsets, but
            // that is ok. This means we will emit source data with previously persisted bindings.
            //
            // We consolidate all the diffs as we go. Below we collect all positive updates and
            // return them.
            *bindings
                .entry((source_timestamp, assigned_timestamp))
                .or_default() += diff;
        }

        let bindings: Vec<_> = bindings
            .drain()
            .filter(|(binding, diff)| {
                if *diff < 0 || *diff > 1 {
                    panic!(
                        "Binding with invalid diff. Binding {:?}, diff: {}.",
                        binding, diff
                    );
                }
                *diff == 1
            })
            .map(|(binding, _diff)| binding)
            .collect();

        Ok((starting_offsets, bindings))
    }
/// Restores the latest partition offsets and timestamp bindings from the given `StreamReadHandle`.
///
/// This restores partition offsets that are not beyond the given `upper_data_seal_ts`, to ensure
/// that the updates we emit from the source are consistent with the state that we have in the
/// persistent data collection. We do restore bindings up to `upper_bindings_seal_ts`, which is
/// potentially beyond `upper_data_seal_ts` because we seal the latter before sealing the former.
/// This means we will emit source data with previously persisted bindings, which is a valid thing
/// to do.
///
/// When `write` is `Some(StreamWriteHandle)`, this will also emit retractions for updates that are
/// beyond the `upper_bindings_seal_ts`. This should be used when restoring to clean up unsealed
/// updates from previous attempts.
Restores the latest partition offsets and timestamp bindings from the given `StreamReadHandle`.
This restores partition offsets that are not beyond the given `upper_data_seal_ts`, to ensure
that the updates we emit from the source are consistent with the state that we have in the
persistent data collection. We do restore bindings up to `upper_bindings_seal_ts`, which is
potentially beyond `upper_data_seal_ts` because we seal the latter before sealing the former.
This means we will emit source data with previously persisted bindings, which is a valid thing
to do.


Restores the latest partition offsets and timestamp bindings from the given ` StreamReadHandle ` . This restores partition offsets that are not beyond the given ` upper_data_seal_ts ` to ensure that the updates we emit from the source are consistent with the state that we have in the persistent data collection . We do restore bindings up to ` upper_bindings_seal_ts ` which is potentially beyond ` upper_data_seal_ts ` because we seal the latter before sealing the former . This means we will emit source data with previously persisted bindings which is a valid thing to do .
fn internal_restore(
        &self,
        read: StreamReadHandle<SourceTimestamp, AssignedTimestamp>,
        upper_bindings_seal_ts: u64,
        upper_data_seal_ts: u64,
    ) -> Result<
        (
            HashMap<PartitionId, Option<MzOffset>>,
            Vec<(SourceTimestamp, AssignedTimestamp)>,
        ),
        persist::error::Error,
    > {
        assert!(upper_bindings_seal_ts >= upper_data_seal_ts);
        let mut bindings: HashMap<_, isize> = HashMap::new();
        let mut starting_offsets = HashMap::new();
        let snapshot = read.snapshot()?;
        let buf = snapshot.into_iter().collect::<Result<Vec<_>, _>>()?;
        for ((source_timestamp, assigned_timestamp), ts, diff) in buf.into_iter() {
            // Only restore starting offsets that are not beyond the uppser_data_seal_ts. This is the
            // timestamp up to which we have sealed the persistent collection storing the actual source
            // data/updates.
            if ts < upper_data_seal_ts {
                starting_offsets
                    .entry(source_timestamp.partition.clone())
                    .and_modify(|current_offset| {
                        match current_offset {
                            Some(current_offset) if source_timestamp.offset > *current_offset => {
                                *current_offset = source_timestamp.offset;
                            }
                            _ => (), // ignore "older" starting offsets
                        }
                    })
                    .or_insert_with(|| Some(source_timestamp.offset));
            } else {
                // If the binding is beyond the upper seal timestamp, we don't want to use its
                // offset as a starting offset. We still want to record that we know about the
                // partition so that we can add bindings to `timestamp_histories`. We do this,
                // because we restore all bindings, ignoring whether they are before or after the
                // seal timestamp.
                starting_offsets
                    .entry(source_timestamp.partition.clone())
                    .or_insert_with(|| None);
            }
            // Collect all the bindings. The bindings are potentially beyond the starting offsets, but
            // that is ok. This means we will emit source data with previously persisted bindings.
            //
            // We consolidate all the diffs as we go. Below we collect all positive updates and
            // return them.
            *bindings
                .entry((source_timestamp, assigned_timestamp))
                .or_default() += diff;
        }
        let bindings: Vec<_> = bindings
            .drain()
            .filter(|(binding, diff)| {
                if *diff < 0 || *diff > 1 {
                    panic!(
                        "Binding with invalid diff. Binding {:?}, diff: {}.",
                        binding, diff
                    );
                }
                *diff == 1
            })
            .map(|(binding, _diff)| binding)
            .collect();
        Ok((starting_offsets, bindings))
    }
fn internal_restore ( & self , read : StreamReadHandle < SourceTimestamp , AssignedTimestamp > , upper_bindings_seal_ts : u64 , upper_data_seal_ts : u64 , ) -> Result < ( HashMap < PartitionId , Option < MzOffset > > , Vec < ( SourceTimestamp , AssignedTimestamp ) > , ) , persist :: error :: Error , > { assert ! ( upper_bindings_seal_ts upper_data_seal_ts ) ; let mut bindings : HashMap < _ , isize > = HashMap :: new ( ) ; let mut starting_offsets = HashMap :: new ( ) ; let snapshot = read . snapshot ( ) ? ; let buf = snapshot . into_iter ( ) . collect :: < Result < Vec < _ > , _ > > ( ) ? ; for ( ( source_timestamp , assigned_timestamp ) , ts , diff ) in buf . into_iter ( ) { // Only restore starting offsets that are not beyond the uppser_data_seal_ts. This is the // timestamp up to which we have sealed the persistent collection storing the actual source // data/updates. if ts < upper_data_seal_ts { starting_offsets . entry ( source_timestamp . partition . clone ( ) ) . and_modify ( | current_offset | { match current_offset { Some ( current_offset ) if source_timestamp . offset > * current_offset => { * current_offset = source_timestamp . offset ; } _ => ( ) , // ignore "older" starting offsets } } ) . or_insert_with ( | | Some ( source_timestamp . offset ) ) ; } else { // If the binding is beyond the upper seal timestamp, we don't want to use its // offset as a starting offset. We still want to record that we know about the // partition so that we can add bindings to `timestamp_histories`. We do this, // because we restore all bindings, ignoring whether they are before or after the // seal timestamp. starting_offsets . entry ( source_timestamp . partition . clone ( ) ) . or_insert_with ( | | None ) ; } // Collect all the bindings. The bindings are potentially beyond the starting offsets, but // that is ok. This means we will emit source data with previously persisted bindings. // // We consolidate all the diffs as we go. Below we collect all positive updates and // return them. * bindings . entry ( ( source_timestamp , assigned_timestamp ) ) . or_default ( ) += diff ; } let bindings : Vec < _ > = bindings . drain ( ) . filter ( | ( binding , diff ) | { if * diff < 0 || * diff > 1 { panic ! ( " " binding diff ) ; } * diff == 1 } ) . map ( | ( binding , _diff ) | binding ) . collect ( ) ; Ok ( ( starting_offsets , bindings ) ) }
Restores the latest partition offsets and timestamp bindings from the given `StreamReadHandle`.
Restores the latest partition offsets and timestamp bindings from the given ` StreamReadHandle ` .
&self
read
StreamReadHandle<SourceTimestamp, AssignedTimestamp>
upper_bindings_seal_ts
u64
upper_data_seal_ts
u64
&self
read
StreamReadHandle<SourceTimestamp, AssignedTimestamp>
upper_bindings_seal_ts
u64
upper_data_seal_ts
u64