reifydb-core 0.10.1

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

use arrow_array::ArrayRef;
use arrow_schema::FieldRef;
use reifydb_codec::row::pod::EncodedPodRow;
use reifydb_value::{
	Result,
	value::{
		Value,
		column_view::{ColumnView, ViewData},
		container::wide_int_array::wide_at,
		datetime::DateTime,
		sumtype::SumTypeId,
		value_type::ValueType,
	},
};
use serde::{Deserialize, Serialize};

use crate::{
	common::TimeSource,
	interface::catalog::{
		column::Column,
		id::{NamespaceId, SeriesId},
		key::PrimaryKey,
	},
	return_internal_error,
	value::column::{builder::ColumnBuilder, factory},
};

#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[derive(Default)]
pub enum TimestampPrecision {
	#[default]
	Millisecond = 0,
	Microsecond = 1,
	Nanosecond = 2,
	Second = 3,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum SeriesKey {
	DateTime {
		column: String,
		precision: TimestampPrecision,
	},
	Integer {
		column: String,
	},
}

impl SeriesKey {
	pub fn column(&self) -> &str {
		match self {
			SeriesKey::DateTime {
				column,
				..
			} => column,
			SeriesKey::Integer {
				column,
			} => column,
		}
	}

	pub fn key_to_u64(&self, value: Value) -> Option<u64> {
		match value {
			Value::Int1(v) => u64::try_from(v).ok(),
			Value::Int2(v) => u64::try_from(v).ok(),
			Value::Int4(v) => u64::try_from(v).ok(),
			Value::Int8(v) => u64::try_from(v).ok(),
			Value::Int16(v) => u64::try_from(v).ok(),
			Value::Uint1(v) => Some(v as u64),
			Value::Uint2(v) => Some(v as u64),
			Value::Uint4(v) => Some(v as u64),
			Value::Uint8(v) => Some(v),
			Value::Uint16(v) => u64::try_from(v).ok(),
			Value::DateTime(dt) => self.datetime_key(dt),
			_ => None,
		}
	}

	fn datetime_key(&self, datetime: DateTime) -> Option<u64> {
		let nanos = datetime.to_nanos();
		match self {
			SeriesKey::DateTime {
				precision,
				..
			} => u64::try_from(match precision {
				TimestampPrecision::Second => nanos.div_euclid(1_000_000_000),
				TimestampPrecision::Millisecond => nanos.div_euclid(1_000_000),
				TimestampPrecision::Microsecond => nanos.div_euclid(1_000),
				TimestampPrecision::Nanosecond => nanos,
			})
			.ok(),
			_ => u64::try_from(nanos).ok(),
		}
	}

	pub fn keys_to_u64(&self, view: &ColumnView<'_>) -> Vec<Option<u64>> {
		let nulls = view.logical_nulls();
		(0..view.len())
			.map(|row| {
				if nulls.as_ref().is_some_and(|nulls| nulls.is_null(row)) {
					return None;
				}
				match &view.data {
					ViewData::Int1(array) => u64::try_from(array.value(row)).ok(),
					ViewData::Int2(array) => u64::try_from(array.value(row)).ok(),
					ViewData::Int4(array) => u64::try_from(array.value(row)).ok(),
					ViewData::Int8(array) => u64::try_from(array.value(row)).ok(),
					ViewData::Int16(array) => {
						wide_at::<i128>(array, row).and_then(|v| u64::try_from(v).ok())
					}
					ViewData::Uint1(array) => Some(array.value(row) as u64),
					ViewData::Uint2(array) => Some(array.value(row) as u64),
					ViewData::Uint4(array) => Some(array.value(row) as u64),
					ViewData::Uint8(array) => Some(array.value(row)),
					ViewData::Uint16(array) => {
						wide_at::<u128>(array, row).and_then(|v| u64::try_from(v).ok())
					}
					ViewData::DateTime(array) => {
						self.datetime_key(DateTime::from_nanos(array.value(row)))
					}
					_ => None,
				}
			})
			.collect()
	}

