reifydb_core/key/
row.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{borrow::Cow, cmp::Ordering, collections::Bound};
5
6use reifydb_codec::{
7	key::{
8		ByteSink,
9		deserializer::KeyDeserializer,
10		encoded::{EncodedKey, EncodedKeyRange},
11		serializer::KeySerializer,
12	},
13	row::shape::fingerprint::RowShapeFingerprint,
14};
15use reifydb_macro::KeyCodec;
16use reifydb_value::value::{partition::Partition, row_number::RowNumber};
17use serde::{Deserialize, Serialize};
18use smallvec::{SmallVec, smallvec};
19
20use super::{KeyRangeCodec, KeyTag};
21use crate::{
22	interface::catalog::{object::ObjectId, storage::StorageId},
23	key::{
24		any::{Field, KeyFields, RawEncoding, TaggedKey, Width},
25		bound::{TaggedKeyBoundRange, object_fields},
26		catalog::{KeyDeserializerCatalogExt, KeySerializerCatalogExt},
27		sort_run::SortRun,
28		typed::{
29			BoundedKey, DenseKey,
30			direction::{Asc, Desc},
31		},
32	},
33	metrics::heap::HeapSize,
34};
35
36#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
37#[key(tag = Row)]
38pub struct RowKey {
39	pub storage: StorageId,
40	pub row: RowNumber,
41}
42
43#[derive(Debug, Clone, PartialEq)]
44pub struct RowKeyRange {
45	pub storage: StorageId,
46}
47
48impl RowKeyRange {
49	fn decode_key(key: &EncodedKey) -> Option<Self> {
50		let mut de = KeyDeserializer::from_bytes(key.as_slice());
51
52		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
53		if kind != Self::TAG {
54			return None;
55		}
56
57		let storage = StorageId::from_object(de.read_object_id().ok()?)?;
58
59		Some(RowKeyRange {
60			storage,
61		})
62	}
63
64	pub fn storage_scan(storage: StorageId) -> TaggedKeyBoundRange {
65		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage)))
66	}
67
68	pub fn scan_range(storage: StorageId, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
69		Self::storage_scan(storage).resume_after(last)
70	}
71
72	pub fn scan_range_rev(storage: StorageId, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
73		Self::storage_scan(storage).resume_before(last)
74	}
75}
76
77impl KeyRangeCodec for RowKeyRange {
78	const TAG: KeyTag = KeyTag::Row;
79
80	fn start(&self) -> Option<EncodedKey> {
81		let mut serializer = KeySerializer::with_capacity(10);
82		serializer.extend_u8(Self::TAG as u8).extend_object_id(self.storage);
83		Some(serializer.to_encoded_key())
84	}
85
86	fn end(&self) -> Option<EncodedKey> {
87		let mut serializer = KeySerializer::with_capacity(10);
88		serializer.extend_u8(Self::TAG as u8).extend_object_id(ObjectId::from(self.storage).prev());
89		Some(serializer.to_encoded_key())
90	}
91
92	fn decode(range: &EncodedKeyRange) -> (Option<Self>, Option<Self>)
93	where
94		Self: Sized,
95	{
96		let start_key = match &range.start {
97			Bound::Included(key) | Bound::Excluded(key) => Self::decode_key(key),
98			Bound::Unbounded => None,
99		};
100
101		let end_key = match &range.end {
102			Bound::Included(key) | Bound::Excluded(key) => Self::decode_key(key),
103			Bound::Unbounded => None,
104		};
105
106		(start_key, end_key)
107	}
108}
109
110impl RowKey {
111	pub fn new(storage: impl Into<StorageId>, row: impl Into<RowNumber>) -> Self {
112		Self {
113			storage: storage.into(),
114			row: row.into(),
115		}
116	}
117
118	pub fn encoded(storage: impl Into<StorageId>, row: impl Into<RowNumber>) -> EncodedKey {
119		Self {
120			storage: storage.into(),
121			row: row.into(),
122		}
123		.encode()
124	}
125
126	pub fn full_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
127		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
128	}
129
130	pub fn storage_start(storage: impl Into<StorageId>) -> EncodedKey {
131		let mut serializer = KeySerializer::with_capacity(10);
132		serializer.extend_u8(RowKey::TAG as u8).extend_object_id(storage.into());
133		serializer.to_encoded_key()
134	}
135
136	pub fn storage_end(storage: impl Into<StorageId>) -> EncodedKey {
137		let mut serializer = KeySerializer::with_capacity(10);
138		serializer.extend_u8(RowKey::TAG as u8).extend_object_id(ObjectId::from(storage.into()).prev());
139		serializer.to_encoded_key()
140	}
141}
142
143#[derive(Debug, Clone, PartialEq, Eq, Hash)]
144pub struct SortedViewRowKey {
145	pub storage: StorageId,
146	pub run: SortRun,
147	pub row: Asc<RowNumber>,
148}
149
150impl SortedViewRowKey {
151	pub const TAG: KeyTag = KeyTag::SortedViewRow;
152
153	pub fn encode(&self) -> EncodedKey {
154		let mut serializer = KeySerializer::with_capacity(20 + self.run.len());
155		serializer.extend_u8(Self::TAG as u8).extend_object_id(self.storage);
156		extend_sort_run(&mut serializer, &self.run);
157		serializer.extend_raw(&self.row.0.0.to_be_bytes());
158		serializer.to_encoded_key()
159	}
160
161	pub fn decode(key: &EncodedKey) -> Option<Self> {
162		let mut de = KeyDeserializer::from_bytes(key.as_slice());
163		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
164		if kind != Self::TAG {
165			return None;
166		}
167		let storage = StorageId::from_object(de.read_object_id().ok()?)?;
168		let run = read_sort_run(&mut de)?;
169		let row = read_row_tail(&mut de)?;
170		if !de.is_empty() {
171			return None;
172		}
173		Some(Self {
174			storage,
175			run,
176			row,
177		})
178	}
179}
180
181impl Ord for SortedViewRowKey {
182	fn cmp(&self, other: &Self) -> Ordering {
183		storage_order(self.storage)
184			.cmp(&storage_order(other.storage))
185			.then_with(|| self.run.cmp(&other.run))
186			.then_with(|| self.row.cmp(&other.row))
187	}
188}
189
190impl PartialOrd for SortedViewRowKey {
191	fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
192		Some(self.cmp(other))
193	}
194}
195
196impl SortedViewRowKey {
197	pub fn new(storage: impl Into<StorageId>, run: SortRun, row: RowNumber) -> Self {
198		Self {
199			storage: storage.into(),
200			run,
201			row: Asc(row),
202		}
203	}
204
205	pub fn encoded(storage: impl Into<StorageId>, run: SortRun, row: RowNumber) -> EncodedKey {
206		Self::new(storage, run, row).encode()
207	}
208
209	pub fn storage_start(storage: impl Into<StorageId>) -> EncodedKey {
210		let mut serializer = KeySerializer::with_capacity(10);
211		serializer.extend_u8(Self::TAG as u8).extend_object_id(storage.into());
212		serializer.to_encoded_key()
213	}
214
215	pub fn storage_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
216		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
217	}
218
219	pub fn scan_range(storage: impl Into<StorageId>, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
220		Self::storage_scan(storage).resume_after(last)
221	}
222
223	pub fn storage_of(key: &EncodedKey) -> Option<StorageId> {
224		let mut de = KeyDeserializer::from_bytes(key.as_slice());
225		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
226		if kind != Self::TAG {
227			return None;
228		}
229		StorageId::from_object(de.read_object_id().ok()?)
230	}
231
232	pub fn row_of(key: &EncodedKey) -> Option<RowNumber> {
233		let bytes = key.as_slice();
234		if KeyTag::of(bytes)? != Self::TAG {
235			return None;
236		}
237		let tail = bytes.len().checked_sub(8)?;
238		Some(RowNumber(u64::from_be_bytes(bytes[tail..].try_into().ok()?)))
239	}
240
241	pub fn range_storage_of(range: &EncodedKeyRange) -> Option<StorageId> {
242		let start = bound_storage_of(&range.start, Self::TAG)?;
243		bound_storage_of(&range.end, Self::TAG)?;
244		Some(start)
245	}
246}
247
248const SORT_RUN_MARKER: u8 = 0x00;
249const SORT_RUN_ZERO: u8 = 0xff;
250const SORT_RUN_END: u8 = 0x00;
251
252pub(crate) fn encode_sort_run<B: ByteSink>(run: &[u8], out: &mut B) {
253	for &byte in run {
254		if byte == SORT_RUN_MARKER {
255			out.extend_from_slice(&[SORT_RUN_MARKER, SORT_RUN_ZERO]);
256		} else {
257			out.push(byte);
258		}
259	}
260	out.extend_from_slice(&[SORT_RUN_MARKER, SORT_RUN_END]);
261}
262
263fn extend_sort_run(serializer: &mut KeySerializer, run: &SortRun) {
264	encode_sort_run(run.as_slice(), serializer);
265}
266
267fn read_sort_run(de: &mut KeyDeserializer) -> Option<SortRun> {
268	let mut run: Vec<u8> = Vec::new();
269	loop {
270		let byte = de.read_raw(1).ok()?[0];
271		if byte != SORT_RUN_MARKER {
272			run.push(byte);
273			continue;
274		}
275		match de.read_raw(1).ok()?[0] {
276			SORT_RUN_END => return Some(SortRun::new(run)),
277			SORT_RUN_ZERO => run.push(0x00),
278			_ => return None,
279		}
280	}
281}
282
283fn read_row_tail(de: &mut KeyDeserializer) -> Option<Asc<RowNumber>> {
284	let bytes: [u8; 8] = de.read_raw(8).ok()?.try_into().ok()?;
285	Some(Asc(RowNumber(u64::from_be_bytes(bytes))))
286}
287
288fn storage_order(storage: StorageId) -> (u8, Desc<u64>) {
289	(ObjectId::from(storage).type_tag(), Desc(storage.as_u64()))
290}
291
292fn bound_storage_of(bound: &Bound<EncodedKey>, kind: KeyTag) -> Option<StorageId> {
293	let key = match bound {
294		Bound::Included(key) | Bound::Excluded(key) => key,
295		Bound::Unbounded => return None,
296	};
297	let mut de = KeyDeserializer::from_bytes(key.as_slice());
298	if KeyTag::try_from(de.read_u8().ok()?).ok()? != kind {
299		return None;
300	}
301	StorageId::from_object(de.read_object_id().ok()?)
302}
303
304#[derive(Debug, Clone, PartialEq, Eq, Hash)]
305pub struct PartitionedSortedViewRowKey {
306	pub storage: StorageId,
307	pub partition: Partition,
308	pub run: SortRun,
309	pub row: Asc<RowNumber>,
310}
311
312impl PartitionedSortedViewRowKey {
313	pub const TAG: KeyTag = KeyTag::PartitionedSortedViewRow;
314
315	pub fn encode(&self) -> EncodedKey {
316		let mut serializer = KeySerializer::with_capacity(36 + self.run.len());
317		serializer.extend_u8(Self::TAG as u8).extend_object_id(self.storage).extend_u128(self.partition.0);
318		extend_sort_run(&mut serializer, &self.run);
319		serializer.extend_raw(&self.row.0.0.to_be_bytes());
320		serializer.to_encoded_key()
321	}
322
323	pub fn decode(key: &EncodedKey) -> Option<Self> {
324		let mut de = KeyDeserializer::from_bytes(key.as_slice());
325		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
326		if kind != Self::TAG {
327			return None;
328		}
329		let storage = StorageId::from_object(de.read_object_id().ok()?)?;
330		let partition = Partition(de.read_u128().ok()?);
331		let run = read_sort_run(&mut de)?;
332		let row = read_row_tail(&mut de)?;
333		if !de.is_empty() {
334			return None;
335		}
336		Some(Self {
337			storage,
338			partition,
339			run,
340			row,
341		})
342	}
343}
344
345impl Ord for PartitionedSortedViewRowKey {
346	fn cmp(&self, other: &Self) -> Ordering {
347		storage_order(self.storage)
348			.cmp(&storage_order(other.storage))
349			.then_with(|| Desc(self.partition).cmp(&Desc(other.partition)))
350			.then_with(|| self.run.cmp(&other.run))
351			.then_with(|| self.row.cmp(&other.row))
352	}
353}
354
355impl PartialOrd for PartitionedSortedViewRowKey {
356	fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
357		Some(self.cmp(other))
358	}
359}
360
361impl PartitionedSortedViewRowKey {
362	pub fn new(storage: impl Into<StorageId>, partition: Partition, run: SortRun, row: RowNumber) -> Self {
363		Self {
364			storage: storage.into(),
365			partition,
366			run,
367			row: Asc(row),
368		}
369	}
370
371	pub fn encoded(
372		storage: impl Into<StorageId>,
373		partition: Partition,
374		run: SortRun,
375		row: RowNumber,
376	) -> EncodedKey {
377		Self::new(storage, partition, run, row).encode()
378	}
379
380	pub fn storage_start(storage: impl Into<StorageId>) -> EncodedKey {
381		let mut serializer = KeySerializer::with_capacity(10);
382		serializer.extend_u8(Self::TAG as u8).extend_object_id(storage.into());
383		serializer.to_encoded_key()
384	}
385
386	pub fn storage_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
387		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
388	}
389
390	pub fn scan_range(storage: impl Into<StorageId>, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
391		Self::storage_scan(storage).resume_after(last)
392	}
393
394	pub fn partition_range(storage: impl Into<StorageId>, partition: Partition) -> TaggedKeyBoundRange {
395		TaggedKeyBoundRange::prefix(
396			Self::TAG,
397			object_fields(ObjectId::from(storage.into()))
398				.into_iter()
399				.chain([Field::UDesc(Width::U128, partition.0)])
400				.collect::<Vec<_>>(),
401		)
402	}
403
404	pub fn partition_scan_range(
405		storage: impl Into<StorageId>,
406		partition: Partition,
407		last: Option<&TaggedKey>,
408	) -> TaggedKeyBoundRange {
409		Self::partition_range(storage, partition).resume_after(last)
410	}
411
412	pub fn storage_of(key: &EncodedKey) -> Option<StorageId> {
413		let mut de = KeyDeserializer::from_bytes(key.as_slice());
414		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
415		if kind != Self::TAG {
416			return None;
417		}
418		StorageId::from_object(de.read_object_id().ok()?)
419	}
420
421	pub fn row_of(key: &EncodedKey) -> Option<RowNumber> {
422		let bytes = key.as_slice();
423		if KeyTag::of(bytes)? != Self::TAG {
424			return None;
425		}
426		let tail = bytes.len().checked_sub(8)?;
427		Some(RowNumber(u64::from_be_bytes(bytes[tail..].try_into().ok()?)))
428	}
429
430	pub fn range_storage_of(range: &EncodedKeyRange) -> Option<StorageId> {
431		let start = bound_storage_of(&range.start, Self::TAG)?;
432		bound_storage_of(&range.end, Self::TAG)?;
433		Some(start)
434	}
435}
436
437#[cfg(test)]
438mod sorted_view_row_key_tests {
439	use std::ops::RangeBounds;
440
441	use reifydb_codec::key::encoded::EncodedKey;
442	use reifydb_value::value::{Value, partition::Partition, row_number::RowNumber};
443
444	use super::{PartitionedSortedViewRowKey, RowKey, SortedViewRowKey, TaggedKey};
445	use crate::{interface::catalog::storage::StorageId, key::sort_run::SortRun};
446
447	fn part(v: &str) -> Partition {
448		Partition::of(&[Value::Utf8(v.to_string())])
449	}
450
451	fn sorted_view(storage: StorageId, sort: &[u8], row: RowNumber) -> EncodedKey {
452		SortedViewRowKey::encoded(storage, SortRun::new(sort), row)
453	}
454
455	fn sorted_view_cursor(storage: StorageId, sort: &[u8], row: RowNumber) -> TaggedKey {
456		TaggedKey::from(SortedViewRowKey::new(storage, SortRun::new(sort), row))
457	}
458
459	fn partitioned(storage: StorageId, partition: Partition, sort: &[u8], row: RowNumber) -> EncodedKey {
460		PartitionedSortedViewRowKey::encoded(storage, partition, SortRun::new(sort), row)
461	}
462
463	#[test]
464	fn test_row_comes_from_the_tail_not_the_sort_prefix() {
465		// The sort payload sits between the storage and the row, so reading a fixed offset picks up
466		// sort bytes instead: two rows that sort differently would report the same row number.
467		let storage = StorageId::view(3);
468		let a = sorted_view(storage, &[0xAA; 8], RowNumber(7));
469		let b = sorted_view(storage, &[0xBB; 24], RowNumber(9));
470
471		assert_eq!(SortedViewRowKey::row_of(&a), Some(RowNumber(7)));
472		assert_eq!(SortedViewRowKey::row_of(&b), Some(RowNumber(9)));
473		assert_eq!(SortedViewRowKey::storage_of(&a), Some(storage));
474	}
475
476	#[test]
477	fn test_row_of_rejects_a_plain_row_key() {
478		// A plain row key ends in the keycode-inverted row, so reading its tail raw yields a
479		// different number entirely; the kind check is what keeps the two keyspaces apart.
480		let plain = RowKey::encoded(StorageId::view(3), RowNumber(7));
481		assert_eq!(SortedViewRowKey::row_of(&plain), None);
482		assert_eq!(PartitionedSortedViewRowKey::row_of(&plain), None);
483	}
484
485	#[test]
486	fn test_scan_range_covers_its_storage_and_nothing_else() {
487		let storage = StorageId::view(3);
488		let range = SortedViewRowKey::scan_range(storage, None).encode();
489
490		assert!(range.contains(&sorted_view(storage, &[0x00; 8], RowNumber(1))));
491		assert!(range.contains(&sorted_view(storage, &[0xFF; 8], RowNumber(u64::MAX))));
492		assert!(!range.contains(&sorted_view(StorageId::view(4), &[0x00; 8], RowNumber(1))));
493	}
494
495	#[test]
496	fn test_scan_range_resumes_strictly_after_the_last_key() {
497		// Resuming inclusively re-serves the last row of the previous chunk, duplicating it in the view.
498		let storage = StorageId::view(3);
499		let last = sorted_view_cursor(storage, &[0x40; 8], RowNumber(5));
500		let range = SortedViewRowKey::scan_range(storage, Some(&last)).encode();
501
502		assert!(!range.contains(&last.encode()));
503		assert!(range.contains(&sorted_view(storage, &[0x41; 8], RowNumber(1))));
504	}
505
506	#[test]
507	fn test_sort_prefix_orders_the_keyspace() {
508		// The sorted view key exists so a plain forward scan returns the view already sorted; if the
509		// row bytes outranked the sort prefix the scan would come back in insertion order.
510		let storage = StorageId::view(3);
511		let early_sort_late_row = sorted_view(storage, &[0x10; 8], RowNumber(999));
512		let late_sort_early_row = sorted_view(storage, &[0x20; 8], RowNumber(1));
513
514		assert!(early_sort_late_row < late_sort_early_row);
515	}
516
517	#[test]
518	fn test_partition_range_contains_only_its_partition() {
519		let storage = StorageId::view(3);
520		let range = PartitionedSortedViewRowKey::partition_range(storage, part("us")).encode();
521
522		assert!(range.contains(&partitioned(storage, part("us"), &[0x10; 8], RowNumber(1))));
523		assert!(!range.contains(&partitioned(storage, part("eu"), &[0x10; 8], RowNumber(1))));
524	}
525
526	#[test]
527	fn test_partitioned_row_comes_from_the_tail() {
528		let storage = StorageId::view(3);
529		let key = partitioned(storage, part("us"), &[0xAA; 8], RowNumber(42));
530
531		assert_eq!(PartitionedSortedViewRowKey::row_of(&key), Some(RowNumber(42)));
532		assert_eq!(PartitionedSortedViewRowKey::storage_of(&key), Some(storage));
533		assert_eq!(SortedViewRowKey::row_of(&key), None);
534	}
535}
536
537#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
538pub struct StorageRowKey(pub Desc<RowNumber>);
539
540impl From<RowKey> for StorageRowKey {
541	fn from(key: RowKey) -> Self {
542		StorageRowKey::new(key.row)
543	}
544}
545
546impl StorageRowKey {
547	pub fn new(row: RowNumber) -> Self {
548		StorageRowKey(Desc(row))
549	}
550
551	pub fn row(self) -> RowNumber {
552		self.0.0
553	}
554
555	pub fn with_storage(self, storage: StorageId) -> RowKey {
556		RowKey {
557			storage,
558			row: self.row(),
559		}
560	}
561}
562
563impl HeapSize for StorageRowKey {
564	fn heap_size(&self) -> usize {
565		0
566	}
567}
568
569impl BoundedKey for StorageRowKey {
570	fn low() -> Self {
571		StorageRowKey(<Desc<RowNumber> as BoundedKey>::low())
572	}
573}
574
575impl DenseKey for StorageRowKey {
576	fn successor(&self) -> Option<Self> {
577		self.0.successor().map(StorageRowKey)
578	}
579}
580
581#[cfg(test)]
582pub mod row_key_tests {
583	use reifydb_value::value::row_number::RowNumber;
584
585	use super::{RowKey, StorageRowKey};
586	use crate::{
587		interface::catalog::storage::StorageId,
588		key::typed::{BoundedKey, DenseKey},
589	};
590
591	#[test]
592	fn test_encode_decode() {
593		let key = RowKey {
594			storage: StorageId::table(0xABCD),
595			row: RowNumber(0x123456789ABCDEF0),
596		};
597		let encoded = key.encode();
598
599		let expected: Vec<u8> = vec![
600			0xFC, 0x01, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x54, 0x32, 0xED, 0xCB, 0xA9, 0x87, 0x65, 0x43,
601			0x21, 0x0F,
602		];
603
604		assert_eq!(encoded.as_slice(), expected);
605
606		let key = RowKey::decode(&encoded).unwrap();
607		assert_eq!(key.storage, StorageId::table(0xABCD));
608		assert_eq!(key.row, 0x123456789ABCDEF0);
609	}
610
611	#[test]
612	fn test_encode_decode_view() {
613		// Without the view tag narrowing back through `from_object`, a view's row key decodes to `None`.
614		let key = RowKey {
615			storage: StorageId::view(0xABCD),
616			row: RowNumber(0x123456789ABCDEF0),
617		};
618		let encoded = key.encode();
619
620		let expected: Vec<u8> = vec![
621			0xFC, 0x02, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x54, 0x32, 0xED, 0xCB, 0xA9, 0x87, 0x65, 0x43,
622			0x21, 0x0F,
623		];
624
625		assert_eq!(encoded.as_slice(), expected);
626
627		let key = RowKey::decode(&encoded).unwrap();
628		assert_eq!(key.storage, StorageId::view(0xABCD));
629		assert_eq!(key.row, 0x123456789ABCDEF0);
630	}
631
632	#[test]
633	fn test_order_preserving() {
634		let key1 = RowKey {
635			storage: StorageId::table(1),
636			row: RowNumber(100),
637		};
638		let key2 = RowKey {
639			storage: StorageId::table(1),
640			row: RowNumber(200),
641		};
642		let key3 = RowKey {
643			storage: StorageId::table(2),
644			row: RowNumber(1),
645		};
646
647		let encoded1 = key1.encode();
648		let encoded2 = key2.encode();
649		let encoded3 = key3.encode();
650
651		assert!(encoded3 < encoded2, "ordering not preserved");
652		assert!(encoded2 < encoded1, "ordering not preserved");
653	}
654
655	#[test]
656	fn test_row_ident_roundtrip() {
657		let key = RowKey {
658			storage: StorageId::table(7),
659			row: RowNumber(42),
660		};
661
662		// dropping storage and re-supplying the same value must recover the original key
663		let ident: StorageRowKey = key.clone().into();
664		let restored = ident.with_storage(key.storage);
665		assert_eq!(restored, key);
666	}
667
668	#[test]
669	fn test_row_ident_ordering_matches_the_encoded_key() {
670		let storage = StorageId::table(1);
671		let one = StorageRowKey::new(RowNumber(1));
672		let two = StorageRowKey::new(RowNumber(2));
673
674		// the narrow identity stands in for the encoded key in the tiers, so it must sort the same way:
675		// descending by row number, not ascending
676		assert!(two < one);
677		assert_eq!(
678			two < one,
679			RowKey::encoded(storage, RowNumber(2)).as_slice()
680				< RowKey::encoded(storage, RowNumber(1)).as_slice()
681		);
682	}
683
684	#[test]
685	fn test_row_ident_low_is_the_greatest_row() {
686		// low() names the first key a scan meets. Under descending order that is the highest row,
687		// and a scan seeded from the lowest row would start past every key it meant to cover.
688		assert_eq!(<StorageRowKey as BoundedKey>::low(), StorageRowKey::new(RowNumber(u64::MAX)));
689	}
690
691	#[test]
692	fn test_row_ident_successor_is_the_next_key_in_scan_order() {
693		// nothing may sort between a key and its successor, or an exclusive upper end drops a row
694		let ident = StorageRowKey::new(RowNumber(5));
695		let next = ident.successor().unwrap();
696		assert_eq!(next, StorageRowKey::new(RowNumber(4)));
697		assert!(next > ident);
698		assert!(StorageRowKey::new(RowNumber(3)) > next);
699	}
700
701	#[test]
702	fn test_row_ident_successor_runs_out_at_row_zero() {
703		// row zero is the last key in descending order, so it has no successor to hand back
704		assert_eq!(StorageRowKey::new(RowNumber(0)).successor(), None);
705	}
706}
707
708#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
709#[key(tag = RowSequence)]
710pub struct RowSequenceKey {
711	pub storage: StorageId,
712}
713
714impl RowSequenceKey {
715	pub fn new(storage: StorageId) -> Self {
716		Self {
717			storage,
718		}
719	}
720
721	pub fn encoded(storage: impl Into<StorageId>) -> EncodedKey {
722		Self::new(storage.into()).encode()
723	}
724
725	pub fn full_scan() -> TaggedKeyBoundRange {
726		TaggedKeyBoundRange::kind(Self::TAG)
727	}
728}
729
730#[cfg(test)]
731pub mod row_sequence_key_tests {
732	use super::RowSequenceKey;
733	use crate::interface::catalog::storage::StorageId;
734
735	#[test]
736	fn test_encode_decode() {
737		let key = RowSequenceKey {
738			storage: StorageId::table(0xABCD),
739		};
740		let encoded = key.encode();
741		let expected = vec![0xF7, 0x01, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x54, 0x32];
742		assert_eq!(encoded.as_slice(), expected);
743
744		let key = RowSequenceKey::decode(&encoded).unwrap();
745		assert_eq!(key.storage, StorageId::table(0xABCD));
746	}
747
748	#[test]
749	fn test_encode_decode_view() {
750		// A view owns its row numbering; the view tag must survive the narrowing back through decode.
751		let key = RowSequenceKey {
752			storage: StorageId::view(0xABCD),
753		};
754		let encoded = key.encode();
755		let expected = vec![0xF7, 0x02, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x54, 0x32];
756		assert_eq!(encoded.as_slice(), expected);
757
758		let key = RowSequenceKey::decode(&encoded).unwrap();
759		assert_eq!(key.storage, StorageId::view(0xABCD));
760	}
761}
762
763#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, KeyCodec, Hash)]
764#[key(tag = RowSettings)]
765pub struct RowSettingsKey {
766	pub storage: StorageId,
767}
768
769impl RowSettingsKey {
770	pub fn new(storage: StorageId) -> Self {
771		Self {
772			storage,
773		}
774	}
775
776	pub fn encoded(storage: StorageId) -> EncodedKey {
777		Self::new(storage).encode()
778	}
779
780	pub fn full_scan() -> TaggedKeyBoundRange {
781		TaggedKeyBoundRange::kind(Self::TAG)
782	}
783}
784
785#[cfg(test)]
786pub mod row_settings_key_tests {
787	use super::*;
788	use crate::interface::catalog::id::{RingBufferId, SeriesId, TableId, ViewId};
789
790	#[test]
791	fn test_row_settings_key_encoding() {
792		let key = RowSettingsKey {
793			storage: StorageId::Table(TableId(42)),
794		};
795
796		let encoded = key.encode();
797		let decoded = RowSettingsKey::decode(&encoded).unwrap();
798		assert_eq!(key, decoded);
799	}
800
801	#[test]
802	fn test_row_settings_key_roundtrip_view() {
803		// A view owns its rows, so its settings key must survive the tag round trip like any other storage.
804		let key = RowSettingsKey {
805			storage: StorageId::View(ViewId(13)),
806		};
807
808		let encoded = key.encode();
809		let decoded = RowSettingsKey::decode(&encoded).unwrap();
810		assert_eq!(key, decoded);
811	}
812
813	#[test]
814	fn test_row_settings_key_roundtrip_ringbuffer() {
815		let key = RowSettingsKey {
816			storage: StorageId::RingBuffer(RingBufferId(99)),
817		};
818
819		let encoded = key.encode();
820		let decoded = RowSettingsKey::decode(&encoded).unwrap();
821		assert_eq!(key, decoded);
822	}
823
824	#[test]
825	fn test_row_settings_key_roundtrip_series() {
826		let key = RowSettingsKey {
827			storage: StorageId::Series(SeriesId(7)),
828		};
829
830		let encoded = key.encode();
831		let decoded = RowSettingsKey::decode(&encoded).unwrap();
832		assert_eq!(key, decoded);
833	}
834}
835
836#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
837#[key(tag = RowShape)]
838pub struct RowShapeKey {
839	pub fingerprint: RowShapeFingerprint,
840}
841
842impl RowShapeKey {
843	pub fn new(fingerprint: RowShapeFingerprint) -> Self {
844		Self {
845			fingerprint,
846		}
847	}
848
849	pub fn encoded(fingerprint: RowShapeFingerprint) -> EncodedKey {
850		Self {
851			fingerprint,
852		}
853		.encode()
854	}
855
856	pub fn full_scan() -> TaggedKeyBoundRange {
857		TaggedKeyBoundRange::kind(Self::TAG)
858	}
859}
860
861#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
862#[key(tag = RowShapeField)]
863pub struct RowShapeFieldKey {
864	pub shape_fingerprint: RowShapeFingerprint,
865	pub field_index: u16,
866}
867
868impl RowShapeFieldKey {
869	pub fn new(shape_fingerprint: RowShapeFingerprint, field_index: u16) -> Self {
870		Self {
871			shape_fingerprint,
872			field_index,
873		}
874	}
875
876	pub fn encoded(shape_fingerprint: RowShapeFingerprint, field_index: u16) -> EncodedKey {
877		Self::new(shape_fingerprint, field_index).encode()
878	}
879
880	pub fn scan_for_shape(fingerprint: RowShapeFingerprint) -> TaggedKeyBoundRange {
881		TaggedKeyBoundRange::prefix(Self::TAG, [Field::UDesc(Width::U64, fingerprint.as_u64() as u128)])
882	}
883}
884
885#[cfg(test)]
886mod row_shape_key_tests {
887	use super::*;
888
889	#[test]
890	fn test_shape_key_encode_decode() {
891		let key = RowShapeKey {
892			fingerprint: RowShapeFingerprint::new(0xDEADBEEFCAFEBABE),
893		};
894		let encoded = key.encode();
895		let decoded = RowShapeKey::decode(&encoded).unwrap();
896		assert_eq!(decoded.fingerprint, RowShapeFingerprint::new(0xDEADBEEFCAFEBABE));
897	}
898
899	#[test]
900	fn test_shape_field_key_encode_decode() {
901		let key = RowShapeFieldKey {
902			shape_fingerprint: RowShapeFingerprint::new(0x1234567890ABCDEF),
903			field_index: 42,
904		};
905		let encoded = key.encode();
906		let decoded = RowShapeFieldKey::decode(&encoded).unwrap();
907		assert_eq!(decoded.shape_fingerprint, RowShapeFingerprint::new(0x1234567890ABCDEF));
908		assert_eq!(decoded.field_index, 42);
909	}
910}
911
912#[derive(Debug, Clone, PartialEq, KeyCodec, Hash)]
913#[key(tag = PartitionedRow)]
914pub struct PartitionedRowKey {
915	pub storage: StorageId,
916	pub partition: Partition,
917	pub row: RowNumber,
918}
919
920impl PartitionedRowKey {
921	pub fn new(storage: impl Into<StorageId>, partition: Partition, row: RowNumber) -> Self {
922		Self {
923			storage: storage.into(),
924			partition,
925			row,
926		}
927	}
928
929	pub fn encoded(storage: impl Into<StorageId>, partition: Partition, row: RowNumber) -> EncodedKey {
930		Self::new(storage, partition, row).encode()
931	}
932
933	pub fn storage_start(storage: impl Into<StorageId>) -> EncodedKey {
934		let mut serializer = KeySerializer::with_capacity(10);
935		serializer.extend_u8(PartitionedRowKey::TAG as u8).extend_object_id(storage.into());
936		serializer.to_encoded_key()
937	}
938
939	pub fn storage_end(storage: impl Into<StorageId>) -> EncodedKey {
940		let mut serializer = KeySerializer::with_capacity(10);
941		serializer
942			.extend_u8(PartitionedRowKey::TAG as u8)
943			.extend_object_id(ObjectId::from(storage.into()).prev());
944		serializer.to_encoded_key()
945	}
946
947	pub fn storage_of(key: &EncodedKey) -> Option<StorageId> {
948		let mut de = KeyDeserializer::from_bytes(key.as_slice());
949		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
950		if kind != Self::TAG {
951			return None;
952		}
953		StorageId::from_object(de.read_object_id().ok()?)
954	}
955
956	pub fn full_scan(storage: impl Into<StorageId>) -> TaggedKeyBoundRange {
957		TaggedKeyBoundRange::prefix(Self::TAG, object_fields(ObjectId::from(storage.into())))
958	}
959
960	pub fn scan_range(storage: impl Into<StorageId>, last: Option<&TaggedKey>) -> TaggedKeyBoundRange {
961		Self::full_scan(storage).resume_after(last)
962	}
963
964	pub fn partition_range(storage: impl Into<StorageId>, partition: Partition) -> TaggedKeyBoundRange {
965		TaggedKeyBoundRange::prefix(
966			Self::TAG,
967			object_fields(ObjectId::from(storage.into()))
968				.into_iter()
969				.chain([Field::UDesc(Width::U128, partition.0)])
970				.collect::<Vec<_>>(),
971		)
972	}
973
974	pub fn partition_scan_range(
975		storage: impl Into<StorageId>,
976		partition: Partition,
977		last: Option<&TaggedKey>,
978	) -> TaggedKeyBoundRange {
979		Self::partition_range(storage, partition).resume_after(last)
980	}
981}
982
983#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
984pub struct StoragePartitionedRowKey {
985	pub partition: Desc<Partition>,
986	pub row: Desc<RowNumber>,
987}
988
989impl StoragePartitionedRowKey {
990	pub fn new(partition: Partition, row: RowNumber) -> Self {
991		Self {
992			partition: Desc(partition),
993			row: Desc(row),
994		}
995	}
996
997	pub fn partition(self) -> Partition {
998		self.partition.0
999	}
1000
1001	pub fn row(self) -> RowNumber {
1002		self.row.0
1003	}
1004
1005	pub fn partition_hi(self) -> u64 {
1006		(self.partition().0 >> 64) as u64
1007	}
1008
1009	pub fn partition_lo(self) -> u64 {
1010		self.partition().0 as u64
1011	}
1012
1013	pub fn from_halves(partition_hi: u64, partition_lo: u64, row: RowNumber) -> Self {
1014		Self::new(Partition(((partition_hi as u128) << 64) | partition_lo as u128), row)
1015	}
1016
1017	pub fn with_storage(self, storage: StorageId) -> PartitionedRowKey {
1018		PartitionedRowKey {
1019			storage,
1020			partition: self.partition(),
1021			row: self.row(),
1022		}
1023	}
1024}
1025
1026impl From<PartitionedRowKey> for StoragePartitionedRowKey {
1027	fn from(key: PartitionedRowKey) -> Self {
1028		StoragePartitionedRowKey::new(key.partition, key.row)
1029	}
1030}
1031
1032impl HeapSize for StoragePartitionedRowKey {
1033	fn heap_size(&self) -> usize {
1034		0
1035	}
1036}
1037
1038impl BoundedKey for StoragePartitionedRowKey {
1039	fn low() -> Self {
1040		Self {
1041			partition: <Desc<Partition> as BoundedKey>::low(),
1042			row: <Desc<RowNumber> as BoundedKey>::low(),
1043		}
1044	}
1045}
1046
1047impl DenseKey for StoragePartitionedRowKey {
1048	fn successor(&self) -> Option<Self> {
1049		if let Some(row) = self.row.successor() {
1050			return Some(Self {
1051				partition: self.partition,
1052				row,
1053			});
1054		}
1055		Some(Self {
1056			partition: self.partition.successor()?,
1057			row: <Desc<RowNumber> as BoundedKey>::low(),
1058		})
1059	}
1060}
1061
1062#[derive(Debug, Clone, PartialEq)]
1063pub struct PartitionedRowKeyRange {
1064	pub storage: StorageId,
1065}
1066
1067impl PartitionedRowKeyRange {
1068	fn decode_key(key: &EncodedKey) -> Option<Self> {
1069		let mut de = KeyDeserializer::from_bytes(key.as_slice());
1070
1071		let kind: KeyTag = de.read_u8().ok()?.try_into().ok()?;
1072		if kind != Self::TAG {
1073			return None;
1074		}
1075
1076		let storage = StorageId::from_object(de.read_object_id().ok()?)?;
1077
1078		Some(PartitionedRowKeyRange {
1079			storage,
1080		})
1081	}
1082}
1083
1084impl KeyRangeCodec for PartitionedRowKeyRange {
1085	const TAG: KeyTag = KeyTag::PartitionedRow;
1086
1087	fn start(&self) -> Option<EncodedKey> {
1088		let mut serializer = KeySerializer::with_capacity(10);
1089		serializer.extend_u8(Self::TAG as u8).extend_object_id(self.storage);
1090		Some(serializer.to_encoded_key())
1091	}
1092
1093	fn end(&self) -> Option<EncodedKey> {
1094		let mut serializer = KeySerializer::with_capacity(10);
1095		serializer.extend_u8(Self::TAG as u8).extend_object_id(ObjectId::from(self.storage).prev());
1096		Some(serializer.to_encoded_key())
1097	}
1098
1099	fn decode(range: &EncodedKeyRange) -> (Option<Self>, Option<Self>)
1100	where
1101		Self: Sized,
1102	{
1103		let start_key = match &range.start {
1104			Bound::Included(key) | Bound::Excluded(key) => Self::decode_key(key),
1105			Bound::Unbounded => None,
1106		};
1107
1108		let end_key = match &range.end {
1109			Bound::Included(key) | Bound::Excluded(key) => Self::decode_key(key),
1110			Bound::Unbounded => None,
1111		};
1112
1113		(start_key, end_key)
1114	}
1115}
1116
1117#[cfg(test)]
1118mod partitioned_row_key_tests {
1119	use std::ops::RangeBounds;
1120
1121	use reifydb_codec::key::{encoded::EncodedKey, serializer::KeySerializer};
1122	use reifydb_value::value::{Value, partition::Partition, row_number::RowNumber};
1123
1124	use super::{PartitionedRowKey, RowKey, StoragePartitionedRowKey};
1125	use crate::{
1126		interface::catalog::{
1127			id::{TableId, ViewId},
1128			object::ObjectId,
1129			storage::StorageId,
1130		},
1131		key::{
1132			catalog::KeySerializerCatalogExt,
1133			typed::{BoundedKey, DenseKey},
1134		},
1135	};
1136
1137	fn part(v: &str) -> Partition {
1138		Partition::of(&[Value::Utf8(v.to_string())])
1139	}
1140
1141	#[test]
1142	fn test_table_roundtrip() {
1143		let key = PartitionedRowKey {
1144			storage: StorageId::Table(TableId(7)),
1145			partition: part("us"),
1146			row: RowNumber(42),
1147		};
1148		let decoded = PartitionedRowKey::decode(&key.encode()).unwrap();
1149		assert_eq!(decoded, key);
1150	}
1151
1152	#[test]
1153	fn test_view_roundtrip() {
1154		// A view owns its rows directly; without the view tag narrowing back the key decodes to `None`.
1155		let key = PartitionedRowKey {
1156			storage: StorageId::View(ViewId(11)),
1157			partition: part("us"),
1158			row: RowNumber(42),
1159		};
1160		let decoded = PartitionedRowKey::decode(&key.encode()).unwrap();
1161		assert_eq!(decoded, key);
1162	}
1163
1164	#[test]
1165	fn test_storage_of() {
1166		let key = PartitionedRowKey::encoded(StorageId::Table(TableId(42)), part("us"), RowNumber(1));
1167		assert_eq!(PartitionedRowKey::storage_of(&key), Some(StorageId::Table(TableId(42))));
1168	}
1169
1170	#[test]
1171	fn test_storage_of_rejects_a_rowless_object() {
1172		// A vtable owns no rows, so its tag must not narrow into a table of the same numeric id.
1173		let mut serializer = KeySerializer::with_capacity(10);
1174		serializer.extend_u8(PartitionedRowKey::TAG as u8).extend_object_id(ObjectId::vtable(42));
1175		assert_eq!(PartitionedRowKey::storage_of(&serializer.to_encoded_key()), None);
1176	}
1177
1178	#[test]
1179	fn test_partition_rows_cluster_together() {
1180		let storage = StorageId::Table(TableId(1));
1181		let us_a = PartitionedRowKey::encoded(storage, part("us"), RowNumber(1));
1182		let us_b = PartitionedRowKey::encoded(storage, part("us"), RowNumber(2));
1183		let eu = PartitionedRowKey::encoded(storage, part("eu"), RowNumber(1));
1184
1185		let mut keys = [us_a.clone(), us_b.clone(), eu.clone()];
1186		keys.sort();
1187		let us_positions: Vec<usize> =
1188			keys.iter().enumerate().filter(|(_, k)| **k == us_a || **k == us_b).map(|(i, _)| i).collect();
1189		assert_eq!(us_positions[1] - us_positions[0], 1, "us partition rows must be contiguous");
1190	}
1191
1192	#[test]
1193	fn test_partition_range_contains_only_its_partition() {
1194		let storage = StorageId::Table(TableId(1));
1195		let range = PartitionedRowKey::partition_range(storage, part("us")).encode();
1196		let us = PartitionedRowKey::encoded(storage, part("us"), RowNumber(500));
1197		let eu = PartitionedRowKey::encoded(storage, part("eu"), RowNumber(1));
1198		assert!(range.contains(&us), "us row must be inside the us partition range");
1199		assert!(!range.contains(&eu), "eu row must be outside the us partition range");
1200	}
1201
1202	#[test]
1203	fn test_partitioned_row_ident_roundtrip() {
1204		let key = PartitionedRowKey {
1205			storage: StorageId::Table(TableId(7)),
1206			partition: part("us"),
1207			row: RowNumber(42),
1208		};
1209
1210		// dropping storage and re-supplying it must recover the original key, halves included
1211		let ident: StoragePartitionedRowKey = key.clone().into();
1212		let restored = ident.with_storage(key.storage);
1213		assert_eq!(restored, key);
1214	}
1215
1216	#[test]
1217	fn test_partitioned_row_ident_halves_split_correctly() {
1218		let partition = Partition(0x1122334455667788_99AABBCCDDEEFF00);
1219		let ident = StoragePartitionedRowKey::new(partition, RowNumber(1));
1220
1221		// the two native halves must reassemble into the exact original 128-bit value
1222		assert_eq!(ident.partition_hi(), 0x1122334455667788);
1223		assert_eq!(ident.partition_lo(), 0x99AABBCCDDEEFF00);
1224		assert_eq!(ident.partition(), partition);
1225		assert_eq!(
1226			StoragePartitionedRowKey::from_halves(ident.partition_hi(), ident.partition_lo(), RowNumber(1)),
1227			ident
1228		);
1229	}
1230
1231	#[test]
1232	fn test_partitioned_row_ident_ordering_matches_field_order() {
1233		let lower_partition = StoragePartitionedRowKey::new(Partition(1), RowNumber(999));
1234		let higher_partition = StoragePartitionedRowKey::new(Partition(2), RowNumber(1));
1235		let same_partition_lower_row = StoragePartitionedRowKey::new(Partition(2), RowNumber(1));
1236		let same_partition_higher_row = StoragePartitionedRowKey::new(Partition(2), RowNumber(2));
1237
1238		// partition must dominate row in ordering, matching PartitionedRowKey's field order, and both
1239		// run descending so the identity sorts exactly as the encoded key does
1240		assert!(higher_partition < lower_partition);
1241		assert!(same_partition_higher_row < same_partition_lower_row);
1242	}
1243
1244	#[test]
1245	fn test_partitioned_row_ident_successor_carries_into_the_partition() {
1246		// row zero ends a partition. Without the carry the scan stops at that boundary and never
1247		// reaches the next partition, which is the whole point of an ordered walk across many.
1248		let last_row = StoragePartitionedRowKey::new(Partition(5), RowNumber(0));
1249		let next = last_row.successor().unwrap();
1250
1251		assert_eq!(next, StoragePartitionedRowKey::new(Partition(4), RowNumber(u64::MAX)));
1252		assert!(next > last_row);
1253	}
1254
1255	#[test]
1256	fn test_partitioned_row_ident_low_is_the_greatest_partition_and_row() {
1257		assert_eq!(
1258			<StoragePartitionedRowKey as BoundedKey>::low(),
1259			StoragePartitionedRowKey::new(Partition(u128::MAX), RowNumber(u64::MAX))
1260		);
1261	}
1262
1263	#[test]
1264	fn test_decode_rejects_trailing_bytes() {
1265		// A key that is longer than the fixed layout must not decode as this type, even when its
1266		// first byte is a matching kind: without the check the extra bytes are ignored and the
1267		// field offsets read a different key's payload as a valid row.
1268		let exact = RowKey::encoded(StorageId::table(7), RowNumber(42));
1269		assert_eq!(exact.as_slice().len(), 18);
1270		assert_eq!(
1271			RowKey::decode(&exact),
1272			Some(RowKey {
1273				storage: StorageId::table(7),
1274				row: RowNumber(42)
1275			})
1276		);
1277
1278		let mut longer = exact.as_slice().to_vec();
1279		longer.push(0x00);
1280		assert_eq!(RowKey::decode(&EncodedKey::new(longer)), None);
1281	}
1282
1283	#[test]
1284	fn test_decode_rejects_a_longer_key_that_shares_the_kind_byte() {
1285		// The live shape: a sorted view writes kind ++ storage ++ sort values ++ row. It carries
1286		// KeyTag::Row, so a length-blind decode claims it and reads sort payload as the row
1287		// number, silently aliasing two distinct rows onto one storage row key.
1288		let mut sorted_view = RowKey::encoded(StorageId::view(3), RowNumber(1)).as_slice().to_vec();
1289		sorted_view.extend_from_slice(&[0xAA; 8]);
1290		sorted_view.extend_from_slice(&99u64.to_be_bytes());
1291
1292		assert_eq!(RowKey::decode(&EncodedKey::new(sorted_view)), None);
1293	}
1294}
1295
1296#[cfg(test)]
1297mod sorted_view_run_tests {
1298	use reifydb_codec::key::encoded::EncodedKey;
1299	use reifydb_value::value::{partition::Partition, row_number::RowNumber};
1300
1301	use super::{PartitionedSortedViewRowKey, SortRun, SortedViewRowKey};
1302	use crate::interface::catalog::storage::StorageId;
1303
1304	fn run(bytes: &[u8]) -> SortRun {
1305		SortRun::new(bytes)
1306	}
1307
1308	#[test]
1309	fn test_a_run_round_trips_through_the_terminator() {
1310		// The run is spliced raw today and cannot be decoded; without the terminator the row tail and
1311		// the run cannot be told apart and decode has to guess where one ends.
1312		let key = SortedViewRowKey::new(StorageId::view(3), run(&[0x10, 0x00, 0xff, 0x00]), RowNumber(42));
1313		let decoded = SortedViewRowKey::decode(&key.encode()).unwrap();
1314		assert_eq!(decoded, key);
1315		assert_eq!(decoded.run.as_slice(), &[0x10, 0x00, 0xff, 0x00]);
1316	}
1317
1318	#[test]
1319	fn test_an_empty_run_round_trips() {
1320		let key = SortedViewRowKey::new(StorageId::view(3), run(&[]), RowNumber(1));
1321		assert_eq!(SortedViewRowKey::decode(&key.encode()), Some(key));
1322	}
1323
1324	#[test]
1325	fn test_a_partitioned_run_round_trips() {
1326		let key = PartitionedSortedViewRowKey::new(
1327			StorageId::view(3),
1328			Partition(0x1122334455667788_99AABBCCDDEEFF00),
1329			run(&[0x00, 0x00, 0x01]),
1330			RowNumber(7),
1331		);
1332		assert_eq!(PartitionedSortedViewRowKey::decode(&key.encode()), Some(key));
1333	}
1334
1335	#[test]
1336	fn test_decode_refuses_the_other_kind_and_trailing_bytes() {
1337		// Both kinds end in a raw eight byte row, so a length blind decode would read one as the other.
1338		let partitioned = PartitionedSortedViewRowKey::encoded(
1339			StorageId::view(3),
1340			Partition(1),
1341			run(&[0x10]),
1342			RowNumber(1),
1343		);
1344		assert_eq!(SortedViewRowKey::decode(&partitioned), None);
1345
1346		let mut longer = SortedViewRowKey::encoded(StorageId::view(3), run(&[0x10]), RowNumber(1)).to_vec();
1347		longer.push(0x00);
1348		assert_eq!(SortedViewRowKey::decode(&EncodedKey::new(longer)), None);
1349	}
1350
1351	#[test]
1352	fn test_ord_matches_the_encoded_byte_order() {
1353		// A view is read by a plain forward scan, so the struct order and the byte order must agree
1354		// everywhere or the scan returns rows in an order the planner never asked for.
1355		let mut keys = vec![
1356			SortedViewRowKey::new(StorageId::table(3), run(&[0x10]), RowNumber(1)),
1357			SortedViewRowKey::new(StorageId::view(3), run(&[0x10]), RowNumber(1)),
1358			SortedViewRowKey::new(StorageId::view(4), run(&[0x10]), RowNumber(1)),
1359			SortedViewRowKey::new(StorageId::view(3), run(&[0x10]), RowNumber(2)),
1360			SortedViewRowKey::new(StorageId::view(3), run(&[0x10, 0x00]), RowNumber(0)),
1361			SortedViewRowKey::new(StorageId::view(3), run(&[0x10, 0x01]), RowNumber(0)),
1362			SortedViewRowKey::new(StorageId::view(3), run(&[0x00]), RowNumber(0)),
1363			SortedViewRowKey::new(StorageId::view(3), run(&[]), RowNumber(u64::MAX)),
1364		];
1365		keys.sort();
1366
1367		let encoded: Vec<EncodedKey> = keys.iter().map(|key| key.encode()).collect();
1368		let mut sorted_bytes = encoded.clone();
1369		sorted_bytes.sort();
1370		assert_eq!(encoded, sorted_bytes);
1371	}
1372
1373	#[test]
1374	fn test_partitioned_ord_matches_the_encoded_byte_order() {
1375		let mut keys = vec![
1376			PartitionedSortedViewRowKey::new(StorageId::view(3), Partition(1), run(&[0x10]), RowNumber(1)),
1377			PartitionedSortedViewRowKey::new(StorageId::view(3), Partition(2), run(&[0x10]), RowNumber(1)),
1378			PartitionedSortedViewRowKey::new(StorageId::view(3), Partition(2), run(&[0x00]), RowNumber(1)),
1379			PartitionedSortedViewRowKey::new(StorageId::view(3), Partition(2), run(&[0x10]), RowNumber(0)),
1380			PartitionedSortedViewRowKey::new(StorageId::view(4), Partition(2), run(&[0x10]), RowNumber(0)),
1381		];
1382		keys.sort();
1383
1384		let encoded: Vec<EncodedKey> = keys.iter().map(|key| key.encode()).collect();
1385		let mut sorted_bytes = encoded.clone();
1386		sorted_bytes.sort();
1387		assert_eq!(encoded, sorted_bytes);
1388	}
1389
1390	#[test]
1391	fn test_a_shorter_run_sorts_before_the_run_that_extends_it() {
1392		// Without a terminator the row tail of the shorter key lands against the extra run bytes of the
1393		// longer one, so a high row number could drag a row past the whole group that follows it.
1394		let short = SortedViewRowKey::new(StorageId::view(3), run(&[0x10]), RowNumber(u64::MAX));
1395		let long = SortedViewRowKey::new(StorageId::view(3), run(&[0x10, 0x00]), RowNumber(0));
1396
1397		assert!(short < long);
1398		assert!(short.encode() < long.encode());
1399	}
1400
1401	#[test]
1402	fn test_the_helpers_still_read_the_new_layout() {
1403		let storage = StorageId::view(3);
1404		let key = SortedViewRowKey::encoded(storage, run(&[0x00, 0xAA]), RowNumber(9));
1405		assert_eq!(SortedViewRowKey::storage_of(&key), Some(storage));
1406		assert_eq!(SortedViewRowKey::row_of(&key), Some(RowNumber(9)));
1407
1408		let partitioned =
1409			PartitionedSortedViewRowKey::encoded(storage, Partition(5), run(&[0x00, 0xAA]), RowNumber(9));
1410		assert_eq!(PartitionedSortedViewRowKey::storage_of(&partitioned), Some(storage));
1411		assert_eq!(PartitionedSortedViewRowKey::row_of(&partitioned), Some(RowNumber(9)));
1412		assert_eq!(SortedViewRowKey::row_of(&partitioned), None);
1413	}
1414}
1415
1416impl KeyFields for SortedViewRowKey {
1417	fn fields(&self) -> SmallVec<[Field<'_>; 6]> {
1418		smallvec![
1419			Field::UAsc(Width::U8, ObjectId::from(self.storage).type_tag() as u128),
1420			Field::UDesc(Width::U64, ObjectId::from(self.storage).as_u64() as u128),
1421			Field::RawAsc(RawEncoding::SortRun, Cow::Borrowed(self.run.as_slice())),
1422			Field::UAsc(Width::U64, self.row.0.0 as u128),
1423		]
1424	}
1425}
1426
1427impl KeyFields for PartitionedSortedViewRowKey {
1428	fn fields(&self) -> SmallVec<[Field<'_>; 6]> {
1429		smallvec![
1430			Field::UAsc(Width::U8, ObjectId::from(self.storage).type_tag() as u128),
1431			Field::UDesc(Width::U64, ObjectId::from(self.storage).as_u64() as u128),
1432			Field::UDesc(Width::U128, self.partition.0),
1433			Field::RawAsc(RawEncoding::SortRun, Cow::Borrowed(self.run.as_slice())),
1434			Field::UAsc(Width::U64, self.row.0.0 as u128),
1435		]
1436	}
1437}
reifydb_core/key/row.rs

reifydb_core/key/
row.rs
