reifydb_value/value/container/

varlen.rs

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

use std::{
	fmt::{self, Debug},
	result::Result as StdResult,
};

use serde::{Deserialize, Deserializer, Serialize, Serializer, ser::SerializeSeq};
use serde_bytes::{ByteBuf, Bytes};

use crate::reifydb_assertions;

pub struct VarlenContainer {
	data: Vec<u8>,
	offsets: Vec<u64>,
}

impl Clone for VarlenContainer {
	fn clone(&self) -> Self {
		Self {
			data: self.data.clone(),
			offsets: self.offsets.clone(),
		}
	}
}

impl Debug for VarlenContainer {
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
		f.debug_struct("VarlenContainer")
			.field("len", &self.len())
			.field("data_bytes", &self.data.len())
			.finish()
	}
}

impl PartialEq for VarlenContainer {
	fn eq(&self, other: &Self) -> bool {
		if self.len() != other.len() {
			return false;
		}
		for i in 0..self.len() {
			if self.get_bytes(i) != other.get_bytes(i) {
				return false;
			}
		}
		true
	}
}

impl VarlenContainer {
	pub fn from_byte_slices<'a, I>(items: I) -> Self
	where
		I: IntoIterator<Item = &'a [u8]>,
	{
		let mut offsets: Vec<u64> = vec![0];
		let mut data: Vec<u8> = Vec::new();
		for item in items {
			data.extend_from_slice(item);
			offsets.push(data.len() as u64);
		}
		Self {
			data,
			offsets,
		}
	}

	pub fn from_repeated_bytes(item: &[u8], count: usize) -> Self {
		let mut data: Vec<u8> = Vec::with_capacity(item.len() * count);
		let mut offsets: Vec<u64> = Vec::with_capacity(count + 1);
		offsets.push(0);
		for _ in 0..count {
			data.extend_from_slice(item);
			offsets.push(data.len() as u64);
		}
		Self {
			data,
			offsets,
		}
	}

	pub fn with_capacity(item_capacity: usize, data_capacity: usize) -> Self {
		let mut offsets = Vec::with_capacity(item_capacity + 1);
		offsets.push(0);
		Self {
			data: Vec::with_capacity(data_capacity),
			offsets,
		}
	}

	pub fn empty() -> Self {
		Self {
			data: Vec::new(),
			offsets: vec![0],
		}
	}

	pub fn from_raw_parts(data: Vec<u8>, offsets: Vec<u64>) -> Self {
		reifydb_assertions! {
			assert!(!offsets.is_empty(), "offsets must always have offsets[0] = 0");
			assert_eq!(offsets[0], 0, "offsets[0] must be 0");
			assert_eq!(*offsets.last().unwrap() as usize, data.len(), "offsets[len] must equal data.len()");
		}
		Self {
			data,
			offsets,
		}
	}
}

impl VarlenContainer {
	pub fn from_storage_parts(data: Vec<u8>, offsets: Vec<u64>) -> Self {
		reifydb_assertions! {
			assert!(
				!offsets.is_empty(),
				"offsets must always include the leading 0; got empty offsets"
			);
		}
		Self {
			data,
			offsets,
		}
	}

	pub fn len(&self) -> usize {
		self.offsets.len().saturating_sub(1)
	}

	pub fn is_empty(&self) -> bool {
		self.len() == 0
	}

	pub fn data_byte_len(&self) -> usize {
		self.data.len()
	}

	pub fn data_bytes(&self) -> &[u8] {
		self.data.as_slice()
	}

	pub fn offsets(&self) -> &[u64] {
		self.offsets.as_slice()
	}

	pub fn get_bytes(&self, idx: usize) -> Option<&[u8]> {
		if idx >= self.len() {
			return None;
		}
		let start = self.offsets.as_slice()[idx] as usize;
		let end = self.offsets.as_slice()[idx + 1] as usize;
		self.data.as_slice().get(start..end)
	}

	pub fn capacity(&self) -> usize {
		self.offsets.capacity().saturating_sub(1)
	}

