Remove backfill_backwards in db cluster migration and make retention ts the snapshot ts (#41783)

Removes `backfill_backwards` algorithm from IndexWriter

GitOrigin-RevId: f7760fe6fc5d0b4cfbce2e640218c4ff4173a631

Author: emmaling27

crates/common/src/testing/persistence_test_suite.rs

@@ -21,11 +21,13 @@ use proptest::collection::size_range;
 use serde_json::json;
 use value::{
     assert_val,
+    sha256::Sha256,
     val,
     ConvexObject,
     ConvexValue,
     DeveloperDocumentId,
     InternalDocumentId,
+    InternalId,
     ResolvedDocumentId,
     TableMapping,
     TabletId,
@@ -41,7 +43,10 @@ use crate::{
         CreationTime,
         ResolvedDocument,
     },
-    index::IndexKey,
+    index::{
+        IndexEntry,
+        IndexKey,
+    },
     interval::{
         BinaryKey,
         End,
@@ -1671,6 +1676,19 @@ pub async fn persistence_enforce_retention<P: Persistence>(p: Arc<P>) -> anyhow:
     let results: Vec<_> = stream.try_collect::<Vec<_>>().await?;
     assert_eq!(results, vec![]);
 
+    // delete_index_entries no-ops when trying to delete nonexistent indexes.
+    let deleted = p
+        .delete_index_entries(vec![IndexEntry {
+            key_prefix: vec![],
+            key_suffix: None,
+            key_sha256: Sha256::new().finalize().to_vec(),
+            deleted: false,
+            index_id: InternalId([1; 16]),
+            ts: Timestamp::must(123),
+        }])
+        .await?;
+    assert_eq!(deleted, 0);
+
     Ok(())
 }
 

crates/common/src/testing/test_persistence.rs

@@ -181,8 +181,7 @@ impl Persistence for TestPersistence {
         for expired_row in expired_rows {
             if index
                 .get_mut(&expired_row.index_id)
-                .unwrap()
-                .remove(&(IndexKeyBytes(expired_row.key_prefix), expired_row.ts))
+                .and_then(|ix| ix.remove(&(IndexKeyBytes(expired_row.key_prefix), expired_row.ts)))
                 .is_some()
             {
                 total_deleted += 1;

crates/database/src/database_index_workers/index_writer.rs

@@ -192,87 +192,6 @@ impl<RT: Runtime> IndexWriter<RT> {
         }
     }
 
-    /// Backfill in two steps: first a snapshot at the current time, and then
-    /// walking the log. After the current snapshot is backfilled, index
-    /// snapshot reads at >=ts are valid. The subsequent walking of the log
-    /// extends the earliest allowed snapshot into the past.
-    ///
-    /// The goal of this backfill is to make snapshot reads of `index_name`
-    /// valid between the range of [retention_cutoff, snapshot_ts].
-    /// To support:
-    /// 1. Latest documents written before retention_cutoff. They are still
-    ///    latest at `ts = snapshot_ts`, so we compute and write index entries
-    ///    when we walk the `snapshot_ts` snapshot.
-    /// 2. Document changes between retention_cutoff and `snapshot_ts`. These
-    ///    are handled by walking the documents log for this range in reverse
-    ///    and creating index entries. When walking the documents log we start
-    ///    at `snapshot_ts`.
-    /// 3. Documents that were latest as of retention_cutoff but were
-    ///    overwritten before `snapshot_ts`. These are handled when walking the
-    ///    documents log and finding an overwrite.
-    /// 4. Document changes after `snapshot_ts`. These are handled by active
-    ///    writes, assuming `snapshot_ts` is after the index was created. If
-    ///    there are no active writes, then `backfill_forwards` must be called
-    ///    with a timestamp <= `snapshot_ts`.
-    ///
-    /// Takes a an optional database to update progress on the index backfill
-    pub async fn backfill_from_retention_cutoff(
-        &self,
-        snapshot_ts: RepeatableTimestamp,
-        index_metadata: &IndexRegistry,
-        index_selector: IndexSelector,
-        concurrency: usize,
-    ) -> anyhow::Result<()> {
-        let pause_client = self.runtime.pause_client();
-        pause_client.wait(PERFORM_BACKFILL_LABEL).await;
-        // Backfill in two steps: first create index entries for all latest documents,
-        // then create index entries for all documents in the retention range.
-
-        stream::iter(index_selector.iterate_tables().map(Ok))
-            .try_for_each_concurrent(concurrency, |table_id| {
-                let index_metadata = index_metadata.clone();
-                let index_selector = index_selector.clone();
-                let self_ = (*self).clone();
-                try_join("index_backfill_table_snapshot", async move {
-                    self_
-                        .backfill_exact_snapshot_of_table(
-                            snapshot_ts,
-                            &index_selector,
-                            &index_metadata,
-                            table_id,
-                            None,
-                        )
-                        .await
-                        .map(|_docs_indexed| ())
-                })
-            })
-            .await?;
-
-        let mut min_backfilled_ts = snapshot_ts;
-
-        // Retry until min_snapshot_ts passes min_backfilled_ts, at which point we
-        // have backfilled the full range of snapshots within retention.
-        loop {
-            let min_snapshot_ts = self.retention_validator.min_snapshot_ts().await?;
-            if min_snapshot_ts >= min_backfilled_ts {
-                break;
-            }
-            // NOTE: ordering Desc is important, to keep the range of valid snapshots
-            // contiguous. If we backfilled in order Asc, then we might see a
-            // document creation before its tombstone, and that document would be
-            // visible at snapshots where it should be deleted.
-            min_backfilled_ts = self
-                .backfill_backwards(
-                    min_backfilled_ts,
-                    *min_snapshot_ts,
-                    index_metadata,
-                    &index_selector,
-                )
-                .await?;
-        }
-        Ok(())
-    }
-
     /// Backfill indexes based on a snapshot at the current time.  After the
     /// current snapshot is backfilled, index snapshot reads at >=ts are
     /// valid.
@@ -384,7 +303,6 @@ impl<RT: Runtime> IndexWriter<RT> {
             index_registry,
             ReceiverStream::new(index_update_rx),
             index_selector,
-            false, /* deduplicate */
         );
         let (docs_indexed, _) = future::try_join(producer, consumer).await?;
         Ok(docs_indexed)
@@ -402,7 +320,7 @@ impl<RT: Runtime> IndexWriter<RT> {
     /// - `index_selector`: Subset of `index_registry` to backfill.
     ///
     /// Preconditions:
-    /// - The selected indexes are fully backfilled for all revisions less than
+    /// - The selected indexes are fully backfilled for all revisions at
     ///   `start_ts`.
     ///
     /// Postconditions:
@@ -439,7 +357,6 @@ impl<RT: Runtime> IndexWriter<RT> {
             index_registry,
             ReceiverStream::new(rx),
             index_selector,
-            false, /* deduplicate */
         );
 
         // Consider ourselves successful if both the producer and consumer exit
@@ -553,7 +470,6 @@ impl<RT: Runtime> IndexWriter<RT> {
             index_registry,
             ReceiverStream::new(rx),
             index_selector,
-            true, /* deduplicate */
         );
 
         // Consider ourselves successful if both the reader and writer exit
@@ -568,7 +484,6 @@ impl<RT: Runtime> IndexWriter<RT> {
         index_registry: &IndexRegistry,
         revision_pairs: impl Stream<Item = RevisionPair>,
         index_selector: &IndexSelector,
-        deduplicate: bool,
     ) -> anyhow::Result<()> {
         let mut last_logged = self.runtime.system_time();
         let mut last_checkpointed = self.runtime.system_time();
@@ -591,7 +506,7 @@ impl<RT: Runtime> IndexWriter<RT> {
                 let cursor = should_send_progress
                     .then(|| chunk.last().map(|revision_pair| revision_pair.id))
                     .flatten();
-                let mut index_updates: Vec<PersistenceIndexEntry> = chunk
+                let index_updates: Vec<PersistenceIndexEntry> = chunk
                     .iter()
                     .flat_map(|revision_pair| {
                         index_registry
@@ -618,12 +533,6 @@ impl<RT: Runtime> IndexWriter<RT> {
                     let delay = not_until.wait_time_from(self.runtime.monotonic_now().into());
                     self.runtime.wait(delay).await;
                 }
-                if deduplicate {
-                    // There might be duplicates in the index entries from the
-                    // backfill backwards algorithm that need to be deduped.
-                    index_updates.sort_unstable();
-                    index_updates.dedup();
-                }
                 persistence
                     .write(&[], &index_updates, ConflictStrategy::Overwrite)
                     .await?;