reifydb_core/interface/catalog/
series.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::row::pod::EncodedPodRow;
5use reifydb_value::{
6	Result,
7	value::{Value, datetime::DateTime, sumtype::SumTypeId, value_type::ValueType},
8};
9use serde::{Deserialize, Serialize};
10
11use crate::{
12	common::TimeSource,
13	interface::catalog::{
14		column::Column,
15		id::{NamespaceId, SeriesId},
16		key::PrimaryKey,
17	},
18	return_internal_error,
19	value::column::buffer::ColumnBuffer,
20};
21
22#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
23#[serde(rename_all = "lowercase")]
24#[derive(Default)]
25pub enum TimestampPrecision {
26	#[default]
27	Millisecond = 0,
28	Microsecond = 1,
29	Nanosecond = 2,
30	Second = 3,
31}
32
33#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
34pub enum SeriesKey {
35	DateTime {
36		column: String,
37		precision: TimestampPrecision,
38	},
39	Integer {
40		column: String,
41	},
42}
43
44impl SeriesKey {
45	pub fn column(&self) -> &str {
46		match self {
47			SeriesKey::DateTime {
48				column,
49				..
50			} => column,
51			SeriesKey::Integer {
52				column,
53			} => column,
54		}
55	}
56
57	pub fn decode(key_kind: u8, precision_raw: u8, column: String) -> Self {
58		match key_kind {
59			1 => SeriesKey::Integer {
60				column,
61			},
62			_ => {
63				let precision = match precision_raw {
64					1 => TimestampPrecision::Microsecond,
65					2 => TimestampPrecision::Nanosecond,
66					3 => TimestampPrecision::Second,
67					_ => TimestampPrecision::Millisecond,
68				};
69				SeriesKey::DateTime {
70					column,
71					precision,
72				}
73			}
74		}
75	}
76}
77
78#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
79pub struct Series {
80	pub id: SeriesId,
81	pub namespace: NamespaceId,
82	pub name: String,
83	pub columns: Vec<Column>,
84	pub tag: Option<SumTypeId>,
85	pub key: SeriesKey,
86	pub primary_key: Option<PrimaryKey>,
87	pub partition_by: Vec<String>,
88	pub underlying: bool,
89	pub time: TimeSource,
90}
91
92impl Series {
93	pub fn name(&self) -> &str {
94		&self.name
95	}
96
97	pub fn key_column_type(&self) -> Option<ValueType> {
98		let key_col_name = self.key.column();
99		self.columns.iter().find(|c| c.name == key_col_name).map(|c| c.constraint.get_type())
100	}
101
102	pub fn key_to_u64(&self, value: Value) -> Option<u64> {
103		match value {
104			Value::Int1(v) => u64::try_from(v).ok(),
105			Value::Int2(v) => u64::try_from(v).ok(),
106			Value::Int4(v) => u64::try_from(v).ok(),
107			Value::Int8(v) => u64::try_from(v).ok(),
108			Value::Int16(v) => u64::try_from(v).ok(),
109			Value::Uint1(v) => Some(v as u64),
110			Value::Uint2(v) => Some(v as u64),
111			Value::Uint4(v) => Some(v as u64),
112			Value::Uint8(v) => Some(v),
113			Value::Uint16(v) => u64::try_from(v).ok(),
114			Value::DateTime(dt) => {
115				let nanos = dt.to_nanos();
116				match &self.key {
117					SeriesKey::DateTime {
118						precision,
119						..
120					} => Some(match precision {
121						TimestampPrecision::Second => nanos / 1_000_000_000,
122						TimestampPrecision::Millisecond => nanos / 1_000_000,
123						TimestampPrecision::Microsecond => nanos / 1_000,
124						TimestampPrecision::Nanosecond => nanos,
125					}),
126					_ => Some(nanos),
127				}
128			}
129			_ => None,
130		}
131	}
132
133	pub fn key_from_u64(&self, v: u64) -> Value {
134		let ty = self.key_column_type();
135		match ty.as_ref() {
136			Some(ValueType::Int1) => Value::Int1(v as i8),
137			Some(ValueType::Int2) => Value::Int2(v as i16),
138			Some(ValueType::Int4) => Value::Int4(v as i32),
139			Some(ValueType::Int8) => Value::Int8(v as i64),
140			Some(ValueType::Uint1) => Value::Uint1(v as u8),
141			Some(ValueType::Uint2) => Value::Uint2(v as u16),
142			Some(ValueType::Uint4) => Value::Uint4(v as u32),
143			Some(ValueType::Uint8) => Value::Uint8(v),
144			Some(ValueType::Uint16) => Value::Uint16(v as u128),
145			Some(ValueType::Int16) => Value::Int16(v as i128),
146			Some(ValueType::DateTime) => {
147				let nanos: u64 = match &self.key {
148					SeriesKey::DateTime {
149						precision,
150						..
151					} => match precision {
152						TimestampPrecision::Second => v * 1_000_000_000,
153						TimestampPrecision::Millisecond => v * 1_000_000,
154						TimestampPrecision::Microsecond => v * 1_000,
155						TimestampPrecision::Nanosecond => v,
156					},
157					_ => v,
158				};
159				Value::DateTime(DateTime::from_nanos(nanos))
160			}
161			_ => Value::Uint8(v),
162		}
163	}
164
165	pub fn key_column_data(&self, keys: Vec<u64>) -> ColumnBuffer {
166		let key_type = self.key_column_type();
167		match &key_type {
168			Some(ty) => {
169				let mut data = ColumnBuffer::with_capacity(ty.clone(), keys.len());
170				for k in keys {
171					data.push_value(self.key_from_u64(k));
172				}
173				data
174			}
175			None => ColumnBuffer::uint8(keys),
176		}
177	}
178
179	pub fn data_columns(&self) -> impl Iterator<Item = &Column> {
180		let key_column = self.key.column().to_string();
181		self.columns.iter().filter(move |c| c.name != key_column)
182	}
183}
184
185#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
186pub struct SeriesMetadata {
187	pub row_count: u64,
188	pub oldest_key: u64,
189	pub newest_key: u64,
190	pub sequence_counter: u64,
191}
192
193impl SeriesMetadata {
194	pub fn new() -> Self {
195		Self {
196			row_count: 0,
197			oldest_key: 0,
198			newest_key: 0,
199			sequence_counter: 0,
200		}
201	}
202}
203
204impl Default for SeriesMetadata {
205	fn default() -> Self {
206		Self::new()
207	}
208}
209
210const SERIES_METADATA_WIDTH: usize = 32;
211
212pub fn encode_series_metadata(metadata: &SeriesMetadata) -> EncodedPodRow {
213	let mut bytes = Vec::with_capacity(SERIES_METADATA_WIDTH);
214	bytes.extend_from_slice(&metadata.row_count.to_be_bytes());
215	bytes.extend_from_slice(&metadata.oldest_key.to_be_bytes());
216	bytes.extend_from_slice(&metadata.newest_key.to_be_bytes());
217	bytes.extend_from_slice(&metadata.sequence_counter.to_be_bytes());
218	EncodedPodRow::new(&bytes)
219}
220
221pub fn decode_series_metadata(row: &EncodedPodRow) -> Result<SeriesMetadata> {
222	let bytes = row.body();
223	if bytes.len() != SERIES_METADATA_WIDTH {
224		return_internal_error!(
225			"Series metadata is {} bytes wide, expected {}. This indicates a corrupt metadata row.",
226			bytes.len(),
227			SERIES_METADATA_WIDTH
228		)
229	}
230	Ok(SeriesMetadata {
231		row_count: u64::from_be_bytes(bytes[0..8].try_into().unwrap()),
232		oldest_key: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
233		newest_key: u64::from_be_bytes(bytes[16..24].try_into().unwrap()),
234		sequence_counter: u64::from_be_bytes(bytes[24..32].try_into().unwrap()),
235	})
236}
237
238#[cfg(test)]
239mod series_metadata_tests {
240	use super::*;
241
242	#[test]
243	fn every_field_survives_a_round_trip_at_the_declared_width() {
244		let metadata = SeriesMetadata {
245			row_count: 42,
246			oldest_key: 100,
247			newest_key: 900,
248			sequence_counter: 7,
249		};
250
251		let row = encode_series_metadata(&metadata);
252
253		assert_eq!(row.len(), SERIES_METADATA_WIDTH);
254		assert_eq!(decode_series_metadata(&row).unwrap(), metadata);
255	}
256
257	#[test]
258	fn the_key_bounds_do_not_swap_because_they_select_which_buckets_materialise() {
259		let metadata = SeriesMetadata {
260			row_count: 1,
261			oldest_key: 1,
262			newest_key: u64::MAX,
263			sequence_counter: 0,
264		};
265
266		let decoded = decode_series_metadata(&encode_series_metadata(&metadata)).unwrap();
267
268		assert_eq!(decoded.oldest_key, 1);
269		assert_eq!(decoded.newest_key, u64::MAX);
270	}
271
272	#[test]
273	fn a_row_of_the_wrong_width_is_rejected_rather_than_rewinding_the_sequence_counter() {
274		assert!(decode_series_metadata(&EncodedPodRow::new(&[0u8; 31])).is_err());
275		assert!(decode_series_metadata(&EncodedPodRow::new(&[0u8; 33])).is_err());
276		assert!(decode_series_metadata(&EncodedPodRow::new(&[0u8; 40])).is_err());
277	}
278}
reifydb_core/interface/catalog/series.rs

reifydb_core/interface/catalog/
series.rs
