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
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
//! [`MaterializedSet`] — a materialized view's incremental member set,
//! and the [`Membership`] verdicts that drive it.
use crate::value::IndexValue;
use kevy_text::SortOrder;
/// One shard's materialized result set: ordered `(order_value, key)`
/// members with the bounded top-K discipline (keep `K + Δ` where
/// `Δ = K/4`; underflow requests a local rebuild from the base
/// indexes — RFC §2).
///
/// ```
/// use kevy_index::{IndexValue, MaterializedSet, Membership, SortOrder};
///
/// // top 4 by score, highest first; the set holds K + K/4 = 5
/// let mut hot = MaterializedSet::new(4, SortOrder::Desc);
/// for (i, score) in [3, 9, 1, 7, 5, 8].into_iter().enumerate() {
/// let key = format!("post:{i}");
/// hot.apply(key.as_bytes(), Membership::Member(Some(IndexValue::I64(score))));
/// }
/// assert_eq!(hot.len(), 5, "the lowest score was evicted");
/// let top: Vec<_> = hot.page(None, 2).into_iter().map(|(v, _)| v).collect();
/// assert_eq!(top, [IndexValue::I64(9), IndexValue::I64(8)]);
/// assert!(hot.approx_bytes() > 0);
///
/// // removals that drop the set below K ask the caller for a rebuild
/// hot.apply(b"post:1", Membership::NonMember);
/// let underflow = hot.apply(b"post:3", Membership::NonMember);
/// assert!(underflow && hot.len() == 3);
///
/// hot.clear();
/// assert!(hot.is_empty());
/// ```
#[derive(Debug, Default)]
pub struct MaterializedSet {
set: std::collections::BTreeSet<(IndexValue, Vec<u8>)>,
back: std::collections::HashMap<Vec<u8>, IndexValue>,
/// 0 = unbounded.
top_k: u32,
/// DESC view: the bound keeps the LARGEST members (evict the
/// smallest past the cap); ASC keeps the smallest.
order: SortOrder,
order_excluded: u64,
}
/// One key's standing against a materialized view, as
/// [`MaterializedSet::apply`] takes it.
///
/// ```
/// use kevy_index::{IndexValue, MaterializedSet, Membership, SortOrder};
/// let mut m = MaterializedSet::new(0, SortOrder::Asc);
/// m.apply(b"k1", Membership::Member(Some(IndexValue::I64(1))));
/// m.apply(b"k2", Membership::Member(None));
/// assert_eq!((m.len(), m.order_excluded()), (1, 1));
/// m.apply(b"k1", Membership::NonMember);
/// assert!(m.is_empty());
/// ```
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum Membership {
/// The key matches the view's tree, with its value in the view's
/// order index — `None` when it has none there, which excludes it
/// (counted, not an error).
///
/// ```
/// use kevy_index::{IndexValue, MaterializedSet, Membership, SortOrder};
/// let mut m = MaterializedSet::new(0, SortOrder::Asc);
/// m.apply(b"a", Membership::Member(Some(IndexValue::I64(2))));
/// m.apply(b"a", Membership::Member(Some(IndexValue::I64(5))));
/// assert_eq!(m.page(None, 9), vec![(IndexValue::I64(5), b"a".to_vec())], "re-applying moves the key");
/// ```
Member(Option<IndexValue>),
/// The key does not match the tree, or no longer exists.
///
/// ```
/// use kevy_index::{IndexValue, MaterializedSet, Membership, SortOrder};
/// let mut m = MaterializedSet::new(0, SortOrder::Asc);
/// m.apply(b"a", Membership::Member(Some(IndexValue::I64(2))));
/// m.apply(b"a", Membership::NonMember);
/// assert!(m.is_empty());
/// ```
NonMember,
}
impl MaterializedSet {
/// New set with the declared bound (0 = unbounded) and order
/// direction (the bound evicts from the view's WORST end).
///
/// ```
/// use kevy_index::{MaterializedSet, SortOrder};
/// assert!(MaterializedSet::new(10, SortOrder::Desc).is_empty());
/// ```
pub fn new(top_k: u32, order: SortOrder) -> Self {
Self { top_k, order, ..Default::default() }
}
/// Members excluded because they are absent from the order index.
///
/// ```
/// use kevy_index::{MaterializedSet, Membership, SortOrder};
/// let mut m = MaterializedSet::new(0, SortOrder::Asc);
/// m.apply(b"k", Membership::Member(None));
/// assert_eq!(m.order_excluded(), 1);
/// ```
pub fn order_excluded(&self) -> u64 {
self.order_excluded
}
fn cap(&self) -> usize {
if self.top_k == 0 { usize::MAX } else { (self.top_k + self.top_k / 4) as usize }
}
/// Apply one key's membership verdict + order value. Returns
/// `true` if the set UNDERFLOWED below K after a removal (the
/// caller must schedule a local rebuild).
pub fn apply(&mut self, key: &[u8], membership: Membership) -> bool {
let (member, order) = match membership {
Membership::Member(order) => (true, order),
Membership::NonMember => (false, None),
};
// Bounded fast path: a NON-member of a full top-K set whose
// value is worse than the current worst can neither enter nor
// change anything — one comparison, no tree ops, no allocs.
// This is the write-tax fast path for hot-list views (most
// writes touch rows outside the top K).
if self.top_k != 0
&& member
&& !self.back.contains_key(key)
&& self.set.len() >= self.cap()
&& let Some(v) = &order
{
let enters = if self.order == SortOrder::Desc {
self.set.iter().next().is_some_and(|(worst, _)| v > worst)
} else {
self.set.iter().next_back().is_some_and(|(worst, _)| v < worst)
};
if !enters {
return false;
}
}
if let Some(old) = self.back.remove(key) {
self.set.remove(&(old, key.to_vec()));
}
match (member, order) {
(true, Some(v)) => {
self.back.insert(key.to_vec(), v.clone());
self.set.insert((v, key.to_vec()));
self.evict_past_cap();
false
}
(true, None) => {
self.order_excluded += 1;
false
}
_ => self.top_k != 0 && self.set.len() < self.top_k as usize,
}
}
/// Bound: evict the view's WORST member past K+Δ — the largest
/// for ASC, the SMALLEST for DESC.
fn evict_past_cap(&mut self) {
if self.set.len() > self.cap() {
let worst = if self.order == SortOrder::Desc {
self.set.iter().next().cloned()
} else {
self.set.iter().next_back().cloned()
};
if let Some(w) = worst {
self.set.remove(&w);
self.back.remove(&w.1);
}
}
}
/// Ordered page in the set's own direction: ascending from just past
/// `after`, or DESCENDING from just below it (a DESC view must take
/// each shard's LARGEST members — taking the ascending head and
/// reversing at the merge yields the wrong member set).
///
/// ```
/// use kevy_index::{IndexValue, MaterializedSet, Membership, SortOrder};
/// let mut m = MaterializedSet::new(0, SortOrder::Desc);
/// for (k, v) in [(b"a", 1), (b"b", 2)] {
/// m.apply(k, Membership::Member(Some(IndexValue::I64(v))));
/// }
/// assert_eq!(m.page(None, 1), vec![(IndexValue::I64(2), b"b".to_vec())]);
/// ```
pub fn page(
&self,
after: Option<&(IndexValue, Vec<u8>)>,
limit: usize,
) -> Vec<(IndexValue, Vec<u8>)> {
if self.order == SortOrder::Desc {
let iter: Box<dyn Iterator<Item = &(IndexValue, Vec<u8>)>> = match after {
Some(c) => Box::new(
self.set
.range((std::ops::Bound::Unbounded, std::ops::Bound::Excluded(c.clone())))
.rev(),
),
None => Box::new(self.set.iter().rev()),
};
return iter.take(limit).cloned().collect();
}
let iter: Box<dyn Iterator<Item = &(IndexValue, Vec<u8>)>> = match after {
Some(c) => Box::new(
self.set.range((std::ops::Bound::Excluded(c.clone()), std::ops::Bound::Unbounded)),
),
None => Box::new(self.set.iter()),
};
iter.take(limit).cloned().collect()
}
/// Member count.
pub fn len(&self) -> usize {
self.set.len()
}
/// Empty?
pub fn is_empty(&self) -> bool {
self.set.is_empty()
}
/// Wipe (rebuild path).
pub fn clear(&mut self) {
self.set.clear();
self.back.clear();
}
/// Approximate heap bytes (RFC §5 formula's measured side).
pub fn approx_bytes(&self) -> u64 {
self.set.iter().map(|(v, k)| (v.approx_bytes() + k.len() + 48) as u64).sum()
}
}