Store index keys instead of full documents in the write log (#38204)

GitOrigin-RevId: 24809bc92c1f28433969528a8209ba602cb9d623

Author: Nicolapps

crates/database/benches/is_stale.rs

@@ -4,11 +4,6 @@ use std::str::FromStr;
 
 use anyhow::Result;
 use common::{
-    document::{
-        CreationTime,
-        PackedDocument,
-        ResolvedDocument,
-    },
     index::IndexKey,
     interval::{
         End,
@@ -20,16 +15,13 @@ use common::{
     testing::TestIdGenerator,
     types::{
         IndexDescriptor,
-        PersistenceVersion,
         TableName,
         TabletIndexName,
         Timestamp,
     },
     value::{
         id_v6::DeveloperDocumentId,
-        ConvexObject,
         ConvexValue,
-        FieldName,
         TabletIdAndTableNumber,
     },
 };
@@ -42,13 +34,15 @@ use criterion::{
 use database::{
     write_log::{
         new_write_log,
+        DocumentIndexKeysUpdate,
         LogWriter,
-        PackedDocumentUpdate,
         WriteSource,
     },
     ReadSet,
     TransactionReadSet,
 };
+use indexing::index_registry::DocumentIndexKeys;
+use maplit::btreemap;
 use search::{
     query::TextQueryTerm,
     FilterConditionRead,
@@ -68,8 +62,7 @@ fn create_test_setup() -> Result<(
     let table_name: TableName = "test_table".parse()?;
     let table_id_and_number = id_generator.user_table_id(&table_name);
 
-    let (_log_owner, _log_reader, log_writer) =
-        new_write_log(Timestamp::must(1000), PersistenceVersion::default());
+    let (_log_owner, _log_reader, log_writer) = new_write_log(Timestamp::must(1000));
 
     Ok((log_writer, id_generator, table_id_and_number, table_name))
 }
@@ -97,26 +90,22 @@ fn create_write_log_with_standard_index_writes(num_writes: usize) -> Result<(Log
             26 + ((i * 7) % 75)
         };
 
-        let document_update = PackedDocumentUpdate {
+        let index_key = IndexKey::new(vec![val!(value as i64)], id.into());
+        let document_keys =
+            DocumentIndexKeys::with_standard_index_for_test(index_name.clone(), index_key);
+
+        let writes = vec![(
             id,
-            old_document: None,
-            new_document: Some(PackedDocument::pack(&ResolvedDocument::new(
+            DocumentIndexKeysUpdate {
                 id,
-                CreationTime::try_from(1.)?,
-                ConvexObject::try_from(
-                    [(
-                        FieldName::from_str("value").unwrap(),
-                        ConvexValue::Int64(value as i64),
-                    )]
-                    .into_iter()
-                    .collect::<std::collections::BTreeMap<_, _>>(),
-                )?,
-            )?)),
-        };
+                old_document_keys: None,
+                new_document_keys: Some(document_keys),
+            },
+        )];
 
         log_writer.append(
             Timestamp::must((1001 + i) as i32),
-            vec![(id, document_update)].into(),
+            writes.into(),
             WriteSource::unknown(),
         );
     }
@@ -203,6 +192,7 @@ fn create_write_log_with_search_index_writes(num_writes: usize) -> Result<(LogWr
     // Add writes to the log
     for i in 0..num_writes {
         let id = id_generator.user_generate(&table_name);
+        let search_field: FieldPath = "content".parse()?;
 
         // Generate random text content - some will contain "target" word, but not all
         // documents that match filter conditions
@@ -222,45 +212,35 @@ fn create_write_log_with_search_index_writes(num_writes: usize) -> Result<(LogWr
             text_words.insert(insert_pos, "target");
         }
 
-        let search_value = text_words.join(" ");
-
-        // Create document with search content and filterable fields
-        let mut document_content = std::collections::BTreeMap::<FieldName, ConvexValue>::new();
-        document_content.insert(
-            FieldName::from_str("content")?,
-            ConvexValue::String(search_value.try_into()?),
-        );
-        document_content.insert(
-            FieldName::from_str("category")?,
-            ConvexValue::String(
+        let filter_values = btreemap! {
+            "category".parse()? => FilterValue::from_search_value(Some(&ConvexValue::String(
                 if i % 3 == 0 { "important" } else { "normal" }
                     .to_string()
                     .try_into()?,
-            ),
-        );
-        document_content.insert(
-            FieldName::from_str("priority")?,
-            ConvexValue::Int64((i % 5 + 1) as i64),
-        );
-        document_content.insert(
-            FieldName::from_str("active")?,
-            ConvexValue::Boolean(i % 2 == 0),
-        );
+            ))),
+            "priority".parse()? => FilterValue::from_search_value(Some(&ConvexValue::Int64((i % 5 + 1) as i64))),
+            "active".parse()? => FilterValue::from_search_value(Some(&ConvexValue::Boolean(i % 2 == 0))),
+        };
 
-        let value = ConvexObject::try_from(document_content)?;
-        let creation_time = CreationTime::try_from(1.)?;
-        let document = ResolvedDocument::new(id, creation_time, value)?;
-        let packed_document = PackedDocument::pack(&document);
+        let document_keys = DocumentIndexKeys::with_search_index_for_test_with_filters(
+            index_name.clone(),
+            search_field,
+            text_words.join(" ").try_into()?,
+            filter_values,
+        );
 
-        let document_update = PackedDocumentUpdate {
+        let writes = vec![(
             id,
-            old_document: None,
-            new_document: Some(packed_document),
-        };
+            DocumentIndexKeysUpdate {
+                id,
+                old_document_keys: None,
+                new_document_keys: Some(document_keys),
+            },
+        )];
 
         log_writer.append(
             Timestamp::must((1001 + i) as i32),
-            vec![(id, document_update)].into(),
+            writes.into(),
             WriteSource::unknown(),
         );
     }

crates/database/benches/subscriptions.rs

@@ -19,16 +19,11 @@ use std::{
 };
 
 use common::{
-    document::{
-        CreationTime,
-        PackedDocument,
-        ResolvedDocument,
-    },
     testing::TestIdGenerator,
     types::{
         GenericIndexName,
         IndexDescriptor,
-        PersistenceVersion,
+        TabletIndexName,
     },
 };
 use criterion::{
@@ -45,6 +40,7 @@ use humansize::{
     FormatSize,
     BINARY,
 };
+use indexing::index_registry::DocumentIndexKeys;
 use itertools::Itertools;
 use search::{
     convex_en,
@@ -56,7 +52,7 @@ use search::{
 use serde::Deserialize;
 use tokio::runtime::Runtime;
 use value::{
-    assert_obj,
+    ConvexString,
     DeveloperDocumentId,
     FieldPath,
     InternalId,
@@ -92,7 +88,7 @@ fn prefix_and_max_distances() -> Vec<(bool, FuzzyDistance)> {
 fn load_datasets(
     table_id: TabletIdAndTableNumber,
     max_size: usize,
-) -> anyhow::Result<BTreeMap<String, (Vec<PackedDocument>, Vec<String>)>> {
+) -> anyhow::Result<BTreeMap<String, (Vec<(ResolvedDocumentId, DocumentIndexKeys)>, Vec<String>)>> {
     let mut next_id = 0u64;
     let mut alloc_id = || {
         let mut result = [0; 16];
@@ -137,10 +133,16 @@ fn load_datasets(
                     *frequency_map.entry(token.text.clone()).or_default() += 1;
                 }
             }
-            let value = assert_obj!("body" => d.text);
-            let creation_time = CreationTime::try_from(1.)?;
-            let document = ResolvedDocument::new(id, creation_time, value)?;
-            documents.push(PackedDocument::pack(&document));
+
+            let field_path = FieldPath::from_str("body")?;
+            documents.push((
+                id,
+                DocumentIndexKeys::with_search_index_for_test(
+                    index_name(table_id.tablet_id),
+                    field_path,
+                    ConvexString::try_from(d.text).unwrap(),
+                ),
+            ));
         }
 
         let terms_by_frequency: Vec<String> = frequency_map
@@ -165,17 +167,18 @@ fn load_datasets(
     Ok(loaded)
 }
 
+fn index_name(tablet_id: TabletId) -> TabletIndexName {
+    GenericIndexName::new(tablet_id, IndexDescriptor::new("index").unwrap()).unwrap()
+}
+
 fn create_subscription_token(
     tablet_id: TabletId,
     prefix: bool,
     max_distance: FuzzyDistance,
     token: String,
 ) -> Token {
-    let index_name: GenericIndexName<TabletId> =
-        GenericIndexName::new(tablet_id, IndexDescriptor::new("index").unwrap()).unwrap();
-
     Token::text_search_token(
-        index_name,
+        index_name(tablet_id),
         FieldPath::from_str("body").unwrap(),
         vec![TextQueryTerm::Fuzzy {
             token,
@@ -261,12 +264,12 @@ fn bench_query(c: &mut Criterion) {
                 data,
                 |b, documents| {
                     b.to_async(&rt).iter(|| async {
-                        for doc in documents {
+                        for (doc_id, doc_index_keys) in documents {
                             let mut to_notify = BTreeSet::new();
-                            subscription_manager.overlapping_for_testing(
-                                doc,
+                            subscription_manager.overlapping(
+                                doc_id,
+                                doc_index_keys,
                                 &mut to_notify,
-                                PersistenceVersion::V5,
                             );
                         }
                     })

crates/database/src/committer.rs

@@ -134,6 +134,7 @@ use crate::{
     },
     transaction::FinalTransaction,
     write_log::{
+        index_keys_from_full_documents,
         LogWriter,
         PackedDocumentUpdate,
         PendingWriteHandle,
@@ -834,12 +835,19 @@ impl<RT: Runtime> Committer<RT> {
 
         // Write transaction state at the commit ts to the document store.
         metrics::commit_rows(ordered_updates.len() as u64);
-        let timer = metrics::write_log_append_timer();
-        let size = ordered_updates.heap_size();
-        self.log.append(commit_ts, ordered_updates, write_source);
+
+        let timer = metrics::pending_writes_to_write_log_timer();
+        // See the comment in `overlaps_index_keys` for why it’s safe
+        // to use indexes from the current snapshot.
+        let writes = index_keys_from_full_documents(ordered_updates, &new_snapshot.index_registry);
+        let size = writes.heap_size();
         drop(timer);
         metrics::write_log_commit_bytes(size);
 
+        let timer = metrics::write_log_append_timer();
+        self.log.append(commit_ts, writes, write_source);
+        drop(timer);
+
         if let Some(table_summaries) = new_snapshot.table_summaries.as_ref() {
             metrics::log_num_keys(table_summaries.num_user_documents);
             metrics::log_document_store_size(table_summaries.user_size);

crates/database/src/database.rs

@@ -824,9 +824,8 @@ impl<RT: Runtime> Database<RT> {
         .await?;
 
         let persistence_reader = persistence.reader();
-        let (log_owner, log_reader, log_writer) = new_write_log(*ts, persistence_reader.version());
-        let subscriptions =
-            SubscriptionsWorker::start(log_owner, runtime.clone(), persistence_reader.version());
+        let (log_owner, log_reader, log_writer) = new_write_log(*ts);
+        let subscriptions = SubscriptionsWorker::start(log_owner, runtime.clone());
         let usage_counter = UsageCounter::new(usage_events);
         let committer = Committer::start(
             log_writer,

crates/database/src/metrics.rs

@@ -246,6 +246,29 @@ pub fn commit_is_stale_timer() -> StatusTimer {
     StatusTimer::new(&DATABASE_COMMIT_IS_STALE_SECONDS)
 }
 
+register_convex_counter!(
+    DATABASE_MISSING_INDEX_KEY_TOTAL,
+    "Number of times an index was not found in DocumentIndexKeys"
+);
+pub fn log_missing_index_key() {
+    // This record cases where some index was expected to be present in the
+    // DocumentIndexKeys, but was not.
+    //
+    // This shouldn’t be a problem, because:
+    // - When creating a new index:
+    //   - If the write is committed _before_ the index is created, then
+    //     subscriptions that use this new index will not need to be invalidated
+    //     because of the write (since they were created after the index was
+    //     created, thus after the write was committed)
+    //   - If the write is committed _after_ the index is created, then the entry in
+    //     the write log will be created with a snapshot that includes the index.
+    // - When deleting an index, reaching this means that the query is trying to
+    //   read from an index that has been deleted. Hence, the conflict will be
+    //   detected over the read on the system document.
+
+    log_counter(&DATABASE_MISSING_INDEX_KEY_TOTAL, 1);
+}
+
 register_convex_histogram!(
     DATABASE_COMMIT_PREPARE_WRITES_SECONDS,
     "Time to prepare writes",
@@ -289,6 +312,14 @@ pub fn write_log_append_timer() -> Timer<VMHistogram> {
     Timer::new(&DATABASE_APPLY_DOCUMENT_STORE_APPEND_SECONDS)
 }
 
+register_convex_histogram!(
+    DATABASE_PENDING_WRITES_TO_WRITE_LOG_SECONDS,
+    "Time to convert writes from PendingWrites to WriteLog"
+);
+pub fn pending_writes_to_write_log_timer() -> Timer<VMHistogram> {
+    Timer::new(&DATABASE_PENDING_WRITES_TO_WRITE_LOG_SECONDS)
+}
+
 register_convex_histogram!(
     DATABASE_WRITE_LOG_COMMIT_BYTES,
     "Total size of all write log entries for a commit"