	pub fn heap_size(&self) -> usize {
		self.data.capacity() + self.offsets.capacity() * size_of::<u64>()
	}

	pub fn data(&self) -> &Vec<u8> {
		&self.data
	}

	pub fn offsets_data(&self) -> &Vec<u64> {
		&self.offsets
	}

	pub fn clear_generic(&mut self) {
		self.data.clear();
		self.offsets.clear();
		self.offsets.push(0u64);
	}
}

impl VarlenContainer {
	pub fn clear(&mut self) {
		let data_mut = &mut self.data;
		data_mut.clear();
		let offsets_mut = &mut self.offsets;
		offsets_mut.clear();
		offsets_mut.push(0);
	}

	pub fn push_bytes(&mut self, bytes: &[u8]) {
		let data_mut = &mut self.data;
		data_mut.extend_from_slice(bytes);
		let new_end = data_mut.len() as u64;
		self.offsets.push(new_end);
	}

	pub fn extend_from(&mut self, other: &Self) {
		let base = self.data.len() as u64;
		let other_offsets = other.offsets.as_slice();
		let other_data = other.data.as_slice();

		let data_mut = &mut self.data;
		data_mut.extend_from_slice(other_data);

		let offsets_mut = &mut self.offsets;

		offsets_mut.extend(other_offsets.iter().skip(1).map(|&o| base + o));
	}

	pub fn filter_in_place<F: FnMut(usize) -> bool>(&mut self, mut keep: F) {
		let len = self.len();
		let mut new_data = Vec::with_capacity(self.data_byte_len());
		let mut new_offsets = Vec::with_capacity(len + 1);
		new_offsets.push(0);
		for i in 0..len {
			if keep(i) {
				let bytes = self.get_bytes(i).unwrap_or(&[]);
				new_data.extend_from_slice(bytes);
				new_offsets.push(new_data.len() as u64);
			}
		}
		self.data = new_data;
		self.offsets = new_offsets;
	}

	pub fn reorder_in_place(&mut self, indices: &[usize]) {
		let mut new_data = Vec::with_capacity(self.data_byte_len());
		let mut new_offsets = Vec::with_capacity(indices.len() + 1);
		new_offsets.push(0);
		for &idx in indices {
			let bytes = self.get_bytes(idx).unwrap_or(&[]);
			new_data.extend_from_slice(bytes);
			new_offsets.push(new_data.len() as u64);
		}
		self.data = new_data;
		self.offsets = new_offsets;
	}

	pub fn take_n(&self, n: usize) -> Self {
		let n = n.min(self.len());
		let end_byte = self.offsets.as_slice()[n] as usize;
		let new_data: Vec<u8> = self.data.as_slice()[..end_byte].to_vec();
		let new_offsets: Vec<u64> = self.offsets.as_slice()[..=n].to_vec();
		Self::from_raw_parts(new_data, new_offsets)
	}

	pub fn slice(&self, start: usize, end: usize) -> Self {
		let len = self.len();
		let start = start.min(len);
		let end = end.min(len);
		if start >= end {
			return Self::empty();
		}
		let start_byte = self.offsets.as_slice()[start] as usize;
		let end_byte = self.offsets.as_slice()[end] as usize;
		let new_data: Vec<u8> = self.data.as_slice()[start_byte..end_byte].to_vec();

		let new_offsets: Vec<u64> =
			self.offsets.as_slice()[start..=end].iter().map(|o| *o - start_byte as u64).collect();
		Self::from_raw_parts(new_data, new_offsets)
	}
}

impl Serialize for VarlenContainer {
	fn serialize<Ser: Serializer>(&self, serializer: Ser) -> StdResult<Ser::Ok, Ser::Error> {
		let mut seq = serializer.serialize_seq(Some(self.len()))?;
		for i in 0..self.len() {
			let bytes = self.get_bytes(i).unwrap_or(&[]);
			seq.serialize_element(Bytes::new(bytes))?;
		}
		seq.end()
	}
}

