reifydb_core/encoded/shape/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4//! Row-shape descriptor: the schema-of-bytes that explains how to interpret an `EncodedRow`.
5//!
6//! A `RowShape` lists every field (name, type constraint, byte offset, byte size, alignment) so storage backends,
7//! replication, and CDC can address fields without consulting the catalog. Submodules cover shape consolidation across
8//! rows of the same logical schema, schema evolution rules for adding and removing fields, structural fingerprinting
9//! used by plan caches and migration tooling, and conversion routines from typed schemas.
10//!
11//! Invariant: the `SHAPE_HEADER_SIZE` constant and the packed-mode bit layout (mode bit, length mask, offset mask) are
12//! part of the wire format. Reordering or resizing any of these regions silently breaks every row written under the
13//! previous layout.
14
15pub mod cache;
16pub mod consolidate;
17pub mod evolution;
18pub mod fingerprint;
19mod from;
20
21use std::{
22	alloc::{Layout, alloc_zeroed, handle_alloc_error},
23	fmt,
24	fmt::Debug,
25	iter,
26	ops::Deref,
27	ptr,
28	sync::{Arc, LazyLock, OnceLock},
29};
30
31use reifydb_value::{
32	reifydb_assertions,
33	util::cowvec::CowVec,
34	value::{constraint::TypeConstraint, value_type::ValueType},
35};
36use serde::{Deserialize, Serialize};
37
38use super::row::EncodedRow;
39use crate::encoded::shape::fingerprint::{RowShapeFingerprint, compute_fingerprint};
40
41pub const SHAPE_HEADER_SIZE: usize = 24;
42
43const PACKED_MODE_DYNAMIC: u128 = 0x80000000000000000000000000000000;
44const PACKED_MODE_MASK: u128 = 0x80000000000000000000000000000000;
45const PACKED_OFFSET_MASK: u128 = 0x0000000000000000FFFFFFFFFFFFFFFF;
46const PACKED_LENGTH_MASK: u128 = 0x7FFFFFFFFFFFFFFF0000000000000000;
47
48#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
49pub struct RowShapeField {
50	pub name: String,
51
52	pub constraint: TypeConstraint,
53
54	pub offset: u32,
55
56	pub size: u32,
57
58	pub align: u8,
59}
60
61impl RowShapeField {
62	pub fn new(name: impl Into<String>, constraint: TypeConstraint) -> Self {
63		let storage_type = constraint.storage_type();
64		Self {
65			name: name.into(),
66			constraint,
67			offset: 0,
68			size: storage_type.size() as u32,
69			align: storage_type.alignment() as u8,
70		}
71	}
72
73	pub fn unconstrained(name: impl Into<String>, field_type: ValueType) -> Self {
74		Self::new(name, TypeConstraint::unconstrained(field_type))
75	}
76}
77
78pub struct RowShape(Arc<Inner>);
79
80#[derive(Debug, Serialize, Deserialize)]
81pub struct Inner {
82	pub fingerprint: RowShapeFingerprint,
83
84	pub fields: Vec<RowShapeField>,
85
86	#[serde(skip)]
87	cached_layout: OnceLock<(usize, usize)>,
88}
89
90impl PartialEq for Inner {
91	fn eq(&self, other: &Self) -> bool {
92		self.fingerprint == other.fingerprint && self.fields == other.fields
93	}
94}
95
96impl Eq for Inner {}
97
98impl Deref for RowShape {
99	type Target = Inner;
100
101	fn deref(&self) -> &Self::Target {
102		&self.0
103	}
104}
105
106impl Clone for RowShape {
107	fn clone(&self) -> Self {
108		Self(self.0.clone())
109	}
110}
111
112impl Debug for RowShape {
113	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
114		self.0.fmt(f)
115	}
116}
117
118impl PartialEq for RowShape {
119	fn eq(&self, other: &Self) -> bool {
120		self.0.as_ref() == other.0.as_ref()
121	}
122}
123
124impl Eq for RowShape {}
125
126impl RowShape {
127	pub fn new(fields: Vec<RowShapeField>) -> Self {
128		let fields = Self::compute_layout(fields);
129		let fingerprint = compute_fingerprint(&fields);
130
131		Self(Arc::new(Inner {
132			fingerprint,
133			fields,
134			cached_layout: OnceLock::new(),
135		}))
136	}
137
138	pub fn from_parts(fingerprint: RowShapeFingerprint, fields: Vec<RowShapeField>) -> Self {
139		Self(Arc::new(Inner {
140			fingerprint,
141			fields,
142			cached_layout: OnceLock::new(),
143		}))
144	}
145
146	pub fn fingerprint(&self) -> RowShapeFingerprint {
147		self.fingerprint
148	}
149
150	pub fn fields(&self) -> &[RowShapeField] {
151		&self.fields
152	}
153
154	pub fn field_count(&self) -> usize {
155		self.fields.len()
156	}
157
158	pub fn find_field(&self, name: &str) -> Option<&RowShapeField> {
159		self.fields.iter().find(|f| f.name == name)
160	}
161
162	pub fn find_field_index(&self, name: &str) -> Option<usize> {
163		self.fields.iter().position(|f| f.name == name)
164	}
165
166	pub fn get_field(&self, index: usize) -> Option<&RowShapeField> {
167		self.fields.get(index)
168	}
169
170	pub fn get_field_name(&self, index: usize) -> Option<&str> {
171		self.fields.get(index).map(|f| f.name.as_str())
172	}
173
174	pub fn field_names(&self) -> impl Iterator<Item = &str> {
175		self.fields.iter().map(|f| f.name.as_str())
176	}
177
178	fn compute_layout(mut fields: Vec<RowShapeField>) -> Vec<RowShapeField> {
179		let bitvec_size = fields.len().div_ceil(8);
180		let mut offset: u32 = (SHAPE_HEADER_SIZE + bitvec_size) as u32;
181
182		for field in fields.iter_mut() {
183			let storage_type = field.constraint.storage_type();
184			field.size = storage_type.size() as u32;
185			field.align = storage_type.alignment() as u8;
186
187			let align = field.align as u32;
188			if align > 0 {
189				offset = (offset + align - 1) & !(align - 1);
190			}
191
192			field.offset = offset;
193			offset += field.size;
194		}
195
196		fields
197	}
198
199	pub fn bitvec_size(&self) -> usize {
200		self.fields.len().div_ceil(8)
201	}
202
203	pub fn data_offset(&self) -> usize {
204		SHAPE_HEADER_SIZE + self.bitvec_size()
205	}
206
207	fn get_cached_layout(&self) -> (usize, usize) {
208		*self.cached_layout.get_or_init(|| {
209			let max_align = self.fields.iter().map(|f| f.align as usize).max().unwrap_or(1);
210
211			let total_size = if self.fields.is_empty() {
212				SHAPE_HEADER_SIZE + self.bitvec_size()
213			} else {
214				let last_field = &self.fields[self.fields.len() - 1];
215				let end = last_field.offset as usize + last_field.size as usize;
216
217				Self::align_up(end, max_align)
218			};
219
220			(total_size, max_align)
221		})
222	}
223
224	pub fn total_static_size(&self) -> usize {
225		self.get_cached_layout().0
226	}
227
228	pub fn dynamic_section_start(&self) -> usize {
229		self.total_static_size()
230	}
231
232	pub fn dynamic_section_size(&self, row: &EncodedRow) -> usize {
233		row.len().saturating_sub(self.total_static_size())
234	}
235
236	pub(crate) fn read_dynamic_ref(&self, row: &EncodedRow, index: usize) -> Option<(usize, usize)> {
237		if !row.is_defined(index) {
238			return None;
239		}
240		let field = &self.fields()[index];
241		match field.constraint.get_type().inner_type() {
242			ValueType::Utf8 | ValueType::Blob | ValueType::Any => {
243				let ref_slice = &row.as_slice()[field.offset as usize..field.offset as usize + 8];
244				let offset =
245					u32::from_le_bytes([ref_slice[0], ref_slice[1], ref_slice[2], ref_slice[3]])
246						as usize;
247				let length =
248					u32::from_le_bytes([ref_slice[4], ref_slice[5], ref_slice[6], ref_slice[7]])
249						as usize;
250				Some((offset, length))
251			}
252			ValueType::Int | ValueType::Uint | ValueType::Decimal => {
253				let packed = unsafe {
254					(row.as_ptr().add(field.offset as usize) as *const u128).read_unaligned()
255				};
256				let packed = u128::from_le(packed);
257				if packed & PACKED_MODE_MASK != 0 {
258					let offset = (packed & PACKED_OFFSET_MASK) as usize;
259					let length = ((packed & PACKED_LENGTH_MASK) >> 64) as usize;
260					Some((offset, length))
261				} else {
262					None
263				}
264			}
265			_ => None,
266		}
267	}
268
269	pub(crate) fn write_dynamic_ref(&self, row: &mut EncodedRow, index: usize, offset: usize, length: usize) {
270		let field = &self.fields()[index];
271		match field.constraint.get_type().inner_type() {
272			ValueType::Utf8 | ValueType::Blob | ValueType::Any => {
273				let ref_slice = &mut row.0.make_mut()[field.offset as usize..field.offset as usize + 8];
274				ref_slice[0..4].copy_from_slice(&(offset as u32).to_le_bytes());
275				ref_slice[4..8].copy_from_slice(&(length as u32).to_le_bytes());
276			}
277			ValueType::Int | ValueType::Uint | ValueType::Decimal => {
278				let offset_part = (offset as u128) & PACKED_OFFSET_MASK;
279				let length_part = ((length as u128) << 64) & PACKED_LENGTH_MASK;
280				let packed = PACKED_MODE_DYNAMIC | offset_part | length_part;
281				unsafe {
282					ptr::write_unaligned(
283						row.0.make_mut().as_mut_ptr().add(field.offset as usize) as *mut u128,
284						packed.to_le(),
285					);
286				}
287			}
288			_ => {}
289		}
290	}
291
292	pub(crate) fn replace_dynamic_data(&self, row: &mut EncodedRow, index: usize, new_data: &[u8]) {
293		if let Some((old_offset, old_length)) = self.read_dynamic_ref(row, index) {
294			let delta = new_data.len() as isize - old_length as isize;
295
296			let refs_to_update: Vec<(usize, usize, usize)> = if delta != 0 {
297				self.fields()
298					.iter()
299					.enumerate()
300					.filter(|(i, _)| *i != index && row.is_defined(*i))
301					.filter_map(|(i, _)| {
302						self.read_dynamic_ref(row, i)
303							.filter(|(off, _)| *off > old_offset)
304							.map(|(off, len)| (i, off, len))
305					})
306					.collect()
307			} else {
308				vec![]
309			};
310
311			let dynamic_start = self.dynamic_section_start();
312			let abs_start = dynamic_start + old_offset;
313			let abs_end = abs_start + old_length;
314			row.0.make_mut().splice(abs_start..abs_end, new_data.iter().copied());
315
316			self.write_dynamic_ref(row, index, old_offset, new_data.len());
317
318			for (i, off, len) in refs_to_update {
319				let new_off = (off as isize + delta) as usize;
320				self.write_dynamic_ref(row, i, new_off, len);
321			}
322		} else {
323			let dynamic_offset = self.dynamic_section_size(row);
324			row.0.extend_from_slice(new_data);
325			self.write_dynamic_ref(row, index, dynamic_offset, new_data.len());
326		}
327		row.set_valid(index, true);
328	}
329
330	pub(crate) fn remove_dynamic_data(&self, row: &mut EncodedRow, index: usize) {
331		if let Some((old_offset, old_length)) = self.read_dynamic_ref(row, index) {
332			let refs_to_update: Vec<(usize, usize, usize)> = self
333				.fields()
334				.iter()
335				.enumerate()
336				.filter(|(i, _)| *i != index && row.is_defined(*i))
337				.filter_map(|(i, _)| {
338					self.read_dynamic_ref(row, i)
339						.filter(|(off, _)| *off > old_offset)
340						.map(|(off, len)| (i, off, len))
341				})
342				.collect();
343
344			let dynamic_start = self.dynamic_section_start();
345			let abs_start = dynamic_start + old_offset;
346			let abs_end = abs_start + old_length;
347			row.0.make_mut().splice(abs_start..abs_end, iter::empty());
348
349			for (i, off, len) in refs_to_update {
350				let new_off = off - old_length;
351				self.write_dynamic_ref(row, i, new_off, len);
352			}
353		}
354	}
355
356	pub fn allocate(&self) -> EncodedRow {
357		let (total_size, max_align) = self.get_cached_layout();
358		let layout = Layout::from_size_align(total_size, max_align).unwrap();
359		unsafe {
360			let ptr = alloc_zeroed(layout);
361			if ptr.is_null() {
362				handle_alloc_error(layout);
363			}
364			let vec = Vec::from_raw_parts(ptr, total_size, total_size);
365			let mut row = EncodedRow(CowVec::new(vec));
366			row.set_fingerprint(self.fingerprint);
367			reifydb_assertions! {
368				assert!(
369					row.len() == total_size,
370					"allocated row length does not match the shape total_static_size, so any field accessor using pre-computed offsets will read from garbage memory (row_len={} total_size={})",
371					row.len(),
372					total_size
373				);
374			}
375			row
376		}
377	}
378
379	fn align_up(offset: usize, align: usize) -> usize {
380		(offset + align).saturating_sub(1) & !(align.saturating_sub(1))
381	}
382
383	pub fn set_none(&self, row: &mut EncodedRow, index: usize) {
384		self.remove_dynamic_data(row, index);
385		row.set_valid(index, false);
386	}
387
388	pub fn testing(types: &[ValueType]) -> Self {
389		RowShape::new(
390			types.iter()
391				.enumerate()
392				.map(|(i, t)| RowShapeField::unconstrained(format!("f{}", i), t.clone()))
393				.collect(),
394		)
395	}
396
397	pub fn operator_state() -> Self {
398		OPERATOR_STATE_SHAPE.clone()
399	}
400}
401
402static OPERATOR_STATE_SHAPE: LazyLock<RowShape> =
403	LazyLock::new(|| RowShape::new(vec![RowShapeField::unconstrained("state", ValueType::Blob)]));
404
405#[cfg(test)]
406mod tests {
407	use super::*;
408
409	#[test]
410	fn test_shape_creation() {
411		let fields = vec![
412			RowShapeField::unconstrained("id", ValueType::Int8),
413			RowShapeField::unconstrained("name", ValueType::Utf8),
414			RowShapeField::unconstrained("active", ValueType::Boolean),
415		];
416
417		let shape = RowShape::new(fields);
418
419		assert_eq!(shape.field_count(), 3);
420		assert_eq!(shape.fields()[0].name, "id");
421		assert_eq!(shape.fields()[1].name, "name");
422		assert_eq!(shape.fields()[2].name, "active");
423	}
424
425	#[test]
426	fn test_shape_fingerprint_deterministic() {
427		let fields1 = vec![
428			RowShapeField::unconstrained("a", ValueType::Int4),
429			RowShapeField::unconstrained("b", ValueType::Utf8),
430		];
431
432		let fields2 = vec![
433			RowShapeField::unconstrained("a", ValueType::Int4),
434			RowShapeField::unconstrained("b", ValueType::Utf8),
435		];
436
437		let shape1 = RowShape::new(fields1);
438		let shape2 = RowShape::new(fields2);
439
440		assert_eq!(shape1.fingerprint(), shape2.fingerprint());
441	}
442
443	#[test]
444	fn test_shape_fingerprint_different_for_different_shapes() {
445		let fields1 = vec![RowShapeField::unconstrained("a", ValueType::Int4)];
446		let fields2 = vec![RowShapeField::unconstrained("a", ValueType::Int8)];
447
448		let shape1 = RowShape::new(fields1);
449		let shape2 = RowShape::new(fields2);
450
451		assert_ne!(shape1.fingerprint(), shape2.fingerprint());
452	}
453
454	#[test]
455	fn test_find_field() {
456		let fields = vec![
457			RowShapeField::unconstrained("id", ValueType::Int8),
458			RowShapeField::unconstrained("name", ValueType::Utf8),
459		];
460
461		let shape = RowShape::new(fields);
462
463		assert!(shape.find_field("id").is_some());
464		assert!(shape.find_field("name").is_some());
465		assert!(shape.find_field("missing").is_none());
466	}
467}
reifydb_core/encoded/shape/mod.rs

reifydb_core/encoded/shape/
mod.rs
