Merge deletes correctly for text indexes (#27302)

This PR fixes a bug where we were not updating the deleted terms metadata when reprocessing deletes that happened in a race with a compaction. I added a `merge_deletes` method to the search index trait and had to thread through repeatable persistence to look up previous revisions for text search. It's a shame that we have to re-walk the document log and previous revisions in this race, but it also seems complicated to store the deletes and try to remap them to the right segments because we don't know how many compactions occur while we're writing the new segment to disk.

GitOrigin-RevId: e7df528990f8d36bd25b0be11a6b841e192e55ce

Author: emmaling27

crates/common/src/bootstrap_model/index/developer_index_config.rs

@@ -12,8 +12,8 @@ use super::{
         SerializedDeveloperDatabaseIndexConfig,
     },
     text_index::{
-        DeveloperSearchIndexConfig,
-        SerializedDeveloperSearchIndexConfig,
+        DeveloperTextIndexConfig,
+        SerializedDeveloperTextIndexConfig,
     },
     vector_index::{
         DeveloperVectorIndexConfig,
@@ -30,7 +30,7 @@ pub enum DeveloperIndexConfig {
     Database(DeveloperDatabaseIndexConfig),
 
     /// Full text search index.
-    Search(DeveloperSearchIndexConfig),
+    Search(DeveloperTextIndexConfig),
 
     Vector(DeveloperVectorIndexConfig),
 }
@@ -60,7 +60,7 @@ enum SerializedDeveloperIndexConfig {
     },
     Search {
         #[serde(flatten)]
-        config: SerializedDeveloperSearchIndexConfig,
+        config: SerializedDeveloperTextIndexConfig,
     },
     Vector {
         #[serde(flatten)]

crates/common/src/bootstrap_model/index/index_config.rs

@@ -14,8 +14,8 @@ use super::{
         SerializedDeveloperDatabaseIndexConfig,
     },
     text_index::{
-        DeveloperSearchIndexConfig,
-        SerializedDeveloperSearchIndexConfig,
+        DeveloperTextIndexConfig,
+        SerializedDeveloperTextIndexConfig,
         SerializedTextIndexState,
         TextIndexState,
     },
@@ -42,7 +42,7 @@ pub enum IndexConfig {
 
     /// Full text search index.
     Search {
-        developer_config: DeveloperSearchIndexConfig,
+        developer_config: DeveloperTextIndexConfig,
 
         /// Whether the index is fully backfilled or not on disk.
         on_disk_state: TextIndexState,
@@ -165,7 +165,7 @@ pub enum SerializedIndexConfig {
     #[serde(rename_all = "camelCase")]
     Search {
         #[serde(flatten)]
-        developer_config: SerializedDeveloperSearchIndexConfig,
+        developer_config: SerializedDeveloperTextIndexConfig,
         on_disk_state: SerializedTextIndexState,
     },
     #[serde(rename_all = "camelCase")]

crates/common/src/bootstrap_model/index/index_metadata.rs

@@ -31,7 +31,7 @@ use super::{
 };
 use crate::{
     bootstrap_model::index::text_index::{
-        DeveloperSearchIndexConfig,
+        DeveloperTextIndexConfig,
         TextIndexBackfillState,
         TextIndexState,
     },
@@ -85,7 +85,7 @@ impl<T: IndexTableIdentifier> IndexMetadata<T> {
     ) -> Self {
         Self::new_search_index(
             name,
-            DeveloperSearchIndexConfig {
+            DeveloperTextIndexConfig {
                 search_field,
                 filter_fields,
             },
@@ -118,7 +118,7 @@ impl<T: IndexTableIdentifier> IndexMetadata<T> {
 
     pub fn new_search_index(
         name: GenericIndexName<T>,
-        developer_config: DeveloperSearchIndexConfig,
+        developer_config: DeveloperTextIndexConfig,
         on_disk_state: TextIndexState,
     ) -> Self {
         Self {

crates/common/src/bootstrap_model/index/text_index/index_config.rs

@@ -10,7 +10,7 @@ use crate::paths::FieldPath;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 #[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
-pub struct DeveloperSearchIndexConfig {
+pub struct DeveloperTextIndexConfig {
     /// The field to index for full text search.
     pub search_field: FieldPath,
 
@@ -20,26 +20,26 @@ pub struct DeveloperSearchIndexConfig {
 
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct SerializedDeveloperSearchIndexConfig {
+pub struct SerializedDeveloperTextIndexConfig {
     search_field: String,
     filter_fields: Vec<String>,
 }
 
-impl TryFrom<DeveloperSearchIndexConfig> for SerializedDeveloperSearchIndexConfig {
+impl TryFrom<DeveloperTextIndexConfig> for SerializedDeveloperTextIndexConfig {
     type Error = anyhow::Error;
 
-    fn try_from(config: DeveloperSearchIndexConfig) -> anyhow::Result<Self> {
+    fn try_from(config: DeveloperTextIndexConfig) -> anyhow::Result<Self> {
         Ok(Self {
             search_field: config.search_field.into(),
             filter_fields: config.filter_fields.into_iter().map(String::from).collect(),
         })
     }
 }
 
-impl TryFrom<SerializedDeveloperSearchIndexConfig> for DeveloperSearchIndexConfig {
+impl TryFrom<SerializedDeveloperTextIndexConfig> for DeveloperTextIndexConfig {
     type Error = anyhow::Error;
 
-    fn try_from(config: SerializedDeveloperSearchIndexConfig) -> anyhow::Result<Self> {
+    fn try_from(config: SerializedDeveloperTextIndexConfig) -> anyhow::Result<Self> {
         Ok(Self {
             search_field: config.search_field.parse()?,
             filter_fields: config
@@ -51,16 +51,13 @@ impl TryFrom<SerializedDeveloperSearchIndexConfig> for DeveloperSearchIndexConfi
     }
 }
 
-codegen_convex_serialization!(
-    DeveloperSearchIndexConfig,
-    SerializedDeveloperSearchIndexConfig
-);
+codegen_convex_serialization!(DeveloperTextIndexConfig, SerializedDeveloperTextIndexConfig);
 
-impl TryFrom<pb::searchlight::SearchIndexConfig> for DeveloperSearchIndexConfig {
+impl TryFrom<pb::searchlight::SearchIndexConfig> for DeveloperTextIndexConfig {
     type Error = anyhow::Error;
 
     fn try_from(proto: pb::searchlight::SearchIndexConfig) -> anyhow::Result<Self> {
-        Ok(DeveloperSearchIndexConfig {
+        Ok(DeveloperTextIndexConfig {
             search_field: proto
                 .search_field_path
                 .ok_or_else(|| anyhow::format_err!("Missing search_field_path"))?
@@ -76,8 +73,8 @@ impl TryFrom<pb::searchlight::SearchIndexConfig> for DeveloperSearchIndexConfig
     }
 }
 
-impl From<DeveloperSearchIndexConfig> for pb::searchlight::SearchIndexConfig {
-    fn from(config: DeveloperSearchIndexConfig) -> Self {
+impl From<DeveloperTextIndexConfig> for pb::searchlight::SearchIndexConfig {
+    fn from(config: DeveloperTextIndexConfig) -> Self {
         pb::searchlight::SearchIndexConfig {
             search_field_path: Some(config.search_field.into()),
             filter_fields: config

crates/common/src/bootstrap_model/index/text_index/mod.rs

@@ -9,8 +9,8 @@ pub use self::{
         TextIndexBackfillState,
     },
     index_config::{
-        DeveloperSearchIndexConfig,
-        SerializedDeveloperSearchIndexConfig,
+        DeveloperTextIndexConfig,
+        SerializedDeveloperTextIndexConfig,
     },
     index_snapshot::{
         FragmentedTextSegment,
@@ -36,9 +36,9 @@ mod tests {
         #![proptest_config(ProptestConfig { cases: 64 * env_config("CONVEX_PROPTEST_MULTIPLIER", 1), failure_persistence: None, .. ProptestConfig::default() })]
 
         #[test]
-        fn test_developer_search_index_config_roundtrips(v in any::<DeveloperSearchIndexConfig>()) {
+        fn test_developer_search_index_config_roundtrips(v in any::<DeveloperTextIndexConfig>()) {
                 assert_roundtrips::<
-                DeveloperSearchIndexConfig,
+                DeveloperTextIndexConfig,
                 pb::searchlight::SearchIndexConfig
             >(v);
         }

crates/database/src/bootstrap_model/index.rs

@@ -14,7 +14,7 @@ use common::{
         },
         index_validation_error,
         text_index::{
-            DeveloperSearchIndexConfig,
+            DeveloperTextIndexConfig,
             TextIndexState,
         },
         vector_index::{
@@ -897,7 +897,7 @@ impl<'a, RT: Runtime> IndexModel<'a, RT> {
                 } => IndexMetadata::new_backfilling(*self.tx.begin_timestamp(), index_name, fields),
                 IndexConfig::Search {
                     developer_config:
-                        DeveloperSearchIndexConfig {
+                        DeveloperTextIndexConfig {
                             search_field,
                             filter_fields,
                         },

crates/database/src/index_workers/index_meta.rs

@@ -40,9 +40,6 @@ use value::{
 use super::search_flusher::MultipartBuildType;
 use crate::Snapshot;
 
-pub trait PreviousSegmentsType: Send {
-    fn maybe_delete_document(&mut self, convex_id: InternalId) -> anyhow::Result<()>;
-}
 pub trait SegmentType<T: SearchIndex> {
     fn id(&self) -> &str;
 
@@ -57,13 +54,13 @@ pub trait SearchIndex: Clone + Debug {
     type Segment: SegmentType<Self> + Clone + Debug + Send + 'static;
     type NewSegment: Send;
 
-    type PreviousSegments: PreviousSegmentsType;
+    type PreviousSegments: Send;
 
     type Statistics: SegmentStatistics;
 
-    type BuildIndexArgs: Clone + Send;
+    type BuildIndexArgs: Clone + Send + 'static;
 
-    type Schema: Send + Sync;
+    type Schema: Send + Sync + 'static;
 
     /// Returns the generalized `SearchIndexConfig` if it matches the type of
     /// the parser (e.g. Text vs Vector) and `None` otherwise.
@@ -140,6 +137,16 @@ pub trait SearchIndex: Clone + Debug {
         config: &Self::DeveloperConfig,
         segments: Vec<Self::Segment>,
     ) -> anyhow::Result<Self::Segment>;
+
+    async fn merge_deletes<RT: Runtime>(
+        runtime: &RT,
+        previous_segments: &mut Self::PreviousSegments,
+        document_stream: DocumentStream<'_>,
+        repeatable_persistence: &RepeatablePersistence,
+        build_index_args: Self::BuildIndexArgs,
+        schema: Self::Schema,
+        document_log_lower_bound: Timestamp,
+    ) -> anyhow::Result<()>;
 }
 
 pub trait SegmentStatistics: Default + Debug {

crates/database/src/index_workers/search_compactor.rs

@@ -198,6 +198,7 @@ impl<RT: Runtime, T: SearchIndex> SearchIndexCompactor<RT, T> {
                 segments_to_compact.clone(),
                 new_segment.clone(),
                 *SEARCH_WORKER_PASSIVE_PAGES_PER_SECOND,
+                T::new_schema(&job.developer_config),
             )
             .await?;
         let total_compacted_segments = total_compacted_segments as u64;

crates/database/src/index_workers/search_worker.rs

@@ -8,6 +8,7 @@ use common::{
     knobs::{
         BUILD_MULTI_SEGMENT_TEXT_INDEXES,
         DATABASE_WORKERS_POLL_INTERVAL,
+        MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
     },
     persistence::PersistenceReader,
     runtime::{
@@ -47,6 +48,7 @@ use crate::{
             new_text_flusher,
             TextIndexFlusher2,
         },
+        BuildTextIndexArgs,
         TextIndexMetadataWriter,
     },
     vector_index_worker::{
@@ -55,6 +57,7 @@ use crate::{
             VectorIndexCompactor,
         },
         flusher::new_vector_flusher,
+        BuildVectorIndexArgs,
     },
     Database,
     TextIndexFlusher,
@@ -99,10 +102,22 @@ impl<RT: Runtime> SearchIndexWorkers<RT> {
         let vector_index_metadata_writer = SearchIndexMetadataWriter::new(
             runtime.clone(),
             database.clone(),
+            reader.clone(),
             search_storage.clone(),
+            BuildVectorIndexArgs {
+                full_scan_threshold_bytes: *MULTI_SEGMENT_FULL_SCAN_THRESHOLD_KB,
+            },
+        );
+        let text_index_metadata_writer = TextIndexMetadataWriter::new(
+            runtime.clone(),
+            database.clone(),
+            reader.clone(),
+            search_storage.clone(),
+            BuildTextIndexArgs {
+                search_storage: search_storage.clone(),
+                segment_term_metadata_fetcher: segment_term_metadata_fetcher.clone(),
+            },
         );
-        let text_index_metadata_writer =
-            TextIndexMetadataWriter::new(runtime.clone(), database.clone(), search_storage.clone());
         let vector_flush = retry_loop_expect_occs_and_overloaded(
             "VectorFlusher",
             runtime.clone(),