Skip to main content

reifydb_core/value/index/
set.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{f64, ptr};
5
6#[cfg(reifydb_assertions)]
7use reifydb_value::value::value_type::ValueType;
8use reifydb_value::{
9	reifydb_assertions,
10	value::{
11		date::Date,
12		datetime::DateTime,
13		duration::Duration,
14		identity::IdentityId,
15		time::Time,
16		uuid::{Uuid4, Uuid7},
17	},
18};
19use uuid::Uuid;
20
21use crate::{
22	sort::SortDirection,
23	value::index::{encoded::EncodedIndexKey, shape::IndexShape},
24};
25
26impl IndexShape {
27	pub fn set_bool(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<bool>) {
28		let field = &self.fields[index];
29		reifydb_assertions! {
30			assert_eq!(field.value, ValueType::Boolean);
31		}
32		key.set_valid(index, true);
33
34		let byte_value = match field.direction {
35			SortDirection::Asc => {
36				if value.into() {
37					1u8
38				} else {
39					0u8
40				}
41			}
42			SortDirection::Desc => {
43				if value.into() {
44					0u8
45				} else {
46					1u8
47				}
48			}
49		};
50
51		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
52		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), byte_value) }
53	}
54
55	pub fn set_f32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<f32>) {
56		let field = &self.fields[index];
57		reifydb_assertions! {
58			assert_eq!(field.value, ValueType::Float4);
59		}
60		key.set_valid(index, true);
61
62		let v = value.into();
63		let mut bytes = v.to_bits().to_be_bytes();
64
65		if v.is_sign_negative() {
66			for b in bytes.iter_mut() {
67				*b = !*b;
68			}
69		} else {
70			bytes[0] ^= 0x80;
71		}
72
73		if field.direction == SortDirection::Desc {
74			for b in bytes.iter_mut() {
75				*b = !*b;
76			}
77		}
78
79		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
80		unsafe {
81			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
82		}
83	}
84
85	pub fn set_f64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<f64>) {
86		let field = &self.fields[index];
87		reifydb_assertions! {
88			assert_eq!(field.value, ValueType::Float8);
89		}
90		key.set_valid(index, true);
91
92		let v = value.into();
93		let mut bytes = v.to_bits().to_be_bytes();
94
95		if v.is_sign_negative() {
96			for b in bytes.iter_mut() {
97				*b = !*b;
98			}
99		} else {
100			bytes[0] ^= 0x80;
101		}
102
103		if field.direction == SortDirection::Desc {
104			for b in bytes.iter_mut() {
105				*b = !*b;
106			}
107		}
108
109		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
110		unsafe {
111			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
112		}
113	}
114
115	pub fn set_i8(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i8>) {
116		let field = &self.fields[index];
117		reifydb_assertions! {
118			assert_eq!(field.value, ValueType::Int1);
119		}
120		key.set_valid(index, true);
121
122		let mut bytes = value.into().to_be_bytes();
123
124		match field.direction {
125			SortDirection::Asc => {
126				bytes[0] ^= 0x80;
127			}
128			SortDirection::Desc => {
129				bytes[0] ^= 0x80;
130				for b in bytes.iter_mut() {
131					*b = !*b;
132				}
133			}
134		}
135
136		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
137		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), bytes[0]) }
138	}
139
140	pub fn set_i16(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i16>) {
141		let field = &self.fields[index];
142		reifydb_assertions! {
143			assert_eq!(field.value, ValueType::Int2);
144		}
145		key.set_valid(index, true);
146
147		let mut bytes = value.into().to_be_bytes();
148
149		match field.direction {
150			SortDirection::Asc => {
151				bytes[0] ^= 0x80;
152			}
153			SortDirection::Desc => {
154				bytes[0] ^= 0x80;
155				for b in bytes.iter_mut() {
156					*b = !*b;
157				}
158			}
159		}
160
161		// SAFETY: IndexShapeInner::new puts field.offset + 2 inside total_size, so the copy stays in bounds.
162		unsafe {
163			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 2);
164		}
165	}
166
167	pub fn set_i32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i32>) {
168		let field = &self.fields[index];
169		reifydb_assertions! {
170			assert_eq!(field.value, ValueType::Int4);
171		}
172		key.set_valid(index, true);
173
174		let mut bytes = value.into().to_be_bytes();
175
176		match field.direction {
177			SortDirection::Asc => {
178				bytes[0] ^= 0x80;
179			}
180			SortDirection::Desc => {
181				bytes[0] ^= 0x80;
182				for b in bytes.iter_mut() {
183					*b = !*b;
184				}
185			}
186		}
187
188		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
189		unsafe {
190			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
191		}
192	}
193
194	pub fn set_i64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i64>) {
195		let field = &self.fields[index];
196		reifydb_assertions! {
197			assert_eq!(field.value, ValueType::Int8);
198		}
199		key.set_valid(index, true);
200
201		let mut bytes = value.into().to_be_bytes();
202
203		match field.direction {
204			SortDirection::Asc => {
205				bytes[0] ^= 0x80;
206			}
207			SortDirection::Desc => {
208				bytes[0] ^= 0x80;
209				for b in bytes.iter_mut() {
210					*b = !*b;
211				}
212			}
213		}
214
215		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
216		unsafe {
217			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
218		}
219	}
220
221	pub fn set_i128(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i128>) {
222		let field = &self.fields[index];
223		reifydb_assertions! {
224			assert_eq!(field.value, ValueType::Int16);
225		}
226		key.set_valid(index, true);
227
228		let mut bytes = value.into().to_be_bytes();
229
230		match field.direction {
231			SortDirection::Asc => {
232				bytes[0] ^= 0x80;
233			}
234			SortDirection::Desc => {
235				bytes[0] ^= 0x80;
236				for b in bytes.iter_mut() {
237					*b = !*b;
238				}
239			}
240		}
241
242		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
243		unsafe {
244			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
245		}
246	}
247
248	pub fn set_u8(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u8>) {
249		let field = &self.fields[index];
250		reifydb_assertions! {
251			assert_eq!(field.value, ValueType::Uint1);
252		}
253		key.set_valid(index, true);
254
255		let byte = match field.direction {
256			SortDirection::Asc => value.into(),
257			SortDirection::Desc => !value.into(),
258		};
259
260		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
261		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), byte) }
262	}
263
264	pub fn set_u16(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u16>) {
265		let field = &self.fields[index];
266		reifydb_assertions! {
267			assert_eq!(field.value, ValueType::Uint2);
268		}
269		key.set_valid(index, true);
270
271		let bytes = match field.direction {
272			SortDirection::Asc => value.into().to_be_bytes(),
273			SortDirection::Desc => (!value.into()).to_be_bytes(),
274		};
275
276		// SAFETY: IndexShapeInner::new puts field.offset + 2 inside total_size, so the copy stays in bounds.
277		unsafe {
278			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 2);
279		}
280	}
281
282	pub fn set_u32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u32>) {
283		let field = &self.fields[index];
284		reifydb_assertions! {
285			assert_eq!(field.value, ValueType::Uint4);
286		}
287		key.set_valid(index, true);
288
289		let bytes = match field.direction {
290			SortDirection::Asc => value.into().to_be_bytes(),
291			SortDirection::Desc => (!value.into()).to_be_bytes(),
292		};
293
294		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
295		unsafe {
296			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
297		}
298	}
299
300	pub fn set_u64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u64>) {
301		let field = &self.fields[index];
302		reifydb_assertions! {
303			assert_eq!(field.value, ValueType::Uint8);
304		}
305		key.set_valid(index, true);
306
307		let bytes = match field.direction {
308			SortDirection::Asc => value.into().to_be_bytes(),
309			SortDirection::Desc => (!value.into()).to_be_bytes(),
310		};
311
312		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
313		unsafe {
314			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
315		}
316	}
317
318	pub fn set_u128(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u128>) {
319		let field = &self.fields[index];
320		reifydb_assertions! {
321			assert_eq!(field.value, ValueType::Uint16);
322		}
323		key.set_valid(index, true);
324
325		let bytes = match field.direction {
326			SortDirection::Asc => value.into().to_be_bytes(),
327			SortDirection::Desc => (!value.into()).to_be_bytes(),
328		};
329
330		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
331		unsafe {
332			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
333		}
334	}
335
336	pub fn set_row_number(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u64>) {
337		let field = &self.fields[index];
338		reifydb_assertions! {
339			assert_eq!(field.value, ValueType::Uint8);
340		}
341		key.set_valid(index, true);
342
343		let bytes = match field.direction {
344			SortDirection::Asc => value.into().to_be_bytes(),
345			SortDirection::Desc => (!value.into()).to_be_bytes(),
346		};
347
348		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
349		unsafe {
350			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
351		}
352	}
353
354	pub fn set_date(&self, key: &mut EncodedIndexKey, index: usize, value: Date) {
355		let field = &self.fields[index];
356		reifydb_assertions! {
357			assert_eq!(field.value, ValueType::Date);
358		}
359		key.set_valid(index, true);
360
361		let days = value.to_days_since_epoch();
362		let mut bytes = days.to_be_bytes();
363
364		match field.direction {
365			SortDirection::Asc => {
366				bytes[0] ^= 0x80;
367			}
368			SortDirection::Desc => {
369				bytes[0] ^= 0x80;
370				for b in bytes.iter_mut() {
371					*b = !*b;
372				}
373			}
374		}
375
376		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
377		unsafe {
378			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
379		}
380	}
381
382	pub fn set_datetime(&self, key: &mut EncodedIndexKey, index: usize, value: DateTime) {
383		let field = &self.fields[index];
384		reifydb_assertions! {
385			assert_eq!(field.value, ValueType::DateTime);
386		}
387		key.set_valid(index, true);
388
389		let nanos = value.to_nanos();
390		let bytes = match field.direction {
391			SortDirection::Asc => nanos.to_be_bytes(),
392			SortDirection::Desc => (!nanos).to_be_bytes(),
393		};
394
395		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
396		unsafe {
397			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
398		}
399	}
400
401	pub fn set_time(&self, key: &mut EncodedIndexKey, index: usize, value: Time) {
402		let field = &self.fields[index];
403		reifydb_assertions! {
404			assert_eq!(field.value, ValueType::Time);
405		}
406		key.set_valid(index, true);
407
408		let nanos = value.to_nanos_since_midnight();
409		let bytes = match field.direction {
410			SortDirection::Asc => nanos.to_be_bytes(),
411			SortDirection::Desc => (!nanos).to_be_bytes(),
412		};
413
414		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
415		unsafe {
416			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
417		}
418	}
419
420	pub fn set_duration(&self, key: &mut EncodedIndexKey, index: usize, value: Duration) {
421		let field = &self.fields[index];
422		reifydb_assertions! {
423			assert_eq!(field.value, ValueType::Duration);
424		}
425		key.set_valid(index, true);
426
427		let mut months_bytes = value.get_months().to_be_bytes();
428		let mut days_bytes = value.get_days().to_be_bytes();
429		let mut nanos_bytes = value.get_nanos().to_be_bytes();
430
431		match field.direction {
432			SortDirection::Asc => {
433				months_bytes[0] ^= 0x80;
434				days_bytes[0] ^= 0x80;
435				nanos_bytes[0] ^= 0x80;
436			}
437			SortDirection::Desc => {
438				months_bytes[0] ^= 0x80;
439				days_bytes[0] ^= 0x80;
440				nanos_bytes[0] ^= 0x80;
441				for b in months_bytes.iter_mut() {
442					*b = !*b;
443				}
444				for b in days_bytes.iter_mut() {
445					*b = !*b;
446				}
447				for b in nanos_bytes.iter_mut() {
448					*b = !*b;
449				}
450			}
451		}
452
453		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so all three copies fit.
454		unsafe {
455			ptr::copy_nonoverlapping(
456				months_bytes.as_ptr(),
457				key.make_mut().as_mut_ptr().add(field.offset),
458				4,
459			);
460			ptr::copy_nonoverlapping(
461				days_bytes.as_ptr(),
462				key.make_mut().as_mut_ptr().add(field.offset + 4),
463				4,
464			);
465			ptr::copy_nonoverlapping(
466				nanos_bytes.as_ptr(),
467				key.make_mut().as_mut_ptr().add(field.offset + 8),
468				8,
469			);
470		}
471	}
472
473	pub fn set_uuid4(&self, key: &mut EncodedIndexKey, index: usize, value: Uuid4) {
474		let field = &self.fields[index];
475		reifydb_assertions! {
476			assert_eq!(field.value, ValueType::Uuid4);
477		}
478		key.set_valid(index, true);
479
480		let uuid: Uuid = value.into();
481		let uuid_bytes = uuid.as_bytes();
482		let mut bytes = [0u8; 16];
483		bytes.copy_from_slice(uuid_bytes);
484
485		if field.direction == SortDirection::Desc {
486			for b in bytes.iter_mut() {
487				*b = !*b;
488			}
489		}
490
491		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
492		unsafe {
493			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
494		}
495	}
496
497	pub fn set_uuid7(&self, key: &mut EncodedIndexKey, index: usize, value: Uuid7) {
498		let field = &self.fields[index];
499		reifydb_assertions! {
500			assert_eq!(field.value, ValueType::Uuid7);
501		}
502		key.set_valid(index, true);
503
504		let uuid: Uuid = value.into();
505		let uuid_bytes = uuid.as_bytes();
506		let mut bytes = [0u8; 16];
507		bytes.copy_from_slice(uuid_bytes);
508
509		if field.direction == SortDirection::Desc {
510			for b in bytes.iter_mut() {
511				*b = !*b;
512			}
513		}
514
515		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
516		unsafe {
517			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
518		}
519	}
520
521	pub fn set_identity_id(&self, key: &mut EncodedIndexKey, index: usize, value: IdentityId) {
522		let field = &self.fields[index];
523		reifydb_assertions! {
524			assert_eq!(field.value, ValueType::IdentityId);
525		}
526		key.set_valid(index, true);
527
528		let uuid: Uuid = value.0.into();
529		let uuid_bytes = uuid.as_bytes();
530		let mut bytes = [0u8; 16];
531		bytes.copy_from_slice(uuid_bytes);
532
533		if field.direction == SortDirection::Desc {
534			for b in bytes.iter_mut() {
535				*b = !*b;
536			}
537		}
538
539		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
540		unsafe {
541			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
542		}
543	}
544
545	pub fn set_none(&self, key: &mut EncodedIndexKey, index: usize) {
546		let field = &self.fields[index];
547		key.set_valid(index, false);
548
549		let buf = key.make_mut();
550		let start = field.offset;
551		let end = start + field.size;
552		buf[start..end].fill(0);
553	}
554}
555
556#[cfg(test)]
557pub mod tests {
558	use crate::{sort::SortDirection, value::index::shape::IndexShape};
559
560	mod bool {
561		use reifydb_value::value::value_type::ValueType;
562
563		use super::*;
564
565		#[test]
566		fn test_asc() {
567			let layout = IndexShape::new(&[ValueType::Boolean], &[SortDirection::Asc]).unwrap();
568			let mut key_false = layout.allocate_key();
569			let mut key_true = layout.allocate_key();
570
571			layout.set_bool(&mut key_false, 0, false);
572			layout.set_bool(&mut key_true, 0, true);
573
574			assert_eq!(key_false[0] & 0x01, 0x01);
575			assert_eq!(key_true[0] & 0x01, 0x01);
576
577			let offset = layout.fields[0].offset;
578			assert_eq!(key_false[offset], 0);
579			assert_eq!(key_true[offset], 1);
580
581			assert!(key_false.as_slice() < key_true.as_slice());
582		}
583
584		#[test]
585		fn test_desc() {
586			let layout = IndexShape::new(&[ValueType::Boolean], &[SortDirection::Desc]).unwrap();
587			let mut key_false = layout.allocate_key();
588			let mut key_true = layout.allocate_key();
589
590			layout.set_bool(&mut key_false, 0, false);
591			layout.set_bool(&mut key_true, 0, true);
592
593			let offset = layout.fields[0].offset;
594			assert_eq!(key_false[offset], 1); // false becomes 1 in DESC
595			assert_eq!(key_true[offset], 0); // true becomes 0 in DESC
596
597			assert!(key_false.as_slice() > key_true.as_slice());
598		}
599	}
600
601	mod i8 {
602		use reifydb_value::value::value_type::ValueType;
603
604		use crate::{sort::SortDirection, value::index::shape::IndexShape};
605
606		#[test]
607		fn test_asc() {
608			let layout = IndexShape::new(&[ValueType::Int1], &[SortDirection::Asc]).unwrap();
609			let mut key_neg = layout.allocate_key();
610			let mut key_zero = layout.allocate_key();
611			let mut key_pos = layout.allocate_key();
612
613			layout.set_i8(&mut key_neg, 0, -128i8);
614			layout.set_i8(&mut key_zero, 0, 0i8);
615			layout.set_i8(&mut key_pos, 0, 127i8);
616
617			let offset = layout.fields[0].offset;
618			// -128 with sign bit flipped: 0x80 -> 0x00
619			assert_eq!(key_neg[offset], 0x00);
620			// 0 with sign bit flipped: 0x00 -> 0x80
621			assert_eq!(key_zero[offset], 0x80);
622			// 127 with sign bit flipped: 0x7F -> 0xFF
623			assert_eq!(key_pos[offset], 0xFF);
624
625			assert!(key_neg.as_slice() < key_zero.as_slice());
626			assert!(key_zero.as_slice() < key_pos.as_slice());
627		}
628
629		#[test]
630		fn test_desc() {
631			let layout = IndexShape::new(&[ValueType::Int1], &[SortDirection::Desc]).unwrap();
632			let mut key_neg = layout.allocate_key();
633			let mut key_zero = layout.allocate_key();
634			let mut key_pos = layout.allocate_key();
635
636			layout.set_i8(&mut key_neg, 0, -128i8);
637			layout.set_i8(&mut key_zero, 0, 0i8);
638			layout.set_i8(&mut key_pos, 0, 127i8);
639
640			let offset = layout.fields[0].offset;
641			// -128: 0x80 -> flip sign: 0x00 -> invert: 0xFF
642			assert_eq!(key_neg[offset], 0xFF);
643			// 0: 0x00 -> flip sign: 0x80 -> invert: 0x7F
644			assert_eq!(key_zero[offset], 0x7F);
645			// 127: 0x7F -> flip sign: 0xFF -> invert: 0x00
646			assert_eq!(key_pos[offset], 0x00);
647
648			assert!(key_neg.as_slice() > key_zero.as_slice());
649			assert!(key_zero.as_slice() > key_pos.as_slice());
650		}
651	}
652
653	mod i32 {
654		use reifydb_value::value::value_type::ValueType;
655
656		use crate::{sort::SortDirection, value::index::shape::IndexShape};
657
658		#[test]
659		fn test_asc() {
660			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Asc]).unwrap();
661			let mut key_neg = layout.allocate_key();
662			let mut key_zero = layout.allocate_key();
663			let mut key_pos = layout.allocate_key();
664
665			layout.set_i32(&mut key_neg, 0, i32::MIN);
666			layout.set_i32(&mut key_zero, 0, 0i32);
667			layout.set_i32(&mut key_pos, 0, i32::MAX);
668
669			let offset = layout.fields[0].offset;
670			// i32::MIN in big-endian with sign bit flipped
671			assert_eq!(&key_neg[offset..offset + 4], &[0x00, 0x00, 0x00, 0x00]);
672			// 0 with sign bit flipped
673			assert_eq!(&key_zero[offset..offset + 4], &[0x80, 0x00, 0x00, 0x00]);
674			// i32::MAX with sign bit flipped
675			assert_eq!(&key_pos[offset..offset + 4], &[0xFF, 0xFF, 0xFF, 0xFF]);
676
677			assert!(key_neg.as_slice() < key_zero.as_slice());
678			assert!(key_zero.as_slice() < key_pos.as_slice());
679		}
680
681		#[test]
682		fn test_desc() {
683			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Desc]).unwrap();
684			let mut key_neg = layout.allocate_key();
685			let mut key_zero = layout.allocate_key();
686			let mut key_pos = layout.allocate_key();
687
688			layout.set_i32(&mut key_neg, 0, i32::MIN);
689			layout.set_i32(&mut key_zero, 0, 0i32);
690			layout.set_i32(&mut key_pos, 0, i32::MAX);
691
692			let offset = layout.fields[0].offset;
693			// i32::MIN: flip sign then invert all
694			assert_eq!(&key_neg[offset..offset + 4], &[0xFF, 0xFF, 0xFF, 0xFF]);
695			// 0: flip sign then invert all
696			assert_eq!(&key_zero[offset..offset + 4], &[0x7F, 0xFF, 0xFF, 0xFF]);
697			// i32::MAX: flip sign then invert all
698			assert_eq!(&key_pos[offset..offset + 4], &[0x00, 0x00, 0x00, 0x00]);
699
700			assert!(key_neg.as_slice() > key_zero.as_slice());
701			assert!(key_zero.as_slice() > key_pos.as_slice());
702		}
703	}
704
705	mod i64 {
706		use reifydb_value::value::value_type::ValueType;
707
708		use crate::{sort::SortDirection, value::index::shape::IndexShape};
709
710		#[test]
711		fn test_asc() {
712			let layout = IndexShape::new(&[ValueType::Int8], &[SortDirection::Asc]).unwrap();
713			let mut key = layout.allocate_key();
714
715			layout.set_i64(&mut key, 0, -1i64);
716
717			let offset = layout.fields[0].offset;
718			// -1 is all 1s in two's complement; the flipped sign bit gives 0x7F...
719			assert_eq!(&key[offset..offset + 8], &[0x7F, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]);
720		}
721
722		#[test]
723		fn test_desc() {
724			let layout = IndexShape::new(&[ValueType::Int8], &[SortDirection::Desc]).unwrap();
725			let mut key = layout.allocate_key();
726
727			layout.set_i64(&mut key, 0, -1i64);
728
729			let offset = layout.fields[0].offset;
730			// -1: flip sign then invert all
731			assert_eq!(&key[offset..offset + 8], &[0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]);
732		}
733	}
734
735	mod u8 {
736		use reifydb_value::value::value_type::ValueType;
737
738		use crate::{sort::SortDirection, value::index::shape::IndexShape};
739
740		#[test]
741		fn test_asc() {
742			let layout = IndexShape::new(&[ValueType::Uint1], &[SortDirection::Asc]).unwrap();
743			let mut key_min = layout.allocate_key();
744			let mut key_mid = layout.allocate_key();
745			let mut key_max = layout.allocate_key();
746
747			layout.set_u8(&mut key_min, 0, 0u8);
748			layout.set_u8(&mut key_mid, 0, 128u8);
749			layout.set_u8(&mut key_max, 0, 255u8);
750
751			let offset = layout.fields[0].offset;
752			assert_eq!(key_min[offset], 0x00);
753			assert_eq!(key_mid[offset], 0x80);
754			assert_eq!(key_max[offset], 0xFF);
755
756			assert!(key_min.as_slice() < key_mid.as_slice());
757			assert!(key_mid.as_slice() < key_max.as_slice());
758		}
759
760		#[test]
761		fn test_desc() {
762			let layout = IndexShape::new(&[ValueType::Uint1], &[SortDirection::Desc]).unwrap();
763			let mut key_min = layout.allocate_key();
764			let mut key_mid = layout.allocate_key();
765			let mut key_max = layout.allocate_key();
766
767			layout.set_u8(&mut key_min, 0, 0u8);
768			layout.set_u8(&mut key_mid, 0, 128u8);
769			layout.set_u8(&mut key_max, 0, 255u8);
770
771			let offset = layout.fields[0].offset;
772			// Inverted for DESC
773			assert_eq!(key_min[offset], 0xFF);
774			assert_eq!(key_mid[offset], 0x7F);
775			assert_eq!(key_max[offset], 0x00);
776
777			assert!(key_min.as_slice() > key_mid.as_slice());
778			assert!(key_mid.as_slice() > key_max.as_slice());
779		}
780	}
781
782	mod u32 {
783		use reifydb_value::value::value_type::ValueType;
784
785		use crate::{sort::SortDirection, value::index::shape::IndexShape};
786
787		#[test]
788		fn test_asc() {
789			let layout = IndexShape::new(&[ValueType::Uint4], &[SortDirection::Asc]).unwrap();
790			let mut key = layout.allocate_key();
791
792			layout.set_u32(&mut key, 0, 0x12345678u32);
793
794			let offset = layout.fields[0].offset;
795			// Big-endian representation
796			assert_eq!(&key[offset..offset + 4], &[0x12, 0x34, 0x56, 0x78]);
797		}
798
799		#[test]
800		fn test_desc() {
801			let layout = IndexShape::new(&[ValueType::Uint4], &[SortDirection::Desc]).unwrap();
802			let mut key = layout.allocate_key();
803
804			layout.set_u32(&mut key, 0, 0x12345678u32);
805
806			let offset = layout.fields[0].offset;
807			// Inverted for DESC
808			assert_eq!(&key[offset..offset + 4], &[0xED, 0xCB, 0xA9, 0x87]);
809		}
810	}
811
812	mod u64 {
813		use reifydb_value::value::value_type::ValueType;
814
815		use crate::{sort::SortDirection, value::index::shape::IndexShape};
816
817		#[test]
818		fn test_asc() {
819			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Asc]).unwrap();
820			let mut key = layout.allocate_key();
821
822			layout.set_u64(&mut key, 0, u64::MAX);
823
824			let offset = layout.fields[0].offset;
825			assert_eq!(&key[offset..offset + 8], &[0xFF; 8]);
826		}
827
828		#[test]
829		fn test_desc() {
830			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Desc]).unwrap();
831			let mut key = layout.allocate_key();
832
833			layout.set_u64(&mut key, 0, u64::MAX);
834
835			let offset = layout.fields[0].offset;
836			assert_eq!(&key[offset..offset + 8], &[0x00; 8]);
837		}
838	}
839
840	mod f32 {
841		use reifydb_value::value::value_type::ValueType;
842
843		use crate::{sort::SortDirection, value::index::shape::IndexShape};
844
845		#[test]
846		fn test_asc() {
847			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Asc]).unwrap();
848			let mut key_neg = layout.allocate_key();
849			let mut key_zero = layout.allocate_key();
850			let mut key_pos = layout.allocate_key();
851
852			layout.set_f32(&mut key_neg, 0, -1.0f32);
853			layout.set_f32(&mut key_zero, 0, 0.0f32);
854			layout.set_f32(&mut key_pos, 0, 1.0f32);
855
856			let offset = layout.fields[0].offset;
857
858			// -1.0f32: 0xBF800000 -> invert all: 0x407FFFFF
859			assert_eq!(&key_neg[offset..offset + 4], &[0x40, 0x7F, 0xFF, 0xFF]);
860			// 0.0f32: 0x00000000 -> flip sign: 0x80000000
861			assert_eq!(&key_zero[offset..offset + 4], &[0x80, 0x00, 0x00, 0x00]);
862			// 1.0f32: 0x3F800000 -> flip sign: 0xBF800000
863			assert_eq!(&key_pos[offset..offset + 4], &[0xBF, 0x80, 0x00, 0x00]);
864
865			assert!(key_neg.as_slice() < key_zero.as_slice());
866			assert!(key_zero.as_slice() < key_pos.as_slice());
867		}
868
869		#[test]
870		fn test_desc() {
871			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Desc]).unwrap();
872			let mut key_neg = layout.allocate_key();
873			let mut key_pos = layout.allocate_key();
874
875			layout.set_f32(&mut key_neg, 0, -1.0f32);
876			layout.set_f32(&mut key_pos, 0, 1.0f32);
877
878			let offset = layout.fields[0].offset;
879
880			// -1.0f32: ASC encoding then invert for DESC
881			assert_eq!(&key_neg[offset..offset + 4], &[0xBF, 0x80, 0x00, 0x00]);
882			// 1.0f32: ASC encoding then invert for DESC
883			assert_eq!(&key_pos[offset..offset + 4], &[0x40, 0x7F, 0xFF, 0xFF]);
884
885			assert!(key_neg.as_slice() > key_pos.as_slice());
886		}
887	}
888
889	mod f64 {
890		use std::f64::consts::PI;
891
892		use reifydb_value::value::value_type::ValueType;
893
894		use crate::{sort::SortDirection, value::index::shape::IndexShape};
895
896		#[test]
897		fn test_asc() {
898			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Asc]).unwrap();
899			let mut key = layout.allocate_key();
900
901			layout.set_f64(&mut key, 0, PI);
902
903			let offset = layout.fields[0].offset;
904			// PI in IEEE 754: 0x400921FB54442D18 -> flip sign bit
905			assert_eq!(&key[offset..offset + 8], &[0xC0, 0x09, 0x21, 0xFB, 0x54, 0x44, 0x2D, 0x18]);
906		}
907
908		#[test]
909		fn test_desc() {
910			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Desc]).unwrap();
911			let mut key = layout.allocate_key();
912
913			layout.set_f64(&mut key, 0, PI);
914
915			let offset = layout.fields[0].offset;
916			// PI: ASC encoding then invert for DESC
917			assert_eq!(&key[offset..offset + 8], &[0x3F, 0xF6, 0xDE, 0x04, 0xAB, 0xBB, 0xD2, 0xE7]);
918		}
919	}
920
921	mod row_number {
922		use reifydb_value::value::value_type::ValueType;
923
924		use crate::{sort::SortDirection, value::index::shape::IndexShape};
925
926		#[test]
927		fn test_asc() {
928			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Asc]).unwrap();
929			let mut key = layout.allocate_key();
930
931			layout.set_row_number(&mut key, 0, 0x123456789ABCDEFu64);
932
933			let offset = layout.fields[0].offset;
934			assert_eq!(&key[offset..offset + 8], &[0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF]);
935		}
936
937		#[test]
938		fn test_desc() {
939			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Desc]).unwrap();
940			let mut key = layout.allocate_key();
941
942			layout.set_row_number(&mut key, 0, 0x123456789ABCDEFu64);
943
944			let offset = layout.fields[0].offset;
945			// Inverted for DESC
946			assert_eq!(&key[offset..offset + 8], &[0xFE, 0xDC, 0xBA, 0x98, 0x76, 0x54, 0x32, 0x10]);
947		}
948	}
949
950	mod date {
951		use reifydb_value::value::{date::Date, value_type::ValueType};
952
953		use crate::{sort::SortDirection, value::index::shape::IndexShape};
954
955		#[test]
956		fn test_asc() {
957			let layout = IndexShape::new(&[ValueType::Date], &[SortDirection::Asc]).unwrap();
958			let mut key = layout.allocate_key();
959
960			let date = Date::new(2025, 1, 1).unwrap();
961			layout.set_date(&mut key, 0, date);
962
963			let offset = layout.fields[0].offset;
964			// Date is stored as i32 days since epoch with the sign bit flipped.
965			let bytes = &key[offset..offset + 4];
966
967			let mut expected = date.to_days_since_epoch().to_be_bytes();
968			expected[0] ^= 0x80;
969			assert_eq!(bytes, expected);
970		}
971
972		#[test]
973		fn test_desc() {
974			let layout = IndexShape::new(&[ValueType::Date], &[SortDirection::Desc]).unwrap();
975			let mut key = layout.allocate_key();
976
977			let date = Date::new(2025, 1, 1).unwrap();
978			layout.set_date(&mut key, 0, date);
979
980			let offset = layout.fields[0].offset;
981			let bytes = &key[offset..offset + 4];
982
983			// Date with sign bit flipped then all inverted for DESC
984			let mut expected = date.to_days_since_epoch().to_be_bytes();
985			expected[0] ^= 0x80;
986			for b in expected.iter_mut() {
987				*b = !*b;
988			}
989			assert_eq!(bytes, expected);
990		}
991	}
992
993	mod composite {
994		use reifydb_value::value::value_type::ValueType;
995
996		use crate::{sort::SortDirection, value::index::shape::IndexShape};
997
998		#[test]
999		fn test_mixed_directions() {
1000			let layout = IndexShape::new(
1001				&[ValueType::Int4, ValueType::Uint8],
1002				&[SortDirection::Desc, SortDirection::Asc],
1003			)
1004			.unwrap();
1005
1006			let mut key = layout.allocate_key();
1007			layout.set_i32(&mut key, 0, 100i32);
1008			layout.set_u64(&mut key, 1, 200u64);
1009
1010			let offset1 = layout.fields[0].offset;
1011			let mut expected_i32 = 100i32.to_be_bytes();
1012			expected_i32[0] ^= 0x80;
1013			for b in expected_i32.iter_mut() {
1014				*b = !*b;
1015			}
1016			assert_eq!(&key[offset1..offset1 + 4], expected_i32);
1017
1018			let offset2 = layout.fields[1].offset;
1019			let expected_u64 = 200u64.to_be_bytes();
1020			assert_eq!(&key[offset2..offset2 + 8], expected_u64);
1021		}
1022	}
1023
1024	mod uuid4 {
1025		use reifydb_value::value::{uuid::Uuid4, value_type::ValueType};
1026
1027		use crate::{sort::SortDirection, value::index::shape::IndexShape};
1028
1029		#[test]
1030		fn test_asc() {
1031			let layout = IndexShape::new(&[ValueType::Uuid4], &[SortDirection::Asc]).unwrap();
1032			let mut key1 = layout.allocate_key();
1033			let mut key2 = layout.allocate_key();
1034
1035			let uuid1 = Uuid4::generate();
1036			let uuid2 = Uuid4::generate();
1037
1038			layout.set_uuid4(&mut key1, 0, uuid1.clone());
1039			layout.set_uuid4(&mut key2, 0, uuid2.clone());
1040
1041			assert!(key1.is_defined(0));
1042			assert!(key2.is_defined(0));
1043
1044			let offset = layout.fields[0].offset;
1045			let uuid1_bytes: Vec<u8> = uuid1.as_bytes().to_vec();
1046			let uuid2_bytes: Vec<u8> = uuid2.as_bytes().to_vec();
1047
1048			assert_eq!(&key1[offset..offset + 16], &uuid1_bytes[..]);
1049			assert_eq!(&key2[offset..offset + 16], &uuid2_bytes[..]);
1050		}
1051
1052		#[test]
1053		fn test_desc() {
1054			let layout = IndexShape::new(&[ValueType::Uuid4], &[SortDirection::Desc]).unwrap();
1055			let mut key = layout.allocate_key();
1056
1057			let uuid = Uuid4::generate();
1058			layout.set_uuid4(&mut key, 0, uuid.clone());
1059
1060			let offset = layout.fields[0].offset;
1061			let mut expected_bytes = uuid.as_bytes().to_vec();
1062			for b in expected_bytes.iter_mut() {
1063				*b = !*b;
1064			}
1065
1066			assert_eq!(&key[offset..offset + 16], &expected_bytes[..]);
1067		}
1068	}
1069
1070	mod uuid7 {
1071		use reifydb_runtime::context::{
1072			clock::{Clock, MockClock},
1073			rng::Rng,
1074		};
1075		use reifydb_value::value::{uuid::Uuid7, value_type::ValueType};
1076
1077		use crate::{sort::SortDirection, value::index::shape::IndexShape};
1078
1079		fn test_clock_and_rng() -> (MockClock, Clock, Rng) {
1080			let mock = MockClock::from_millis(1000);
1081			let clock = Clock::Mock(mock.clone());
1082			let rng = Rng::seeded(42);
1083			(mock, clock, rng)
1084		}
1085
1086		#[test]
1087		fn test_asc() {
1088			let (mock, clock, rng) = test_clock_and_rng();
1089			let layout = IndexShape::new(&[ValueType::Uuid7], &[SortDirection::Asc]).unwrap();
1090			let mut key1 = layout.allocate_key();
1091			let mut key2 = layout.allocate_key();
1092
1093			let uuid1 = Uuid7::generate(&clock, &rng);
1094			// Advance the clock so the two ids get different timestamps.
1095			mock.advance_millis(10);
1096			let uuid2 = Uuid7::generate(&clock, &rng);
1097
1098			layout.set_uuid7(&mut key1, 0, uuid1.clone());
1099			layout.set_uuid7(&mut key2, 0, uuid2.clone());
1100
1101			assert!(key1.is_defined(0));
1102			assert!(key2.is_defined(0));
1103
1104			let offset = layout.fields[0].offset;
1105			let uuid1_bytes: Vec<u8> = uuid1.as_bytes().to_vec();
1106			let uuid2_bytes: Vec<u8> = uuid2.as_bytes().to_vec();
1107
1108			assert_eq!(&key1[offset..offset + 16], &uuid1_bytes[..]);
1109			assert_eq!(&key2[offset..offset + 16], &uuid2_bytes[..]);
1110
1111			// Uuid7 leads with a timestamp, so the later one must sort greater.
1112			assert!(key1.as_slice() < key2.as_slice());
1113		}
1114
1115		#[test]
1116		fn test_desc() {
1117			let (mock, clock, rng) = test_clock_and_rng();
1118			let layout = IndexShape::new(&[ValueType::Uuid7], &[SortDirection::Desc]).unwrap();
1119			let mut key1 = layout.allocate_key();
1120			let mut key2 = layout.allocate_key();
1121
1122			let uuid1 = Uuid7::generate(&clock, &rng);
1123			// Advance the clock so the two ids get different timestamps.
1124			mock.advance_millis(10);
1125			let uuid2 = Uuid7::generate(&clock, &rng);
1126
1127			layout.set_uuid7(&mut key1, 0, uuid1.clone());
1128			layout.set_uuid7(&mut key2, 0, uuid2.clone());
1129
1130			let offset = layout.fields[0].offset;
1131			let mut expected_bytes1 = uuid1.as_bytes().to_vec();
1132			let mut expected_bytes2 = uuid2.as_bytes().to_vec();
1133			for b in expected_bytes1.iter_mut() {
1134				*b = !*b;
1135			}
1136			for b in expected_bytes2.iter_mut() {
1137				*b = !*b;
1138			}
1139
1140			assert_eq!(&key1[offset..offset + 16], &expected_bytes1[..]);
1141			assert_eq!(&key2[offset..offset + 16], &expected_bytes2[..]);
1142
1143			assert!(key1.as_slice() > key2.as_slice());
1144		}
1145	}
1146
1147	mod identity_id {
1148		use reifydb_runtime::context::{
1149			clock::{Clock, MockClock},
1150			rng::Rng,
1151		};
1152		use reifydb_value::value::{identity::IdentityId, uuid::Uuid7, value_type::ValueType};
1153
1154		use crate::{sort::SortDirection, value::index::shape::IndexShape};
1155
1156		fn test_clock_and_rng() -> (MockClock, Clock, Rng) {
1157			let mock = MockClock::from_millis(1000);
1158			let clock = Clock::Mock(mock.clone());
1159			let rng = Rng::seeded(42);
1160			(mock, clock, rng)
1161		}
1162
1163		#[test]
1164		fn test_asc() {
1165			let (mock, clock, rng) = test_clock_and_rng();
1166			let layout = IndexShape::new(&[ValueType::IdentityId], &[SortDirection::Asc]).unwrap();
1167			let mut key1 = layout.allocate_key();
1168			let mut key2 = layout.allocate_key();
1169
1170			let id1 = IdentityId::generate(&clock, &rng);
1171			// Advance the clock so the two ids get different timestamps.
1172			mock.advance_millis(10);
1173			let id2 = IdentityId::generate(&clock, &rng);
1174
1175			layout.set_identity_id(&mut key1, 0, id1.clone());
1176			layout.set_identity_id(&mut key2, 0, id2.clone());
1177
1178			assert!(key1.is_defined(0));
1179			assert!(key2.is_defined(0));
1180
1181			let offset = layout.fields[0].offset;
1182			let uuid7_1: Uuid7 = id1.into();
1183			let uuid7_2: Uuid7 = id2.into();
1184			let id1_bytes: Vec<u8> = uuid7_1.as_bytes().to_vec();
1185			let id2_bytes: Vec<u8> = uuid7_2.as_bytes().to_vec();
1186
1187			assert_eq!(&key1[offset..offset + 16], &id1_bytes[..]);
1188			assert_eq!(&key2[offset..offset + 16], &id2_bytes[..]);
1189
1190			// IdentityId wraps Uuid7, which leads with a timestamp, so the later one sorts greater.
1191			assert!(key1.as_slice() < key2.as_slice());
1192		}
1193
1194		#[test]
1195		fn test_desc() {
1196			let (mock, clock, rng) = test_clock_and_rng();
1197			let layout = IndexShape::new(&[ValueType::IdentityId], &[SortDirection::Desc]).unwrap();
1198			let mut key1 = layout.allocate_key();
1199			let mut key2 = layout.allocate_key();
1200
1201			let id1 = IdentityId::generate(&clock, &rng);
1202			// Advance the clock so the two ids get different timestamps.
1203			mock.advance_millis(10);
1204			let id2 = IdentityId::generate(&clock, &rng);
1205
1206			layout.set_identity_id(&mut key1, 0, id1.clone());
1207			layout.set_identity_id(&mut key2, 0, id2.clone());
1208
1209			let offset = layout.fields[0].offset;
1210			let uuid7_1: Uuid7 = id1.into();
1211			let uuid7_2: Uuid7 = id2.into();
1212			let mut expected_bytes1 = uuid7_1.as_bytes().to_vec();
1213			let mut expected_bytes2 = uuid7_2.as_bytes().to_vec();
1214			for b in expected_bytes1.iter_mut() {
1215				*b = !*b;
1216			}
1217			for b in expected_bytes2.iter_mut() {
1218				*b = !*b;
1219			}
1220
1221			assert_eq!(&key1[offset..offset + 16], &expected_bytes1[..]);
1222			assert_eq!(&key2[offset..offset + 16], &expected_bytes2[..]);
1223
1224			assert!(key1.as_slice() > key2.as_slice());
1225		}
1226	}
1227
1228	mod undefined {
1229		use reifydb_value::value::value_type::ValueType;
1230
1231		use crate::{sort::SortDirection, value::index::shape::IndexShape};
1232
1233		#[test]
1234		fn test_undefined() {
1235			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Asc]).unwrap();
1236			let mut key = layout.allocate_key();
1237
1238			layout.set_i32(&mut key, 0, 42i32);
1239			assert!(key.is_defined(0));
1240
1241			layout.set_none(&mut key, 0);
1242			assert!(!key.is_defined(0));
1243
1244			let offset = layout.fields[0].offset;
1245			assert_eq!(&key[offset..offset + 4], &[0, 0, 0, 0]);
1246		}
1247	}
1248}