Repository: MystenLabs/sui
File: crates/sui-pg-db/src/store.rs (Rust)

Code before:
            .do_update()
            // When using `do_update`, at least one change needs to be set, so set the pipeline to itself (nothing changes).
            // `excluded` is a virtual table containing the existing row that there was a conflict with.
            .set(watermarks::pipeline.eq(excluded(watermarks::pipeline)))
            .returning((
                watermarks::epoch_hi_inclusive,
                watermarks::checkpoint_hi_inclusive,
                watermarks::tx_hi,
                watermarks::timestamp_ms_hi_inclusive,
            ))
            .get_result(self)
            .await?;

        let committer_watermark = if checkpoint_hi_inclusive < 0 {
            None
        } else {
            Some(store::CommitterWatermark {
                epoch_hi_inclusive: epoch_hi_inclusive as u64,
                checkpoint_hi_inclusive: checkpoint_hi_inclusive as u64,
                tx_hi: tx_hi as u64,
                timestamp_ms_hi_inclusive: timestamp_ms_hi_inclusive as u64,
            })
        };

        Ok(committer_watermark)
    }

    async fn committer_watermark(
        &mut self,
        pipeline_task: &str,
    ) -> anyhow::Result<Option<store::CommitterWatermark>> {
        let watermark: Option<(i64, i64, i64, i64)> = watermarks::table
            .select((
                watermarks::epoch_hi_inclusive,
                watermarks::checkpoint_hi_inclusive,
                watermarks::tx_hi,
                watermarks::timestamp_ms_hi_inclusive,
            ))
            .filter(watermarks::pipeline.eq(pipeline_task))
            .filter(watermarks::checkpoint_hi_inclusive.ge(0))
            .first(self)
            .await
            .optional()?;

        if let Some(watermark) = watermark {
            Ok(Some(store::CommitterWatermark {
                epoch_hi_inclusive: watermark.0 as u64,
                checkpoint_hi_inclusive: watermark.1 as u64,
                tx_hi: watermark.2 as u64,
                timestamp_ms_hi_inclusive: watermark.3 as u64,
            }))

Review comment (suggestion):
i think it would be nice if we could avoid store implementors from having to deal with `checkpoint_hi_inclusive < 0`, which barring negative numbers, means someone configured `--first-checkpoint 0`. In that case, it should be the same as not providing `--first-checkpoint` at all. We can then leverage the existing logic


Code after:
            // When using `do_update`, at least one change needs to be set, so set the pipeline to itself (nothing changes).
            // `excluded` is a virtual table containing the existing row that there was a conflict with.
            .set(watermarks::pipeline.eq(excluded(watermarks::pipeline)))
            .returning(watermarks::checkpoint_hi_inclusive)
            .get_result(self)
            .await?;

        Ok(Some(checkpoint_hi_inclusive as u64))
    }

    async fn committer_watermark(
        &mut self,
        pipeline_task: &str,
    ) -> anyhow::Result<Option<store::CommitterWatermark>> {
        let watermark: Option<(i64, i64, i64, i64)> = watermarks::table
            .select((
                watermarks::epoch_hi_inclusive,
                watermarks::checkpoint_hi_inclusive,
                watermarks::tx_hi,
                watermarks::timestamp_ms_hi_inclusive,
            ))
            .filter(watermarks::pipeline.eq(pipeline_task))
            .first(self)
            .await
            .optional()?;

        if let Some(watermark) = watermark {
            Ok(Some(store::CommitterWatermark {
                epoch_hi_inclusive: watermark.0 as u64,
                checkpoint_hi_inclusive: watermark.1 as u64,
                tx_hi: watermark.2 as u64,
                timestamp_ms_hi_inclusive: watermark.3 as u64,
            }))
        } else {
            Ok(None)
        }
    }

    async fn reader_watermark(
        &mut self,
        pipeline: &'static str,
    ) -> anyhow::Result<Option<store::ReaderWatermark>> {
        let watermark: Option<(i64, i64)> = watermarks::table
            .select((watermarks::checkpoint_hi_inclusive, watermarks::reader_lo))
            .filter(watermarks::pipeline.eq(pipeline))
            .first(self)
            .await
            .optional()?;

        if let Some(watermark) = watermark {
            Ok(Some(store::ReaderWatermark {