1use std::{
5 borrow::Cow,
6 ops::Bound::{self, Included},
7 sync::{
8 Arc,
9 atomic::{AtomicU64, Ordering},
10 },
11};
12
13use reifydb_codec::key::encoded::EncodedKey;
14use reifydb_core::{
15 common::CommitVersion,
16 default,
17 interface::{
18 catalog::storage::StorageId,
19 store::{EntryKind, EntryLayout},
20 },
21 key::{
22 row::{PartitionedRowKey, RowKey, StoragePartitionedRowKey, StorageRowKey},
23 series::{
24 PartitionedSeriesRowKey, PartitionedSeriesRowKeyRange, SeriesRowKey, SeriesRowKeyRange,
25 StoragePartitionedSeriesKey, StorageSeriesKey,
26 },
27 typed::{BoundedKey, DenseKey, Edge, range::KeyRange},
28 },
29 metrics::{collect::MetricsCollector, sample::MetricsSample},
30};
31use reifydb_store::{
32 coverage::{
33 cursor::{Cursor, ServedChunk as TierChunk},
34 interval::Interval,
35 plan::{DEFAULT_GAP_GUARD, Segment},
36 },
37 tier::range::{
38 DEFAULT_COVERAGE_INTERVALS, Materialize, RangeConfig, RangeDomain, RangeMetrics, RangeRows,
39 RangeShardMetrics, RangeTier, RowBytes,
40 },
41};
42use reifydb_store_commit::{MultiVersionScope, RangeBatch, RangeCursor, RawEntry};
43use reifydb_value::{byte_size::ByteSize, reifydb_assertions, util::cowvec::CowVec, value::row_number::RowNumber};
44use tracing::instrument;
45
46#[derive(Clone, Copy, Debug)]
47pub struct MultiRangeConfig {
48 pub shard_bytes: Option<ByteSize>,
49 pub shards: usize,
50 pub gap_guard: usize,
51}
52
53impl MultiRangeConfig {
54 pub fn testing() -> Self {
55 Self {
56 shard_bytes: Some(default::store::MULTI_RANGE_BUFFER_SHARD_TESTING),
57 shards: default::store::MULTI_RANGE_BUFFER_SHARDS_TESTING as usize,
58 gap_guard: DEFAULT_GAP_GUARD,
59 }
60 }
61}
62
63impl From<MultiRangeConfig> for RangeConfig {
64 fn from(config: MultiRangeConfig) -> Self {
65 Self {
66 shard_bytes: config.shard_bytes,
67 shards: config.shards,
68 gap_guard: config.gap_guard,
69 coverage_bytes: None,
70 coverage_intervals: DEFAULT_COVERAGE_INTERVALS,
71 }
72 }
73}
74pub type ServedChunk = reifydb_store::coverage::cursor::ServedChunk<RangeBatch>;
75
76const ROW_BUCKET_SHIFT: u32 = 16;
77const BUCKETS: u64 = 1 << (u64::BITS - ROW_BUCKET_SHIFT);
78
79#[derive(Clone, Copy, Debug)]
80pub struct MultiDomain;
81
82pub trait NarrowLayout: DenseKey + Copy {
83 type Wide;
84
85 fn kind(storage: StorageId) -> EntryKind;
86
87 fn owns(kind: EntryKind) -> Option<StorageId>;
88
89 fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<Self>;
90
91 fn widen(storage: StorageId, key: &Self) -> EncodedKey;
92
93 fn storage_start(storage: StorageId) -> EncodedKey;
94
95 fn storage_end(storage: StorageId) -> EncodedKey;
96}
97
98impl NarrowLayout for StorageRowKey {
99 type Wide = RowKey;
100
101 fn kind(storage: StorageId) -> EntryKind {
102 EntryKind::Source(storage, EntryLayout::Row)
103 }
104
105 fn owns(kind: EntryKind) -> Option<StorageId> {
106 match kind {
107 EntryKind::Source(storage, EntryLayout::Row) => Some(storage),
108 _ => None,
109 }
110 }
111
112 fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<Self> {
113 let storage = Self::owns(kind)?;
114 let row = RowKey::decode(key)?;
115 (row.storage == storage).then(|| StorageRowKey::from(row))
116 }
117
118 fn widen(storage: StorageId, key: &Self) -> EncodedKey {
119 RowKey::encoded(storage, key.row())
120 }
121
122 fn storage_start(storage: StorageId) -> EncodedKey {
123 RowKey::storage_start(storage)
124 }
125
126 fn storage_end(storage: StorageId) -> EncodedKey {
127 RowKey::storage_end(storage)
128 }
129}
130
131impl NarrowLayout for StoragePartitionedRowKey {
132 type Wide = PartitionedRowKey;
133
134 fn kind(storage: StorageId) -> EntryKind {
135 EntryKind::PartitionedSource(storage, EntryLayout::Row)
136 }
137
138 fn owns(kind: EntryKind) -> Option<StorageId> {
139 match kind {
140 EntryKind::PartitionedSource(storage, EntryLayout::Row) => Some(storage),
141 _ => None,
142 }
143 }
144
145 fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<Self> {
146 let storage = Self::owns(kind)?;
147 let row = PartitionedRowKey::decode(key)?;
148 (row.storage == storage).then(|| StoragePartitionedRowKey::from(row))
149 }
150
151 fn widen(storage: StorageId, key: &Self) -> EncodedKey {
152 PartitionedRowKey::encoded(storage, key.partition(), key.row())
153 }
154
155 fn storage_start(storage: StorageId) -> EncodedKey {
156 PartitionedRowKey::storage_start(storage)
157 }
158
159 fn storage_end(storage: StorageId) -> EncodedKey {
160 PartitionedRowKey::storage_end(storage)
161 }
162}
163
164impl NarrowLayout for StorageSeriesKey {
165 type Wide = SeriesRowKey;
166
167 fn kind(storage: StorageId) -> EntryKind {
168 EntryKind::Source(storage, EntryLayout::Series)
169 }
170
171 fn owns(kind: EntryKind) -> Option<StorageId> {
172 match kind {
173 EntryKind::Source(storage, EntryLayout::Series) => Some(storage),
174 _ => None,
175 }
176 }
177
178 fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<Self> {
179 let storage = Self::owns(kind)?;
180 let row = SeriesRowKey::decode(key)?;
181 (row.storage == storage).then(|| StorageSeriesKey::from(row))
182 }
183
184 fn widen(storage: StorageId, key: &Self) -> EncodedKey {
185 key.with_storage(storage).encode()
186 }
187
188 fn storage_start(storage: StorageId) -> EncodedKey {
189 SeriesRowKeyRange::storage_start(storage)
190 }
191
192 fn storage_end(storage: StorageId) -> EncodedKey {
193 SeriesRowKeyRange::storage_end(storage)
194 }
195}
196
197impl NarrowLayout for StoragePartitionedSeriesKey {
198 type Wide = PartitionedSeriesRowKey;
199
200 fn kind(storage: StorageId) -> EntryKind {
201 EntryKind::PartitionedSource(storage, EntryLayout::Series)
202 }
203
204 fn owns(kind: EntryKind) -> Option<StorageId> {
205 match kind {
206 EntryKind::PartitionedSource(storage, EntryLayout::Series) => Some(storage),
207 _ => None,
208 }
209 }
210
211 fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<Self> {
212 let storage = Self::owns(kind)?;
213 let row = PartitionedSeriesRowKey::decode(key)?;
214 (row.storage == storage).then(|| StoragePartitionedSeriesKey::from(row))
215 }
216
217 fn widen(storage: StorageId, key: &Self) -> EncodedKey {
218 key.with_storage(storage).encode()
219 }
220
221 fn storage_start(storage: StorageId) -> EncodedKey {
222 PartitionedSeriesRowKeyRange::storage_start(storage)
223 }
224
225 fn storage_end(storage: StorageId) -> EncodedKey {
226 PartitionedSeriesRowKeyRange::storage_end(storage)
227 }
228}
229
230pub fn resume_after<L: NarrowLayout>(kind: EntryKind, last: &EncodedKey) -> Option<EncodedKey> {
231 let storage = L::owns(kind)?;
232 let next = L::narrow(kind, last)?.successor()?;
233 Some(L::widen(storage, &next))
234}
235
236pub fn narrow_bound_of<L: NarrowLayout>(kind: EntryKind, bytes: &[u8]) -> Option<Edge<L>> {
237 let storage = L::owns(kind)?;
238 if bytes == L::storage_start(storage).as_slice() {
239 return Some(Edge::Bottom);
240 }
241 if bytes >= L::storage_end(storage).as_slice() {
242 return Some(Edge::Top);
243 }
244 L::narrow(kind, &EncodedKey::new(bytes)).map(Edge::Key)
245}
246
247pub fn narrow(kind: EntryKind, key: &EncodedKey) -> Option<StorageRowKey> {
248 StorageRowKey::narrow(kind, key)
249}
250
251pub fn narrow_bound(kind: EntryKind, bytes: &[u8]) -> Option<Edge<StorageRowKey>> {
252 narrow_bound_of::<StorageRowKey>(kind, bytes)
253}
254
255fn stops_in_band(kind: EntryKind, bytes: &[u8]) -> bool {
256 StorageRowKey::owns(kind).is_some_and(|storage| bytes <= StorageRowKey::storage_end(storage).as_slice())
257}
258
259#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
260pub struct PartitionId {
261 pub kind: EntryKind,
262 pub bucket: u64,
263}
264
265impl PartitionId {
266 pub fn of(dimension: EntryKind, key: &StorageRowKey) -> Self {
267 Self {
268 kind: dimension,
269 bucket: bucket_of(key),
270 }
271 }
272
273 fn storage(&self) -> StorageId {
274 match self.kind {
275 EntryKind::Source(storage, _) => storage,
276 _ => panic!("a range partition outside a source entry kind names no row band"),
277 }
278 }
279
280 pub fn span(&self) -> (Edge<StorageRowKey>, Edge<StorageRowKey>) {
281 let start = Edge::Key(bucket_start(self.bucket));
282 let end = match self.bucket + 1 {
283 next if next < BUCKETS => Edge::Key(bucket_start(next)),
284 _ => Edge::Top,
285 };
286 (start, end)
287 }
288}
289
290fn bucket_of(key: &StorageRowKey) -> u64 {
291 !key.row().0 >> ROW_BUCKET_SHIFT
292}
293
294fn bucket_start(bucket: u64) -> StorageRowKey {
295 StorageRowKey::new(RowNumber(!(bucket << ROW_BUCKET_SHIFT)))
296}
297
298#[derive(Clone, Debug)]
299pub struct MultiRow {
300 pub version: CommitVersion,
301 pub value: Option<CowVec<u8>>,
302}
303
304impl RowBytes for MultiRow {
305 fn row_bytes(&self) -> usize {
306 self.value.as_ref().map_or(0, |value| value.len())
307 }
308}
309
310impl RangeDomain for MultiDomain {
311 type Dimension = EntryKind;
312 type Partition = PartitionId;
313 type Key = StorageRowKey;
314 type MetricBucket = ();
315 type Row = MultiRow;
316
317 const METRIC_BUCKETS: usize = 1;
318
319 const SCOPE: &'static str = "multi_range";
320
321 const GAP_SCOPE: &'static str = "multi_range::gaps";
322
323 fn just_past(key: &Self::Key) -> Edge<Self::Key> {
324 Edge::just_past(key)
325 }
326
327 fn partition(dimension: Self::Dimension, key: &Self::Key) -> Self::Partition {
328 PartitionId::of(dimension, key)
329 }
330
331 fn dimension(partition: &Self::Partition) -> Self::Dimension {
332 partition.kind
333 }
334
335 fn span(partition: &Self::Partition) -> (Edge<Self::Key>, Edge<Self::Key>) {
336 partition.span()
337 }
338
339 fn head_band(dimension: Self::Dimension) -> Option<(Edge<Self::Key>, Edge<Self::Key>)> {
340 StorageRowKey::owns(dimension).map(|_| (Edge::Bottom, Edge::Top))
341 }
342
343 fn caches_ranges(partition: &Self::Partition) -> bool {
344 StorageRowKey::owns(partition.kind).is_some() && partition.kind.caches_ranges()
345 }
346
347 fn cache_run_end(_partition: &Self::Partition) -> Edge<Self::Key> {
348 Edge::Top
349 }
350
351 fn supersedes(resident: &Self::Row, incoming: &Self::Row) -> bool {
352 incoming.version >= resident.version
353 }
354
355 fn admits_unproven_writes() -> bool {
356 true
357 }
358
359 fn metric_bucket(_partition: &Self::Partition) -> usize {
360 0
361 }
362
363 fn metric_bucket_at(_index: usize) -> Self::MetricBucket {}
364
365 fn metric_bucket_name(_slot: Self::MetricBucket) -> Cow<'static, str> {
366 Cow::Borrowed("row")
367 }
368}
369
370#[derive(Clone, Copy, Debug)]
371pub struct MultiRangeShardMetrics {
372 pub shard: usize,
373 pub used: ByteSize,
374 pub limit: ByteSize,
375 pub partitions: usize,
376 pub entries: usize,
377 pub complete_partitions: usize,
378 pub counters: RangeMetrics,
379 pub serve: MultiServeMetrics,
380}
381
382#[derive(Clone, Copy, Debug, Default)]
383pub struct MultiServeMetrics {
384 pub served: u64,
385 pub rows: u64,
386 pub head_advances: u64,
387}
388
389#[derive(Default)]
390struct ServeCounters {
391 served: AtomicU64,
392 rows: AtomicU64,
393 head_advances: AtomicU64,
394}
395
396#[derive(Clone)]
397pub struct MultiRangeTier {
398 tier: RangeTier<MultiDomain>,
399 serves: Arc<[ServeCounters]>,
400}
401
402impl MultiRangeTier {
403 pub fn new(config: MultiRangeConfig) -> Option<Self> {
404 let tier = RangeTier::new(config.into())?;
405 let shards = config.shards.max(1);
406 Some(Self {
407 tier,
408 serves: (0..shards).map(|_| ServeCounters::default()).collect(),
409 })
410 }
411
412 pub fn serve_metrics(&self) -> Vec<MultiServeMetrics> {
413 self.serves
414 .iter()
415 .map(|counters| MultiServeMetrics {
416 served: counters.served.load(Ordering::Relaxed),
417 rows: counters.rows.load(Ordering::Relaxed),
418 head_advances: counters.head_advances.load(Ordering::Relaxed),
419 })
420 .collect()
421 }
422
423 pub fn complete_partitions(&self) -> Vec<usize> {
424 self.tier.complete_partitions()
425 }
426
427 #[instrument(name = "store::multi::range::insert", level = "trace", skip_all, fields(table = ?table, version = version.0))]
428 pub fn insert(&self, table: EntryKind, key: EncodedKey, version: CommitVersion, value: Option<CowVec<u8>>) {
429 let Some(key) = narrow(table, &key) else {
430 return;
431 };
432 self.tier.insert(
433 table,
434 key,
435 MultiRow {
436 version,
437 value,
438 },
439 );
440 }
441
442 pub fn invalidate(&self, table: EntryKind, key: &EncodedKey) {
443 let Some(key) = narrow(table, key) else {
444 return;
445 };
446 self.tier.invalidate(table, &key);
447 }
448
449 pub fn clear(&self) {
450 self.tier.clear();
451 }
452
453 pub fn shard_metrics(&self) -> Vec<RangeShardMetrics> {
454 self.tier.shard_metrics()
455 }
456
457 pub fn full_shard_metrics(&self) -> Vec<MultiRangeShardMetrics> {
458 let shards = self.tier.shard_metrics();
459 let serves = self.serve_metrics();
460 let complete = self.complete_partitions();
461 reifydb_assertions! {
462 assert_eq!(
463 (shards.len(), shards.len()),
464 (serves.len(), complete.len()),
465 "every shard must report all three sources, or a shard past the shortest reports zero forever"
466 );
467 }
468 shards.into_iter()
469 .zip(serves)
470 .zip(complete)
471 .map(|((shard, serve), complete_partitions)| MultiRangeShardMetrics {
472 shard: shard.shard,
473 used: shard.used,
474 limit: shard.limit,
475 partitions: shard.partitions,
476 entries: shard.entries,
477 complete_partitions,
478 counters: shard.counters,
479 serve,
480 })
481 .collect()
482 }
483
484 pub fn materialize_scanned_chunk(
485 &self,
486 table: EntryKind,
487 lo: &EncodedKey,
488 through: &EncodedKey,
489 entries: &[RawEntry],
490 ) -> bool {
491 if !table.caches_ranges() {
492 return false;
493 }
494 let (Some(lo), Some(through)) =
495 (narrow_bound(table, lo.as_slice()), narrow_bound(table, through.as_slice()))
496 else {
497 return false;
498 };
499 let rows: RangeRows<MultiDomain> = entries
500 .iter()
501 .filter_map(|entry| {
502 narrow(table, &entry.key).map(|key| {
503 (
504 key,
505 MultiRow {
506 version: entry.version,
507 value: entry.value.clone(),
508 },
509 )
510 })
511 })
512 .collect();
513 let proven = just_past(&through);
514 self.tier.raise_head(table, &lo, &proven, rows.first().map(|(key, _)| key), self.tier.retractions());
515 let Some(start) = anchor(table, &lo, &rows) else {
516 return false;
517 };
518 if !proven.covers(&start) {
519 return false;
520 }
521 let Some(scan) = self.tier.plan_scan(table, &KeyRange::new(Included(start), bound(&through))) else {
522 return false;
523 };
524 let span = Interval::new(Edge::Key(start), proven);
525 matches!(self.tier.materialize(&scan, &span, &rows), Materialize::Materialized)
526 }
527
528 #[allow(clippy::too_many_arguments)]
529 pub fn serve_persistent_chunk(
530 &self,
531 table: EntryKind,
532 cursor: &mut RangeCursor,
533 start: &[u8],
534 end: &[u8],
535 scope: MultiVersionScope,
536 batch_size: usize,
537 descending: bool,
538 ) -> ServedChunk {
539 if descending || !table.caches_ranges() {
540 return ServedChunk::Gap;
541 }
542 let (Some(range_lo), Some(range_hi)) = (narrow_bound(table, start), narrow_bound(table, end)) else {
543 return ServedChunk::Gap;
544 };
545 let hi = just_past(&range_hi);
546 if hi <= range_lo {
547 return ServedChunk::Gap;
548 }
549 let lo = match cursor.last_key() {
550 Some(last) if last.as_slice() >= start => match narrow(table, last) {
551 Some(last) => match last.successor() {
552 Some(next) => Edge::Key(next),
553 None => return ServedChunk::Gap,
554 },
555 None => return ServedChunk::Gap,
556 },
557 _ => range_lo,
558 };
559 if hi <= lo {
560 return ServedChunk::Gap;
561 }
562 let Some(low) = lo.lowest() else {
563 return ServedChunk::Gap;
564 };
565
566 let Some(scan) = self.tier.plan_scan(table, &KeyRange::new(Included(low), bound(&range_hi))) else {
567 return self.chunk_proven_empty(table, &lo, &range_hi, cursor);
568 };
569 let Some(Segment::Resident(segment)) = scan.segments().first() else {
570 return self.chunk_proven_empty(table, &lo, &range_hi, cursor);
571 };
572 let Some(at) = segment.start.anchor() else {
573 return ServedChunk::Gap;
574 };
575 let partition = PartitionId::of(table, &at);
576 let counters = &self.serves[self.tier.shard_index(&partition)];
577 if scan.advanced() {
578 counters.head_advances.fetch_add(1, Ordering::Relaxed);
579 }
580
581 let storage = partition.storage();
582 let mut served = Cursor::<(), StorageRowKey>::new();
583 let TierChunk::Served(rows) = self.tier.serve(&scan, segment, &mut served, batch_size) else {
584 return ServedChunk::Gap;
585 };
586 let out: Vec<RawEntry> = rows
587 .into_iter()
588 .filter(|(_, row)| scope.contains(row.version))
589 .map(|(key, row)| RawEntry {
590 key: StorageRowKey::widen(storage, &key),
591 version: row.version,
592 value: row.value,
593 })
594 .collect();
595
596 let exhausted = served.is_exhausted() && stops_in_band(table, end) && segment.end >= hi;
597 if !exhausted && out.is_empty() {
598 return ServedChunk::Gap;
599 }
600 counters.served.fetch_add(1, Ordering::Relaxed);
601 counters.rows.fetch_add(out.len() as u64, Ordering::Relaxed);
602 served_chunk(out, cursor, exhausted)
603 }
604
605 fn chunk_proven_empty(
606 &self,
607 table: EntryKind,
608 lo: &Edge<StorageRowKey>,
609 range_hi: &Edge<StorageRowKey>,
610 cursor: &mut RangeCursor,
611 ) -> ServedChunk {
612 if self.tier.head_proves_empty(table, lo, range_hi) {
613 return served_chunk(Vec::new(), cursor, true);
614 }
615 ServedChunk::Gap
616 }
617}
618
619fn just_past(end: &Edge<StorageRowKey>) -> Edge<StorageRowKey> {
620 match end {
621 Edge::Key(key) => Edge::just_past(key),
622 other => other.clone(),
623 }
624}
625
626fn bound(end: &Edge<StorageRowKey>) -> Bound<StorageRowKey> {
627 match end {
628 Edge::Bottom => Bound::Excluded(StorageRowKey::low()),
629 Edge::Key(key) | Edge::AfterKey(key) => Included(*key),
630 Edge::Top => Bound::Unbounded,
631 }
632}
633
634fn anchor(table: EntryKind, lo: &Edge<StorageRowKey>, rows: &RangeRows<MultiDomain>) -> Option<StorageRowKey> {
635 match lo {
636 Edge::Bottom => {
637 let (first, _) = rows.first()?;
638 MultiDomain::span(&PartitionId::of(table, first)).0.lowest()
639 }
640 Edge::Key(key) => Some(*key),
641 Edge::AfterKey(_) | Edge::Top => None,
642 }
643}
644
645fn served_chunk(out: Vec<RawEntry>, cursor: &mut RangeCursor, exhausted: bool) -> ServedChunk {
646 reifydb_assertions! {
647 assert!(
648 exhausted || !out.is_empty(),
649 "a chunk that reports more must carry an entry, otherwise last_key never advances and the store's scan loop, which now ends only when every tier cursor is exhausted, spins forever"
650 );
651 }
652 if let Some(last) = out.last() {
653 cursor.advance(last.key.clone());
654 }
655 if exhausted {
656 cursor.finish();
657 }
658 ServedChunk::Served(RangeBatch {
659 entries: out,
660 has_more: !exhausted,
661 })
662}
663
664#[cfg(test)]
665mod tests {
666 use reifydb_core::{
667 common::CommitVersion,
668 interface::{
669 catalog::{id::TableId, storage::StorageId},
670 store::EntryLayout,
671 },
672 key::{
673 row::{RowKey, StorageRowKey},
674 series::SeriesRowKey,
675 typed::range::KeyRange,
676 },
677 };
678 use reifydb_store::coverage::plan::DEFAULT_GAP_GUARD;
679 use reifydb_value::{byte_size::ByteSize, util::cowvec::CowVec, value::row_number::RowNumber};
680
681 use super::{
682 Bound, Edge, EncodedKey, EntryKind, MultiDomain, MultiRangeConfig, MultiRangeTier, MultiVersionScope,
683 PartitionId, ROW_BUCKET_SHIFT, RangeCursor, RangeDomain, RawEntry, Segment, ServedChunk, narrow,
684 narrow_bound,
685 };
686
687 const STORAGE: StorageId = StorageId::Table(TableId(1));
688 const NEIGHBOUR: StorageId = StorageId::Table(TableId(0));
689
690 fn tier() -> MultiRangeTier {
691 MultiRangeTier::new(MultiRangeConfig {
692 shard_bytes: Some(ByteSize::from_mib(1)),
693 shards: 4,
694 gap_guard: DEFAULT_GAP_GUARD,
695 })
696 .expect("a tier with a byte budget must be constructed")
697 }
698
699 const BUCKET: u64 = 1 << ROW_BUCKET_SHIFT;
700
701 fn tight() -> MultiRangeTier {
702 MultiRangeTier::new(MultiRangeConfig {
703 shard_bytes: Some(ByteSize::from_kib(4)),
704 shards: 1,
705 gap_guard: DEFAULT_GAP_GUARD,
706 })
707 .expect("a tier with a byte budget must be constructed")
708 }
709
710 fn row(n: u64) -> EncodedKey {
711 RowKey {
712 storage: STORAGE,
713 row: RowNumber(n),
714 }
715 .encode()
716 }
717
718 fn key(n: u64) -> StorageRowKey {
719 StorageRowKey::new(RowNumber(n))
720 }
721
722 fn series(n: u64) -> EncodedKey {
723 SeriesRowKey {
724 storage: STORAGE,
725 variant_tag: None,
726 key: n,
727 sequence: 0,
728 }
729 .encode()
730 }
731
732 fn source() -> EntryKind {
733 EntryKind::Source(STORAGE, EntryLayout::Row)
734 }
735
736 fn entry(n: u64, version: u64) -> RawEntry {
737 RawEntry {
738 key: row(n),
739 version: CommitVersion(version),
740 value: Some(CowVec::new(version.to_be_bytes().to_vec())),
741 }
742 }
743
744 fn newest() -> MultiVersionScope {
745 MultiVersionScope::AsOf {
746 read: CommitVersion(u64::MAX),
747 }
748 }
749
750 fn storage_start() -> EncodedKey {
751 RowKey::storage_start(STORAGE)
752 }
753
754 fn storage_end() -> EncodedKey {
755 RowKey::storage_end(STORAGE)
756 }
757
758 fn materialize_from_prefix(tier: &MultiRangeTier, rows: &[u64], version: u64) {
762 let entries: Vec<RawEntry> = rows.iter().map(|n| entry(*n, version)).collect();
763 tier.materialize_scanned_chunk(source(), &storage_start(), &storage_end(), &entries);
764 }
765
766 fn serve_whole_storage(tier: &MultiRangeTier, cursor: &mut RangeCursor) -> ServedChunk {
767 tier.serve_persistent_chunk(
768 source(),
769 cursor,
770 storage_start().as_slice(),
771 storage_end().as_slice(),
772 newest(),
773 64,
774 false,
775 )
776 }
777
778 fn head_advances(tier: &MultiRangeTier) -> u64 {
779 tier.serve_metrics().iter().map(|shard| shard.head_advances).sum()
780 }
781
782 fn serve(
785 tier: &MultiRangeTier,
786 cursor: &mut RangeCursor,
787 lo_row: u64,
788 hi_row: u64,
789 batch: usize,
790 ) -> ServedChunk {
791 let start = row(hi_row);
792 let end = row(lo_row);
793 tier.serve_persistent_chunk(source(), cursor, start.as_slice(), end.as_slice(), newest(), batch, false)
794 }
795
796 fn rows_of(chunk: &ServedChunk) -> Vec<u64> {
797 match chunk {
798 ServedChunk::Served(batch) => batch
799 .entries
800 .iter()
801 .map(|e| RowKey::decode(&e.key).expect("a served row key must decode").row.0)
802 .collect(),
803 ServedChunk::Gap => panic!("expected a served chunk, got a gap"),
804 }
805 }
806
807 fn is_gap(chunk: &ServedChunk) -> bool {
808 matches!(chunk, ServedChunk::Gap)
809 }
810
811 fn fill_bucket(tier: &MultiRangeTier, bucket: u64, rows: &[u64], version: u64) {
812 let base = bucket * BUCKET;
815 let mut entries: Vec<RawEntry> = rows.iter().map(|n| entry(*n, version)).collect();
816 entries.sort_by(|left, right| left.key.cmp(&right.key));
817 assert!(
818 tier.materialize_scanned_chunk(source(), &row(base + BUCKET - 1), &row(base), &entries),
819 "a whole-bucket chunk must publish its claim"
820 );
821 }
822
823 #[test]
824 fn a_write_into_a_partition_no_claim_reached_is_still_seated() {
825 let tier = tier();
827
828 tier.insert(source(), RowKey::encoded(STORAGE, 1), CommitVersion(1), Some(CowVec::new(b"v".to_vec())));
829
830 let entries: usize = tier.shard_metrics().iter().map(|shard| shard.entries).sum();
831 assert_eq!(entries, 1, "the write was dropped, so a claim taken across it answers the row absent");
832 }
833
834 #[test]
835 fn a_claim_taken_across_a_declined_write_must_not_answer_that_row_absent() {
836 let tier = tier();
839 let kind = EntryKind::Source(STORAGE, EntryLayout::Row);
840 let flushed = RowKey::encoded(STORAGE, 5);
841 let lo = RowKey::encoded(STORAGE, 9);
842 let through = RowKey::encoded(STORAGE, 1);
843 assert!(
844 lo < through,
845 "row keys encode descending, so the low end of the span is the highest row number"
846 );
847
848 tier.insert(kind, flushed.clone(), CommitVersion(1), Some(CowVec::new(b"flushed".to_vec())));
849
850 let stale = [RawEntry {
851 key: lo.clone(),
852 version: CommitVersion(1),
853 value: Some(CowVec::new(b"scanned".to_vec())),
854 }];
855 assert!(
856 tier.materialize_scanned_chunk(kind, &lo, &through, &stale),
857 "the chunk must claim its span, or the test never reaches the case it is here to pin"
858 );
859
860 let mut cursor = RangeCursor::new();
861 let served = tier.serve_persistent_chunk(
862 kind,
863 &mut cursor,
864 lo.as_slice(),
865 through.as_slice(),
866 MultiVersionScope::AsOf {
867 read: CommitVersion(10),
868 },
869 32,
870 false,
871 );
872 let ServedChunk::Served(batch) = served else {
873 panic!("a claimed span must serve from ram, or the claim bought nothing");
874 };
875 assert!(
876 batch.entries.iter().any(|entry| entry.key == flushed),
877 "the claim outranked a flushed row the persistent read never saw, so the row reads as absent"
878 );
879 }
880
881 #[test]
882 fn the_multi_domain_hands_durability_to_ram_rather_than_declining_a_write() {
883 assert!(
885 MultiDomain::admits_unproven_writes(),
886 "multi hands a flushed row to ram unconditionally, or the row is lost between the buffer and the claim"
887 );
888 }
889
890 #[test]
891 fn a_key_the_domain_cannot_attribute_names_no_partition() {
892 let stray = EncodedKey::new(vec![0u8, 1, 2]);
897 assert_eq!(
898 narrow(EntryKind::Source(STORAGE, EntryLayout::Row), &stray),
899 None,
900 "a key shorter than the band prefix carries no bucket to attribute it by"
901 );
902 assert_eq!(
903 narrow(EntryKind::Multi, &RowKey::encoded(STORAGE, 5)),
904 None,
905 "a row key under a kind with no row band must not be attributed either"
906 );
907 assert_eq!(
908 narrow(EntryKind::Source(NEIGHBOUR, EntryLayout::Row), &RowKey::encoded(STORAGE, 5)),
909 None,
910 "a row key of another storage must not be attributed to this one"
911 );
912 }
913
914 #[test]
915 fn an_older_write_must_not_displace_a_newer_resident_row() {
916 let tier = tier();
919 let kind = EntryKind::Source(STORAGE, EntryLayout::Row);
920 let key = RowKey::encoded(STORAGE, 5);
921 let through = RowKey::encoded(STORAGE, 1);
922
923 let newer = [RawEntry {
924 key: key.clone(),
925 version: CommitVersion(5),
926 value: Some(CowVec::new(b"v5".to_vec())),
927 }];
928 assert!(
929 tier.materialize_scanned_chunk(kind, &key, &through, &newer),
930 "the chunk must claim its span, or the write below never lands on a resident row"
931 );
932
933 tier.insert(kind, key.clone(), CommitVersion(2), Some(CowVec::new(b"v2".to_vec())));
934
935 let mut cursor = RangeCursor::new();
936 let served = tier.serve_persistent_chunk(
937 kind,
938 &mut cursor,
939 key.as_slice(),
940 through.as_slice(),
941 MultiVersionScope::AsOf {
942 read: CommitVersion(10),
943 },
944 32,
945 false,
946 );
947 let ServedChunk::Served(batch) = served else {
948 panic!("the claimed span must serve from ram");
949 };
950 let entry =
951 batch.entries.iter().find(|entry| entry.key == key).expect("the row must still be resident");
952 assert_eq!(entry.version, CommitVersion(5), "the older write must not have displaced the newer row");
953 assert_eq!(entry.value.as_ref().expect("a value, not a tombstone").as_ref(), b"v5");
954 }
955
956 #[test]
957 fn evicting_the_partition_the_head_came_from_leaves_the_head_standing() {
958 let tier = tight();
964 let entries = vec![entry(BUCKET * 4 + 3, 1), entry(BUCKET * 4 + 2, 1), entry(BUCKET * 4 + 1, 1)];
965 assert!(
966 tier.materialize_scanned_chunk(source(), &storage_start(), &storage_end(), &entries),
967 "the chunk must publish its claim, or the test never reaches the case it is here to pin"
968 );
969 assert_eq!(
970 tier.tier.head(source()),
971 Some(Edge::Key(key(BUCKET * 4 + 3))),
972 "the materialize must have recorded a head"
973 );
974 assert!(
975 tier.tier.lookup(source(), &key(BUCKET * 4 + 2)).is_some(),
976 "the materialize must have published a claim"
977 );
978
979 for n in 1..=512 {
980 tier.insert(source(), row(n), CommitVersion(1), Some(CowVec::new(vec![n as u8; 8])));
981 }
982
983 assert!(
984 tier.tier.lookup(source(), &key(BUCKET * 4 + 2)).is_none(),
985 "the evicted partition's claim must be withdrawn, or it answers for rows ram no longer holds"
986 );
987 assert_eq!(
988 tier.tier.head(source()),
989 Some(Edge::Key(key(BUCKET * 4 + 3))),
990 "eviction cannot create a row, so the proof of absence must survive it"
991 );
992 }
993
994 #[test]
995 fn a_row_placed_into_ram_below_the_head_pulls_the_head_back_to_it() {
996 let tier = tier();
1001 tier.tier.raise_head(source(), &Edge::Bottom, &Edge::Top, Some(&key(3)), tier.tier.retractions());
1002
1003 tier.insert(source(), row(7), CommitVersion(1), Some(CowVec::new(vec![1])));
1004
1005 assert_eq!(
1006 tier.tier.head(source()),
1007 Some(Edge::Key(key(7))),
1008 "a row placed inside the head span must pull the head back to it"
1009 );
1010 }
1011
1012 #[test]
1013 fn a_head_raise_that_read_its_token_before_a_withdrawal_publishes_nothing() {
1014 let tier = tier();
1018 let token = tier.tier.retractions();
1019
1020 tier.invalidate(source(), &row(7));
1021
1022 tier.tier.raise_head(source(), &Edge::Bottom, &Edge::Top, Some(&key(3)), token);
1023 assert_eq!(tier.tier.head(source()), None, "a head published across a withdrawal");
1024
1025 tier.tier.raise_head(source(), &Edge::Bottom, &Edge::Top, Some(&key(3)), tier.tier.retractions());
1026 assert_eq!(tier.tier.head(source()), Some(Edge::Key(key(3))), "a fresh token must publish");
1027 }
1028
1029 #[test]
1030 fn a_scan_below_the_row_band_never_raises_a_head_over_it() {
1031 let tier = tier();
1037
1038 assert_eq!(
1039 narrow_bound(source(), series(9).as_slice()),
1040 None,
1041 "a bound below the row band must not resolve to an edge of it"
1042 );
1043 assert!(
1044 !tier.materialize_scanned_chunk(source(), &series(9), &storage_end(), &[]),
1045 "a scan that started below the row band must not be claimed"
1046 );
1047
1048 assert_eq!(
1049 tier.tier.head(source()),
1050 None,
1051 "a scan that never entered the row band proved nothing about it"
1052 );
1053 }
1054
1055 #[test]
1056 fn a_scan_starting_at_a_storage_prefix_serves_once_the_head_names_the_first_row() {
1057 let tier = tier();
1063 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1064
1065 let mut cursor = RangeCursor::new();
1066 let chunk = serve_whole_storage(&tier, &mut cursor);
1067
1068 assert_eq!(rows_of(&chunk), vec![3, 2, 1], "the leading chunk of a prefix scan must serve from ram");
1069 assert!(cursor.is_exhausted(), "the claim reaches the storage end, so nothing is left for persistent");
1070 assert_eq!(
1071 head_advances(&tier),
1072 1,
1073 "the serve must be attributed to the head, not to a claim over the prefix"
1074 );
1075 }
1076
1077 #[test]
1078 fn a_commit_below_the_head_pulls_it_back_and_stops_the_scan_skipping_the_new_row() {
1079 let tier = tier();
1084 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1085 assert_eq!(
1086 tier.tier.head(source()),
1087 Some(Edge::Key(key(3))),
1088 "the materialize must have recorded a head"
1089 );
1090
1091 tier.invalidate(source(), &row(7));
1092
1093 assert_eq!(
1094 tier.tier.head(source()),
1095 Some(Edge::Key(key(7))),
1096 "a row committed inside the head span must pull the head back to it"
1097 );
1098 let mut cursor = RangeCursor::new();
1099 let chunk = serve_whole_storage(&tier, &mut cursor);
1100 assert!(
1101 is_gap(&chunk),
1102 "the span the commit landed in is no longer claimed, so the scan must fall through"
1103 );
1104 assert!(!cursor.is_exhausted(), "a gap must leave the cursor untouched");
1105 }
1106
1107 #[test]
1108 fn the_head_never_moves_a_scan_past_the_end_of_its_own_range() {
1109 let tier = tier();
1113 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1114 assert_eq!(
1115 tier.tier.head(source()),
1116 Some(Edge::Key(key(3))),
1117 "the materialize must have recorded a head"
1118 );
1119
1120 let mut cursor = RangeCursor::new();
1121 let chunk = serve(&tier, &mut cursor, 5, 9, 64);
1122
1123 assert!(rows_of(&chunk).is_empty(), "no row of this storage lies in rows five through nine");
1124 assert!(cursor.is_exhausted(), "the claim spans the whole range, so ram has proven it empty");
1125 assert_eq!(head_advances(&tier), 0, "the head sorts past this range and must not have been used");
1126 }
1127
1128 #[test]
1129 fn a_range_below_the_row_band_is_never_moved_onto_it_by_the_head() {
1130 let tier = tier();
1135 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1136 tier.insert(source(), series(1), CommitVersion(10), Some(CowVec::new(vec![1])));
1137 assert!(
1138 series(1).as_slice() < storage_start().as_slice(),
1139 "the series band must sort below the row band, or this range never crosses the boundary"
1140 );
1141
1142 let mut cursor = RangeCursor::new();
1143 let chunk = tier.serve_persistent_chunk(
1144 source(),
1145 &mut cursor,
1146 series(9).as_slice(),
1147 storage_end().as_slice(),
1148 newest(),
1149 64,
1150 false,
1151 );
1152
1153 assert!(is_gap(&chunk), "a range starting below the row band must never be answered from a row head");
1154 assert!(!cursor.is_exhausted(), "a gap must leave the cursor untouched");
1155 assert_eq!(head_advances(&tier), 0, "the head must not have been applied outside its own band");
1156 }
1157
1158 #[test]
1159 fn an_empty_storage_is_read_from_persistent_once_and_never_again() {
1160 let tier = tier();
1163
1164 let mut first = RangeCursor::new();
1165 assert!(is_gap(&serve_whole_storage(&tier, &mut first)), "nothing is proven before the first scan");
1166
1167 tier.materialize_scanned_chunk(source(), &storage_start(), &storage_end(), &[]);
1168
1169 let mut second = RangeCursor::new();
1170 let chunk = serve_whole_storage(&tier, &mut second);
1171 assert!(!is_gap(&chunk), "the proven-empty storage must never reach the persistent tier again");
1172 assert!(rows_of(&chunk).is_empty(), "a proven-empty range must serve no rows");
1173 assert!(
1174 second.is_exhausted(),
1175 "an empty range that is not exhausted hands the scan straight back to persistent"
1176 );
1177 }
1178
1179 #[test]
1180 fn a_range_ending_on_the_head_is_never_answered_empty() {
1181 let tier = tier();
1184 materialize_from_prefix(&tier, &[5, 3], 10);
1185 assert_eq!(
1186 tier.tier.head(source()),
1187 Some(Edge::Key(key(5))),
1188 "the materialize must name the first row as the head"
1189 );
1190
1191 tier.invalidate(source(), &row(5));
1192 tier.invalidate(source(), &row(3));
1193
1194 let mut cursor = RangeCursor::new();
1195 let chunk = tier.serve_persistent_chunk(
1196 source(),
1197 &mut cursor,
1198 storage_start().as_slice(),
1199 row(5).as_slice(),
1200 newest(),
1201 64,
1202 false,
1203 );
1204 assert!(
1205 is_gap(&chunk),
1206 "a range whose last key is the head itself is not proven empty and the persistent tier still owes it"
1207 );
1208 assert!(!cursor.is_exhausted(), "a gap must leave the cursor untouched");
1209 }
1210
1211 #[test]
1212 fn a_serve_reports_exhausted_only_when_the_claim_reaches_past_the_range_end() {
1213 let intact = tier();
1217 fill_bucket(&intact, 0, &[2, 4, 6], 10);
1218
1219 let mut whole = RangeCursor::new();
1220 let chunk = serve(&intact, &mut whole, 0, BUCKET - 1, 64);
1221 assert_eq!(rows_of(&chunk), vec![6, 4, 2]);
1222 assert!(whole.is_exhausted(), "a claim spanning the whole range has proven the rest of it empty");
1223
1224 let punched = tier();
1225 fill_bucket(&punched, 0, &[2, 4, 6], 10);
1226 punched.invalidate(source(), &row(1));
1227
1228 let mut clipped = RangeCursor::new();
1229 let chunk = serve(&punched, &mut clipped, 0, BUCKET - 1, 64);
1230 assert_eq!(rows_of(&chunk), vec![6, 4, 2], "the rows below the punched key are the same");
1231 assert!(
1232 !clipped.is_exhausted(),
1233 "the claim now ends at the punched key, so the persistent tier still owes the rest"
1234 );
1235 }
1236
1237 #[test]
1238 fn a_claim_that_scanned_to_the_storage_end_reports_exhausted_there() {
1239 let tier = tier();
1242 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1243
1244 let mut cursor = RangeCursor::new();
1245 cursor.advance(row(3));
1246 let chunk = serve_whole_storage(&tier, &mut cursor);
1247
1248 assert!(
1249 cursor.is_exhausted(),
1250 "a claim that scanned to the storage end has proven the rest of it empty"
1251 );
1252 assert_eq!(rows_of(&chunk), vec![2, 1]);
1253 }
1254
1255 #[test]
1256 fn a_claim_punched_short_of_the_storage_end_is_not_exhausted() {
1257 let tier = tier();
1260 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1261 tier.invalidate(source(), &row(1));
1262
1263 let mut cursor = RangeCursor::new();
1264 cursor.advance(row(3));
1265 let chunk = serve_whole_storage(&tier, &mut cursor);
1266
1267 assert!(!cursor.is_exhausted(), "the claim stops at the punched key, which proves nothing past it");
1268 assert_eq!(rows_of(&chunk), vec![2]);
1269 }
1270
1271 #[test]
1272 fn a_claim_stopping_on_its_last_row_rather_than_past_it_is_not_exhausted() {
1273 let tier = tier();
1276 let entries = vec![entry(3, 10), entry(2, 10), entry(1, 10)];
1277 assert!(
1278 tier.materialize_scanned_chunk(source(), &storage_start(), &row(1), &entries),
1279 "the chunk must publish its claim, or the test never reaches the case it is here to pin"
1280 );
1281
1282 let mut cursor = RangeCursor::new();
1283 cursor.advance(row(3));
1284 let chunk = serve_whole_storage(&tier, &mut cursor);
1285
1286 assert!(
1287 !cursor.is_exhausted(),
1288 "the claim stops on the last row it read, which proves nothing past it"
1289 );
1290 assert_eq!(rows_of(&chunk), vec![2, 1]);
1291 }
1292
1293 #[test]
1294 fn a_claim_over_a_partition_that_is_not_the_last_is_not_exhausted_at_the_storage_end() {
1295 let tier = tier();
1299 fill_bucket(&tier, 1, &[BUCKET + 1, BUCKET + 2], 10);
1300 fill_bucket(&tier, 0, &[1, 2], 10);
1301
1302 let mut cursor = RangeCursor::new();
1303 cursor.advance(row(BUCKET + 2));
1304 let chunk = serve_whole_storage(&tier, &mut cursor);
1305
1306 assert!(
1307 !cursor.is_exhausted(),
1308 "the lower partition is a separate claim the persistent tier still owes"
1309 );
1310 assert_eq!(rows_of(&chunk), vec![BUCKET + 1]);
1311 }
1312
1313 #[test]
1314 fn a_range_reaching_past_the_storage_end_is_never_reported_exhausted() {
1315 let tier = tier();
1318 materialize_from_prefix(&tier, &[3, 2, 1], 10);
1319
1320 let end = RowKey::encoded(NEIGHBOUR, 5);
1321 let mut cursor = RangeCursor::new();
1322 cursor.advance(row(3));
1323 let chunk = tier.serve_persistent_chunk(
1324 source(),
1325 &mut cursor,
1326 storage_start().as_slice(),
1327 end.as_slice(),
1328 newest(),
1329 64,
1330 false,
1331 );
1332
1333 assert!(!cursor.is_exhausted(), "the claim says nothing about the storage the range runs on into");
1334 assert_eq!(rows_of(&chunk), vec![2, 1]);
1335 assert!(
1336 end.as_slice() > storage_end().as_slice(),
1337 "the range end must really sort past this storage, or the case under test never arose"
1338 );
1339 }
1340
1341 #[test]
1342 fn a_claim_serves_a_partition_no_longer_covered_end_to_end() {
1343 let tier = tier();
1346 fill_bucket(&tier, 0, &[1, 2, 3, 4, 5], 10);
1347 tier.invalidate(source(), &row(3));
1348
1349 let mut cursor = RangeCursor::new();
1350 let chunk = serve(&tier, &mut cursor, 0, BUCKET - 1, 64);
1351
1352 assert_eq!(
1353 rows_of(&chunk),
1354 vec![5, 4],
1355 "the claim below the punched key must still serve, where a whole-partition claim serves nothing"
1356 );
1357 assert!(!cursor.is_exhausted(), "a claim that stops at the punched key has proven nothing beyond it");
1358 }
1359
1360 #[test]
1361 fn a_scan_starting_at_a_storage_prefix_is_not_claimed_and_falls_through() {
1362 let tier = tier();
1366 fill_bucket(&tier, 0, &[1, 2, 3], 10);
1367
1368 let (lo, hi) = (
1369 narrow_bound(source(), storage_start().as_slice()).expect("a row source narrows its bounds"),
1370 narrow_bound(source(), storage_end().as_slice()).expect("a row source narrows its bounds"),
1371 );
1372 assert_eq!(
1373 (lo.clone(), hi),
1374 (Edge::Bottom, Edge::Top),
1375 "the storage prefix and the storage end are the two edges of the band"
1376 );
1377 let range = KeyRange::new(
1378 Bound::Included(lo.lowest().expect("the bottom edge lowers to the lowest key")),
1379 Bound::Unbounded,
1380 );
1381 let plan = tier.tier.plan_scan(source(), &range).expect("a whole storage must be plannable");
1382 assert!(
1383 matches!(plan.segments().first(), Some(Segment::Gap { .. })),
1384 "a claim reached below the lowest key its materialize observed, down to a prefix nothing proved"
1385 );
1386
1387 let mut cursor = RangeCursor::new();
1388 let chunk = serve_whole_storage(&tier, &mut cursor);
1389 assert!(is_gap(&chunk), "no claim covers the prefix the scan starts at");
1390
1391 cursor.advance(row(3));
1392 let resumed = serve_whole_storage(&tier, &mut cursor);
1393 assert_eq!(rows_of(&resumed), vec![2, 1], "once the cursor is on a real key the claim serves");
1394 }
1395
1396 #[test]
1397 fn a_series_key_is_never_attributed_to_a_row_partition() {
1398 assert!(
1403 series(1).as_slice() < RowKey::storage_start(STORAGE).as_slice(),
1404 "the series band must sort below the row band"
1405 );
1406 assert_eq!(narrow(source(), &series(1)), None, "a series key must name no row partition");
1407 assert_eq!(
1408 narrow(source(), &series(u64::MAX)),
1409 None,
1410 "no series key of the band may be attributed to a row partition"
1411 );
1412 assert_eq!(
1413 narrow_bound(source(), series(1).as_slice()),
1414 None,
1415 "a bound in the series band must resolve to no edge of the row band, or a scan starting there \
1416 slides onto rows it never asked for"
1417 );
1418 }
1419
1420 #[test]
1421 fn the_last_bucket_spans_to_the_top_of_its_own_dimension() {
1422 for (storage, other) in
1429 [(STORAGE, NEIGHBOUR), (NEIGHBOUR, STORAGE), (StorageId::Table(TableId(u64::MAX)), STORAGE)]
1430 {
1431 let kind = EntryKind::Source(storage, EntryLayout::Row);
1432 let last = u64::MAX >> ROW_BUCKET_SHIFT;
1433
1434 let (_, end) = PartitionId {
1435 kind,
1436 bucket: last,
1437 }
1438 .span();
1439 assert!(
1440 matches!(end, Edge::Top),
1441 "the last bucket of {storage:?} must span to the top of its own dimension"
1442 );
1443
1444 for bucket in [0u64, 1] {
1445 let (_, end) = PartitionId {
1446 kind,
1447 bucket,
1448 }
1449 .span();
1450 assert!(
1451 !matches!(end, Edge::Top),
1452 "bucket {bucket} of {storage:?} must stop at a successor, not at the ceiling"
1453 );
1454 }
1455
1456 assert_eq!(
1457 narrow(kind, &RowKey::encoded(other, 1)),
1458 None,
1459 "a row of {other:?} names no key of {storage:?}, so reaching Top retracts nothing of it"
1460 );
1461 }
1462 }
1463}
1464
1465impl MetricsCollector for MultiRangeTier {
1466 fn collect(&self, out: &mut Vec<MetricsSample>) {
1467 self.tier.collect(out);
1468 }
1469}