reifydb_core/value/index/

set.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use std::{f64, ptr};

use reifydb_value::{
	reifydb_assertions,
	value::{
		date::Date,
		datetime::DateTime,
		duration::Duration,
		identity::IdentityId,
		time::Time,
		uuid::{Uuid4, Uuid7},
	},
};
use uuid::Uuid;

use crate::{
	sort::SortDirection,
	value::index::{encoded::EncodedIndexKey, shape::IndexShape},
};

impl IndexShape {
	pub fn set_bool(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<bool>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Boolean);
		}
		key.set_valid(index, true);

		let byte_value = match field.direction {
			SortDirection::Asc => {
				if value.into() {
					1u8
				} else {
					0u8
				}
			}
			SortDirection::Desc => {
				if value.into() {
					0u8
				} else {
					1u8
				}
			}
		};

		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), byte_value) }
	}

	pub fn set_f32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<f32>) {
		let field = &self.fields[index];
		let v = value.into();
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Float4);
			assert_eq!(
				v.to_bits(),
				reifydb_value::value::ordered_f32::OrderedF32::canonical(v).to_bits(),
				"set_f32: field {index} float {v:?} is not canonical"
			);
		}
		key.set_valid(index, true);

		let mut bytes = v.to_bits().to_be_bytes();

		if v.is_sign_negative() {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		} else {
			bytes[0] ^= 0x80;
		}

		if field.direction == SortDirection::Desc {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
		}
	}

	pub fn set_f64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<f64>) {
		let field = &self.fields[index];
		let v = value.into();
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Float8);
			assert_eq!(
				v.to_bits(),
				reifydb_value::value::ordered_f64::OrderedF64::canonical(v).to_bits(),
				"set_f64: field {index} float {v:?} is not canonical"
			);
		}
		key.set_valid(index, true);

		let mut bytes = v.to_bits().to_be_bytes();

		if v.is_sign_negative() {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		} else {
			bytes[0] ^= 0x80;
		}

		if field.direction == SortDirection::Desc {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_i8(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i8>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Int1);
		}
		key.set_valid(index, true);

		let mut bytes = value.into().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), bytes[0]) }
	}

	pub fn set_i16(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i16>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Int2);
		}
		key.set_valid(index, true);

		let mut bytes = value.into().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 2 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 2);
		}
	}

	pub fn set_i32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i32>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Int4);
		}
		key.set_valid(index, true);

		let mut bytes = value.into().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
		}
	}

	pub fn set_i64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i64>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Int8);
		}
		key.set_valid(index, true);

		let mut bytes = value.into().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_i128(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<i128>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Int16);
		}
		key.set_valid(index, true);

		let mut bytes = value.into().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
		}
	}

	pub fn set_u8(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u8>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint1);
		}
		key.set_valid(index, true);

		let byte = match field.direction {
			SortDirection::Asc => value.into(),
			SortDirection::Desc => !value.into(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset inside total_size, so the 1-byte write is in bounds.
		unsafe { ptr::write_unaligned(key.make_mut().as_mut_ptr().add(field.offset), byte) }
	}

	pub fn set_u16(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u16>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint2);
		}
		key.set_valid(index, true);

		let bytes = match field.direction {
			SortDirection::Asc => value.into().to_be_bytes(),
			SortDirection::Desc => (!value.into()).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 2 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 2);
		}
	}

	pub fn set_u32(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u32>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint4);
		}
		key.set_valid(index, true);

		let bytes = match field.direction {
			SortDirection::Asc => value.into().to_be_bytes(),
			SortDirection::Desc => (!value.into()).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
		}
	}

	pub fn set_u64(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u64>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint8);
		}
		key.set_valid(index, true);

		let bytes = match field.direction {
			SortDirection::Asc => value.into().to_be_bytes(),
			SortDirection::Desc => (!value.into()).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_u128(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u128>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint16);
		}
		key.set_valid(index, true);

		let bytes = match field.direction {
			SortDirection::Asc => value.into().to_be_bytes(),
			SortDirection::Desc => (!value.into()).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
		}
	}

	pub fn set_row_number(&self, key: &mut EncodedIndexKey, index: usize, value: impl Into<u64>) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uint8);
		}
		key.set_valid(index, true);

		let bytes = match field.direction {
			SortDirection::Asc => value.into().to_be_bytes(),
			SortDirection::Desc => (!value.into()).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_date(&self, key: &mut EncodedIndexKey, index: usize, value: Date) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Date);
		}
		key.set_valid(index, true);

		let days = value.to_days_since_epoch();
		let mut bytes = days.to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				bytes[0] ^= 0x80;
				for b in bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 4 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 4);
		}
	}

	pub fn set_datetime(&self, key: &mut EncodedIndexKey, index: usize, value: DateTime) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::DateTime);
		}
		key.set_valid(index, true);

		let order = value.to_order();
		let bytes = match field.direction {
			SortDirection::Asc => order.to_be_bytes(),
			SortDirection::Desc => (!order).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_time(&self, key: &mut EncodedIndexKey, index: usize, value: Time) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Time);
		}
		key.set_valid(index, true);

		let nanos = value.to_nanos_since_midnight();
		let bytes = match field.direction {
			SortDirection::Asc => nanos.to_be_bytes(),
			SortDirection::Desc => (!nanos).to_be_bytes(),
		};

		// SAFETY: IndexShapeInner::new puts field.offset + 8 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 8);
		}
	}

	pub fn set_duration(&self, key: &mut EncodedIndexKey, index: usize, value: Duration) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Duration);
		}
		key.set_valid(index, true);

		let mut months_bytes = value.get_months().to_be_bytes();
		let mut days_bytes = value.get_days().to_be_bytes();
		let mut nanos_bytes = value.get_nanos().to_be_bytes();

		match field.direction {
			SortDirection::Asc => {
				months_bytes[0] ^= 0x80;
				days_bytes[0] ^= 0x80;
				nanos_bytes[0] ^= 0x80;
			}
			SortDirection::Desc => {
				months_bytes[0] ^= 0x80;
				days_bytes[0] ^= 0x80;
				nanos_bytes[0] ^= 0x80;
				for b in months_bytes.iter_mut() {
					*b = !*b;
				}
				for b in days_bytes.iter_mut() {
					*b = !*b;
				}
				for b in nanos_bytes.iter_mut() {
					*b = !*b;
				}
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so all three copies fit.
		unsafe {
			ptr::copy_nonoverlapping(
				months_bytes.as_ptr(),
				key.make_mut().as_mut_ptr().add(field.offset),
				4,
			);
			ptr::copy_nonoverlapping(
				days_bytes.as_ptr(),
				key.make_mut().as_mut_ptr().add(field.offset + 4),
				4,
			);
			ptr::copy_nonoverlapping(
				nanos_bytes.as_ptr(),
				key.make_mut().as_mut_ptr().add(field.offset + 8),
				8,
			);
		}
	}

	pub fn set_uuid4(&self, key: &mut EncodedIndexKey, index: usize, value: Uuid4) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uuid4);
		}
		key.set_valid(index, true);

		let uuid: Uuid = value.into();
		let uuid_bytes = uuid.as_bytes();
		let mut bytes = [0u8; 16];
		bytes.copy_from_slice(uuid_bytes);

		if field.direction == SortDirection::Desc {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
		}
	}

	pub fn set_uuid7(&self, key: &mut EncodedIndexKey, index: usize, value: Uuid7) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::Uuid7);
		}
		key.set_valid(index, true);

		let uuid: Uuid = value.into();
		let uuid_bytes = uuid.as_bytes();
		let mut bytes = [0u8; 16];
		bytes.copy_from_slice(uuid_bytes);

		if field.direction == SortDirection::Desc {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
		}
	}

	pub fn set_identity_id(&self, key: &mut EncodedIndexKey, index: usize, value: IdentityId) {
		let field = &self.fields[index];
		reifydb_assertions! {
			assert_eq!(field.value, reifydb_value::value::value_type::ValueType::IdentityId);
		}
		key.set_valid(index, true);

		let uuid: Uuid = value.0.into();
		let uuid_bytes = uuid.as_bytes();
		let mut bytes = [0u8; 16];
		bytes.copy_from_slice(uuid_bytes);

		if field.direction == SortDirection::Desc {
			for b in bytes.iter_mut() {
				*b = !*b;
			}
		}

		// SAFETY: IndexShapeInner::new puts field.offset + 16 inside total_size, so the copy stays in bounds.
		unsafe {
			ptr::copy_nonoverlapping(bytes.as_ptr(), key.make_mut().as_mut_ptr().add(field.offset), 16);
		}
	}

	pub fn set_none(&self, key: &mut EncodedIndexKey, index: usize) {
		let field = &self.fields[index];
		key.set_valid(index, false);

		let buf = key.make_mut();
		let start = field.offset;
		let end = start + field.size;
		buf[start..end].fill(0);
	}
}