impl<'de> Deserialize<'de> for VarlenContainer {
	fn deserialize<D: Deserializer<'de>>(deserializer: D) -> StdResult<Self, D::Error> {
		let items: Vec<ByteBuf> = Vec::deserialize(deserializer)?;
		let total: usize = items.iter().map(|b| b.len()).sum();
		let mut data = Vec::with_capacity(total);
		let mut offsets = Vec::with_capacity(items.len() + 1);
		offsets.push(0);
		for item in items {
			data.extend_from_slice(item.as_slice());
			offsets.push(data.len() as u64);
		}
		Ok(Self::from_raw_parts(data, offsets))
	}
}

impl Default for VarlenContainer {
	fn default() -> Self {
		Self::empty()
	}
}

#[cfg(test)]
mod tests {
	use postcard::{from_bytes as postcard_from_bytes, to_allocvec as postcard_to_allocvec};

	use super::*;

	#[test]
	fn empty_has_zero_len_and_offsets_with_one_zero() {
		let c = VarlenContainer::empty();
		assert_eq!(c.len(), 0);
		assert_eq!(c.offsets(), &[0u64]);
		assert!(c.data_bytes().is_empty());
	}

	#[test]
	fn push_bytes_appends_and_updates_offsets() {
		let mut c = VarlenContainer::empty();
		c.push_bytes(b"hello");
		c.push_bytes(b"");
		c.push_bytes(b"world");
		assert_eq!(c.len(), 3);
		assert_eq!(c.offsets(), &[0u64, 5, 5, 10]);
		assert_eq!(c.data_bytes(), b"helloworld");
		assert_eq!(c.get_bytes(0), Some(b"hello".as_slice()));
		assert_eq!(c.get_bytes(1), Some(b"".as_slice()));
		assert_eq!(c.get_bytes(2), Some(b"world".as_slice()));
		assert_eq!(c.get_bytes(3), None);
	}

	#[test]
	fn from_byte_slices_round_trip() {
		let c = VarlenContainer::from_byte_slices([b"a".as_slice(), b"bc", b"def"]);
		assert_eq!(c.len(), 3);
		assert_eq!(c.data_bytes(), b"abcdef");
		assert_eq!(c.offsets(), &[0u64, 1, 3, 6]);
	}

	#[test]
	fn from_repeated_bytes_matches_explicit_copies() {
		let repeated = VarlenContainer::from_repeated_bytes(b"abc", 3);
		let explicit = VarlenContainer::from_byte_slices([b"abc".as_slice(), b"abc", b"abc"]);
		assert_eq!(repeated, explicit);
		assert_eq!(repeated.len(), 3);
		assert_eq!(repeated.data_bytes(), b"abcabcabc");
		assert_eq!(repeated.offsets(), &[0u64, 3, 6, 9]);
	}

	#[test]
	fn from_repeated_bytes_zero_count_is_empty() {
		let c = VarlenContainer::from_repeated_bytes(b"abc", 0);
		assert_eq!(c.len(), 0);
		assert_eq!(c.offsets(), &[0u64]);
		assert!(c.data_bytes().is_empty());
	}

	#[test]
	fn from_repeated_bytes_empty_item_keeps_count() {
		let c = VarlenContainer::from_repeated_bytes(b"", 4);
		assert_eq!(c.len(), 4);
		assert_eq!(c.get_bytes(0), Some(b"".as_slice()));
		assert_eq!(c.get_bytes(3), Some(b"".as_slice()));
		assert!(c.data_bytes().is_empty());
	}

	#[test]
	fn clear_resets_to_empty_state() {
		let mut c = VarlenContainer::from_byte_slices([b"x".as_slice(), b"y"]);
		c.clear();
		assert_eq!(c.len(), 0);
		assert_eq!(c.offsets(), &[0u64]);
		assert!(c.data_bytes().is_empty());
	}

