1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
// SPDX-License-Identifier: BUSL-1.1
//! KvEngine methods for sorted index lifecycle and query.
//!
//! Extends `KvEngine` with:
//! - `register_sorted_index()` / `drop_sorted_index()` — DDL
//! - `sorted_index_on_put()` / `sorted_index_on_delete()` — auto-maintenance
//! - `sorted_index_rank()` / `sorted_index_top_k()` / etc. — query
use super::engine::KvEngine;
use super::engine_helpers::table_key;
use super::sorted_index::manager::SortedIndexDef;
/// Parameters for [`KvEngine::sorted_index_range`].
#[derive(Debug, Clone, Copy)]
pub struct SortedIndexRangeParams<'a> {
pub database_id: u64,
pub tenant_id: u64,
pub index_name: &'a str,
pub score_min: Option<&'a [u8]>,
pub score_max: Option<&'a [u8]>,
pub now_ms: u64,
}
impl KvEngine {
/// Register a new sorted index with backfill from existing KV data.
///
/// Scans the hash table for all entries, extracts sort key columns,
/// and populates the order-statistic tree. Returns backfill count.
pub fn register_sorted_index(
&mut self,
database_id: u64,
tenant_id: u64,
collection: &str,
def: SortedIndexDef,
) -> u32 {
let tkey = table_key(database_id, tenant_id, collection);
let now_ms = super::current_ms();
// Name the collection even when it holds no rows yet: the checkpoint
// writer recovers a collection's identity from these reverse maps, and
// an unnamed collection gets no checkpoint file — which would drop this
// registration from the checkpoint while WAL truncation deleted the
// record that carries it.
self.hash_to_tenant.entry(tkey).or_insert(tenant_id);
self.hash_to_collection
.entry(tkey)
.or_insert_with(|| collection.to_string());
// Collect existing entries from the hash table for backfill.
let entries: Vec<(Vec<u8>, Vec<u8>)> = self
.tables
.get(&tkey)
.map(|t| {
let (entries, _) = t.scan(0, usize::MAX, now_ms, None);
entries
.into_iter()
.map(|(k, v)| (k.to_vec(), v.to_vec()))
.collect()
})
.unwrap_or_default();
self.sorted_indexes
.register(database_id, tenant_id, def, entries.into_iter())
}
/// Drop a sorted index. Returns `true` if it existed.
pub fn drop_sorted_index(
&mut self,
database_id: u64,
tenant_id: u64,
index_name: &str,
) -> bool {
self.sorted_indexes.drop(database_id, tenant_id, index_name)
}
/// Called after a KV PUT to maintain sorted indexes on this collection.
///
/// `field_values` are the extracted field name/value pairs from the new value.
pub fn sorted_index_on_put(
&mut self,
database_id: u64,
tenant_id: u64,
collection: &str,
primary_key: &[u8],
field_values: &[(String, Vec<u8>)],
) {
let tkey = table_key(database_id, tenant_id, collection);
self.sorted_indexes.on_put(tkey, primary_key, field_values);
}
/// Called after a KV DELETE to maintain sorted indexes on this collection.
pub fn sorted_index_on_delete(
&mut self,
database_id: u64,
tenant_id: u64,
collection: &str,
primary_key: &[u8],
) {
let tkey = table_key(database_id, tenant_id, collection);
self.sorted_indexes.on_delete(tkey, primary_key);
}
/// Check if any sorted indexes exist for this tenant/collection.
pub fn has_sorted_indexes(&self, database_id: u64, tenant_id: u64, collection: &str) -> bool {
let tkey = table_key(database_id, tenant_id, collection);
self.sorted_indexes.has_indexes(tkey)
}
// ── Query methods ──────────────────────────────────────────────────
pub fn sorted_index_rank(
&self,
database_id: u64,
tenant_id: u64,
index_name: &str,
primary_key: &[u8],
now_ms: u64,
) -> Option<u32> {
self.sorted_indexes
.rank(database_id, tenant_id, index_name, primary_key, now_ms)
}
pub fn sorted_index_top_k(
&self,
database_id: u64,
tenant_id: u64,
index_name: &str,
k: u32,
now_ms: u64,
) -> Option<Vec<(u32, Vec<u8>)>> {
self.sorted_indexes
.top_k(database_id, tenant_id, index_name, k, now_ms)
}
pub fn sorted_index_range(
&self,
params: SortedIndexRangeParams<'_>,
) -> Option<Vec<(u32, Vec<u8>)>> {
let SortedIndexRangeParams {
database_id,
tenant_id,
index_name,
score_min,
score_max,
now_ms,
} = params;
self.sorted_indexes.range(
database_id,
tenant_id,
index_name,
score_min,
score_max,
now_ms,
)
}
pub fn sorted_index_count(
&self,
database_id: u64,
tenant_id: u64,
index_name: &str,
now_ms: u64,
) -> Option<u32> {
self.sorted_indexes
.count(database_id, tenant_id, index_name, now_ms)
}
pub fn sorted_index_score(
&self,
database_id: u64,
tenant_id: u64,
index_name: &str,
primary_key: &[u8],
) -> Option<Vec<u8>> {
self.sorted_indexes
.score(database_id, tenant_id, index_name, primary_key)
}
pub fn sorted_index_def(
&self,
database_id: u64,
tenant_id: u64,
index_name: &str,
) -> Option<&SortedIndexDef> {
self.sorted_indexes
.get_def(database_id, tenant_id, index_name)
}
}