Thread prev_ts through database impls (#33277)

1. Merge DatabaseDocumentUpdate into DatabaseLogEntry, so that the same type is inserted into and queried from documents
2. Also modify `fn previous_revisions` to return DatabaseLogEntry
3. Thread prev_ts through all the SQL queries for sqlite/mysql/postgres.

GitOrigin-RevId: 75040afced34435ee6915f571296acd0c18f6334

Author: goffrie

crates/common/src/persistence.rs

@@ -50,7 +50,13 @@ use crate::{
     },
 };
 
-pub type DocumentLogEntry = (Timestamp, InternalDocumentId, Option<ResolvedDocument>);
+#[derive(Debug, Clone, PartialEq)]
+pub struct DocumentLogEntry {
+    pub ts: Timestamp,
+    pub id: InternalDocumentId,
+    pub value: Option<ResolvedDocument>,
+    pub prev_ts: Option<Timestamp>,
+}
 
 pub type DocumentStream<'a> = BoxStream<'a, anyhow::Result<DocumentLogEntry>>;
 
@@ -161,14 +167,6 @@ impl PersistenceGlobalKey {
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
-pub struct DatabaseDocumentUpdate {
-    pub ts: Timestamp,
-    pub id: InternalDocumentId,
-    pub value: Option<ResolvedDocument>,
-    pub prev_ts: Option<Timestamp>,
-}
-
 #[async_trait]
 pub trait Persistence: Sync + Send + 'static {
     /// Whether the persistence layer is freshely created or not.
@@ -179,7 +177,7 @@ pub trait Persistence: Sync + Send + 'static {
     /// Writes documents and the respective derived indexes.
     async fn write(
         &self,
-        documents: Vec<DatabaseDocumentUpdate>,
+        documents: Vec<DocumentLogEntry>,
         indexes: BTreeSet<(Timestamp, DatabaseIndexUpdate)>,
         conflict_strategy: ConflictStrategy,
     ) -> anyhow::Result<()>;
@@ -338,7 +336,7 @@ pub trait PersistenceReader: Send + Sync + 'static {
         retention_validator: Arc<dyn RetentionValidator>,
     ) -> DocumentStream<'_> {
         self.load_documents(range, order, page_size, retention_validator)
-            .try_filter(move |(_, doc_id, _)| future::ready(doc_id.table() == tablet_id))
+            .try_filter(move |doc| future::ready(doc.id.table() == tablet_id))
             .boxed()
     }
 
@@ -352,9 +350,7 @@ pub trait PersistenceReader: Send + Sync + 'static {
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
         retention_validator: Arc<dyn RetentionValidator>,
-    ) -> anyhow::Result<
-        BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
-    >;
+    ) -> anyhow::Result<BTreeMap<(InternalDocumentId, Timestamp), DocumentLogEntry>>;
 
     /// Loads documentIds with respective timestamps that match the
     /// index query criteria.
@@ -431,7 +427,7 @@ pub trait PersistenceReader: Send + Sync + 'static {
         let max_repeatable =
             self.get_persistence_global(PersistenceGlobalKey::MaxRepeatableTimestamp);
         let (max_committed, max_repeatable) = try_join!(stream.try_next(), max_repeatable)?;
-        let max_committed_ts = max_committed.map(|(ts, ..)| ts);
+        let max_committed_ts = max_committed.map(|entry| entry.ts);
         let max_repeatable_ts = max_repeatable.map(Timestamp::try_from).transpose()?;
         let max_ts = cmp::max(max_committed_ts, max_repeatable_ts); // note None < Some
         Ok(max_ts)
@@ -525,7 +521,7 @@ impl RepeatablePersistence {
             *DEFAULT_DOCUMENTS_PAGE_SIZE,
             self.retention_validator.clone(),
         );
-        Box::pin(stream.try_filter(|(ts, ..)| future::ready(*ts <= *self.upper_bound)))
+        Box::pin(stream.try_filter(|entry| future::ready(entry.ts <= *self.upper_bound)))
     }
 
     /// Same as `load_documents` but doesn't use the `RetentionValidator` from
@@ -543,15 +539,13 @@ impl RepeatablePersistence {
             *DEFAULT_DOCUMENTS_PAGE_SIZE,
             retention_validator,
         );
-        Box::pin(stream.try_filter(|(ts, ..)| future::ready(*ts <= *self.upper_bound)))
+        Box::pin(stream.try_filter(|entry| future::ready(entry.ts <= *self.upper_bound)))
     }
 
     pub async fn previous_revisions(
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
-    ) -> anyhow::Result<
-        BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
-    > {
+    ) -> anyhow::Result<BTreeMap<(InternalDocumentId, Timestamp), DocumentLogEntry>> {
         for (_, ts) in &ids {
             // Reading documents <ts, so ts-1 needs to be repeatable.
             anyhow::ensure!(*ts <= self.upper_bound.succ()?);
@@ -566,9 +560,7 @@ impl RepeatablePersistence {
         &self,
         ids: BTreeSet<(InternalDocumentId, Timestamp)>,
         retention_validator: Arc<dyn RetentionValidator>,
-    ) -> anyhow::Result<
-        BTreeMap<(InternalDocumentId, Timestamp), (Timestamp, Option<ResolvedDocument>)>,
-    > {
+    ) -> anyhow::Result<BTreeMap<(InternalDocumentId, Timestamp), DocumentLogEntry>> {
         for (_, ts) in &ids {
             // Reading documents <ts, so ts-1 needs to be repeatable.
             anyhow::ensure!(*ts <= self.upper_bound.succ()?);

crates/common/src/persistence_helpers.rs

@@ -9,7 +9,10 @@ use crate::{
     comparators::AsComparator,
     document::ResolvedDocument,
     knobs::DOCUMENTS_IN_MEMORY,
-    persistence::RepeatablePersistence,
+    persistence::{
+        DocumentLogEntry,
+        RepeatablePersistence,
+    },
     try_chunks::TryChunksExt,
     types::Timestamp,
 };
@@ -41,8 +44,7 @@ impl RevisionPair {
     }
 }
 
-type RevisionStreamEntry =
-    anyhow::Result<(Timestamp, InternalDocumentId, Option<ResolvedDocument>)>;
+type RevisionStreamEntry = anyhow::Result<DocumentLogEntry>;
 
 #[allow(clippy::needless_lifetimes)]
 #[try_stream(ok = RevisionPair, error = anyhow::Error)]
@@ -54,16 +56,26 @@ pub async fn stream_revision_pairs<'a>(
     futures::pin_mut!(documents);
 
     while let Some(read_chunk) = documents.try_next().await? {
-        let ids = read_chunk.iter().map(|(ts, id, _)| (*id, *ts)).collect();
+        // TODO: use prev_ts when it is available
+        let ids = read_chunk
+            .iter()
+            .map(|entry| (entry.id, entry.ts))
+            .collect();
         let mut prev_revs = reader.previous_revisions(ids).await?;
-        for (ts, id, document) in read_chunk {
+        for DocumentLogEntry {
+            ts,
+            id,
+            value: document,
+            ..
+        } in read_chunk
+        {
             let rev = DocumentRevision { ts, document };
             let prev_rev =
                 prev_revs
                     .remove((&id, &ts).as_comparator())
-                    .map(|(prev_ts, prev_document)| DocumentRevision {
-                        ts: prev_ts,
-                        document: prev_document,
+                    .map(|entry| DocumentRevision {
+                        ts: entry.ts,
+                        document: entry.value,
                     });
             yield RevisionPair { id, rev, prev_rev };
         }