	#[test]
	fn extend_from_concatenates_and_rebases_offsets() {
		let mut a = VarlenContainer::from_byte_slices([b"foo".as_slice()]);
		let b = VarlenContainer::from_byte_slices([b"bar".as_slice(), b"baz"]);
		a.extend_from(&b);
		assert_eq!(a.len(), 3);
		assert_eq!(a.data_bytes(), b"foobarbaz");
		assert_eq!(a.offsets(), &[0u64, 3, 6, 9]);
	}

	#[test]
	fn filter_in_place_keeps_matching_elements() {
		let mut c = VarlenContainer::from_byte_slices([b"yes".as_slice(), b"no", b"yes", b"no"]);
		c.filter_in_place(|i| i % 2 == 0);
		assert_eq!(c.len(), 2);
		assert_eq!(c.get_bytes(0), Some(b"yes".as_slice()));
		assert_eq!(c.get_bytes(1), Some(b"yes".as_slice()));
	}

	#[test]
	fn reorder_in_place_handles_oob_as_empty() {
		let mut c = VarlenContainer::from_byte_slices([b"a".as_slice(), b"b"]);
		c.reorder_in_place(&[1, 100, 0]);
		assert_eq!(c.len(), 3);
		assert_eq!(c.get_bytes(0), Some(b"b".as_slice()));
		assert_eq!(c.get_bytes(1), Some(b"".as_slice()));
		assert_eq!(c.get_bytes(2), Some(b"a".as_slice()));
	}

	#[test]
	fn take_n_truncates() {
		let c = VarlenContainer::from_byte_slices([b"a".as_slice(), b"b", b"c"]);
		let t = c.take_n(2);
		assert_eq!(t.len(), 2);
		assert_eq!(t.get_bytes(0), Some(b"a".as_slice()));
		assert_eq!(t.get_bytes(1), Some(b"b".as_slice()));
	}

	#[test]
	fn slice_extracts_subrange_with_rebased_offsets() {
		let c = VarlenContainer::from_byte_slices([b"aa".as_slice(), b"bb", b"cc", b"dd"]);
		let s = c.slice(1, 3);
		assert_eq!(s.len(), 2);
		assert_eq!(s.get_bytes(0), Some(b"bb".as_slice()));
		assert_eq!(s.get_bytes(1), Some(b"cc".as_slice()));
		assert_eq!(s.offsets(), &[0u64, 2, 4]);
	}

	#[test]
	fn serde_round_trip_preserves_content() {
		let original = VarlenContainer::from_byte_slices([b"hello".as_slice(), b"", b"world"]);
		let encoded: Vec<u8> = postcard_to_allocvec(&original).unwrap();
		let decoded: VarlenContainer = postcard_from_bytes(&encoded).unwrap();
		assert_eq!(decoded.len(), 3);
		assert_eq!(decoded.get_bytes(0), Some(b"hello".as_slice()));
		assert_eq!(decoded.get_bytes(1), Some(b"".as_slice()));
		assert_eq!(decoded.get_bytes(2), Some(b"world".as_slice()));
	}

	#[test]
	fn serde_wire_compat_with_vec_of_strings() {
		// Postcard encodes `Vec<String>` and `Vec<&[u8]>` identically (both are length-prefixed
		// sequences of length-prefixed bytes), so the two are wire-compatible.
		let strings = vec!["a".to_string(), "bc".to_string(), "def".to_string()];
		let encoded: Vec<u8> = postcard_to_allocvec(&strings).unwrap();
		let decoded: VarlenContainer = postcard_from_bytes(&encoded).unwrap();
		assert_eq!(decoded.len(), 3);
		assert_eq!(decoded.get_bytes(0), Some(b"a".as_slice()));
		assert_eq!(decoded.get_bytes(1), Some(b"bc".as_slice()));
		assert_eq!(decoded.get_bytes(2), Some(b"def".as_slice()));
	}

	#[test]
	fn equality_compares_logical_content() {
		let a = VarlenContainer::from_byte_slices([b"x".as_slice(), b"y"]);
		let b = VarlenContainer::from_byte_slices([b"x".as_slice(), b"y"]);
		assert_eq!(a, b);
	}
}