	pub fn key_from_u64(&self, v: u64, key_type: Option<ValueType>) -> Value {
		match key_type.as_ref() {
			Some(ValueType::Int1) => Value::Int1(v as i8),
			Some(ValueType::Int2) => Value::Int2(v as i16),
			Some(ValueType::Int4) => Value::Int4(v as i32),
			Some(ValueType::Int8) => Value::Int8(v as i64),
			Some(ValueType::Uint1) => Value::Uint1(v as u8),
			Some(ValueType::Uint2) => Value::Uint2(v as u16),
			Some(ValueType::Uint4) => Value::Uint4(v as u32),
			Some(ValueType::Uint8) => Value::Uint8(v),
			Some(ValueType::Uint16) => Value::Uint16(v as u128),
			Some(ValueType::Int16) => Value::Int16(v as i128),
			Some(ValueType::DateTime) => {
				let nanos: i128 = match self {
					SeriesKey::DateTime {
						precision,
						..
					} => match precision {
						TimestampPrecision::Second => v as i128 * 1_000_000_000,
						TimestampPrecision::Millisecond => v as i128 * 1_000_000,
						TimestampPrecision::Microsecond => v as i128 * 1_000,
						TimestampPrecision::Nanosecond => v as i128,
					},
					_ => v as i128,
				};
				Value::DateTime(DateTime::from_nanos(
					i64::try_from(nanos).expect("series key past the datetime range"),
				))
			}
			_ => Value::Uint8(v),
		}
	}

