reifydb_core/interface/catalog/
series.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use arrow_array::ArrayRef;
5use arrow_schema::FieldRef;
6use reifydb_codec::row::pod::EncodedPodRow;
7use reifydb_value::{
8	Result,
9	value::{
10		Value,
11		column_view::{ColumnView, ViewData},
12		container::wide_int_array::wide_at,
13		datetime::DateTime,
14		sumtype::SumTypeId,
15		value_type::ValueType,
16	},
17};
18use serde::{Deserialize, Serialize};
19
20use crate::{
21	common::TimeSource,
22	interface::catalog::{
23		column::Column,
24		id::{NamespaceId, SeriesId},
25		key::PrimaryKey,
26	},
27	return_internal_error,
28	value::column::{builder::ColumnBuilder, factory},
29};
30
31#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
32#[serde(rename_all = "lowercase")]
33#[derive(Default)]
34pub enum TimestampPrecision {
35	#[default]
36	Millisecond = 0,
37	Microsecond = 1,
38	Nanosecond = 2,
39	Second = 3,
40}
41
42#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
43pub enum SeriesKey {
44	DateTime {
45		column: String,
46		precision: TimestampPrecision,
47	},
48	Integer {
49		column: String,
50	},
51}
52
53impl SeriesKey {
54	pub fn column(&self) -> &str {
55		match self {
56			SeriesKey::DateTime {
57				column,
58				..
59			} => column,
60			SeriesKey::Integer {
61				column,
62			} => column,
63		}
64	}
65
66	pub fn key_to_u64(&self, value: Value) -> Option<u64> {
67		match value {
68			Value::Int1(v) => u64::try_from(v).ok(),
69			Value::Int2(v) => u64::try_from(v).ok(),
70			Value::Int4(v) => u64::try_from(v).ok(),
71			Value::Int8(v) => u64::try_from(v).ok(),
72			Value::Int16(v) => u64::try_from(v).ok(),
73			Value::Uint1(v) => Some(v as u64),
74			Value::Uint2(v) => Some(v as u64),
75			Value::Uint4(v) => Some(v as u64),
76			Value::Uint8(v) => Some(v),
77			Value::Uint16(v) => u64::try_from(v).ok(),
78			Value::DateTime(dt) => self.datetime_key(dt),
79			_ => None,
80		}
81	}
82
83	fn datetime_key(&self, datetime: DateTime) -> Option<u64> {
84		let nanos = datetime.to_nanos();
85		match self {
86			SeriesKey::DateTime {
87				precision,
88				..
89			} => u64::try_from(match precision {
90				TimestampPrecision::Second => nanos.div_euclid(1_000_000_000),
91				TimestampPrecision::Millisecond => nanos.div_euclid(1_000_000),
92				TimestampPrecision::Microsecond => nanos.div_euclid(1_000),
93				TimestampPrecision::Nanosecond => nanos,
94			})
95			.ok(),
96			_ => u64::try_from(nanos).ok(),
97		}
98	}
99
100	pub fn keys_to_u64(&self, view: &ColumnView<'_>) -> Vec<Option<u64>> {
101		let nulls = view.logical_nulls();
102		(0..view.len())
103			.map(|row| {
104				if nulls.as_ref().is_some_and(|nulls| nulls.is_null(row)) {
105					return None;
106				}
107				match &view.data {
108					ViewData::Int1(array) => u64::try_from(array.value(row)).ok(),
109					ViewData::Int2(array) => u64::try_from(array.value(row)).ok(),
110					ViewData::Int4(array) => u64::try_from(array.value(row)).ok(),
111					ViewData::Int8(array) => u64::try_from(array.value(row)).ok(),
112					ViewData::Int16(array) => {
113						wide_at::<i128>(array, row).and_then(|v| u64::try_from(v).ok())
114					}
115					ViewData::Uint1(array) => Some(array.value(row) as u64),
116					ViewData::Uint2(array) => Some(array.value(row) as u64),
117					ViewData::Uint4(array) => Some(array.value(row) as u64),
118					ViewData::Uint8(array) => Some(array.value(row)),
119					ViewData::Uint16(array) => {
120						wide_at::<u128>(array, row).and_then(|v| u64::try_from(v).ok())
121					}
122					ViewData::DateTime(array) => {
123						self.datetime_key(DateTime::from_nanos(array.value(row)))
124					}
125					_ => None,
126				}
127			})
128			.collect()
129	}
130
131	pub fn key_from_u64(&self, v: u64, key_type: Option<ValueType>) -> Value {
132		match key_type.as_ref() {
133			Some(ValueType::Int1) => Value::Int1(v as i8),
134			Some(ValueType::Int2) => Value::Int2(v as i16),
135			Some(ValueType::Int4) => Value::Int4(v as i32),
136			Some(ValueType::Int8) => Value::Int8(v as i64),
137			Some(ValueType::Uint1) => Value::Uint1(v as u8),
138			Some(ValueType::Uint2) => Value::Uint2(v as u16),
139			Some(ValueType::Uint4) => Value::Uint4(v as u32),
140			Some(ValueType::Uint8) => Value::Uint8(v),
141			Some(ValueType::Uint16) => Value::Uint16(v as u128),
142			Some(ValueType::Int16) => Value::Int16(v as i128),
143			Some(ValueType::DateTime) => {
144				let nanos: i128 = match self {
145					SeriesKey::DateTime {
146						precision,
147						..
148					} => match precision {
149						TimestampPrecision::Second => v as i128 * 1_000_000_000,
150						TimestampPrecision::Millisecond => v as i128 * 1_000_000,
151						TimestampPrecision::Microsecond => v as i128 * 1_000,
152						TimestampPrecision::Nanosecond => v as i128,
153					},
154					_ => v as i128,
155				};
156				Value::DateTime(DateTime::from_nanos(
157					i64::try_from(nanos).expect("series key past the datetime range"),
158				))
159			}
160			_ => Value::Uint8(v),
161		}
162	}
163
164	pub fn decode(key_kind: u8, precision_raw: u8, column: String) -> Self {
165		match key_kind {
166			1 => SeriesKey::Integer {
167				column,
168			},
169			_ => {
170				let precision = match precision_raw {
171					1 => TimestampPrecision::Microsecond,
172					2 => TimestampPrecision::Nanosecond,
173					3 => TimestampPrecision::Second,
174					_ => TimestampPrecision::Millisecond,
175				};
176				SeriesKey::DateTime {
177					column,
178					precision,
179				}
180			}
181		}
182	}
183}
184
185#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
186pub struct Series {
187	pub id: SeriesId,
188	pub namespace: NamespaceId,
189	pub name: String,
190	pub columns: Vec<Column>,
191	pub tag: Option<SumTypeId>,
192	pub key: SeriesKey,
193	pub primary_key: Option<PrimaryKey>,
194	pub partition_by: Vec<String>,
195	pub time: TimeSource,
196	pub encoding: Option<SeriesEncoding>,
197}
198
199#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
200pub struct SeriesEncoding {
201	pub name: String,
202	pub config: Vec<u8>,
203}
204
205impl Series {
206	pub fn name(&self) -> &str {
207		&self.name
208	}
209
210	pub fn key_column_type(&self) -> Option<ValueType> {
211		let key_col_name = self.key.column();
212		self.columns.iter().find(|c| c.name == key_col_name).map(|c| c.constraint.get_type())
213	}
214
215	pub fn key_to_u64(&self, value: Value) -> Option<u64> {
216		self.key.key_to_u64(value)
217	}
218
219	pub fn key_from_u64(&self, v: u64) -> Value {
220		self.key.key_from_u64(v, self.key_column_type())
221	}
222
223	pub fn key_column_data(&self, keys: Vec<u64>) -> (FieldRef, ArrayRef) {
224		match self.key_column_type() {
225			Some(ty) => {
226				let mut builder = ColumnBuilder::with_capacity(ty.clone(), keys.len());
227				for k in keys {
228					builder.push_value(self.key.key_from_u64(k, Some(ty.clone())));
229				}
230				builder.finish(self.key.column())
231			}
232			None => factory::uint8(self.key.column(), keys),
233		}
234	}
235
236	pub fn data_columns(&self) -> impl Iterator<Item = &Column> {
237		let key_column = self.key.column().to_string();
238		self.columns.iter().filter(move |c| c.name != key_column)
239	}
240
241	pub fn encoding_user_fields(&self) -> Vec<FieldRef> {
242		let mut fields = vec![self.key_column_data(Vec::new()).0];
243		if self.tag.is_some() {
244			fields.push(factory::uint1("tag", Vec::new()).0);
245		}
246		for column in self.data_columns() {
247			fields.push(ColumnBuilder::with_capacity(column.constraint.get_type(), 0)
248				.finish(&column.name)
249				.0);
250		}
251		fields
252	}
253}
254
255#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
256pub struct SeriesPartitionMetadata {
257	pub row_count: u64,
258	pub oldest_key: u64,
259	pub newest_key: u64,
260	pub sequence_counter: u64,
261	pub last_write_at: DateTime,
262	pub dirty_from_key: u64,
263	pub dirty_to_key: u64,
264}
265
266impl SeriesPartitionMetadata {
267	pub fn new() -> Self {
268		Self {
269			row_count: 0,
270			oldest_key: 0,
271			newest_key: 0,
272			sequence_counter: 0,
273			last_write_at: DateTime::default(),
274			dirty_from_key: u64::MAX,
275			dirty_to_key: 0,
276		}
277	}
278}
279
280impl Default for SeriesPartitionMetadata {
281	fn default() -> Self {
282		Self::new()
283	}
284}
285
286const SERIES_PARTITION_METADATA_WIDTH: usize = 56;
287
288pub fn encode_series_partition_metadata(metadata: &SeriesPartitionMetadata) -> EncodedPodRow {
289	let mut bytes = Vec::with_capacity(SERIES_PARTITION_METADATA_WIDTH);
290	bytes.extend_from_slice(&metadata.row_count.to_be_bytes());
291	bytes.extend_from_slice(&metadata.oldest_key.to_be_bytes());
292	bytes.extend_from_slice(&metadata.newest_key.to_be_bytes());
293	bytes.extend_from_slice(&metadata.sequence_counter.to_be_bytes());
294	bytes.extend_from_slice(&metadata.last_write_at.to_order().to_be_bytes());
295	bytes.extend_from_slice(&metadata.dirty_from_key.to_be_bytes());
296	bytes.extend_from_slice(&metadata.dirty_to_key.to_be_bytes());
297	EncodedPodRow::new(&bytes)
298}
299
300pub fn decode_series_partition_metadata(row: &EncodedPodRow) -> Result<SeriesPartitionMetadata> {
301	let bytes = row.body();
302	if bytes.len() != SERIES_PARTITION_METADATA_WIDTH {
303		return_internal_error!(
304			"Series partition metadata is {} bytes wide, expected {}. This indicates a corrupt metadata row.",
305			bytes.len(),
306			SERIES_PARTITION_METADATA_WIDTH
307		)
308	}
309	Ok(SeriesPartitionMetadata {
310		row_count: u64::from_be_bytes(bytes[0..8].try_into().unwrap()),
311		oldest_key: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
312		newest_key: u64::from_be_bytes(bytes[16..24].try_into().unwrap()),
313		sequence_counter: u64::from_be_bytes(bytes[24..32].try_into().unwrap()),
314		last_write_at: DateTime::from_order(u64::from_be_bytes(bytes[32..40].try_into().unwrap())),
315		dirty_from_key: u64::from_be_bytes(bytes[40..48].try_into().unwrap()),
316		dirty_to_key: u64::from_be_bytes(bytes[48..56].try_into().unwrap()),
317	})
318}
319
320#[cfg(test)]
321mod series_partition_metadata_tests {
322	use super::*;
323
324	#[test]
325	fn every_field_survives_a_round_trip_at_the_declared_width() {
326		let metadata = SeriesPartitionMetadata {
327			row_count: 42,
328			oldest_key: 100,
329			newest_key: 900,
330			sequence_counter: 7,
331			last_write_at: DateTime::from_order(1_700_000_000_000_000_000),
332			dirty_from_key: 512,
333			dirty_to_key: 1024,
334		};
335
336		let row = encode_series_partition_metadata(&metadata);
337
338		assert_eq!(row.len(), SERIES_PARTITION_METADATA_WIDTH);
339		assert_eq!(decode_series_partition_metadata(&row).unwrap(), metadata);
340	}
341
342	#[test]
343	fn the_key_bounds_do_not_swap_because_they_select_which_buckets_materialise() {
344		let metadata = SeriesPartitionMetadata {
345			row_count: 1,
346			oldest_key: 1,
347			newest_key: u64::MAX,
348			sequence_counter: 0,
349			last_write_at: DateTime::default(),
350			dirty_from_key: u64::MAX,
351			dirty_to_key: 0,
352		};
353
354		let decoded = decode_series_partition_metadata(&encode_series_partition_metadata(&metadata)).unwrap();
355
356		assert_eq!(decoded.oldest_key, 1);
357		assert_eq!(decoded.newest_key, u64::MAX);
358	}
359
360	#[test]
361	fn a_row_of_the_wrong_width_is_rejected_rather_than_rewinding_the_sequence_counter() {
362		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 32])).is_err());
363		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 55])).is_err());
364		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 57])).is_err());
365	}
366
367	#[test]
368	fn the_last_write_stamp_does_not_collide_with_the_sequence_counter() {
369		// the two sit next to each other in the row; a width or offset slip would let a
370		// sequence bump read back as a write stamp and strand a bucket forever
371		let metadata = SeriesPartitionMetadata {
372			row_count: 0,
373			oldest_key: 0,
374			newest_key: 0,
375			sequence_counter: u64::MAX,
376			last_write_at: DateTime::from_order(1),
377			dirty_from_key: u64::MAX,
378			dirty_to_key: 0,
379		};
380
381		let decoded = decode_series_partition_metadata(&encode_series_partition_metadata(&metadata)).unwrap();
382
383		assert_eq!(decoded.sequence_counter, u64::MAX);
384		assert_eq!(decoded.last_write_at, DateTime::from_order(1));
385	}
386}
387
388#[cfg(test)]
389mod series_key_tests {
390	use super::*;
391
392	#[test]
393	fn a_negative_datetime_gives_no_series_key() {
394		// nanos before the epoch must never wrap into a huge positive u64 key
395		let key = SeriesKey::DateTime {
396			column: "ts".to_string(),
397			precision: TimestampPrecision::Nanosecond,
398		};
399
400		assert_eq!(key.key_to_u64(Value::DateTime(DateTime::from_nanos(-1))), None);
401	}
402}
403
404#[cfg(test)]
405mod encoding_user_fields_tests {
406	use reifydb_value::value::{
407		constraint::{Constraint, TypeConstraint, bytes::MaxBytes},
408		dictionary::DictionaryId,
409		sumtype::SumTypeId,
410		value_type::ValueType,
411	};
412
413	use super::{Series, SeriesKey};
414	use crate::{
415		common::TimeSource,
416		interface::catalog::{
417			column::{Column, ColumnIndex},
418			id::{ColumnId, NamespaceId, SeriesId},
419		},
420		value::column::builder::ColumnBuilder,
421	};
422
423	fn column(index: u8, name: &str, constraint: TypeConstraint, dictionary_id: Option<DictionaryId>) -> Column {
424		Column {
425			id: ColumnId(index as u64),
426			name: name.to_string(),
427			constraint,
428			properties: vec![],
429			index: ColumnIndex(index),
430			auto_increment: false,
431			dictionary_id,
432		}
433	}
434
435	#[test]
436	fn test_encoding_user_fields_keep_extension_tags() {
437		// Fields built from the bare value type drop the tags the seal scan sets, so bind sees another schema.
438		let data = [
439			(
440				"label",
441				TypeConstraint::with_constraint(
442					ValueType::Utf8,
443					Constraint::MaxBytes(MaxBytes::new(50)),
444				),
445			),
446			("region", TypeConstraint::dictionary(DictionaryId(3), ValueType::Uint4)),
447			("id", TypeConstraint::unconstrained(ValueType::Uuid7)),
448			("maybe", TypeConstraint::unconstrained(ValueType::Option(Box::new(ValueType::Int4)))),
449		];
450		let mut columns = vec![column(0, "ts", TypeConstraint::unconstrained(ValueType::Int8), None)];
451		for (i, (name, constraint)) in data.iter().enumerate() {
452			let dictionary = (*name == "region").then_some(DictionaryId(3));
453			columns.push(column(i as u8 + 1, name, constraint.clone(), dictionary));
454		}
455		let series = Series {
456			id: SeriesId(1),
457			namespace: NamespaceId(1),
458			name: "s".to_string(),
459			columns,
460			tag: Some(SumTypeId(9)),
461			key: SeriesKey::Integer {
462				column: "ts".to_string(),
463			},
464			primary_key: None,
465			partition_by: vec![],
466			time: TimeSource::Processing,
467			encoding: None,
468		};
469
470		let fields = series.encoding_user_fields();
471
472		let names: Vec<&str> = fields.iter().map(|f| f.name().as_str()).collect();
473		assert_eq!(
474			names,
475			vec!["ts", "tag", "label", "region", "id", "maybe"],
476			"key, tag, then data; no system columns"
477		);
478		for (name, constraint) in data.iter() {
479			let scanned = ColumnBuilder::with_capacity(constraint.get_type(), 0).finish(name).0;
480			let field = fields.iter().find(|f| f.name() == name).unwrap();
481			assert_eq!(field, &scanned, "field {name} must equal the seal scan field");
482		}
483		assert!(fields.iter().find(|f| f.name() == "maybe").unwrap().is_nullable());
484		assert!(!fields.iter().find(|f| f.name() == "ts").unwrap().is_nullable());
485	}
486}
reifydb_core/interface/catalog/series.rs

reifydb_core/interface/catalog/
series.rs
