Make LiveFlush and Backfill variants of text and vector search flushers so that backfill doesn't block flushes (#39341)

Writes start failing if the search index flusher can't keep up, which happens sometimes when the search index flusher is blocked on building backfill segments that can take a long time. This PR separates the vector and text flushers into two flushers each - one for live flushes of the in-memory index and one for backfills. This should make it safer to backfill vector indexes on large tables.

GitOrigin-RevId: 3abf4d4a477d77dbf0c7e2f17dd17ea9817a7277

Author: emmaling27

crates/database/src/index_workers/mod.rs

@@ -31,6 +31,28 @@ pub(crate) enum BuildReason {
     VersionMismatch,
 }
 
+/// There are two types of search index flushers: live flushers and backfill
+/// flushers. Live flushers are responsible for flushing the in-memory search
+/// index contents when they get too large or old. Backfill flushers are
+/// responsible for backfilling newly added indexes or indexes on the wrong
+/// version. We separate the flushers so that building backfill segments (which
+/// can take a long time) does not block the in-memory index flushes. In-memory
+/// flushes need to happen quickly or else writes fail when the in-memory
+/// index gets too large.
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub enum FlusherType {
+    LiveFlush,
+    Backfill,
+}
+impl From<BuildReason> for FlusherType {
+    fn from(reason: BuildReason) -> Self {
+        match reason {
+            BuildReason::Backfilling | BuildReason::VersionMismatch => FlusherType::Backfill,
+            BuildReason::TooOld | BuildReason::TooLarge => FlusherType::LiveFlush,
+        }
+    }
+}
+
 impl BuildReason {
     pub fn read_max_pages_per_second(&self) -> NonZeroU32 {
         match self {

crates/database/src/index_workers/search_flusher.rs

@@ -74,6 +74,7 @@ use crate::{
             SearchIndexWriteResult,
         },
         BuildReason,
+        FlusherType,
         MultiSegmentBackfillResult,
     },
     metrics::{
@@ -95,6 +96,7 @@ pub struct SearchFlusher<RT: Runtime, T: SearchIndex> {
     params: Params<RT, T>,
     writer: SearchIndexMetadataWriter<RT, T>,
     _config: PhantomData<T>,
+    flusher_type: FlusherType,
 }
 
 impl<RT: Runtime, T: SearchIndex> Deref for SearchFlusher<RT, T> {
@@ -144,6 +146,7 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
         limits: SearchIndexLimits,
         writer: SearchIndexMetadataWriter<RT, T>,
         build_args: T::BuildIndexArgs,
+        flusher_type: FlusherType,
     ) -> Self {
         Self {
             params: Params {
@@ -156,6 +159,7 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
             },
             writer,
             _config: PhantomData,
+            flusher_type,
         }
     }
 
@@ -306,6 +310,14 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
                 },
             };
             if let Some(build_reason) = needs_backfill {
+                if FlusherType::from(build_reason) != self.flusher_type {
+                    tracing::info!(
+                        "Skipping build for index {name} with id {index_id} and {build_reason:?} \
+                         because it is a {:?} flusher",
+                        self.flusher_type
+                    );
+                    continue;
+                }
                 tracing::info!(
                     "Queueing {} index for rebuild: {name:?} ({build_reason:?})",
                     self.index_type_name()

crates/database/src/index_workers/search_worker.rs

@@ -43,6 +43,7 @@ use crate::{
         search_compactor::CompactionConfig,
         timeout_with_jitter,
         writer::SearchIndexMetadataWriter,
+        FlusherType,
     },
     text_index_worker::{
         compactor::{
@@ -122,8 +123,8 @@ impl SearchIndexWorkers {
                 segment_term_metadata_fetcher: segment_term_metadata_fetcher.clone(),
             },
         );
-        let vector_flush = retry_loop_expect_occs_and_overloaded(
-            "VectorFlusher",
+        let vector_live_flush = retry_loop_expect_occs_and_overloaded(
+            "VectorLiveFlusher",
             runtime.clone(),
             database.clone(),
             // Wait a bit since vector needs time to bootstrap. Makes startup logs a bit cleaner.
@@ -136,6 +137,24 @@ impl SearchIndexWorkers {
                 reader.clone(),
                 search_storage.clone(),
                 vector_index_metadata_writer.clone(),
+                FlusherType::LiveFlush,
+            )),
+        );
+        let vector_backfill_flush = retry_loop_expect_occs_and_overloaded(
+            "VectorBackfillFlusher",
+            runtime.clone(),
+            database.clone(),
+            // Wait a bit since vector needs time to bootstrap. Makes startup logs a bit cleaner.
+            Duration::from_secs(5),
+            *INDEX_WORKERS_INITIAL_BACKOFF,
+            *SEARCH_INDEX_FLUSHER_MAX_BACKOFF,
+            SearchIndexWorker::VectorFlusher(new_vector_flusher(
+                runtime.clone(),
+                database.clone(),
+                reader.clone(),
+                search_storage.clone(),
+                vector_index_metadata_writer.clone(),
+                FlusherType::Backfill,
             )),
         );
         let vector_compact = retry_loop_expect_occs_and_overloaded(
@@ -153,22 +172,41 @@ impl SearchIndexWorkers {
                 vector_index_metadata_writer,
             )),
         );
-        let text_flusher = SearchIndexWorker::TextFlusher(new_text_flusher(
+        let text_live_flusher = SearchIndexWorker::TextFlusher(new_text_flusher(
+            runtime.clone(),
+            database.clone(),
+            reader.clone(),
+            search_storage.clone(),
+            segment_term_metadata_fetcher.clone(),
+            text_index_metadata_writer.clone(),
+            FlusherType::LiveFlush,
+        ));
+        let text_live_flush = retry_loop_expect_occs_and_overloaded(
+            "TextLiveFlusher",
+            runtime.clone(),
+            database.clone(),
+            Duration::ZERO,
+            *INDEX_WORKERS_INITIAL_BACKOFF,
+            *SEARCH_INDEX_FLUSHER_MAX_BACKOFF,
+            text_live_flusher,
+        );
+        let text_backfill_flusher = SearchIndexWorker::TextFlusher(new_text_flusher(
             runtime.clone(),
             database.clone(),
             reader,
             search_storage.clone(),
             segment_term_metadata_fetcher,
             text_index_metadata_writer.clone(),
+            FlusherType::Backfill,
         ));
-        let text_flush = retry_loop_expect_occs_and_overloaded(
-            "SearchFlusher",
+        let text_backfill_flush = retry_loop_expect_occs_and_overloaded(
+            "TextBackfillFlusher",
             runtime.clone(),
             database.clone(),
             Duration::ZERO,
             *INDEX_WORKERS_INITIAL_BACKOFF,
             *SEARCH_INDEX_FLUSHER_MAX_BACKOFF,
-            text_flusher,
+            text_backfill_flusher,
         );
 
         let text_compact = retry_loop_expect_occs_and_overloaded(
@@ -187,15 +225,20 @@ impl SearchIndexWorkers {
             )),
         );
 
-        let vector_flush_handle = runtime.spawn("vector_flush", vector_flush);
+        let vector_backfill_flush_handle =
+            runtime.spawn("vector_backfill_flush", vector_backfill_flush);
+        let vector_live_flush_handle = runtime.spawn("vector_live_flush", vector_live_flush);
         let vector_compact_handle = runtime.spawn("vector_compact", vector_compact);
-        let text_flush_handle = runtime.spawn("text_flush", text_flush);
+        let text_live_flush_handle = runtime.spawn("text_live_flush", text_live_flush);
+        let text_backfill_flush_handle = runtime.spawn("text_backfill_flush", text_backfill_flush);
         let text_compact_handle = runtime.spawn("text_compact", text_compact);
         Self {
             handles: vec![
-                vector_flush_handle,
+                vector_backfill_flush_handle,
+                vector_live_flush_handle,
                 vector_compact_handle,
-                text_flush_handle,
+                text_live_flush_handle,
+                text_backfill_flush_handle,
                 text_compact_handle,
             ],
         }

crates/database/src/tests/randomized_search_tests.rs

@@ -98,6 +98,7 @@ use crate::{
     index_workers::{
         search_compactor::CompactionConfig,
         writer::SearchIndexMetadataWriter,
+        FlusherType,
     },
     query::{
         PaginationOptions,
@@ -261,6 +262,7 @@ impl Scenario {
             self.search_storage.clone(),
             self.build_index_args.segment_term_metadata_fetcher.clone(),
             writer,
+            FlusherType::Backfill,
         );
         flusher.step().await?;
 

crates/database/src/tests/text_test_utils.rs

@@ -56,7 +56,10 @@ use value::{
 };
 
 use crate::{
-    index_workers::search_compactor::CompactionConfig,
+    index_workers::{
+        search_compactor::CompactionConfig,
+        FlusherType,
+    },
     test_helpers::{
         DbFixtures,
         DbFixturesArgs,
@@ -151,16 +154,31 @@ impl TextFixtures {
             self.storage.clone(),
             self.segment_term_metadata_fetcher.clone(),
             self.writer.clone(),
+            FlusherType::Backfill,
         )
     }
 
-    pub fn new_search_flusher(&self) -> TextIndexFlusher<TestRuntime> {
+    pub fn new_backfill_text_flusher(&self) -> TextIndexFlusher<TestRuntime> {
         self.new_search_flusher_builder().set_soft_limit(0).build()
     }
 
-    pub fn new_search_flusher_with_soft_limit(&self) -> TextIndexFlusher<TestRuntime> {
+    pub fn new_live_text_flusher(&self) -> TextIndexFlusher<TestRuntime> {
+        self.new_search_flusher_builder()
+            .set_soft_limit(0)
+            .set_live_flush()
+            .build()
+    }
+
+    pub fn new_backfill_flusher_with_soft_limit(&self) -> TextIndexFlusher<TestRuntime> {
+        self.new_search_flusher_builder()
+            .set_soft_limit(2048)
+            .build()
+    }
+
+    pub fn new_live_flusher_with_soft_limit(&self) -> TextIndexFlusher<TestRuntime> {
         self.new_search_flusher_builder()
             .set_soft_limit(2048)
+            .set_live_flush()
             .build()
     }
 
@@ -354,7 +372,7 @@ impl TextFixtures {
         pause: PauseController,
         label: &'static str,
     ) -> anyhow::Result<()> {
-        let flusher = self.new_search_flusher();
+        let flusher = self.new_live_text_flusher();
         let hold_guard = pause.hold(label);
         let flush = flusher.step();
         let compactor = self.new_compactor();

crates/database/src/tests/usage_tracking.rs

@@ -205,7 +205,7 @@ async fn vectors_in_segment_count_as_usage(rt: TestRuntime) -> anyhow::Result<()
         )
         .await;
 
-    fixtures.new_index_flusher()?.step().await?;
+    fixtures.new_live_index_flusher()?.step().await?;
 
     let storage = fixtures
         .db
@@ -231,7 +231,7 @@ async fn vector_query_counts_bandwidth(rt: TestRuntime) -> anyhow::Result<()> {
         .await?;
     add_document_vec_array(&mut tx, index_name.table(), [3f64, 4f64]).await?;
     fixtures.db.commit(tx).await?;
-    fixtures.new_index_flusher()?.step().await?;
+    fixtures.new_backfill_index_flusher()?.step().await?;
 
     let (_, usage_stats) = fixtures
         .db

crates/database/src/tests/vector_test_utils.rs

@@ -85,6 +85,7 @@ use crate::{
     index_workers::{
         search_compactor::CompactionConfig,
         search_flusher::FLUSH_RUNNING_LABEL,
+        FlusherType,
     },
     test_helpers::DbFixturesArgs,
     vector_index_worker::{
@@ -258,8 +259,18 @@ impl VectorFixtures {
         ))
     }
 
-    pub fn new_index_flusher(&self) -> anyhow::Result<VectorIndexFlusher<TestRuntime>> {
-        self.new_index_flusher_with_full_scan_threshold(*MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB)
+    pub fn new_backfill_index_flusher(&self) -> anyhow::Result<VectorIndexFlusher<TestRuntime>> {
+        self.new_index_flusher_with_full_scan_threshold(
+            *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
+            FlusherType::Backfill,
+        )
+    }
+
+    pub fn new_live_index_flusher(&self) -> anyhow::Result<VectorIndexFlusher<TestRuntime>> {
+        self.new_index_flusher_with_full_scan_threshold(
+            *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
+            FlusherType::LiveFlush,
+        )
     }
 
     pub async fn run_compaction_during_flush(&self, pause: PauseController) -> anyhow::Result<()> {
@@ -272,6 +283,7 @@ impl VectorFixtures {
             0,
             *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
             8,
+            FlusherType::LiveFlush,
         );
         let hold_guard = pause.hold(FLUSH_RUNNING_LABEL);
         let flush = flusher.step();
@@ -290,6 +302,7 @@ impl VectorFixtures {
     pub fn new_index_flusher_with_full_scan_threshold(
         &self,
         full_scan_threshold_kb: usize,
+        flusher_type: FlusherType,
     ) -> anyhow::Result<VectorIndexFlusher<TestRuntime>> {
         Ok(new_vector_flusher_for_tests(
             self.rt.clone(),
@@ -300,12 +313,14 @@ impl VectorFixtures {
             0,
             full_scan_threshold_kb,
             *VECTOR_INDEX_SIZE_SOFT_LIMIT,
+            flusher_type,
         ))
     }
 
     pub fn new_index_flusher_with_incremental_part_threshold(
         &self,
         incremental_part_threshold: usize,
+        // flusher_type: FlusherType,
     ) -> anyhow::Result<VectorIndexFlusher<TestRuntime>> {
         Ok(new_vector_flusher_for_tests(
             self.rt.clone(),
@@ -316,6 +331,7 @@ impl VectorFixtures {
             0,
             *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
             incremental_part_threshold,
+            FlusherType::Backfill,
         ))
     }
 

crates/database/src/tests/vector_tests.rs

@@ -70,6 +70,7 @@ use vector::{
 };
 
 use crate::{
+    index_workers::FlusherType,
     test_helpers::{
         vector_utils::{
             random_vector,
@@ -776,6 +777,7 @@ async fn test_index_backfill_is_incremental(rt: TestRuntime) -> anyhow::Result<(
         *VECTOR_INDEX_SIZE_SOFT_LIMIT,
         *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
         incremental_index_size,
+        FlusherType::Backfill,
     );
 
     let mut backfill_ts = None;
@@ -860,6 +862,7 @@ async fn test_incremental_backfill_with_compaction(rt: TestRuntime) -> anyhow::R
         *VECTOR_INDEX_SIZE_SOFT_LIMIT,
         *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
         incremental_index_size,
+        FlusherType::Backfill,
     );
 
     for _ in 0..num_parts {
@@ -1000,7 +1003,8 @@ async fn test_multi_segment_search_obeys_sorted_order(rt: TestRuntime) -> anyhow
             .add_document_vec_array(index_name.table(), vector)
             .await?;
         ids.push(id);
-        let worker = fixtures.new_index_flusher_with_full_scan_threshold(0)?;
+        let worker =
+            fixtures.new_index_flusher_with_full_scan_threshold(0, FlusherType::LiveFlush)?;
         let (metrics, _) = worker.step().await?;
         assert_eq!(metrics, btreemap! {resolved_index_name.clone() => 1});
     }

crates/database/src/text_index_worker/fast_forward.rs

@@ -131,7 +131,7 @@ pub mod tests {
         } = fixtures
             .insert_backfilling_text_index_with_document()
             .await?;
-        let worker = fixtures.new_search_flusher();
+        let worker = fixtures.new_backfill_text_flusher();
 
         // Backfill the index
         let (metrics, _) = worker.step().await?;