Build a single large text index in the multisegment format when migrating (#26910)

GitOrigin-RevId: e23ab1fb1bf8f154fa17bcea501c2b32ffad0c76

Author: sjudd

crates/database/src/index_workers/search_flusher.rs

@@ -356,13 +356,11 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
                 match snapshot.data {
                     // If we're on an old or unrecognized version, rebuild everything. The formats
                     // are not compatible.
-                    SnapshotData::SingleSegment(_) | SnapshotData::Unknown(_) => (
-                        vec![],
-                        MultipartBuildType::IncrementalComplete {
-                            cursor: None,
-                            backfill_snapshot_ts: new_ts,
-                        },
-                    ),
+                    // TODO(CX-6743): Make this a failure so that we do not inadvertently rebuild
+                    // indexes once we migrate to the multi segment text search index format.
+                    SnapshotData::SingleSegment(_) | SnapshotData::Unknown(_) => {
+                        (vec![], MultipartBuildType::Complete)
+                    },
                     SnapshotData::MultiSegment(ref parts) => {
                         let ts = IndexWorkerMetadataModel::new(&mut tx)
                             .get_fast_forward_ts(snapshot.ts, job.index_id)
@@ -546,6 +544,22 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
                     .boxed();
                 (documents, previous_segments)
             },
+            // TODO(CX-6743): Remove this logic, it's expensive to run and we don't want to do so
+            // inadvertently.
+            MultipartBuildType::Complete => {
+                let table_iterator = params.database.table_iterator(
+                    snapshot_ts,
+                    *VECTOR_INDEX_WORKER_PAGE_SIZE,
+                    None,
+                );
+                (
+                    table_iterator
+                        .stream_documents_in_table(*index_name.table(), by_id, None)
+                        .map_ok(|(doc, ts)| (ts, doc.id().into(), Some(doc)))
+                        .boxed(),
+                    vec![],
+                )
+            },
         };
 
         let mut mutable_previous_segments = T::download_previous_segments(
@@ -626,6 +640,9 @@ pub struct MultiSegmentBuildResult<T: SearchIndex> {
 #[derive(Clone, Copy)]
 pub enum MultipartBuildType {
     Partial(RepeatableTimestamp),
+    // TODO(CX-6743): Remove this build type, it's expensive to run and we don't want to do so
+    // inadvertently.
+    Complete,
     IncrementalComplete {
         cursor: Option<ResolvedDocumentId>,
         backfill_snapshot_ts: RepeatableTimestamp,

crates/database/src/index_workers/writer.rs

@@ -52,6 +52,7 @@ use crate::{
             IndexBuild,
             IndexBuildResult,
         },
+        BuildReason,
         MultiSegmentBackfillResult,
     },
     metrics::{
@@ -392,7 +393,10 @@ impl<RT: Runtime, T: SearchIndex> Inner<RT, T> {
         let mut tx: Transaction<RT> = self.database.begin(Identity::system()).await?;
         let metadata = Self::require_index_metadata(&mut tx, job.metadata_id).await?;
 
-        anyhow::ensure!(metadata.config.is_backfilling());
+        anyhow::ensure!(
+            metadata.config.is_backfilling()
+                || matches!(job.build_reason, BuildReason::VersionMismatch)
+        );
 
         let (developer_config, state) = T::extract_metadata(metadata)?;
 

crates/database/src/text_index_worker/flusher2.rs

@@ -151,19 +151,35 @@ pub(crate) fn new_text_flusher<RT: Runtime>(
 mod tests {
     use common::{
         bootstrap_model::index::{
-            text_index::TextIndexState,
+            text_index::{
+                TextIndexBackfillState,
+                TextIndexSnapshot,
+                TextIndexSnapshotData,
+                TextIndexState,
+                TextSnapshotVersion,
+            },
             IndexConfig,
+            IndexMetadata,
         },
         runtime::testing::TestRuntime,
-        types::TabletIndexName,
+        types::{
+            IndexName,
+            ObjectKey,
+            TabletIndexName,
+        },
     };
     use maplit::btreemap;
     use must_let::must_let;
     use value::TableNamespace;
 
-    use crate::tests::text_test_utils::{
-        IndexData,
-        TextFixtures,
+    use crate::{
+        tests::text_test_utils::{
+            backfilling_text_index,
+            IndexData,
+            TextFixtures,
+        },
+        IndexModel,
+        SystemMetadataModel,
     };
 
     #[convex_macro::test_runtime]
@@ -502,4 +518,101 @@ mod tests {
 
         Ok(())
     }
+
+    #[convex_macro::test_runtime]
+    async fn backfill_with_backfilled_single_segment_format_backfills_with_multi_segment_format(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+        let index_name = create_backfilled_single_segment_text_index(&fixtures).await?;
+
+        fixtures.add_document("cat").await?;
+
+        let mut flusher = fixtures.new_search_flusher2();
+        flusher.step().await?;
+
+        fixtures.enable_index(&index_name).await?;
+
+        let results = fixtures.search(index_name.clone(), "cat").await?;
+        assert_eq!(results.len(), 1);
+        let segments = fixtures.get_segments_metadata(index_name).await?;
+        assert_eq!(segments.len(), 1);
+        Ok(())
+    }
+
+    #[convex_macro::test_runtime]
+    async fn backfill_with_snapshotted_at_single_segment_format_backfills_with_multi_segment_format(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+
+        let index_name = create_backfilled_single_segment_text_index(&fixtures).await?;
+
+        let mut tx = fixtures.db.begin_system().await?;
+        IndexModel::new(&mut tx)
+            .enable_index_for_testing(TableNamespace::Global, &index_name)
+            .await?;
+        fixtures.db.commit(tx).await?;
+
+        fixtures.add_document("cat").await?;
+
+        let mut flusher = fixtures.new_search_flusher2();
+        flusher.step().await?;
+
+        let results = fixtures.search(index_name.clone(), "cat").await?;
+        assert_eq!(results.len(), 1);
+        let segments = fixtures.get_segments_metadata(index_name).await?;
+        assert_eq!(segments.len(), 1);
+        Ok(())
+    }
+
+    async fn create_backfilled_single_segment_text_index(
+        fixtures: &TextFixtures,
+    ) -> anyhow::Result<IndexName> {
+        let mut tx = fixtures.db.begin_system().await?;
+        let metadata = backfilling_text_index()?;
+        let on_disk_state = TextIndexState::Backfilling(TextIndexBackfillState::new());
+        must_let!(let IndexConfig::Search {
+            developer_config,
+            ..
+        } = metadata.config);
+        let doc_id = IndexModel::new(&mut tx)
+            .add_application_index(
+                TableNamespace::Global,
+                IndexMetadata::new_search_index(
+                    metadata.name.clone(),
+                    developer_config,
+                    on_disk_state,
+                ),
+            )
+            .await?;
+
+        fixtures.db.commit(tx).await?;
+        let mut tx = fixtures.db.begin_system().await?;
+        let indexes = IndexModel::new(&mut tx).get_all_indexes().await?;
+        let index = indexes
+            .into_iter()
+            .find(|index| index.id() == doc_id)
+            .unwrap();
+        let (id, value) = index.into_id_and_value();
+        must_let!(let IndexConfig::Search {
+            developer_config,
+            ..
+        } = value.config);
+        let on_disk_state = TextIndexState::Backfilled(TextIndexSnapshot {
+            data: TextIndexSnapshotData::SingleSegment(ObjectKey::try_from("Fake".to_string())?),
+            ts: *tx.begin_timestamp(),
+            version: TextSnapshotVersion::V2UseStringIds,
+        });
+
+        SystemMetadataModel::new_global(&mut tx)
+            .replace(
+                id,
+                IndexMetadata::new_search_index(value.name, developer_config, on_disk_state)
+                    .try_into()?,
+            )
+            .await?;
+        fixtures.db.commit(tx).await?;
+        Ok(metadata.name)
+    }
 }