Skip backfilling to retention cutoff in IndexWorker (#41756)

Changes `perform_backfill` into two functions `backfill_from_ts` and `backfill_from_retention_cutoff`. The index worker should never need to backfill all the way to retention cutoff because new indexes don't get queried until after they are committed. I'll follow up to modify db cluster migration to allow us to get rid of `backfill_from_retention_cutoff` completely.

GitOrigin-RevId: 789db6ecbb4186a383b18fc95f3ee99d4dc9db3c

Author: emmaling27

crates/database/src/database_index_workers/index_writer.rs

@@ -207,7 +207,7 @@ impl<RT: Runtime> IndexWriter<RT> {
     ///    with a timestamp <= `snapshot_ts`.
     ///
     /// Takes a an optional database to update progress on the index backfill
-    pub async fn perform_backfill(
+    pub async fn backfill_from_retention_cutoff(
         &self,
         snapshot_ts: RepeatableTimestamp,
         index_metadata: &IndexRegistry,
@@ -265,6 +265,52 @@ impl<RT: Runtime> IndexWriter<RT> {
         Ok(())
     }
 
+    /// Backfill indexes based on a snapshot at the current time.  After the
+    /// current snapshot is backfilled, index snapshot reads at >=ts are
+    /// valid.
+    ///
+    /// The goal of this backfill is to make snapshot reads of `index_name`
+    /// valid at or after snapshot_ts.
+    /// To support:
+    /// 1. Latest documents as of snapshot_ts.
+    /// 2. 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_ts(
+        &self,
+        snapshot_ts: RepeatableTimestamp,
+        index_metadata: &IndexRegistry,
+        index_selector: IndexSelector,
+        concurrency: usize,
+        database: Option<Database<RT>>,
+    ) -> anyhow::Result<()> {
+        let pause_client = self.runtime.pause_client();
+        pause_client.wait(PERFORM_BACKFILL_LABEL).await;
+        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 database = database.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,
+                            database,
+                        )
+                        .await
+                })
+            })
+            .await?;
+        Ok(())
+    }
+
     /// Backfills exactly the index entries necessary to represent documents
     /// which were latest at `snapshot`. In particular it does not create any
     /// tombstone index entries. And it only does snapshot reads (of `by_id`) at

crates/database/src/database_index_workers/mod.rs

@@ -334,7 +334,7 @@ impl<RT: Runtime> IndexWorker<RT> {
             };
             let index_registry = snapshot.index_registry;
             index_writer
-                .perform_backfill(
+                .backfill_from_ts(
                     ts,
                     &index_registry,
                     index_selector,

crates/database/src/tests/mod.rs

@@ -1916,7 +1916,7 @@ async fn test_index_write(rt: TestRuntime) -> anyhow::Result<()> {
     .await?;
     let index_metadata = database_snapshot.index_registry().clone();
     index_writer
-        .perform_backfill(
+        .backfill_from_ts(
             unchecked_repeatable_ts(ts),
             &index_metadata,
             IndexSelector::All(index_metadata.clone()),