Handle changing term values due to replaced documents in the log (#27102)

When we iterate over the document log, we may encounter a document whose most recent previous revision does not match the version of that document that's in a previous text search segment. We need these to match because we update statistics based on the contents of the document. If the document terms we use to update statistics don't match the document in the previous segment, lots of things break.

To make this work, rather than taking the first previous revision of a document we find, we now look for the first previous revision of a document that's outside the timestamp bounds of the new segment. If the previous revision is within our new segment's bound, we skip processing it. As we continue backwards down the document log, we will then either find a pure add within our segment's time bounds (which we'll skip), or we eventually find a previous revision outside of our time bounds that we can use for the update.

GitOrigin-RevId: 7eb7a742fb77b5a85adcc6bea70f4c2b56adbc29

Author: sjudd

crates/database/src/bootstrap_model/test_facing.rs

@@ -34,6 +34,17 @@ impl<'a, RT: Runtime> TestFacingModel<'a, RT> {
             .await
     }
 
+    #[convex_macro::instrument_future]
+    pub async fn replace(
+        &mut self,
+        id: ResolvedDocumentId,
+        value: ConvexObject,
+    ) -> anyhow::Result<ResolvedDocument> {
+        SystemMetadataModel::new(self.tx, TableNamespace::test_user())
+            .replace(id, value)
+            .await
+    }
+
     /// Insert a new document and immediately read it. Prefer using `insert`
     /// unless you need to read the creation time.
     #[convex_macro::instrument_future]

crates/database/src/index_workers/index_meta.rs

@@ -107,6 +107,7 @@ pub trait SearchIndex: Clone + Debug {
         documents: DocumentStream<'_>,
         reader: RepeatablePersistence,
         previous_segments: &mut Self::PreviousSegments,
+        document_log_lower_bound: Option<Timestamp>,
         build_index_args: Self::BuildIndexArgs,
         multipart_build_type: MultipartBuildType,
     ) -> anyhow::Result<Option<Self::NewSegment>>;

crates/database/src/index_workers/search_flusher.rs

@@ -491,19 +491,23 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
         let mut is_size_exceeded = false;
         let qdrant_schema = T::new_schema(&developer_config);
 
