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_controlled, encode_terms, score_count, ClusterPosting,
23    ClusteredPostingCursor, EncodedScoreCluster, OccurrencePosting,
24};
25use crate::inverted_index::{AnalyzerBindings, IndexedFieldMetadata, IndexedFieldRevision};
26use crate::{PostingCursor, TokenTermKey};
27
28mod accelerators;
29mod controlled;
30mod data;
31mod format;
32mod migration;
33mod mutation;
34mod queries;
35mod read_impl;
36mod rebuild;
37mod storage;
38mod trait_impl;
39mod view;
40pub use storage::OccurrenceStorage;
41use view::{OccurrenceRead, OccurrenceSource};
42
43use migration::{migrate_legacy_forward_postings, migrate_legacy_reverse_postings};
44
45const FORMAT_METADATA_KEY: &str = "inverted_index_format";
46const CLUSTERED_FORMAT_NAME: &str = "clustered-v1";
47const MIGRATION_PAGE_SIZE: usize = 1_024;
48
49type ClusterKey = (FieldName, TokenTermKey, u64);
50type DocumentFields = BTreeMap<FieldName, FieldSnapshot>;
51type StagedDocuments = BTreeMap<DocId, DocumentFields>;
52type ClusterChanges = BTreeMap<ClusterKey, BTreeMap<DocId, Option<OccurrencePosting>>>;
53
54#[derive(Clone)]
55struct FieldSnapshot {
56    metadata: IndexedFieldMetadata,
57    terms: BTreeMap<TokenTermKey, Vec<uqa_core::TokenOccurrence>>,
58}
59
60#[derive(Debug, Clone, Copy)]
61pub(super) struct FieldStats {
62    pub(super) revision: IndexedFieldRevision,
63    pub(super) doc_count: u64,
64    pub(super) total_length: u64,
65}
66
67/// Inverted index implemented over [`KeyValueStore`].
68#[derive(Clone)]
69pub struct KeyValueInvertedIndex {
70    source: OccurrenceSource,
71    table: String,
72    bindings: AnalyzerBindings,
73}
74
75impl KeyValueInvertedIndex {
76    pub fn new(
77        store: Arc<dyn KeyValueStore>,
78        table: impl Into<String>,
79        analyzer: Analyzer,
80    ) -> Self {
81        Self::from_storage(
82            Arc::new(storage::KeyValueOccurrences(store)),
83            table,
84            AnalyzerBindings::new(analyzer),
85        )
86    }
87
88    /// Bind the common occurrence algorithms to a provider's native row projection while preserving the already compiled analyzer revisions.
89    pub fn from_storage(
90        storage: Arc<dyn OccurrenceStorage>,
91        table: impl Into<String>,
92        bindings: AnalyzerBindings,
93    ) -> Self {
94        Self {
95            source: OccurrenceSource::Live(storage),
96            table: table.into(),
97            bindings,
98        }
99    }
100
101    /// Compiled bindings used by this index, including successful native-provider binding changes.
102    pub fn analyzer_bindings(&self) -> &AnalyzerBindings {
103        &self.bindings
104    }
105
106    pub(crate) fn migrate_legacy_storage(store: &dyn KeyValueStore) -> StorageBackendResult<()> {
107        let marker = single_str_key(TAG_METADATA, FORMAT_METADATA_KEY)?;
108        if let Some(format) = store.get(&marker)? {
109            if format == CLUSTERED_FORMAT_NAME.as_bytes() {
110                return Ok(());
111            }
112            return Err(other_error(format!(
113                "unsupported KeyValue inverted-index format `{}`",
114                String::from_utf8_lossy(&format)
115            )));
116        }
117        if store.in_transaction() {
118            return Self::migrate_legacy_storage_in_transaction(store, &marker);
119        }
120
121        store.begin_transaction()?;
122        let migration = Self::migrate_legacy_storage_in_transaction(store, &marker);
123        match migration {
124            Ok(()) => store.commit_transaction(),
125            Err(error) => match store.rollback_transaction() {
126                Ok(()) => Err(error),
127                Err(rollback) => Err(other_error(format!(
128                    "{error}; KeyValue posting migration rollback also failed: {rollback}"
129                ))),
130            },
131        }
132    }
133
134    fn migrate_legacy_storage_in_transaction(
135        store: &dyn KeyValueStore,
136        marker: &[u8],
137    ) -> StorageBackendResult<()> {
138        let posting_count = migrate_legacy_forward_postings(store)?;
139        let reverse_count = migrate_legacy_reverse_postings(store)?;
140        if posting_count != reverse_count {
141            return Err(other_error(format!(
142                "cannot migrate inconsistent KeyValue postings: {posting_count} forward rows and {reverse_count} reverse rows"
143            )));
144        }
145        store.put(marker, &string_value(CLUSTERED_FORMAT_NAME))
146    }
147}
uqa_storage/key_value/inverted_index.rs

uqa_storage/key_value/
inverted_index.rs