#[cfg(test)]
pub mod tests {
	use crate::{sort::SortDirection, value::index::shape::IndexShape};

	mod bool {
		use reifydb_value::value::value_type::ValueType;

		use super::*;

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Boolean], &[SortDirection::Asc]).unwrap();
			let mut key_false = layout.allocate_key();
			let mut key_true = layout.allocate_key();

			layout.set_bool(&mut key_false, 0, false);
			layout.set_bool(&mut key_true, 0, true);

			assert_eq!(key_false[0] & 0x01, 0x01);
			assert_eq!(key_true[0] & 0x01, 0x01);

			let offset = layout.fields[0].offset;
			assert_eq!(key_false[offset], 0);
			assert_eq!(key_true[offset], 1);

			assert!(key_false.as_slice() < key_true.as_slice());
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Boolean], &[SortDirection::Desc]).unwrap();
			let mut key_false = layout.allocate_key();
			let mut key_true = layout.allocate_key();

			layout.set_bool(&mut key_false, 0, false);
			layout.set_bool(&mut key_true, 0, true);

			let offset = layout.fields[0].offset;
			assert_eq!(key_false[offset], 1); // false becomes 1 in DESC
			assert_eq!(key_true[offset], 0); // true becomes 0 in DESC

			assert!(key_false.as_slice() > key_true.as_slice());
		}
	}

	mod i8 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Int1], &[SortDirection::Asc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_zero = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_i8(&mut key_neg, 0, -128i8);
			layout.set_i8(&mut key_zero, 0, 0i8);
			layout.set_i8(&mut key_pos, 0, 127i8);

			let offset = layout.fields[0].offset;
			// -128 with sign bit flipped: 0x80 -> 0x00
			assert_eq!(key_neg[offset], 0x00);
			// 0 with sign bit flipped: 0x00 -> 0x80
			assert_eq!(key_zero[offset], 0x80);
			// 127 with sign bit flipped: 0x7F -> 0xFF
			assert_eq!(key_pos[offset], 0xFF);

			assert!(key_neg.as_slice() < key_zero.as_slice());
			assert!(key_zero.as_slice() < key_pos.as_slice());
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Int1], &[SortDirection::Desc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_zero = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_i8(&mut key_neg, 0, -128i8);
			layout.set_i8(&mut key_zero, 0, 0i8);
			layout.set_i8(&mut key_pos, 0, 127i8);

			let offset = layout.fields[0].offset;
			// -128: 0x80 -> flip sign: 0x00 -> invert: 0xFF
			assert_eq!(key_neg[offset], 0xFF);
			// 0: 0x00 -> flip sign: 0x80 -> invert: 0x7F
			assert_eq!(key_zero[offset], 0x7F);
			// 127: 0x7F -> flip sign: 0xFF -> invert: 0x00
			assert_eq!(key_pos[offset], 0x00);

			assert!(key_neg.as_slice() > key_zero.as_slice());
			assert!(key_zero.as_slice() > key_pos.as_slice());
		}
	}

	mod i32 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Asc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_zero = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_i32(&mut key_neg, 0, i32::MIN);
			layout.set_i32(&mut key_zero, 0, 0i32);
			layout.set_i32(&mut key_pos, 0, i32::MAX);

			let offset = layout.fields[0].offset;
			// i32::MIN in big-endian with sign bit flipped
			assert_eq!(&key_neg[offset..offset + 4], &[0x00, 0x00, 0x00, 0x00]);
			// 0 with sign bit flipped
			assert_eq!(&key_zero[offset..offset + 4], &[0x80, 0x00, 0x00, 0x00]);
			// i32::MAX with sign bit flipped
			assert_eq!(&key_pos[offset..offset + 4], &[0xFF, 0xFF, 0xFF, 0xFF]);

			assert!(key_neg.as_slice() < key_zero.as_slice());
			assert!(key_zero.as_slice() < key_pos.as_slice());
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Desc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_zero = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_i32(&mut key_neg, 0, i32::MIN);
			layout.set_i32(&mut key_zero, 0, 0i32);
			layout.set_i32(&mut key_pos, 0, i32::MAX);

			let offset = layout.fields[0].offset;
			// i32::MIN: flip sign then invert all
			assert_eq!(&key_neg[offset..offset + 4], &[0xFF, 0xFF, 0xFF, 0xFF]);
			// 0: flip sign then invert all
			assert_eq!(&key_zero[offset..offset + 4], &[0x7F, 0xFF, 0xFF, 0xFF]);
			// i32::MAX: flip sign then invert all
			assert_eq!(&key_pos[offset..offset + 4], &[0x00, 0x00, 0x00, 0x00]);

			assert!(key_neg.as_slice() > key_zero.as_slice());
			assert!(key_zero.as_slice() > key_pos.as_slice());
		}
	}

	mod i64 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Int8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_i64(&mut key, 0, -1i64);

			let offset = layout.fields[0].offset;
			// -1 is all 1s in two's complement; the flipped sign bit gives 0x7F...
			assert_eq!(&key[offset..offset + 8], &[0x7F, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Int8], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_i64(&mut key, 0, -1i64);

			let offset = layout.fields[0].offset;
			// -1: flip sign then invert all
			assert_eq!(&key[offset..offset + 8], &[0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]);
		}
	}

	mod u8 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Uint1], &[SortDirection::Asc]).unwrap();
			let mut key_min = layout.allocate_key();
			let mut key_mid = layout.allocate_key();
			let mut key_max = layout.allocate_key();

			layout.set_u8(&mut key_min, 0, 0u8);
			layout.set_u8(&mut key_mid, 0, 128u8);
			layout.set_u8(&mut key_max, 0, 255u8);

			let offset = layout.fields[0].offset;
			assert_eq!(key_min[offset], 0x00);
			assert_eq!(key_mid[offset], 0x80);
			assert_eq!(key_max[offset], 0xFF);

			assert!(key_min.as_slice() < key_mid.as_slice());
			assert!(key_mid.as_slice() < key_max.as_slice());
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Uint1], &[SortDirection::Desc]).unwrap();
			let mut key_min = layout.allocate_key();
			let mut key_mid = layout.allocate_key();
			let mut key_max = layout.allocate_key();

			layout.set_u8(&mut key_min, 0, 0u8);
			layout.set_u8(&mut key_mid, 0, 128u8);
			layout.set_u8(&mut key_max, 0, 255u8);

			let offset = layout.fields[0].offset;
			// Inverted for DESC
			assert_eq!(key_min[offset], 0xFF);
			assert_eq!(key_mid[offset], 0x7F);
			assert_eq!(key_max[offset], 0x00);

			assert!(key_min.as_slice() > key_mid.as_slice());
			assert!(key_mid.as_slice() > key_max.as_slice());
		}
	}

	mod u32 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Uint4], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_u32(&mut key, 0, 0x12345678u32);

			let offset = layout.fields[0].offset;
			// Big-endian representation
			assert_eq!(&key[offset..offset + 4], &[0x12, 0x34, 0x56, 0x78]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Uint4], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_u32(&mut key, 0, 0x12345678u32);

			let offset = layout.fields[0].offset;
			// Inverted for DESC
			assert_eq!(&key[offset..offset + 4], &[0xED, 0xCB, 0xA9, 0x87]);
		}
	}

	mod u64 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_u64(&mut key, 0, u64::MAX);

			let offset = layout.fields[0].offset;
			assert_eq!(&key[offset..offset + 8], &[0xFF; 8]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_u64(&mut key, 0, u64::MAX);

			let offset = layout.fields[0].offset;
			assert_eq!(&key[offset..offset + 8], &[0x00; 8]);
		}
	}

	mod f32 {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Asc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_zero = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_f32(&mut key_neg, 0, -1.0f32);
			layout.set_f32(&mut key_zero, 0, 0.0f32);
			layout.set_f32(&mut key_pos, 0, 1.0f32);

			let offset = layout.fields[0].offset;

			// -1.0f32: 0xBF800000 -> invert all: 0x407FFFFF
			assert_eq!(&key_neg[offset..offset + 4], &[0x40, 0x7F, 0xFF, 0xFF]);
			// 0.0f32: 0x00000000 -> flip sign: 0x80000000
			assert_eq!(&key_zero[offset..offset + 4], &[0x80, 0x00, 0x00, 0x00]);
			// 1.0f32: 0x3F800000 -> flip sign: 0xBF800000
			assert_eq!(&key_pos[offset..offset + 4], &[0xBF, 0x80, 0x00, 0x00]);

			assert!(key_neg.as_slice() < key_zero.as_slice());
			assert!(key_zero.as_slice() < key_pos.as_slice());
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Desc]).unwrap();
			let mut key_neg = layout.allocate_key();
			let mut key_pos = layout.allocate_key();

			layout.set_f32(&mut key_neg, 0, -1.0f32);
			layout.set_f32(&mut key_pos, 0, 1.0f32);

			let offset = layout.fields[0].offset;

			// -1.0f32: ASC encoding then invert for DESC
			assert_eq!(&key_neg[offset..offset + 4], &[0xBF, 0x80, 0x00, 0x00]);
			// 1.0f32: ASC encoding then invert for DESC
			assert_eq!(&key_pos[offset..offset + 4], &[0x40, 0x7F, 0xFF, 0xFF]);

			assert!(key_neg.as_slice() > key_pos.as_slice());
		}

		#[test]
		#[cfg(reifydb_assertions)]
		#[should_panic(expected = "is not canonical")]
		fn test_negative_zero_panics() {
			// -0.0 and 0.0 encode to different key bytes, so a raw -0.0 must never reach a key.
			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();
			layout.set_f32(&mut key, 0, -0.0f32);
		}

		#[test]
		#[cfg(reifydb_assertions)]
		#[should_panic(expected = "is not canonical")]
		fn test_negative_nan_panics() {
			// Each NaN bit pattern encodes to its own key bytes, so only the canonical NaN may reach a key.
			let layout = IndexShape::new(&[ValueType::Float4], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();
			layout.set_f32(&mut key, 0, f32::from_bits(f32::NAN.to_bits() | 0x8000_0000));
		}
	}

	mod f64 {
		use std::f64::consts::PI;

		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_f64(&mut key, 0, PI);

			let offset = layout.fields[0].offset;
			// PI in IEEE 754: 0x400921FB54442D18 -> flip sign bit
			assert_eq!(&key[offset..offset + 8], &[0xC0, 0x09, 0x21, 0xFB, 0x54, 0x44, 0x2D, 0x18]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_f64(&mut key, 0, PI);

			let offset = layout.fields[0].offset;
			// PI: ASC encoding then invert for DESC
			assert_eq!(&key[offset..offset + 8], &[0x3F, 0xF6, 0xDE, 0x04, 0xAB, 0xBB, 0xD2, 0xE7]);
		}

		#[test]
		#[cfg(reifydb_assertions)]
		#[should_panic(expected = "is not canonical")]
		fn test_negative_zero_panics() {
			// -0.0 and 0.0 encode to different key bytes, so a raw -0.0 must never reach a key.
			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();
			layout.set_f64(&mut key, 0, -0.0f64);
		}

		#[test]
		#[cfg(reifydb_assertions)]
		#[should_panic(expected = "is not canonical")]
		fn test_negative_nan_panics() {
			// Each NaN bit pattern encodes to its own key bytes, so only the canonical NaN may reach a key.
			let layout = IndexShape::new(&[ValueType::Float8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();
			layout.set_f64(&mut key, 0, f64::from_bits(f64::NAN.to_bits() | 0x8000_0000_0000_0000));
		}
	}

	mod row_number {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_row_number(&mut key, 0, 0x123456789ABCDEFu64);

			let offset = layout.fields[0].offset;
			assert_eq!(&key[offset..offset + 8], &[0x01, 0x23, 0x45, 0x67, 0x89, 0xAB, 0xCD, 0xEF]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Uint8], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_row_number(&mut key, 0, 0x123456789ABCDEFu64);

			let offset = layout.fields[0].offset;
			// Inverted for DESC
			assert_eq!(&key[offset..offset + 8], &[0xFE, 0xDC, 0xBA, 0x98, 0x76, 0x54, 0x32, 0x10]);
		}
	}

	mod date {
		use reifydb_value::value::{date::Date, value_type::ValueType};

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Date], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			let date = Date::new(2025, 1, 1).unwrap();
			layout.set_date(&mut key, 0, date);

			let offset = layout.fields[0].offset;
			// Date is stored as i32 days since epoch with the sign bit flipped.
			let bytes = &key[offset..offset + 4];

			let mut expected = date.to_days_since_epoch().to_be_bytes();
			expected[0] ^= 0x80;
			assert_eq!(bytes, expected);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Date], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			let date = Date::new(2025, 1, 1).unwrap();
			layout.set_date(&mut key, 0, date);

			let offset = layout.fields[0].offset;
			let bytes = &key[offset..offset + 4];

			// Date with sign bit flipped then all inverted for DESC
			let mut expected = date.to_days_since_epoch().to_be_bytes();
			expected[0] ^= 0x80;
			for b in expected.iter_mut() {
				*b = !*b;
			}
			assert_eq!(bytes, expected);
		}
	}

	mod composite {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_mixed_directions() {
			let layout = IndexShape::new(
				&[ValueType::Int4, ValueType::Uint8],
				&[SortDirection::Desc, SortDirection::Asc],
			)
			.unwrap();

			let mut key = layout.allocate_key();
			layout.set_i32(&mut key, 0, 100i32);
			layout.set_u64(&mut key, 1, 200u64);

			let offset1 = layout.fields[0].offset;
			let mut expected_i32 = 100i32.to_be_bytes();
			expected_i32[0] ^= 0x80;
			for b in expected_i32.iter_mut() {
				*b = !*b;
			}
			assert_eq!(&key[offset1..offset1 + 4], expected_i32);

			let offset2 = layout.fields[1].offset;
			let expected_u64 = 200u64.to_be_bytes();
			assert_eq!(&key[offset2..offset2 + 8], expected_u64);
		}
	}

	mod uuid4 {
		use reifydb_value::value::{uuid::Uuid4, value_type::ValueType};

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_asc() {
			let layout = IndexShape::new(&[ValueType::Uuid4], &[SortDirection::Asc]).unwrap();
			let mut key1 = layout.allocate_key();
			let mut key2 = layout.allocate_key();

			let uuid1 = Uuid4::generate();
			let uuid2 = Uuid4::generate();

			layout.set_uuid4(&mut key1, 0, uuid1);
			layout.set_uuid4(&mut key2, 0, uuid2);

			assert!(key1.is_defined(0));
			assert!(key2.is_defined(0));

			let offset = layout.fields[0].offset;
			let uuid1_bytes: Vec<u8> = uuid1.as_bytes().to_vec();
			let uuid2_bytes: Vec<u8> = uuid2.as_bytes().to_vec();

			assert_eq!(&key1[offset..offset + 16], &uuid1_bytes[..]);
			assert_eq!(&key2[offset..offset + 16], &uuid2_bytes[..]);
		}

		#[test]
		fn test_desc() {
			let layout = IndexShape::new(&[ValueType::Uuid4], &[SortDirection::Desc]).unwrap();
			let mut key = layout.allocate_key();

			let uuid = Uuid4::generate();
			layout.set_uuid4(&mut key, 0, uuid);

			let offset = layout.fields[0].offset;
			let mut expected_bytes = uuid.as_bytes().to_vec();
			for b in expected_bytes.iter_mut() {
				*b = !*b;
			}

			assert_eq!(&key[offset..offset + 16], &expected_bytes[..]);
		}
	}

	mod uuid7 {
		use reifydb_runtime::context::{
			clock::{Clock, MockClock},
			rng::Rng,
		};
		use reifydb_value::value::{uuid::Uuid7, value_type::ValueType};

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		fn test_clock_and_rng() -> (MockClock, Clock, Rng) {
			let mock = MockClock::from_millis(1000);
			let clock = Clock::Mock(mock.clone());
			let rng = Rng::seeded(42);
			(mock, clock, rng)
		}

		#[test]
		fn test_asc() {
			let (mock, clock, rng) = test_clock_and_rng();
			let layout = IndexShape::new(&[ValueType::Uuid7], &[SortDirection::Asc]).unwrap();
			let mut key1 = layout.allocate_key();
			let mut key2 = layout.allocate_key();

			let uuid1 = Uuid7::generate(&clock, &rng);
			// Advance the clock so the two ids get different timestamps.
			mock.advance_millis(10);
			let uuid2 = Uuid7::generate(&clock, &rng);

			layout.set_uuid7(&mut key1, 0, uuid1);
			layout.set_uuid7(&mut key2, 0, uuid2);

			assert!(key1.is_defined(0));
			assert!(key2.is_defined(0));

			let offset = layout.fields[0].offset;
			let uuid1_bytes: Vec<u8> = uuid1.as_bytes().to_vec();
			let uuid2_bytes: Vec<u8> = uuid2.as_bytes().to_vec();

			assert_eq!(&key1[offset..offset + 16], &uuid1_bytes[..]);
			assert_eq!(&key2[offset..offset + 16], &uuid2_bytes[..]);

			// Uuid7 leads with a timestamp, so the later one must sort greater.
			assert!(key1.as_slice() < key2.as_slice());
		}

		#[test]
		fn test_desc() {
			let (mock, clock, rng) = test_clock_and_rng();
			let layout = IndexShape::new(&[ValueType::Uuid7], &[SortDirection::Desc]).unwrap();
			let mut key1 = layout.allocate_key();
			let mut key2 = layout.allocate_key();

			let uuid1 = Uuid7::generate(&clock, &rng);
			// Advance the clock so the two ids get different timestamps.
			mock.advance_millis(10);
			let uuid2 = Uuid7::generate(&clock, &rng);

			layout.set_uuid7(&mut key1, 0, uuid1);
			layout.set_uuid7(&mut key2, 0, uuid2);

			let offset = layout.fields[0].offset;
			let mut expected_bytes1 = uuid1.as_bytes().to_vec();
			let mut expected_bytes2 = uuid2.as_bytes().to_vec();
			for b in expected_bytes1.iter_mut() {
				*b = !*b;
			}
			for b in expected_bytes2.iter_mut() {
				*b = !*b;
			}

			assert_eq!(&key1[offset..offset + 16], &expected_bytes1[..]);
			assert_eq!(&key2[offset..offset + 16], &expected_bytes2[..]);

			assert!(key1.as_slice() > key2.as_slice());
		}
	}

	mod identity_id {
		use reifydb_runtime::context::{
			clock::{Clock, MockClock},
			rng::Rng,
		};
		use reifydb_value::value::{identity::IdentityId, uuid::Uuid7, value_type::ValueType};

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		fn test_clock_and_rng() -> (MockClock, Clock, Rng) {
			let mock = MockClock::from_millis(1000);
			let clock = Clock::Mock(mock.clone());
			let rng = Rng::seeded(42);
			(mock, clock, rng)
		}

		#[test]
		fn test_asc() {
			let (mock, clock, rng) = test_clock_and_rng();
			let layout = IndexShape::new(&[ValueType::IdentityId], &[SortDirection::Asc]).unwrap();
			let mut key1 = layout.allocate_key();
			let mut key2 = layout.allocate_key();

			let id1 = IdentityId::generate(&clock, &rng);
			// Advance the clock so the two ids get different timestamps.
			mock.advance_millis(10);
			let id2 = IdentityId::generate(&clock, &rng);

			layout.set_identity_id(&mut key1, 0, id1);
			layout.set_identity_id(&mut key2, 0, id2);

			assert!(key1.is_defined(0));
			assert!(key2.is_defined(0));

			let offset = layout.fields[0].offset;
			let uuid7_1: Uuid7 = id1.into();
			let uuid7_2: Uuid7 = id2.into();
			let id1_bytes: Vec<u8> = uuid7_1.as_bytes().to_vec();
			let id2_bytes: Vec<u8> = uuid7_2.as_bytes().to_vec();

			assert_eq!(&key1[offset..offset + 16], &id1_bytes[..]);
			assert_eq!(&key2[offset..offset + 16], &id2_bytes[..]);

			// IdentityId wraps Uuid7, which leads with a timestamp, so the later one sorts greater.
			assert!(key1.as_slice() < key2.as_slice());
		}

		#[test]
		fn test_desc() {
			let (mock, clock, rng) = test_clock_and_rng();
			let layout = IndexShape::new(&[ValueType::IdentityId], &[SortDirection::Desc]).unwrap();
			let mut key1 = layout.allocate_key();
			let mut key2 = layout.allocate_key();

			let id1 = IdentityId::generate(&clock, &rng);
			// Advance the clock so the two ids get different timestamps.
			mock.advance_millis(10);
			let id2 = IdentityId::generate(&clock, &rng);

			layout.set_identity_id(&mut key1, 0, id1);
			layout.set_identity_id(&mut key2, 0, id2);

			let offset = layout.fields[0].offset;
			let uuid7_1: Uuid7 = id1.into();
			let uuid7_2: Uuid7 = id2.into();
			let mut expected_bytes1 = uuid7_1.as_bytes().to_vec();
			let mut expected_bytes2 = uuid7_2.as_bytes().to_vec();
			for b in expected_bytes1.iter_mut() {
				*b = !*b;
			}
			for b in expected_bytes2.iter_mut() {
				*b = !*b;
			}

			assert_eq!(&key1[offset..offset + 16], &expected_bytes1[..]);
			assert_eq!(&key2[offset..offset + 16], &expected_bytes2[..]);

			assert!(key1.as_slice() > key2.as_slice());
		}
	}

	mod undefined {
		use reifydb_value::value::value_type::ValueType;

		use crate::{sort::SortDirection, value::index::shape::IndexShape};

		#[test]
		fn test_undefined() {
			let layout = IndexShape::new(&[ValueType::Int4], &[SortDirection::Asc]).unwrap();
			let mut key = layout.allocate_key();

			layout.set_i32(&mut key, 0, 42i32);
			assert!(key.is_defined(0));

			layout.set_none(&mut key, 0);
			assert!(!key.is_defined(0));

			let offset = layout.fields[0].offset;
			assert_eq!(&key[offset..offset + 4], &[0, 0, 0, 0]);
		}
	}
}