Eliminating unnecessary post filtering.

Search — Eliminating Unnecessary Post-Filtering

Today, after the background threads have searched the indexes, the resulting keys are transferred back to the main thread to generate the result. Because indexes are in a different time domain than the key database, it’s possible that the key database has been changed from what was in the index (i.e., was modified). Currently this situation is addressed by having the mainthread apply the search criteria to each individual result key to validate it’s appearance in the final result, aka post-filtering. This validation is expensive and almost always unnecessary. What’s needed is a simple bookkeeping machinery to enable the mainthread to cheaply determine that a result key must be validated or not.

An abstract solution

Examining the problem in the abstract. We have three different domains: 1/ the Valkey key database, 2/ the search mutation queue and 3/ the search indexes.

Thanks to Valkey, mutations are inherently serial, so we can assign a monotonically increasing sequence number to each mutation. From the key database perspective, mutations are atomic, meaning that each key can have a single — current — value which is the sequence number of the latest (last) mutation and represents its current value.

The mutation queue contains key modifications that are visible in the database but have not yet been inserted in to the index. Thus the mutation queue can be represented by a mapping of keys to a vector of sequence numbers and/or a mapping of sequence numbers to keys.

The indexes are also atomically updated on a per-key basis. Thus, like the database, the index can have a mapping from key to a single sequence number.

As described above, a query operation has two phases: 1/ index-based lookup which generates a list of keys. 2/ mainthread based processing of the list of keys. During phase 1/ result keys have the sequence numbers fetched from the index to sequence number mapping and transmitted to the second phase. In phase 2, the result keys have their current sequence numbers fetched from the database mapping and are compared to the equivalent sequence number that was transmitted from phase 1. If the two sequence numbers are equal, then we know that the index was up to date for this key and no revalidation is required. If the numbers are unequal, then revalidation is required.

A cheaper abstract solution

Let’s add two more constraints. 1/ Updates from the mutation queue to the index must be performed in sequence number order. 2/ Index mutations are mutually exclusive with query operations. Now, the index-based mapping can simply be replaced by a single value which, i.e., an index-wide sequence number. The query operation now doesn’t need to transmit a sequence number per key, it only needs to transmit the single index-wide sequence number.

The discerning reader will realize that the index-wide sequence number is equal to the largest sequence number in the index map.

A slightly more expensive solution that maps better into the implementation.

Let’s add one more constraint. The query operation is allowed to have false positives, i.e., it’s acceptable for the sequence number comparison done by the query operation to falsely indicate that a re-validation is required. This wastes time but is not incorrect. This constraint allows the single index-side sequence number to be smaller than the largest sequence number that would appear in the abstract per-key index map.

This constraint allows unordered insertions from the mutation queue into the indexes, provided that the index-wide sequence number is no larger than the smallest remaining mutation queue entry. This is a key capability to support QoS for ingestions.

The ability to support unordered insertions also enables the collapsing of multiple queue entries for the same key into a single entry that’s tagged by the maximum sequence number of the collapsed entries.

Proposed implementation

Objects stored in the IndexSchema.

struct IndexSchema {
   ...
   // These variables are owned by MainThread
   size_t NextSeqNum{0};
   absl::FlatHashMap<InternStringPtr, size_t> CurrentSeqNum;
   
   // These variables are owned by writer threads
   size_t CurrentIndexSeqNum{0}; // captured by reader threads.
   IntrusiveList<MutationQueueEntry> mqe_sequence_list;
   
};

Objects stored in the MutationQueueEntry

struct MutationQueueEntry {
   ...
   size_t IndexSeqNum;
   MutationQueueEntry *next{nullptr}; // For intrusive list
   MutationQueueEntry *prev{nullptr}; // For intrusive list
};

Operationally, the mainthread only performs two operations: upsert and delete (backfill is an upsert). Both operations in the main thread modify the mutation queue.

For the upsert case:

CurrentSeqNum[Key] = NextSeqNum++;  // Update the map
InsertMutationQueue(kUpsert, Key, CurrentSeqNum[Key], ...); // Queue the mutation

For the delete case:

CurrentSeqNum.erase(Key); // remove from map
InsertMutationQueue(kDelete, Key, NextSeqNum++, ...); // Queue the mutation

Mutation Queue Insert

The mutation queue insert is done from the main thread using a mutex to lock out writer threads.

void InsertMutationQueue(operation, Key, SequenceNumber) {
    
    // ... Lock the mutex for the mutation queue...
    
    MutationQueueEntry *this_entry;
    if (mqe.contains(Key)) {
        // This is the case where we're collapsing two entries.
        //
        // remove this entry from the sequence list, we're going to reinsert
        // it below
        //
        this_entry = meq.find(Key);
        meq_sequence_list.remove(this_entry); // Crash if it's not there ;-)
    } else {
        //
        // Need new entry
        //
        this_entry = new MutationQueueEntry;
    }
    //
    // This ensure that the new or collapsed entry has the highest sequence number
    // and is always at the end of the list.
    //
    this_entry -> IndexSeqNum = SequenceNumber;
    mqe_sequence_list.push_back(this_entry);  // Move to end of list
}

Mutation Queue Removal

The mutation queue removal is what happens at the end of an index update.

void CompleteMutationQueueEntry(MutationQueueEntry *entry) {
   // ... Lock the mutex for the mutation queue
   
   if (entry == mqe_sequence_list.front()) {
      // Since we're removing the oldest entry, we can advance the indexes
      IndexSeqNum = entry->IndexSeqNum;
   }
   mqe_sequence_list.remove(entry); // crash if it's not there
   delete entry;
}