	pub fn decode(key_kind: u8, precision_raw: u8, column: String) -> Self {
		match key_kind {
			1 => SeriesKey::Integer {
				column,
			},
			_ => {
				let precision = match precision_raw {
					1 => TimestampPrecision::Microsecond,
					2 => TimestampPrecision::Nanosecond,
					3 => TimestampPrecision::Second,
					_ => TimestampPrecision::Millisecond,
				};
				SeriesKey::DateTime {
					column,
					precision,
				}
			}
		}
	}
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Series {
	pub id: SeriesId,
	pub namespace: NamespaceId,
	pub name: String,
	pub columns: Vec<Column>,
	pub tag: Option<SumTypeId>,
	pub key: SeriesKey,
	pub primary_key: Option<PrimaryKey>,
	pub partition_by: Vec<String>,
	pub time: TimeSource,
	pub encoding: Option<SeriesEncoding>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SeriesEncoding {
	pub name: String,
	pub config: Vec<u8>,
}

impl Series {
	pub fn name(&self) -> &str {
		&self.name
	}

	pub fn key_column_type(&self) -> Option<ValueType> {
		let key_col_name = self.key.column();
		self.columns.iter().find(|c| c.name == key_col_name).map(|c| c.constraint.get_type())
	}

	pub fn key_to_u64(&self, value: Value) -> Option<u64> {
		self.key.key_to_u64(value)
	}

	pub fn key_from_u64(&self, v: u64) -> Value {
		self.key.key_from_u64(v, self.key_column_type())
	}

	pub fn key_column_data(&self, keys: Vec<u64>) -> (FieldRef, ArrayRef) {
		match self.key_column_type() {
			Some(ty) => {
				let mut builder = ColumnBuilder::with_capacity(ty.clone(), keys.len());
				for k in keys {
					builder.push_value(self.key.key_from_u64(k, Some(ty.clone())));
				}
				builder.finish(self.key.column())
			}
			None => factory::uint8(self.key.column(), keys),
		}
	}

	pub fn data_columns(&self) -> impl Iterator<Item = &Column> {
		let key_column = self.key.column().to_string();
		self.columns.iter().filter(move |c| c.name != key_column)
	}

	pub fn encoding_user_fields(&self) -> Vec<FieldRef> {
		let mut fields = vec![self.key_column_data(Vec::new()).0];
		if self.tag.is_some() {
			fields.push(factory::uint1("tag", Vec::new()).0);
		}
		for column in self.data_columns() {
			fields.push(ColumnBuilder::with_capacity(column.constraint.get_type(), 0)
				.finish(&column.name)
				.0);
		}
		fields
	}
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SeriesPartitionMetadata {
	pub row_count: u64,
	pub oldest_key: u64,
	pub newest_key: u64,
	pub sequence_counter: u64,
	pub last_write_at: DateTime,
	pub dirty_from_key: u64,
	pub dirty_to_key: u64,
}

impl SeriesPartitionMetadata {
	pub fn new() -> Self {
		Self {
			row_count: 0,
			oldest_key: 0,
			newest_key: 0,
			sequence_counter: 0,
			last_write_at: DateTime::default(),
			dirty_from_key: u64::MAX,
			dirty_to_key: 0,
		}
	}
}

impl Default for SeriesPartitionMetadata {
	fn default() -> Self {
		Self::new()
	}
}

const SERIES_PARTITION_METADATA_WIDTH: usize = 56;

pub fn encode_series_partition_metadata(metadata: &SeriesPartitionMetadata) -> EncodedPodRow {
	let mut bytes = Vec::with_capacity(SERIES_PARTITION_METADATA_WIDTH);
	bytes.extend_from_slice(&metadata.row_count.to_be_bytes());
	bytes.extend_from_slice(&metadata.oldest_key.to_be_bytes());
	bytes.extend_from_slice(&metadata.newest_key.to_be_bytes());
	bytes.extend_from_slice(&metadata.sequence_counter.to_be_bytes());
	bytes.extend_from_slice(&metadata.last_write_at.to_order().to_be_bytes());
	bytes.extend_from_slice(&metadata.dirty_from_key.to_be_bytes());
	bytes.extend_from_slice(&metadata.dirty_to_key.to_be_bytes());
	EncodedPodRow::new(&bytes)
}

pub fn decode_series_partition_metadata(row: &EncodedPodRow) -> Result<SeriesPartitionMetadata> {
	let bytes = row.body();
	if bytes.len() != SERIES_PARTITION_METADATA_WIDTH {
		return_internal_error!(
			"Series partition metadata is {} bytes wide, expected {}. This indicates a corrupt metadata row.",
			bytes.len(),
			SERIES_PARTITION_METADATA_WIDTH
		)
	}
	Ok(SeriesPartitionMetadata {
		row_count: u64::from_be_bytes(bytes[0..8].try_into().unwrap()),
		oldest_key: u64::from_be_bytes(bytes[8..16].try_into().unwrap()),
		newest_key: u64::from_be_bytes(bytes[16..24].try_into().unwrap()),
		sequence_counter: u64::from_be_bytes(bytes[24..32].try_into().unwrap()),
		last_write_at: DateTime::from_order(u64::from_be_bytes(bytes[32..40].try_into().unwrap())),
		dirty_from_key: u64::from_be_bytes(bytes[40..48].try_into().unwrap()),
		dirty_to_key: u64::from_be_bytes(bytes[48..56].try_into().unwrap()),
	})
}

#[cfg(test)]
mod series_partition_metadata_tests {
	use super::*;

	#[test]
	fn every_field_survives_a_round_trip_at_the_declared_width() {
		let metadata = SeriesPartitionMetadata {
			row_count: 42,
			oldest_key: 100,
			newest_key: 900,
			sequence_counter: 7,
			last_write_at: DateTime::from_order(1_700_000_000_000_000_000),
			dirty_from_key: 512,
			dirty_to_key: 1024,
		};

		let row = encode_series_partition_metadata(&metadata);

		assert_eq!(row.len(), SERIES_PARTITION_METADATA_WIDTH);
		assert_eq!(decode_series_partition_metadata(&row).unwrap(), metadata);
	}

	#[test]
	fn the_key_bounds_do_not_swap_because_they_select_which_buckets_materialise() {
		let metadata = SeriesPartitionMetadata {
			row_count: 1,
			oldest_key: 1,
			newest_key: u64::MAX,
			sequence_counter: 0,
			last_write_at: DateTime::default(),
			dirty_from_key: u64::MAX,
			dirty_to_key: 0,
		};

		let decoded = decode_series_partition_metadata(&encode_series_partition_metadata(&metadata)).unwrap();

		assert_eq!(decoded.oldest_key, 1);
		assert_eq!(decoded.newest_key, u64::MAX);
	}

	#[test]
	fn a_row_of_the_wrong_width_is_rejected_rather_than_rewinding_the_sequence_counter() {
		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 32])).is_err());
		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 55])).is_err());
		assert!(decode_series_partition_metadata(&EncodedPodRow::new(&[0u8; 57])).is_err());
	}

	#[test]
	fn the_last_write_stamp_does_not_collide_with_the_sequence_counter() {
		// the two sit next to each other in the row; a width or offset slip would let a
		// sequence bump read back as a write stamp and strand a bucket forever
		let metadata = SeriesPartitionMetadata {
			row_count: 0,
			oldest_key: 0,
			newest_key: 0,
			sequence_counter: u64::MAX,
			last_write_at: DateTime::from_order(1),
			dirty_from_key: u64::MAX,
			dirty_to_key: 0,
		};

		let decoded = decode_series_partition_metadata(&encode_series_partition_metadata(&metadata)).unwrap();

		assert_eq!(decoded.sequence_counter, u64::MAX);
		assert_eq!(decoded.last_write_at, DateTime::from_order(1));
	}
}

