[PERF]: better locking of uncommitted tracking maps (decrease compact…

…ion time by 3x) (#2736)

rust/index/src/fulltext/types.rs

@@ -101,45 +101,45 @@ impl<'me> FullTextIndexWriter<'me> {
                     // Readers are uninitialized until the first compaction finishes
                     // so there is a case when this is none hence not an error.
                     None => 0,
-                    Some(reader) => match reader.get_frequencies_for_token(token).await {
-                        Ok(frequency) => frequency,
-                        // New token so start with frequency of 0.
-                        Err(_) => 0,
-                    },
+                    Some(reader) => {
+                        match reader.get_frequencies_for_token(token.text.as_str()).await {
+                            Ok(frequency) => frequency,
+                            // New token so start with frequency of 0.
+                            Err(_) => 0,
+                        }
+                    }
                 };
                 uncommitted_frequencies
-                    .insert(token.to_string(), (frequency as i32, frequency as i32));
+                    .insert(token.text.clone(), (frequency as i32, frequency as i32));
             }
         }
+
         let mut uncommitted_postings = self.uncommitted_postings.lock().await;
-        match uncommitted_postings.positional_postings.get(token) {
-            Some(_) => {
-                // This should never happen -- if uncommitted has the token, then
-                // uncommitted_frequencies should have had it as well.
-                tracing::error!(
-                    "Error populating frequencies and posting lists from previous version"
-                );
-                return Err(FullTextIndexError::InvariantViolation);
+        for token in tokens {
+            if uncommitted_postings
+                .positional_postings
+                .contains_key(&token.text)
+            {
+                continue;
             }
-            None => {
-                let results = match &self.full_text_index_reader {
-                    // Readers are uninitialized until the first compaction finishes
-                    // so there is a case when this is none hence not an error.
-                    None => vec![],
-                    Some(reader) => match reader.get_all_results_for_token(token).await {
-                        Ok(results) => results,
-                        // New token so start with empty postings list.
-                        Err(_) => vec![],
-                    },
-                };
-                let mut doc_and_positions = HashMap::new();
-                for result in results {
-                    doc_and_positions.insert(result.0, result.1);
-                }
-                uncommitted_postings
-                    .positional_postings
-                    .insert(token.to_string(), doc_and_positions);
+
+            let results = match &self.full_text_index_reader {
+                // Readers are uninitialized until the first compaction finishes
+                // so there is a case when this is none hence not an error.
+                None => vec![],
+                Some(reader) => match reader.get_all_results_for_token(&token.text).await {
+                    Ok(results) => results,
+                    // New token so start with empty postings list.
+                    Err(_) => vec![],
+                },
+            };
+            let mut doc_and_positions = HashMap::new();
+            for result in results {
+                doc_and_positions.insert(result.0, result.1);
             }
+            uncommitted_postings
+                .positional_postings
+                .insert(token.text.clone(), doc_and_positions);
         }
 
         let mut uncommitted_postings = self.uncommitted_postings.lock().await;