uqa_storage/key_value/
inverted_index.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Clustered occurrence indexes over an ordered key/value store.
8
9use super::codec::{
10    blob_to_positions, decode_u64_value, doc_length_key, key_with_tag, other_error,
11    posting_cluster_positions_key, posting_cluster_score_key, posting_document_key, read_str,
12    read_u64, reverse_posting_key, single_str_key, string_value,
13};
14use super::occurrence_keys as keys;
15use super::{
16    Analyzer, AnalyzerPhase, Arc, BTreeMap, BTreeSet, DocId, FieldName, IndexStats, InvertedIndex,
17    KeyValueBatch, KeyValueStore, Payload, PostingEntry, PostingList, StorageBackendResult,
18    TAG_METADATA, TAG_POSTING, TAG_REVERSE_POSTING,
19};
20use crate::clustered_postings::{
21    cluster_id, decode_all_scores, decode_occurrence_cluster, decode_term_keys, encode_cluster,
22    encode_occurrence_cluster, encode_term_keys, encode_terms, score_count, ClusterPosting,
23    ClusteredPostingCursor, EncodedScoreCluster, OccurrencePosting,
24};
25use crate::inverted_index::{
26    analyze_index_field, AnalyzerBindings, IndexedFieldMetadata, IndexedFieldRevision,
27};
28use crate::{PostingCursor, TokenTermKey};
29
30mod controlled;
31mod data;
32mod format;
33mod migration;
34mod mutation;
35mod queries;
36mod rebuild;
37mod trait_impl;
38
39use migration::{migrate_legacy_forward_postings, migrate_legacy_reverse_postings};
40
41const FORMAT_METADATA_KEY: &str = "inverted_index_format";
42const CLUSTERED_FORMAT_NAME: &str = "clustered-v1";
43const MIGRATION_PAGE_SIZE: usize = 1_024;
44
45type ClusterKey = (FieldName, TokenTermKey, u64);
46type DocumentFields = BTreeMap<FieldName, FieldSnapshot>;
47type StagedDocuments = BTreeMap<DocId, DocumentFields>;
48type ClusterChanges = BTreeMap<ClusterKey, BTreeMap<DocId, Option<OccurrencePosting>>>;
49
50#[derive(Clone)]
51struct FieldSnapshot {
52    metadata: IndexedFieldMetadata,
53    terms: BTreeMap<TokenTermKey, Vec<uqa_core::TokenOccurrence>>,
54}
55
56#[derive(Debug, Clone, Copy)]
57struct FieldStats {
58    revision: IndexedFieldRevision,
59    doc_count: u64,
60    total_length: u64,
61}
62
63/// Inverted index implemented over [`KeyValueStore`].
64#[derive(Clone)]
65pub struct KeyValueInvertedIndex {
66    store: Arc<dyn KeyValueStore>,
67    table: String,
68    bindings: AnalyzerBindings,
69}
70
71impl KeyValueInvertedIndex {
72    pub fn new(
73        store: Arc<dyn KeyValueStore>,
74        table: impl Into<String>,
75        analyzer: Analyzer,
76    ) -> Self {
77        Self {
78            store,
79            table: table.into(),
80            bindings: AnalyzerBindings::new(analyzer),
81        }
82    }
83
84    pub(crate) fn migrate_legacy_storage(store: &dyn KeyValueStore) -> StorageBackendResult<()> {
85        let marker = single_str_key(TAG_METADATA, FORMAT_METADATA_KEY)?;
86        if let Some(format) = store.get(&marker)? {
87            if format == CLUSTERED_FORMAT_NAME.as_bytes() {
88                return Ok(());
89            }
90            return Err(other_error(format!(
91                "unsupported KeyValue inverted-index format `{}`",
92                String::from_utf8_lossy(&format)
93            )));
94        }
95        if store.in_transaction() {
96            return Self::migrate_legacy_storage_in_transaction(store, &marker);
97        }
98
99        store.begin_transaction()?;
100        let migration = Self::migrate_legacy_storage_in_transaction(store, &marker);
101        match migration {
102            Ok(()) => store.commit_transaction(),
103            Err(error) => match store.rollback_transaction() {
104                Ok(()) => Err(error),
105                Err(rollback) => Err(other_error(format!(
106                    "{error}; KeyValue posting migration rollback also failed: {rollback}"
107                ))),
108            },
109        }
110    }
111
112    fn migrate_legacy_storage_in_transaction(
113        store: &dyn KeyValueStore,
114        marker: &[u8],
115    ) -> StorageBackendResult<()> {
116        let posting_count = migrate_legacy_forward_postings(store)?;
117        let reverse_count = migrate_legacy_reverse_postings(store)?;
118        if posting_count != reverse_count {
119            return Err(other_error(format!(
120                "cannot migrate inconsistent KeyValue postings: {posting_count} forward rows and {reverse_count} reverse rows"
121            )));
122        }
123        store.put(marker, &string_value(CLUSTERED_FORMAT_NAME))
124    }
125}
uqa_storage/key_value/inverted_index.rs

uqa_storage/key_value/
inverted_index.rs