#[cfg(test)]
mod series_key_tests {
	use super::*;

	#[test]
	fn a_negative_datetime_gives_no_series_key() {
		// nanos before the epoch must never wrap into a huge positive u64 key
		let key = SeriesKey::DateTime {
			column: "ts".to_string(),
			precision: TimestampPrecision::Nanosecond,
		};

		assert_eq!(key.key_to_u64(Value::DateTime(DateTime::from_nanos(-1))), None);
	}
}

#[cfg(test)]
mod encoding_user_fields_tests {
	use reifydb_value::value::{
		constraint::{Constraint, TypeConstraint, bytes::MaxBytes},
		dictionary::DictionaryId,
		sumtype::SumTypeId,
		value_type::ValueType,
	};

	use super::{Series, SeriesKey};
	use crate::{
		common::TimeSource,
		interface::catalog::{
			column::{Column, ColumnIndex},
			id::{ColumnId, NamespaceId, SeriesId},
		},
		value::column::builder::ColumnBuilder,
	};

	fn column(index: u8, name: &str, constraint: TypeConstraint, dictionary_id: Option<DictionaryId>) -> Column {
		Column {
			id: ColumnId(index as u64),
			name: name.to_string(),
			constraint,
			properties: vec![],
			index: ColumnIndex(index),
			auto_increment: false,
			dictionary_id,
		}
	}

	#[test]
	fn test_encoding_user_fields_keep_extension_tags() {
		// Fields built from the bare value type drop the tags the seal scan sets, so bind sees another schema.
		let data = [
			(
				"label",
				TypeConstraint::with_constraint(
					ValueType::Utf8,
					Constraint::MaxBytes(MaxBytes::new(50)),
				),
			),
			("region", TypeConstraint::dictionary(DictionaryId(3), ValueType::Uint4)),
			("id", TypeConstraint::unconstrained(ValueType::Uuid7)),
			("maybe", TypeConstraint::unconstrained(ValueType::Option(Box::new(ValueType::Int4)))),
		];
		let mut columns = vec![column(0, "ts", TypeConstraint::unconstrained(ValueType::Int8), None)];
		for (i, (name, constraint)) in data.iter().enumerate() {
			let dictionary = (*name == "region").then_some(DictionaryId(3));
			columns.push(column(i as u8 + 1, name, constraint.clone(), dictionary));
		}
		let series = Series {
			id: SeriesId(1),
			namespace: NamespaceId(1),
			name: "s".to_string(),
			columns,
			tag: Some(SumTypeId(9)),
			key: SeriesKey::Integer {
				column: "ts".to_string(),
			},
			primary_key: None,
			partition_by: vec![],
			time: TimeSource::Processing,
			encoding: None,
		};

		let fields = series.encoding_user_fields();

		let names: Vec<&str> = fields.iter().map(|f| f.name().as_str()).collect();
		assert_eq!(
			names,
			vec!["ts", "tag", "label", "region", "id", "maybe"],
			"key, tag, then data; no system columns"
		);
		for (name, constraint) in data.iter() {
			let scanned = ColumnBuilder::with_capacity(constraint.get_type(), 0).finish(name).0;
			let field = fields.iter().find(|f| f.name() == name).unwrap();
			assert_eq!(field, &scanned, "field {name} must equal the seal scan field");
		}
		assert!(fields.iter().find(|f| f.name() == "maybe").unwrap().is_nullable());
		assert!(!fields.iter().find(|f| f.name() == "ts").unwrap().is_nullable());
	}
}