+        let mut lower_bound_ts: Option<Timestamp> = None;
         let (documents, previous_segments) = match build_type {
-            MultipartBuildType::Partial(last_ts) => (
-                params.database.load_documents_in_table(
-                    *index_name.table(),
-                    TimestampRange::new((
-                        Bound::Excluded(*last_ts),
-                        Bound::Included(*snapshot_ts),
-                    ))?,
-                    T::partial_document_order(),
-                    &row_rate_limiter,
-                ),
-                previous_segments,
-            ),
+            MultipartBuildType::Partial(last_ts) => {
+                lower_bound_ts = Some(*last_ts);
+                (
+                    params.database.load_documents_in_table(
+                        *index_name.table(),
+                        TimestampRange::new((
+                            Bound::Excluded(*last_ts),
+                            Bound::Included(*snapshot_ts),
+                        ))?,
+                        T::partial_document_order(),
+                        &row_rate_limiter,
+                    ),
+                    previous_segments,
+                )
+            },
             MultipartBuildType::IncrementalComplete {
                 cursor,
                 backfill_snapshot_ts,
@@ -584,6 +588,7 @@ impl<RT: Runtime, T: SearchIndex + 'static> SearchFlusher<RT, T> {
             documents,
             persistence,
             &mut mutable_previous_segments,
+            lower_bound_ts,
             build_index_args,
             build_type,
         )

crates/database/src/tests/text_test_utils.rs

@@ -269,6 +269,21 @@ impl TextFixtures {
         Ok(segments.clone())
     }
 
+    pub async fn replace_document(
+        &self,
+        doc_id: ResolvedDocumentId,
+        text: &str,
+    ) -> anyhow::Result<()> {
+        let mut tx = self.db.begin_system().await?;
+        let document = assert_obj!(
+            "text" => text,
+            "channel" => "#general",
+        );
+        tx.replace_inner(doc_id, document).await?;
+        self.db.commit(tx).await?;
+        Ok(())
+    }
+
     pub async fn add_document(&self, text: &str) -> anyhow::Result<ResolvedDocumentId> {
         let table_name = TABLE_NAME.parse::<TableName>()?;
         let mut tx = self.db.begin_system().await?;

crates/database/src/text_index_worker/flusher2.rs

@@ -519,6 +519,115 @@ mod tests {
         Ok(())
     }
 
+    #[convex_macro::test_runtime]
+    async fn backfill_insert_then_replace_delete_one_segment_doesnt_panic(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+        let IndexData { index_name, .. } = fixtures.insert_backfilling_text_index().await?;
+        let mut flusher = fixtures.new_search_flusher2();
+
+        let doc_id = fixtures.add_document("cat").await?;
+        fixtures.replace_document(doc_id, "new_text").await?;
+        let mut tx = fixtures.db.begin_system().await?;
+        tx.delete_inner(doc_id).await?;
+        fixtures.db.commit(tx).await?;
+
+        flusher.step().await?;
+        fixtures.enable_index(&index_name).await?;
+
+        let results = fixtures.search(index_name.clone(), "cat").await?;
+        assert!(results.is_empty());
+        let results = fixtures.search(index_name, "new_text").await?;
+        assert!(results.is_empty());
+
+        Ok(())
+    }
+
+    #[convex_macro::test_runtime]
+    async fn backfill_insert_then_replace_delete_second_segment_doesnt_panic(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+        let IndexData { index_name, .. } = fixtures.insert_backfilling_text_index().await?;
+        let mut flusher = fixtures.new_search_flusher2();
+
+        let doc_id = fixtures.add_document("cat").await?;
+        flusher.step().await?;
+
+        fixtures.replace_document(doc_id, "new_text").await?;
+
+        let mut tx = fixtures.db.begin_system().await?;
+        tx.delete_inner(doc_id).await?;
+        fixtures.db.commit(tx).await?;
+
+        flusher.step().await?;
+        fixtures.enable_index(&index_name).await?;
+
+        let results = fixtures.search(index_name.clone(), "cat").await?;
+        assert!(results.is_empty());
+        let results = fixtures.search(index_name, "new_text").await?;
+        assert!(results.is_empty());
+
+        Ok(())
+    }
+
+    #[convex_macro::test_runtime]
+    async fn backfill_insert_replace_replace_delete_doesnt_panic(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+        let IndexData { index_name, .. } = fixtures.insert_backfilling_text_index().await?;
+        let mut flusher = fixtures.new_search_flusher2();
+
+        let doc_id = fixtures.add_document("cat").await?;
+        flusher.step().await?;
+
+        fixtures.replace_document(doc_id, "new_text").await?;
+        fixtures.replace_document(doc_id, "really_new_text").await?;
+
+        let mut tx = fixtures.db.begin_system().await?;
+        tx.delete_inner(doc_id).await?;
+        fixtures.db.commit(tx).await?;
+
+        flusher.step().await?;
+        fixtures.enable_index(&index_name).await?;
+
+        let results = fixtures.search(index_name, "really_new_text").await?;
+        assert!(results.is_empty());
+
+        Ok(())
+    }
+
+    #[convex_macro::test_runtime]
+    async fn backfill_insert_replace_replace_delete_different_segments_doesnt_panic(
+        rt: TestRuntime,
+    ) -> anyhow::Result<()> {
+        let fixtures = TextFixtures::new(rt).await?;
+        let IndexData { index_name, .. } = fixtures.insert_backfilling_text_index().await?;
+        let mut flusher = fixtures.new_search_flusher2();
+
+        let doc_id = fixtures.add_document("cat").await?;
+        flusher.step().await?;
+
+        fixtures.replace_document(doc_id, "new_text").await?;
+        flusher.step().await?;
+        fixtures.replace_document(doc_id, "really_new_text").await?;
+        flusher.step().await?;
+
+        let mut tx = fixtures.db.begin_system().await?;
+        tx.delete_inner(doc_id).await?;
+        fixtures.db.commit(tx).await?;
+
+        flusher.step().await?;
+        fixtures.enable_index(&index_name).await?;
+
+        let results = fixtures.search(index_name, "really_new_text").await?;
+        assert!(results.is_empty());
+
+        Ok(())
+    }
+
     #[convex_macro::test_runtime]
     async fn backfill_with_backfilled_single_segment_format_backfills_with_multi_segment_format(
         rt: TestRuntime,

crates/database/src/text_index_worker/text_meta.rs

@@ -56,6 +56,7 @@ use search::{
     UpdatableTextSegment,
 };
 use storage::Storage;
+use sync_types::Timestamp;
 use value::InternalId;
 
 use crate::{
@@ -218,6 +219,7 @@ impl SearchIndex for TextSearchIndex {
         documents: DocumentStream<'_>,
         reader: RepeatablePersistence,
         previous_segments: &mut Self::PreviousSegments,
+        lower_bound_ts: Option<Timestamp>,
         BuildTextIndexArgs {
             search_storage,
             segment_term_metadata_fetcher,
@@ -254,6 +256,7 @@ impl SearchIndex for TextSearchIndex {
             previous_segments,
             segment_term_metadata_fetcher,
             search_storage,
+            lower_bound_ts,
         )
         .await
     }

crates/database/src/vector_index_worker/vector_meta.rs

@@ -42,6 +42,7 @@ use search::{
     Searcher,
 };
 use storage::Storage;
+use sync_types::Timestamp;
 use value::InternalId;
 use vector::{
     qdrant_segments::VectorDiskSegmentValues,
@@ -221,6 +222,7 @@ impl SearchIndex for VectorSearchIndex {
         documents: DocumentStream<'_>,
         _reader: RepeatablePersistence,
         previous_segments: &mut Self::PreviousSegments,
+        _document_log_lower_bound: Option<Timestamp>,
         BuildVectorIndexArgs {
             full_scan_threshold_bytes,
         }: Self::BuildIndexArgs,