1use std::collections::Bound;
5
6use reifydb_codec::key::{
7 deserializer::KeyDeserializer,
8 encoded::{EncodedKey, EncodedKeyRange},
9};
10use reifydb_macro::KeyCodec;
11use reifydb_value::value::partition::Partition;
12
13use super::KeyTag;
14use crate::{
15 interface::catalog::{id::SeriesId, object::ObjectId, storage::StorageId},
16 key::{
17 any::{Field, KeyFields, TaggedKey, Width},
18 bound::{OwnedField, TaggedKeyBound, TaggedKeyBoundRange, object_fields},
19 catalog::{KeyDeserializerCatalogExt, KeySerializerCatalogExt},
20 typed::{BoundedKey, DenseKey, direction::Desc},
21 },
22 metrics::heap::HeapSize,
23};
24
25#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
26#[key(tag = Series)]
27pub struct SeriesKey {
28 pub series: SeriesId,
29}
30
31impl SeriesKey {
32 pub fn new(series: SeriesId) -> Self {
33 Self {
34 series,
35 }
36 }
37
38 pub fn encoded(series: impl Into<SeriesId>) -> EncodedKey {
39 Self::new(series.into()).encode()
40 }
41
42 pub fn full_scan() -> TaggedKeyBoundRange {
43 TaggedKeyBoundRange::kind(Self::TAG)
44 }
45}
46
47#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
48#[key(tag = SeriesPartitionMetadata)]
49pub struct SeriesPartitionMetadataKey {
50 pub storage: StorageId,
51 pub partition: Partition,
52}
53
54impl SeriesPartitionMetadataKey {
55 pub fn new(storage: impl Into<StorageId>, partition: Partition) -> Self {
56 Self {
57 storage: storage.into(),
58 partition,
59 }
60 }
61
62 pub fn encoded(storage: impl Into<StorageId>, partition: Partition) -> EncodedKey {
63 Self::new(storage, partition).encode()
64 }
65
66 pub fn full_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
67 TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
68 }
69}
70
71#[cfg(test)]
72mod series_metadata_key_tests {
73 use std::collections::Bound;
74
75 use reifydb_codec::key::{encoded::EncodedKey, serializer::KeySerializer};
76
77 use super::{KeyTag, Partition, SeriesKey, SeriesPartitionMetadataKey};
78 use crate::key::bound::TaggedKeyBoundRange;
79
80 fn in_range(range: &TaggedKeyBoundRange, key: &EncodedKey) -> bool {
81 let start = match &range.start {
82 Bound::Included(b) => *key >= b.encode(),
83 Bound::Excluded(b) => *key > b.encode(),
84 Bound::Unbounded => true,
85 };
86 let end = match &range.end {
87 Bound::Included(b) => *key <= b.encode(),
88 Bound::Excluded(b) => *key < b.encode(),
89 Bound::Unbounded => true,
90 };
91 start && end
92 }
93 use crate::interface::catalog::{
94 id::{SeriesId, ViewId},
95 storage::StorageId,
96 };
97
98 #[test]
99 fn test_metadata_key_roundtrip_series() {
100 let key = SeriesPartitionMetadataKey {
102 storage: StorageId::Series(SeriesId(7)),
103 partition: Partition(0x0123_4567_89AB_CDEF_FEDC_BA98_7654_3210),
104 };
105 assert_eq!(SeriesPartitionMetadataKey::decode(&key.encode()).unwrap(), key);
106 }
107
108 #[test]
109 fn test_metadata_key_roundtrip_view() {
110 let key = SeriesPartitionMetadataKey {
112 storage: StorageId::View(ViewId(7)),
113 partition: Partition::default(),
114 };
115 assert_eq!(SeriesPartitionMetadataKey::decode(&key.encode()).unwrap(), key);
116 }
117
118 #[test]
119 fn test_metadata_key_separates_a_series_from_a_view_with_the_same_id() {
120 assert_ne!(
122 SeriesPartitionMetadataKey::encoded(SeriesId(7), Partition::default()),
123 SeriesPartitionMetadataKey::encoded(ViewId(7), Partition::default())
124 );
125 }
126
127 #[test]
128 fn test_metadata_full_scan_covers_every_partition_of_one_series() {
129 let range = SeriesPartitionMetadataKey::full_scan(SeriesId(7));
132 for partition in [Partition::default(), Partition(1), Partition(u128::MAX)] {
133 let key = SeriesPartitionMetadataKey::encoded(SeriesId(7), partition);
134 assert!(in_range(&range, &key), "partition {:?} must be inside the scan", partition);
135 }
136 assert!(
137 !in_range(&range, &SeriesPartitionMetadataKey::encoded(SeriesId(8), Partition::default())),
138 "another series must fall outside"
139 );
140 }
141
142 #[test]
143 fn test_metadata_key_separates_two_partitions_of_one_series() {
144 assert_ne!(
147 SeriesPartitionMetadataKey::encoded(SeriesId(7), Partition(1)),
148 SeriesPartitionMetadataKey::encoded(SeriesId(7), Partition(2))
149 );
150 }
151
152 #[test]
153 fn test_series_key_matches_legacy_byte_layout() {
154 for id in [SeriesId(0), SeriesId(1), SeriesId(u64::MAX)] {
155 let mut legacy = KeySerializer::with_capacity(9);
156 legacy.extend_u8(KeyTag::Series as u8).extend_u64(id);
157 assert_eq!(legacy.to_encoded_key().as_slice(), SeriesKey::encoded(id).as_slice());
158 }
159 }
160}
161
162#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
163#[key(tag = SeriesRow)]
164pub struct SeriesRowKey {
165 pub storage: StorageId,
166 pub variant_tag: Option<u8>,
167 pub key: u64,
168 pub sequence: u64,
169}
170
171#[derive(Debug, Clone)]
172pub struct SeriesRowKeyRange {
173 pub storage: StorageId,
174 pub variant_tag: Option<u8>,
175 pub key_start: Option<u64>,
176 pub key_end: Option<u64>,
177}
178
179impl SeriesRowKeyRange {
180 pub fn storage_start(storage: StorageId) -> EncodedKey {
181 TaggedKeyBound::prefix(SeriesRowKey::TAG, object_fields(ObjectId::from(storage))).encode()
182 }
183
184 pub fn storage_end(storage: StorageId) -> EncodedKey {
185 TaggedKeyBound::prefix(SeriesRowKey::TAG, object_fields(ObjectId::from(storage).prev())).encode()
186 }
187
188 pub fn full_scan(storage: StorageId, variant_tag: Option<u8>) -> TaggedKeyBoundRange {
189 SeriesRowKeyRange {
190 storage,
191 variant_tag,
192 key_start: None,
193 key_end: None,
194 }
195 .bounds()
196 }
197
198 pub fn scan_range(
199 storage: StorageId,
200 tagged: bool,
201 variant_tag: Option<u8>,
202 key_start: Option<u64>,
203 key_end: Option<u64>,
204 last: Option<&TaggedKey>,
205 ) -> TaggedKeyBoundRange {
206 assert!(
207 !(tagged && variant_tag.is_none() && (key_start.is_some() || key_end.is_some())),
208 "a key range over a tagged series needs a variant tag: the key sits after the tag in the row \
209 key, so no single range spans every tag and the range would silently match nothing"
210 );
211 if matches!(key_end, Some(0)) {
212 return TaggedKeyBoundRange::empty(SeriesRowKey::TAG);
213 }
214
215 SeriesRowKeyRange {
216 storage,
217 variant_tag,
218 key_start,
219 key_end,
220 }
221 .bounds()
222 .resume_after(last)
223 }
224
225 pub fn decode_storage(key: &EncodedKey) -> Option<StorageId> {
226 let mut de = KeyDeserializer::from_bytes(key.as_slice());
227
228 let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
229 if kind != SeriesRowKey::TAG {
230 return None;
231 }
232
233 StorageId::from_object(de.read_object_id().ok()?)
234 }
235
236 pub fn decode(range: &EncodedKeyRange) -> (Option<StorageId>, Option<StorageId>) {
237 let start = match &range.start {
238 Bound::Included(key) | Bound::Excluded(key) => Self::decode_storage(key),
239 Bound::Unbounded => None,
240 };
241
242 let end = match &range.end {
243 Bound::Included(key) | Bound::Excluded(key) => Self::decode_storage(key),
244 Bound::Unbounded => None,
245 };
246
247 (start, end)
248 }
249
250 fn head_fields(&self, tagged: bool) -> Vec<OwnedField> {
251 let mut fields = object_fields(ObjectId::from(self.storage)).to_vec();
252 match self.variant_tag {
253 Some(tag) => {
254 fields.push(Field::UDesc(Width::U8, 1));
255 fields.push(Field::UDesc(Width::U8, tag as u128));
256 }
257 None if tagged || self.key_start.is_some() || self.key_end.is_some() => {
258 fields.push(Field::UDesc(Width::U8, 0));
259 fields.push(Field::UDesc(Width::U8, 0));
260 }
261 None => {}
262 }
263 fields
264 }
265
266 fn bounds(&self) -> TaggedKeyBoundRange {
267 let mut start = self.head_fields(false);
268 if let Some(key_val) = self.key_end {
269 start.push(Field::UDesc(Width::U64, (key_val - 1) as u128));
270 }
271 let start = Bound::Included(TaggedKeyBound::prefix(SeriesRowKey::TAG, start));
272
273 let end = match self.key_start {
274 Some(key_val) => {
275 let mut end = self.head_fields(true);
276 end.push(Field::UDesc(Width::U64, key_val as u128));
277 end.push(Field::UDesc(Width::U64, 0));
278 Bound::Included(TaggedKeyBound::prefix(SeriesRowKey::TAG, end))
279 }
280 None => Bound::Excluded(TaggedKeyBound::prefix_end(
281 SeriesRowKey::TAG,
282 object_fields(ObjectId::from(self.storage)),
283 )),
284 };
285
286 TaggedKeyBoundRange {
287 start,
288 end,
289 }
290 }
291}
292
293#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
294pub struct StorageSeriesKey {
295 pub variant_tag: Desc<Option<u8>>,
296 pub key: Desc<u64>,
297 pub sequence: Desc<u64>,
298}
299
300impl StorageSeriesKey {
301 pub fn new(variant_tag: Option<u8>, key: u64, sequence: u64) -> Self {
302 Self {
303 variant_tag: Desc(variant_tag),
304 key: Desc(key),
305 sequence: Desc(sequence),
306 }
307 }
308
309 pub fn variant_tag(self) -> Option<u8> {
310 self.variant_tag.0
311 }
312
313 pub fn key(self) -> u64 {
314 self.key.0
315 }
316
317 pub fn sequence(self) -> u64 {
318 self.sequence.0
319 }
320
321 pub fn to_sql_columns(self) -> SeriesKeyColumns {
322 SeriesKeyColumns {
323 variant_tag: series_variant_tag_to_sql(self.variant_tag()),
324 key: series_key_to_sql(self.key()),
325 sequence: series_sequence_to_sql(self.sequence()),
326 }
327 }
328
329 pub fn from_sql_columns(columns: SeriesKeyColumns) -> Self {
330 Self::new(
331 series_variant_tag_from_sql(columns.variant_tag),
332 series_key_from_sql(columns.key),
333 series_sequence_from_sql(columns.sequence),
334 )
335 }
336
337 pub fn with_storage(self, storage: StorageId) -> SeriesRowKey {
338 SeriesRowKey {
339 storage,
340 variant_tag: self.variant_tag(),
341 key: self.key(),
342 sequence: self.sequence(),
343 }
344 }
345}
346
347impl From<SeriesRowKey> for StorageSeriesKey {
348 fn from(key: SeriesRowKey) -> Self {
349 StorageSeriesKey::new(key.variant_tag, key.key, key.sequence)
350 }
351}
352
353impl HeapSize for StorageSeriesKey {
354 fn heap_size(&self) -> usize {
355 0
356 }
357}
358
359impl BoundedKey for StorageSeriesKey {
360 fn low() -> Self {
361 Self {
362 variant_tag: lowest_variant_tag(),
363 key: <Desc<u64> as BoundedKey>::low(),
364 sequence: <Desc<u64> as BoundedKey>::low(),
365 }
366 }
367}
368
369impl DenseKey for StorageSeriesKey {
370 fn successor(&self) -> Option<Self> {
371 if let Some(sequence) = self.sequence.successor() {
372 return Some(Self {
373 variant_tag: self.variant_tag,
374 key: self.key,
375 sequence,
376 });
377 }
378 if let Some(key) = self.key.successor() {
379 return Some(Self {
380 variant_tag: self.variant_tag,
381 key,
382 sequence: <Desc<u64> as BoundedKey>::low(),
383 });
384 }
385 Some(Self {
386 variant_tag: next_variant_tag(self.variant_tag)?,
387 key: <Desc<u64> as BoundedKey>::low(),
388 sequence: <Desc<u64> as BoundedKey>::low(),
389 })
390 }
391}
392
393fn lowest_variant_tag() -> Desc<Option<u8>> {
394 Desc(Some(u8::MAX))
395}
396
397fn next_variant_tag(tag: Desc<Option<u8>>) -> Option<Desc<Option<u8>>> {
398 match tag.0 {
399 Some(0) => Some(Desc(None)),
400 Some(value) => Some(Desc(Some(value - 1))),
401 None => None,
402 }
403}
404
405#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
406pub struct SeriesKeyColumns {
407 pub variant_tag: i64,
408 pub key: i64,
409 pub sequence: i64,
410}
411
412#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
413pub struct PartitionedSeriesKeyColumns {
414 pub partition_hi: i64,
415 pub partition_lo: i64,
416 pub variant_tag: i64,
417 pub key: i64,
418 pub sequence: i64,
419}
420
421const SERIES_VARIANT_TAG_NONE_SQL: i64 = (((!0u8) as i64) << 8) | ((!0u8) as i64);
422
423pub fn series_variant_tag_to_sql(variant_tag: Option<u8>) -> i64 {
424 match variant_tag {
425 Some(tag) => (((!1u8) as i64) << 8) | ((!tag) as i64),
426 None => SERIES_VARIANT_TAG_NONE_SQL,
427 }
428}
429
430pub fn series_variant_tag_from_sql(value: i64) -> Option<u8> {
431 if value == SERIES_VARIANT_TAG_NONE_SQL {
432 return None;
433 }
434 Some(!(value as u8))
435}
436
437pub fn series_key_to_sql(key: u64) -> i64 {
438 desc_u64_to_sql(key)
439}
440
441pub fn series_key_from_sql(value: i64) -> u64 {
442 desc_u64_from_sql(value)
443}
444
445pub fn series_sequence_to_sql(sequence: u64) -> i64 {
446 desc_u64_to_sql(sequence)
447}
448
449pub fn series_sequence_from_sql(value: i64) -> u64 {
450 desc_u64_from_sql(value)
451}
452
453pub fn series_partition_half_to_sql(half: u64) -> i64 {
454 desc_u64_to_sql(half)
455}
456
457pub fn series_partition_half_from_sql(value: i64) -> u64 {
458 desc_u64_from_sql(value)
459}
460
461fn desc_u64_to_sql(value: u64) -> i64 {
462 ((!value) ^ (1u64 << 63)) as i64
463}
464
465fn desc_u64_from_sql(value: i64) -> u64 {
466 !((value as u64) ^ (1u64 << 63))
467}
468
469#[cfg(test)]
470mod row_key_range_tests {
471 use std::ops::RangeBounds;
472
473 use super::*;
474 use crate::{
475 interface::catalog::id::{SeriesId, ViewId},
476 key::{KeyRangeCodec, row::RowKeyRange},
477 };
478
479 #[test]
480 fn test_encode_decode_without_tag() {
481 let key = SeriesRowKey {
483 storage: StorageId::Series(SeriesId(42)),
484 variant_tag: None,
485 key: 1706745600000,
486 sequence: 1,
487 };
488 let encoded = key.encode();
489 let decoded = SeriesRowKey::decode(&encoded).unwrap();
490 assert_eq!(decoded.storage, StorageId::Series(SeriesId(42)));
491 assert_eq!(decoded.variant_tag, None);
492 assert_eq!(decoded.key, 1706745600000);
493 assert_eq!(decoded.sequence, 1);
494 }
495
496 #[test]
497 fn test_encode_decode_with_tag() {
498 let key = SeriesRowKey {
500 storage: StorageId::Series(SeriesId(42)),
501 variant_tag: Some(3),
502 key: 1706745600000,
503 sequence: 5,
504 };
505 let encoded = key.encode();
506 let decoded = SeriesRowKey::decode(&encoded).unwrap();
507 assert_eq!(decoded.storage, StorageId::Series(SeriesId(42)));
508 assert_eq!(decoded.variant_tag, Some(3));
509 assert_eq!(decoded.key, 1706745600000);
510 assert_eq!(decoded.sequence, 5);
511 }
512
513 #[test]
514 fn test_a_view_storage_round_trips() {
515 let key = SeriesRowKey {
517 storage: StorageId::View(ViewId(42)),
518 variant_tag: Some(7),
519 key: 900,
520 sequence: 2,
521 };
522 let decoded = SeriesRowKey::decode(&key.encode()).unwrap();
523 assert_eq!(decoded, key);
524 }
525
526 #[test]
527 fn test_full_scan_range_still_names_its_storage() {
528 let range = SeriesRowKeyRange::full_scan(StorageId::Series(SeriesId(42)), None).encode();
530 let (start, end) = SeriesRowKeyRange::decode(&range);
531 assert_eq!(start, Some(StorageId::Series(SeriesId(42))));
532 assert_eq!(
533 end,
534 Some(StorageId::Series(SeriesId(41))),
535 "the exclusive end brackets the next series id down"
536 );
537 }
538
539 #[test]
540 fn test_untagged_full_scan_covers_tagged_rows() {
541 let storage = StorageId::Series(SeriesId(9));
543 let range = SeriesRowKeyRange::full_scan(storage, None).encode();
544 let untagged = SeriesRowKey {
545 storage,
546 variant_tag: None,
547 key: 500,
548 sequence: 0,
549 }
550 .encode();
551 let tagged = SeriesRowKey {
552 storage,
553 variant_tag: Some(4),
554 key: 500,
555 sequence: 0,
556 }
557 .encode();
558
559 assert!(range.contains(&untagged));
560 assert!(range.contains(&tagged), "an untagged full scan must still see tagged rows");
561 }
562
563 #[test]
564 fn test_scan_range_brackets_the_rows_it_selects() {
565 let storage = StorageId::Series(SeriesId(1));
567 let range = SeriesRowKeyRange::scan_range(storage, false, None, Some(100), Some(200), None).encode();
568 let inside = SeriesRowKey {
569 storage,
570 variant_tag: None,
571 key: 150,
572 sequence: 1,
573 }
574 .encode();
575 let below = SeriesRowKey {
576 storage,
577 variant_tag: None,
578 key: 99,
579 sequence: 1,
580 }
581 .encode();
582 let above = SeriesRowKey {
583 storage,
584 variant_tag: None,
585 key: 201,
586 sequence: 1,
587 }
588 .encode();
589
590 assert!(range.contains(&inside));
591 assert!(!range.contains(&below));
592 assert!(!range.contains(&above));
593 }
594
595 #[test]
596 fn test_row_key_range_never_claims_a_series_range() {
597 let range = SeriesRowKeyRange::full_scan(StorageId::Series(SeriesId(42)), None).encode();
599 assert_eq!(RowKeyRange::decode(&range), (None, None));
600 }
601
602 #[test]
603 fn test_ordering_by_key() {
604 let key1 = SeriesRowKey {
606 storage: StorageId::Series(SeriesId(1)),
607 variant_tag: None,
608 key: 100,
609 sequence: 0,
610 };
611 let key2 = SeriesRowKey {
612 storage: StorageId::Series(SeriesId(1)),
613 variant_tag: None,
614 key: 200,
615 sequence: 0,
616 };
617 let e1 = key1.encode();
618 let e2 = key2.encode();
619
620 assert!(e1 > e2, "key descending ordering not preserved");
621 }
622
623 #[test]
624 fn test_ordering_by_sequence() {
625 let key1 = SeriesRowKey {
627 storage: StorageId::Series(SeriesId(1)),
628 variant_tag: None,
629 key: 100,
630 sequence: 1,
631 };
632 let key2 = SeriesRowKey {
633 storage: StorageId::Series(SeriesId(1)),
634 variant_tag: None,
635 key: 100,
636 sequence: 2,
637 };
638 let e1 = key1.encode();
639 let e2 = key2.encode();
640
641 assert!(e1 > e2, "sequence descending ordering not preserved");
642 }
643
644 #[test]
645 fn test_half_bounded_untagged_range_excludes_tagged_rows() {
646 let range = SeriesRowKeyRange::scan_range(
650 StorageId::Series(SeriesId(7)),
651 false,
652 None,
653 Some(100),
654 None,
655 None,
656 )
657 .encode();
658
659 let untagged = SeriesRowKey {
660 storage: StorageId::Series(SeriesId(7)),
661 variant_tag: None,
662 key: 500,
663 sequence: 1,
664 }
665 .encode();
666 let tagged = SeriesRowKey {
667 storage: StorageId::Series(SeriesId(7)),
668 variant_tag: Some(3),
669 key: 500,
670 sequence: 1,
671 }
672 .encode();
673
674 assert!(range.contains(&untagged), "an untagged row above the lower bound must stay in range");
675 assert!(!range.contains(&tagged), "a tagged row must never leak into an untagged key-bounded range");
676 }
677}
678
679#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
680#[key(tag = PartitionedSeriesRow)]
681pub struct PartitionedSeriesRowKey {
682 pub storage: StorageId,
683 pub partition: Partition,
684 pub variant_tag: Option<u8>,
685 pub key: u64,
686 pub sequence: u64,
687}
688
689impl PartitionedSeriesRowKey {
690 pub fn new(
691 storage: impl Into<StorageId>,
692 partition: Partition,
693 variant_tag: Option<u8>,
694 key: u64,
695 sequence: u64,
696 ) -> Self {
697 Self {
698 storage: storage.into(),
699 partition,
700 variant_tag,
701 key,
702 sequence,
703 }
704 }
705
706 pub fn encoded(
707 storage: impl Into<StorageId>,
708 partition: Partition,
709 variant_tag: Option<u8>,
710 key: u64,
711 sequence: u64,
712 ) -> EncodedKey {
713 Self::new(storage, partition, variant_tag, key, sequence).encode()
714 }
715
716 pub fn storage_of(key: &EncodedKey) -> Option<StorageId> {
717 let mut de = KeyDeserializer::from_bytes(key.as_slice());
718 let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
719 if kind != Self::TAG {
720 return None;
721 }
722 StorageId::from_object(de.read_object_id().ok()?)
723 }
724}
725
726#[derive(Debug, Clone)]
727pub struct PartitionedSeriesRowKeyRange {
728 pub storage: StorageId,
729 pub partition: Partition,
730 pub variant_tag: Option<u8>,
731 pub key_start: Option<u64>,
732 pub key_end: Option<u64>,
733}
734
735impl PartitionedSeriesRowKeyRange {
736 pub fn storage_start(storage: impl Into<StorageId>) -> EncodedKey {
737 TaggedKeyBound::prefix(PartitionedSeriesRowKey::TAG, object_fields(ObjectId::from(storage.into())))
738 .encode()
739 }
740
741 pub fn storage_end(storage: impl Into<StorageId>) -> EncodedKey {
742 TaggedKeyBound::prefix(
743 PartitionedSeriesRowKey::TAG,
744 object_fields(ObjectId::from(storage.into()).prev()),
745 )
746 .encode()
747 }
748
749 pub fn full_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
750 TaggedKeyBoundRange::prefix(PartitionedSeriesRowKey::TAG, object_fields(ObjectId::from(storage.into())))
751 }
752
753 pub fn full_scan_range(storage: impl Into<StorageId>, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
754 Self::full_scan(storage).resume_after(last)
755 }
756
757 pub fn partition_range(storage: impl Into<StorageId>, partition: Partition) -> TaggedKeyBoundRange {
758 TaggedKeyBoundRange::prefix(
759 PartitionedSeriesRowKey::TAG,
760 Self::partition_fields(storage.into(), partition),
761 )
762 }
763
764 pub fn partition_scan_range(
765 storage: impl Into<StorageId>,
766 partition: Partition,
767 last: Option<&TaggedKey>,
768 ) -> TaggedKeyBoundRange {
769 Self::partition_range(storage, partition).resume_after(last)
770 }
771
772 pub fn scan_range(
773 storage: impl Into<StorageId>,
774 partition: Partition,
775 tagged: bool,
776 variant_tag: Option<u8>,
777 key_start: Option<u64>,
778 key_end: Option<u64>,
779 last: Option<&TaggedKey>,
780 ) -> TaggedKeyBoundRange {
781 assert!(
782 !(tagged && variant_tag.is_none() && (key_start.is_some() || key_end.is_some())),
783 "a key range over a tagged series needs a variant tag: the key sits after the tag in the row \
784 key, so no single range spans every tag and the range would silently match nothing"
785 );
786 if matches!(key_end, Some(0)) {
787 return TaggedKeyBoundRange::empty(PartitionedSeriesRowKey::TAG);
788 }
789
790 PartitionedSeriesRowKeyRange {
791 storage: storage.into(),
792 partition,
793 variant_tag,
794 key_start,
795 key_end,
796 }
797 .bounds()
798 .resume_after(last)
799 }
800
801 pub fn decode_storage(key: &EncodedKey) -> Option<StorageId> {
802 PartitionedSeriesRowKey::storage_of(key)
803 }
804
805 pub fn decode(range: &EncodedKeyRange) -> (Option<StorageId>, Option<StorageId>) {
806 let start = match &range.start {
807 Bound::Included(key) | Bound::Excluded(key) => Self::decode_storage(key),
808 Bound::Unbounded => None,
809 };
810
811 let end = match &range.end {
812 Bound::Included(key) | Bound::Excluded(key) => Self::decode_storage(key),
813 Bound::Unbounded => None,
814 };
815
816 (start, end)
817 }
818
819 fn partition_fields(storage: StorageId, partition: Partition) -> Vec<OwnedField> {
820 let mut fields = object_fields(ObjectId::from(storage)).to_vec();
821 fields.push(Field::UDesc(Width::U128, partition.0));
822 fields
823 }
824
825 fn head_fields(&self, tagged: bool) -> Vec<OwnedField> {
826 let mut fields = Self::partition_fields(self.storage, self.partition);
827 match self.variant_tag {
828 Some(tag) => {
829 fields.push(Field::UDesc(Width::U8, 1));
830 fields.push(Field::UDesc(Width::U8, tag as u128));
831 }
832 None if tagged || self.key_start.is_some() || self.key_end.is_some() => {
833 fields.push(Field::UDesc(Width::U8, 0));
834 fields.push(Field::UDesc(Width::U8, 0));
835 }
836 None => {}
837 }
838 fields
839 }
840
841 fn bounds(&self) -> TaggedKeyBoundRange {
842 let mut start = self.head_fields(false);
843 if let Some(key_val) = self.key_end {
844 start.push(Field::UDesc(Width::U64, (key_val - 1) as u128));
845 }
846 let start = Bound::Included(TaggedKeyBound::prefix(PartitionedSeriesRowKey::TAG, start));
847
848 let end = match self.key_start {
849 Some(key_val) => {
850 let mut end = self.head_fields(true);
851 end.push(Field::UDesc(Width::U64, key_val as u128));
852 end.push(Field::UDesc(Width::U64, 0));
853 Bound::Included(TaggedKeyBound::prefix(PartitionedSeriesRowKey::TAG, end))
854 }
855 None => Bound::Excluded(TaggedKeyBound::prefix_end(
856 PartitionedSeriesRowKey::TAG,
857 Self::partition_fields(self.storage, self.partition),
858 )),
859 };
860
861 TaggedKeyBoundRange {
862 start,
863 end,
864 }
865 }
866}
867
868#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
869pub struct StoragePartitionedSeriesKey {
870 pub partition: Desc<Partition>,
871 pub variant_tag: Desc<Option<u8>>,
872 pub key: Desc<u64>,
873 pub sequence: Desc<u64>,
874}
875
876impl StoragePartitionedSeriesKey {
877 pub fn new(partition: Partition, variant_tag: Option<u8>, key: u64, sequence: u64) -> Self {
878 Self {
879 partition: Desc(partition),
880 variant_tag: Desc(variant_tag),
881 key: Desc(key),
882 sequence: Desc(sequence),
883 }
884 }
885
886 pub fn partition(self) -> Partition {
887 self.partition.0
888 }
889
890 pub fn variant_tag(self) -> Option<u8> {
891 self.variant_tag.0
892 }
893
894 pub fn key(self) -> u64 {
895 self.key.0
896 }
897
898 pub fn sequence(self) -> u64 {
899 self.sequence.0
900 }
901
902 pub fn partition_hi(self) -> u64 {
903 (self.partition().0 >> 64) as u64
904 }
905
906 pub fn partition_lo(self) -> u64 {
907 self.partition().0 as u64
908 }
909
910 pub fn from_halves(
911 partition_hi: u64,
912 partition_lo: u64,
913 variant_tag: Option<u8>,
914 key: u64,
915 sequence: u64,
916 ) -> Self {
917 Self::new(Partition(((partition_hi as u128) << 64) | partition_lo as u128), variant_tag, key, sequence)
918 }
919
920 pub fn to_sql_columns(self) -> PartitionedSeriesKeyColumns {
921 PartitionedSeriesKeyColumns {
922 partition_hi: series_partition_half_to_sql(self.partition_hi()),
923 partition_lo: series_partition_half_to_sql(self.partition_lo()),
924 variant_tag: series_variant_tag_to_sql(self.variant_tag()),
925 key: series_key_to_sql(self.key()),
926 sequence: series_sequence_to_sql(self.sequence()),
927 }
928 }
929
930 pub fn from_sql_columns(columns: PartitionedSeriesKeyColumns) -> Self {
931 Self::from_halves(
932 series_partition_half_from_sql(columns.partition_hi),
933 series_partition_half_from_sql(columns.partition_lo),
934 series_variant_tag_from_sql(columns.variant_tag),
935 series_key_from_sql(columns.key),
936 series_sequence_from_sql(columns.sequence),
937 )
938 }
939
940 pub fn with_storage(self, storage: StorageId) -> PartitionedSeriesRowKey {
941 PartitionedSeriesRowKey {
942 storage,
943 partition: self.partition(),
944 variant_tag: self.variant_tag(),
945 key: self.key(),
946 sequence: self.sequence(),
947 }
948 }
949}
950
951impl From<PartitionedSeriesRowKey> for StoragePartitionedSeriesKey {
952 fn from(key: PartitionedSeriesRowKey) -> Self {
953 StoragePartitionedSeriesKey::new(key.partition, key.variant_tag, key.key, key.sequence)
954 }
955}
956
957impl HeapSize for StoragePartitionedSeriesKey {
958 fn heap_size(&self) -> usize {
959 0
960 }
961}
962
963impl BoundedKey for StoragePartitionedSeriesKey {
964 fn low() -> Self {
965 Self {
966 partition: <Desc<Partition> as BoundedKey>::low(),
967 variant_tag: lowest_variant_tag(),
968 key: <Desc<u64> as BoundedKey>::low(),
969 sequence: <Desc<u64> as BoundedKey>::low(),
970 }
971 }
972}
973
974impl DenseKey for StoragePartitionedSeriesKey {
975 fn successor(&self) -> Option<Self> {
976 if let Some(sequence) = self.sequence.successor() {
977 return Some(Self {
978 sequence,
979 ..*self
980 });
981 }
982 if let Some(key) = self.key.successor() {
983 return Some(Self {
984 key,
985 sequence: <Desc<u64> as BoundedKey>::low(),
986 ..*self
987 });
988 }
989 if let Some(variant_tag) = next_variant_tag(self.variant_tag) {
990 return Some(Self {
991 partition: self.partition,
992 variant_tag,
993 key: <Desc<u64> as BoundedKey>::low(),
994 sequence: <Desc<u64> as BoundedKey>::low(),
995 });
996 }
997 Some(Self {
998 partition: self.partition.successor()?,
999 variant_tag: lowest_variant_tag(),
1000 key: <Desc<u64> as BoundedKey>::low(),
1001 sequence: <Desc<u64> as BoundedKey>::low(),
1002 })
1003 }
1004}
1005
1006#[cfg(test)]
1007#[allow(clippy::disallowed_methods)]
1008mod partitioned_row_key_tests {
1009 use std::ops::RangeBounds;
1010
1011 use reifydb_value::value::{Value, partition::Partition, row_number::RowNumber};
1012
1013 use super::*;
1014 use crate::{
1015 interface::catalog::id::{SeriesId, TableId, ViewId},
1016 key::row::PartitionedRowKey,
1017 };
1018
1019 fn part(v: &str) -> Partition {
1020 Partition::of(&[Value::Utf8(v.to_string())])
1021 }
1022
1023 #[test]
1024 fn test_round_trip_without_tag() {
1025 let key = PartitionedSeriesRowKey {
1027 storage: StorageId::Series(SeriesId(3)),
1028 partition: part("btc"),
1029 variant_tag: None,
1030 key: 1_700_000_000,
1031 sequence: 9,
1032 };
1033 let decoded = PartitionedSeriesRowKey::decode(&key.encode()).unwrap();
1034 assert_eq!(decoded, key);
1035 }
1036
1037 #[test]
1038 fn test_round_trip_with_tag() {
1039 let key = PartitionedSeriesRowKey {
1041 storage: StorageId::Series(SeriesId(3)),
1042 partition: part("eth"),
1043 variant_tag: Some(5),
1044 key: 42,
1045 sequence: 0,
1046 };
1047 let decoded = PartitionedSeriesRowKey::decode(&key.encode()).unwrap();
1048 assert_eq!(decoded, key);
1049 }
1050
1051 #[test]
1052 fn test_a_view_storage_round_trips() {
1053 let key = PartitionedSeriesRowKey {
1055 storage: StorageId::View(ViewId(11)),
1056 partition: part("us"),
1057 variant_tag: Some(1),
1058 key: 77,
1059 sequence: 4,
1060 };
1061 let decoded = PartitionedSeriesRowKey::decode(&key.encode()).unwrap();
1062 assert_eq!(decoded, key);
1063 }
1064
1065 #[test]
1066 fn test_storage_of() {
1067 let key = PartitionedSeriesRowKey::encoded(StorageId::Series(SeriesId(42)), part("us"), None, 1, 0);
1069 assert_eq!(PartitionedSeriesRowKey::storage_of(&key), Some(StorageId::Series(SeriesId(42))));
1070 }
1071
1072 #[test]
1073 fn test_ordering_by_key_is_descending() {
1074 let storage = StorageId::Series(SeriesId(1));
1076 let low = PartitionedSeriesRowKey::encoded(storage, part("us"), None, 100, 0);
1077 let high = PartitionedSeriesRowKey::encoded(storage, part("us"), None, 200, 0);
1078
1079 assert!(low > high, "key descending ordering not preserved");
1080 }
1081
1082 #[test]
1083 fn test_untagged_full_scan_covers_tagged_rows() {
1084 let storage = StorageId::Series(SeriesId(9));
1086 let range = PartitionedSeriesRowKeyRange::full_scan(storage).encode();
1087 let untagged = PartitionedSeriesRowKey::encoded(storage, part("us"), None, 500, 0);
1088 let tagged = PartitionedSeriesRowKey::encoded(storage, part("us"), Some(4), 500, 0);
1089
1090 assert!(range.contains(&untagged));
1091 assert!(range.contains(&tagged), "an untagged full scan must still see tagged rows");
1092 }
1093
1094 #[test]
1095 fn test_scan_range_brackets_the_rows_it_selects() {
1096 let storage = StorageId::Series(SeriesId(1));
1098 let partition = part("us");
1099 let range = PartitionedSeriesRowKeyRange::scan_range(
1100 storage,
1101 partition,
1102 false,
1103 None,
1104 Some(100),
1105 Some(200),
1106 None,
1107 )
1108 .encode();
1109 let inside = PartitionedSeriesRowKey::encoded(storage, partition, None, 150, 1);
1110 let below = PartitionedSeriesRowKey::encoded(storage, partition, None, 99, 1);
1111 let above = PartitionedSeriesRowKey::encoded(storage, partition, None, 201, 1);
1112
1113 assert!(range.contains(&inside));
1114 assert!(!range.contains(&below));
1115 assert!(!range.contains(&above));
1116 }
1117
1118 #[test]
1119 fn test_scan_range_never_crosses_into_another_partition() {
1120 let storage = StorageId::Series(SeriesId(1));
1122 let range = PartitionedSeriesRowKeyRange::scan_range(
1123 storage,
1124 part("us"),
1125 false,
1126 None,
1127 Some(100),
1128 Some(200),
1129 None,
1130 )
1131 .encode();
1132 let other = PartitionedSeriesRowKey::encoded(storage, part("eu"), None, 150, 1);
1133
1134 assert!(!range.contains(&other), "an in-bounds key of another partition must stay outside");
1135 }
1136
1137 #[test]
1138 fn test_partition_range_covers_every_key_of_its_partition() {
1139 let storage = StorageId::Series(SeriesId(1));
1141 let range = PartitionedSeriesRowKeyRange::partition_range(storage, part("us")).encode();
1142 let untagged = PartitionedSeriesRowKey::encoded(storage, part("us"), None, 1, 0);
1143 let tagged = PartitionedSeriesRowKey::encoded(storage, part("us"), Some(9), u64::MAX, u64::MAX);
1144 let other = PartitionedSeriesRowKey::encoded(storage, part("eu"), None, 1, 0);
1145
1146 assert!(range.contains(&untagged));
1147 assert!(range.contains(&tagged));
1148 assert!(!range.contains(&other));
1149 }
1150
1151 #[test]
1152 fn test_full_scan_range_resumes_across_partitions() {
1153 let storage = StorageId::Series(SeriesId(1));
1156 let cursor = TaggedKey::from(PartitionedSeriesRowKey::new(storage, part("us"), None, 200, 0));
1157 let range = PartitionedSeriesRowKeyRange::full_scan_range(storage, Some(&cursor)).encode();
1158
1159 assert!(!range.contains(&cursor.encode()));
1160 assert!(range.contains(&PartitionedSeriesRowKey::encoded(storage, part("us"), None, 100, 0)));
1161 assert!(
1162 range.contains(&PartitionedSeriesRowKey::encoded(storage, part("eu"), None, 100, 0))
1163 || range.contains(&PartitionedSeriesRowKey::encoded(storage, part("eu"), None, 300, 0)),
1164 "a resumed object-wide scan must still be able to reach another partition"
1165 );
1166 }
1167
1168 #[test]
1169 fn test_full_scan_range_without_a_cursor_covers_every_partition() {
1170 let storage = StorageId::Series(SeriesId(1));
1173 let range = PartitionedSeriesRowKeyRange::full_scan_range(storage, None).encode();
1174
1175 assert!(range.contains(&PartitionedSeriesRowKey::encoded(storage, part("us"), None, 1, 0)));
1176 assert!(range.contains(&PartitionedSeriesRowKey::encoded(storage, part("eu"), Some(3), 9, 9)));
1177 assert!(!range.contains(&PartitionedSeriesRowKey::encoded(
1178 StorageId::Series(SeriesId(2)),
1179 part("us"),
1180 None,
1181 1,
1182 0
1183 )));
1184 }
1185
1186 #[test]
1187 fn test_partition_scan_range_resumes_after_the_cursor() {
1188 let storage = StorageId::Series(SeriesId(1));
1190 let partition = part("us");
1191 let cursor = TaggedKey::from(PartitionedSeriesRowKey::new(storage, partition, None, 200, 0));
1192 let range =
1193 PartitionedSeriesRowKeyRange::partition_scan_range(storage, partition, Some(&cursor)).encode();
1194 let next = PartitionedSeriesRowKey::encoded(storage, partition, None, 100, 0);
1195
1196 assert!(!range.contains(&cursor.encode()));
1197 assert!(range.contains(&next));
1198 }
1199
1200 #[test]
1201 fn test_the_two_partitioned_kinds_do_not_share_a_keyspace() {
1202 let series =
1204 PartitionedSeriesRowKey::encoded(StorageId::Series(SeriesId(1)), part("us"), Some(2), 7, 9);
1205 let row = PartitionedRowKey::encoded(StorageId::Table(TableId(1)), part("us"), RowNumber(9));
1206
1207 assert_ne!(series.as_slice()[0], row.as_slice()[0]);
1208 assert!(PartitionedRowKey::decode(&series).is_none());
1209 assert!(PartitionedSeriesRowKey::decode(&row).is_none());
1210 }
1211
1212 #[test]
1213 fn test_half_bounded_untagged_range_excludes_tagged_rows() {
1214 let range = PartitionedSeriesRowKeyRange::scan_range(
1217 StorageId::Series(SeriesId(7)),
1218 part("us"),
1219 false,
1220 None,
1221 Some(100),
1222 None,
1223 None,
1224 )
1225 .encode();
1226
1227 let untagged = PartitionedSeriesRowKey {
1228 storage: StorageId::Series(SeriesId(7)),
1229 partition: part("us"),
1230 variant_tag: None,
1231 key: 500,
1232 sequence: 1,
1233 }
1234 .encode();
1235 let tagged = PartitionedSeriesRowKey {
1236 storage: StorageId::Series(SeriesId(7)),
1237 partition: part("us"),
1238 variant_tag: Some(3),
1239 key: 500,
1240 sequence: 1,
1241 }
1242 .encode();
1243
1244 assert!(range.contains(&untagged), "an untagged row above the lower bound must stay in range");
1245 assert!(!range.contains(&tagged), "a tagged row must never leak into an untagged key-bounded range");
1246 }
1247}
1248
1249#[cfg(test)]
1250mod storage_series_key_tests {
1251 use reifydb_value::value::partition::Partition;
1252
1253 use super::{PartitionedSeriesRowKey, SeriesRowKey, StorageId, StoragePartitionedSeriesKey, StorageSeriesKey};
1254 use crate::{
1255 interface::catalog::id::SeriesId,
1256 key::typed::{BoundedKey, DenseKey},
1257 };
1258
1259 const STORAGE: StorageId = StorageId::Series(SeriesId(7));
1260
1261 fn series(variant_tag: Option<u8>, key: u64, sequence: u64) -> StorageSeriesKey {
1262 StorageSeriesKey::new(variant_tag, key, sequence)
1263 }
1264
1265 #[test]
1266 fn storage_key_order_matches_the_encoded_byte_order() {
1267 let mut keys = [
1270 series(None, 5, 1),
1271 series(Some(0), 5, 1),
1272 series(Some(9), 5, 1),
1273 series(Some(9), 5, 2),
1274 series(Some(9), 6, 1),
1275 ];
1276 let mut encoded: Vec<_> = keys.iter().map(|it| it.with_storage(STORAGE).encode().to_vec()).collect();
1277 keys.sort();
1278 encoded.sort();
1279
1280 let reordered: Vec<_> = keys.iter().map(|it| it.with_storage(STORAGE).encode().to_vec()).collect();
1281 assert_eq!(reordered, encoded);
1282 }
1283
1284 #[test]
1285 fn a_tagged_row_sorts_before_an_untagged_one() {
1286 assert!(series(Some(0), 1, 1) < series(None, 1, 1));
1289 assert!(series(Some(255), 1, 1) < series(Some(0), 1, 1));
1290 }
1291
1292 #[test]
1293 fn low_is_the_first_key_the_encoder_can_produce() {
1294 let low = <StorageSeriesKey as BoundedKey>::low();
1295 assert_eq!(low, series(Some(u8::MAX), u64::MAX, u64::MAX));
1296 for other in [series(None, 0, 0), series(Some(0), 0, 0), series(Some(200), 7, 3)] {
1297 assert!(low < other, "low must not sort above a real key");
1298 }
1299 }
1300
1301 #[test]
1302 fn successor_walks_the_sequence_then_the_key_then_the_tag() {
1303 assert_eq!(series(Some(9), 5, 2).successor(), Some(series(Some(9), 5, 1)));
1306 assert_eq!(series(Some(9), 5, 0).successor(), Some(series(Some(9), 4, u64::MAX)));
1307 assert_eq!(series(Some(9), 0, 0).successor(), Some(series(Some(8), u64::MAX, u64::MAX)));
1308 assert_eq!(series(Some(0), 0, 0).successor(), Some(series(None, u64::MAX, u64::MAX)));
1309 assert_eq!(series(None, 0, 0).successor(), None);
1310 }
1311
1312 #[test]
1313 fn every_successor_is_the_immediate_next_storage_key() {
1314 let mut walk = vec![series(Some(1), 1, 1)];
1315 for _ in 0..4 {
1316 walk.push(walk.last().unwrap().successor().unwrap());
1317 }
1318 assert!(walk.windows(2).all(|pair| pair[0] < pair[1]), "the walk must ascend: {walk:?}");
1319 }
1320
1321 #[test]
1322 fn partitioned_storage_key_order_matches_the_encoded_byte_order() {
1323 let mut keys = [
1324 StoragePartitionedSeriesKey::new(Partition(2), None, 5, 1),
1325 StoragePartitionedSeriesKey::new(Partition(2), Some(3), 5, 1),
1326 StoragePartitionedSeriesKey::new(Partition(1), Some(3), 5, 1),
1327 StoragePartitionedSeriesKey::new(Partition(2), Some(3), 5, 2),
1328 ];
1329 let mut encoded: Vec<_> = keys.iter().map(|it| it.with_storage(STORAGE).encode().to_vec()).collect();
1330 keys.sort();
1331 encoded.sort();
1332
1333 let reordered: Vec<_> = keys.iter().map(|it| it.with_storage(STORAGE).encode().to_vec()).collect();
1334 assert_eq!(reordered, encoded);
1335 }
1336
1337 #[test]
1338 fn partitioned_successor_carries_into_the_partition() {
1339 let last_of_partition = StoragePartitionedSeriesKey::new(Partition(5), None, 0, 0);
1341 assert_eq!(
1342 last_of_partition.successor(),
1343 Some(StoragePartitionedSeriesKey::new(Partition(4), Some(u8::MAX), u64::MAX, u64::MAX))
1344 );
1345 assert_eq!(StoragePartitionedSeriesKey::new(Partition(0), None, 0, 0).successor(), None);
1346 }
1347
1348 #[test]
1349 fn a_storage_key_round_trips_through_its_row_key() {
1350 let key = SeriesRowKey {
1351 storage: STORAGE,
1352 variant_tag: Some(4),
1353 key: 900,
1354 sequence: 12,
1355 };
1356 assert_eq!(StorageSeriesKey::from(key.clone()).with_storage(STORAGE), key);
1357
1358 let partitioned = PartitionedSeriesRowKey::new(STORAGE, Partition(3), None, 900, 12);
1359 assert_eq!(StoragePartitionedSeriesKey::from(partitioned.clone()).with_storage(STORAGE), partitioned);
1360 }
1361}