pub fn build(self) -> Result<ReplayPlan> {
        let Self {
            replay_ranges,
            last_partition_checkpoints,
        } = self;

        // sanity-check partition checkpoints
        for ((table_name, partition_key), partition_checkpoint) in &last_partition_checkpoints {
            for (sequencer_id, min_max) in &partition_checkpoint.sequencer_numbers {
                let database_wide_min_max = replay_ranges.get(sequencer_id).context(
                    PartitionCheckpointWithoutDatabase {
                        table_name: Arc::clone(table_name),
                        partition_key: Arc::clone(partition_key),
                        sequencer_id: *sequencer_id,
                    },
                )?;

                let sequence_missing = match (min_max.min(), database_wide_min_max.min()) {
                    (Some(min), Some(db_min)) => min < db_min,
                    (Some(_min), None) => true,
                    _ => false,
                };
                if sequence_missing {
                    warn!(
                        %table_name,
                        %partition_key,
                        sequencer_id,
                        database_checkpoint_range=%database_wide_min_max,
                        partition_checkpoint_range=%min_max,
                        "Last partition checkpoint states that there is unpersisted data, but the final database state deems it fully persisted. What happened to these sequence numbers?",
                    );
                }

                if min_max.max() > database_wide_min_max.max() {
                    return Err(Error::PartitionCheckpointMaximumAfterDatabase {
                        partition_checkpoint_sequence_number: min_max.max(),
                        database_checkpoint_sequence_number: database_wide_min_max.max(),
                        table_name: Arc::clone(table_name),
                        partition_key: Arc::clone(partition_key),
                        sequencer_id: *sequencer_id,
                    });
                }
            }
        }

        Ok(ReplayPlan {
            replay_ranges,
            last_partition_checkpoints,
        })
    }
/// Build plan that is then used for replay.
Build plan that is then used for replay.
Build plan that is then used for replay .
pub fn build(self) -> Result<ReplayPlan> {
        let Self {
            replay_ranges,
            last_partition_checkpoints,
        } = self;
        // sanity-check partition checkpoints
        for ((table_name, partition_key), partition_checkpoint) in &last_partition_checkpoints {
            for (sequencer_id, min_max) in &partition_checkpoint.sequencer_numbers {
                let database_wide_min_max = replay_ranges.get(sequencer_id).context(
                    PartitionCheckpointWithoutDatabase {
                        table_name: Arc::clone(table_name),
                        partition_key: Arc::clone(partition_key),
                        sequencer_id: *sequencer_id,
                    },
                )?;
                let sequence_missing = match (min_max.min(), database_wide_min_max.min()) {
                    (Some(min), Some(db_min)) => min < db_min,
                    (Some(_min), None) => true,
                    _ => false,
                };
                if sequence_missing {
                    warn!(
                        %table_name,
                        %partition_key,
                        sequencer_id,
                        database_checkpoint_range=%database_wide_min_max,
                        partition_checkpoint_range=%min_max,
                        "Last partition checkpoint states that there is unpersisted data, but the final database state deems it fully persisted. What happened to these sequence numbers?",
                    );
                }
                if min_max.max() > database_wide_min_max.max() {
                    return Err(Error::PartitionCheckpointMaximumAfterDatabase {
                        partition_checkpoint_sequence_number: min_max.max(),
                        database_checkpoint_sequence_number: database_wide_min_max.max(),
                        table_name: Arc::clone(table_name),
                        partition_key: Arc::clone(partition_key),
                        sequencer_id: *sequencer_id,
                    });
                }
            }
        }
        Ok(ReplayPlan {
            replay_ranges,
            last_partition_checkpoints,
        })
    }
pub fn build ( self ) -> Result < ReplayPlan > { let Self { replay_ranges , last_partition_checkpoints , } = self ; // sanity-check partition checkpoints for ( ( table_name , partition_key ) , partition_checkpoint ) in & last_partition_checkpoints { for ( sequencer_id , min_max ) in & partition_checkpoint . sequencer_numbers { let database_wide_min_max = replay_ranges . get ( sequencer_id ) . context ( PartitionCheckpointWithoutDatabase { table_name : Arc :: clone ( table_name ) , partition_key : Arc :: clone ( partition_key ) , sequencer_id : * sequencer_id , } , ) ? ; let sequence_missing = match ( min_max . min ( ) , database_wide_min_max . min ( ) ) { ( Some ( min ) , Some ( db_min ) ) => min < db_min , ( Some ( _min ) , None ) => true , _ => false , } ; if sequence_missing { warn ! ( table_name partition_key sequencer_id database_checkpoint_range database_wide_min_max partition_checkpoint_range min_max " " ) ; } if min_max . max ( ) > database_wide_min_max . max ( ) { return Err ( Error :: PartitionCheckpointMaximumAfterDatabase { partition_checkpoint_sequence_number : min_max . max ( ) , database_checkpoint_sequence_number : database_wide_min_max . max ( ) , table_name : Arc :: clone ( table_name ) , partition_key : Arc :: clone ( partition_key ) , sequencer_id : * sequencer_id , } ) ; } } } Ok ( ReplayPlan { replay_ranges , last_partition_checkpoints , } ) }
Build plan that is then used for replay.
Build plan that is then used for replay .
self
self