reifydb_value/value/
system_columns.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::HashMap,
6	fmt::{self, Display, Formatter},
7	slice,
8	sync::{Arc, LazyLock},
9};
10
11use arrow_array::{
12	Array, ArrayRef, FixedSizeBinaryArray, RecordBatch, RecordBatchOptions, TimestampNanosecondArray, UInt64Array,
13};
14use arrow_schema::{ArrowError, Field, FieldRef, Schema};
15
16use crate::{
17	Result,
18	error::Error,
19	value::{
20		column_view::ColumnView,
21		container::{temporal_array::datetimes, wide_int_array::wides},
22		datetime::DateTime,
23		partition::Partition,
24		row_number::RowNumber,
25		value_type::{
26			ValueType,
27			field::{FieldType, field_error, to_field},
28		},
29	},
30};
31
32#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum SystemColumn {
34	RowNumbers,
35	Partitions,
36	CreatedAt,
37	UpdatedAt,
38	Time,
39	CommitVersion,
40}
41
42impl SystemColumn {
43	pub const ALL: [SystemColumn; 6] = [
44		SystemColumn::RowNumbers,
45		SystemColumn::Partitions,
46		SystemColumn::CreatedAt,
47		SystemColumn::UpdatedAt,
48		SystemColumn::Time,
49		SystemColumn::CommitVersion,
50	];
51
52	pub const fn name(self) -> &'static str {
53		match self {
54			SystemColumn::RowNumbers => "#rownum",
55			SystemColumn::Partitions => "#partition",
56			SystemColumn::CreatedAt => "#created_at",
57			SystemColumn::UpdatedAt => "#updated_at",
58			SystemColumn::Time => "#time",
59			SystemColumn::CommitVersion => "#commit_version",
60		}
61	}
62
63	pub const fn ty(self) -> ValueType {
64		match self {
65			SystemColumn::RowNumbers => ValueType::Uint8,
66			SystemColumn::Partitions => ValueType::Uint16,
67			SystemColumn::CreatedAt => ValueType::DateTime,
68			SystemColumn::UpdatedAt => ValueType::DateTime,
69			SystemColumn::Time => ValueType::DateTime,
70			SystemColumn::CommitVersion => ValueType::Uint8,
71		}
72	}
73
74	pub fn from_name(name: &str) -> Option<SystemColumn> {
75		SystemColumn::ALL.into_iter().find(|column| column.name() == name)
76	}
77
78	fn rank(self) -> usize {
79		SystemColumn::ALL.iter().position(|column| *column == self).expect("every system column is in ALL")
80	}
81}
82
83static SYSTEM_FIELDS: LazyLock<[[FieldRef; 2]; 6]> = LazyLock::new(|| {
84	SystemColumn::ALL.map(|column| {
85		[false, true].map(|holds_nones| {
86			let value_type = match holds_nones {
87				true => ValueType::Option(Box::new(column.ty())),
88				false => column.ty(),
89			};
90			Arc::new(to_field(
91				column.name(),
92				&FieldType {
93					value_type: Some(value_type),
94					..FieldType::default()
95				},
96			))
97		})
98	})
99});
100
101pub fn system_field(column: SystemColumn, holds_nones: bool) -> FieldRef {
102	SYSTEM_FIELDS[column.rank()][usize::from(holds_nones)].clone()
103}
104
105impl Display for SystemColumn {
106	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
107		f.write_str(self.name())
108	}
109}
110
111pub fn system_column(batch: &RecordBatch, column: SystemColumn) -> Option<&ArrayRef> {
112	position(batch, column.name()).map(|index| batch.column(index))
113}
114
115pub fn row_numbers(batch: &RecordBatch) -> Result<&[RowNumber]> {
116	let Some(array) = present::<UInt64Array>(batch, SystemColumn::RowNumbers)? else {
117		return Ok(&[]);
118	};
119	let values: &[u64] = array.values();
120	// SAFETY: RowNumber is repr(transparent) over u64 with no niche, so the cast keeps the bounds and lifetime.
121	Ok(unsafe { slice::from_raw_parts(values.as_ptr().cast::<RowNumber>(), values.len()) })
122}
123
124pub fn partitions(batch: &RecordBatch) -> Result<Vec<Partition>> {
125	let Some(array) = present::<FixedSizeBinaryArray>(batch, SystemColumn::Partitions)? else {
126		return Ok(Vec::new());
127	};
128	Ok(wides::<u128>(array).into_iter().map(Partition).collect())
129}
130
131pub fn created_at(batch: &RecordBatch) -> Result<&[DateTime]> {
132	datetime_column(batch, SystemColumn::CreatedAt)
133}
134
135pub fn updated_at(batch: &RecordBatch) -> Result<&[DateTime]> {
136	datetime_column(batch, SystemColumn::UpdatedAt)
137}
138
139pub fn time(batch: &RecordBatch) -> Result<&[DateTime]> {
140	datetime_column(batch, SystemColumn::Time)
141}
142
143#[derive(Debug, Clone, Copy, PartialEq, Eq)]
144pub struct RowTimes<'a> {
145	values: &'a [DateTime],
146	defined: Option<(&'a [u8], usize)>,
147}
148
149impl<'a> RowTimes<'a> {
150	pub const EMPTY: Self = Self {
151		values: &[],
152		defined: None,
153	};
154
155	pub fn new(values: &'a [DateTime], defined: Option<(&'a [u8], usize)>) -> Self {
156		Self {
157			values,
158			defined,
159		}
160	}
161
162	pub fn len(&self) -> usize {
163		self.values.len()
164	}
165
166	pub fn is_empty(&self) -> bool {
167		self.values.is_empty()
168	}
169
170	pub fn get(&self, index: usize) -> Option<DateTime> {
171		let value = *self.values.get(index)?;
172		match self.defined {
173			Some((bits, offset)) => {
174				let bit = offset + index;
175				bits.get(bit / 8).is_some_and(|byte| byte & (1 << (bit % 8)) != 0).then_some(value)
176			}
177			None => Some(value),
178		}
179	}
180
181	pub fn iter(&self) -> impl Iterator<Item = Option<DateTime>> + '_ {
182		(0..self.len()).map(|index| self.get(index))
183	}
184
185	pub fn values(&self) -> &'a [DateTime] {
186		self.values
187	}
188
189	pub fn defined(&self) -> Option<(&'a [u8], usize)> {
190		self.defined
191	}
192}
193
194#[allow(clippy::disallowed_methods)]
195pub fn row_times(batch: &RecordBatch) -> Result<RowTimes<'_>> {
196	let Some(array) = system_column(batch, SystemColumn::Time) else {
197		return Ok(RowTimes::EMPTY);
198	};
199	let array = array.as_any().downcast_ref::<TimestampNanosecondArray>().ok_or_else(|| {
200		field_error(format!(
201			"system column {} holds arrow type {}",
202			SystemColumn::Time.name(),
203			array.data_type()
204		))
205	})?;
206	Ok(RowTimes::new(datetimes(array), array.nulls().map(|nulls| (nulls.validity(), nulls.offset()))))
207}
208
209pub fn require_row_numbers(batch: &RecordBatch) -> Result<&[RowNumber]> {
210	require(batch, SystemColumn::RowNumbers)?;
211	row_numbers(batch)
212}
213
214pub fn require_created_at(batch: &RecordBatch) -> Result<&[DateTime]> {
215	require(batch, SystemColumn::CreatedAt)?;
216	created_at(batch)
217}
218
219pub fn require_updated_at(batch: &RecordBatch) -> Result<&[DateTime]> {
220	require(batch, SystemColumn::UpdatedAt)?;
221	updated_at(batch)
222}
223
224pub fn require_time(batch: &RecordBatch) -> Result<&[DateTime]> {
225	require(batch, SystemColumn::Time)?;
226	time(batch)
227}
228
229pub fn commit_versions(batch: &RecordBatch) -> Result<&[u64]> {
230	match present::<UInt64Array>(batch, SystemColumn::CommitVersion)? {
231		Some(array) => Ok(array.values()),
232		None => Ok(&[]),
233	}
234}
235
236pub fn with_system_column(batch: RecordBatch, column: SystemColumn, array: ArrayRef) -> Result<RecordBatch> {
237	stamp_system_columns(batch, vec![(column, array)])
238}
239
240pub fn stamp_system_columns(batch: RecordBatch, stamps: Vec<(SystemColumn, ArrayRef)>) -> Result<RecordBatch> {
241	let schema = batch.schema();
242	let fields: Vec<FieldRef> = schema.fields().iter().cloned().collect();
243	stamped(schema.metadata().clone(), fields, batch.columns().to_vec(), batch.num_rows(), stamps)
244}
245
246pub fn restamp_row_numbers(batch: &RecordBatch, keep: &[SystemColumn], row_numbers: ArrayRef) -> Result<RecordBatch> {
247	let schema = batch.schema_ref();
248	let (fields, columns): (Vec<FieldRef>, Vec<ArrayRef>) = schema
249		.fields()
250		.iter()
251		.zip(batch.columns())
252		.filter(|(field, _)| SystemColumn::from_name(field.name()).is_none_or(|column| keep.contains(&column)))
253		.map(|(field, array)| (field.clone(), array.clone()))
254		.unzip();
255	stamped(
256		schema.metadata().clone(),
257		fields,
258		columns,
259		batch.num_rows(),
260		vec![(SystemColumn::RowNumbers, row_numbers)],
261	)
262}
263
264fn stamped(
265	metadata: HashMap<String, String>,
266	mut fields: Vec<FieldRef>,
267	mut columns: Vec<ArrayRef>,
268	num_rows: usize,
269	stamps: Vec<(SystemColumn, ArrayRef)>,
270) -> Result<RecordBatch> {
271	let mut row_count = num_rows;
272	for (column, array) in stamps {
273		let field = system_field(column, array.logical_null_count() > 0);
274		if fields.is_empty() && row_count == 0 {
275			row_count = array.len();
276		}
277		match fields.iter().position(|other| other.name() == column.name()) {
278			Some(index) => {
279				fields[index] = field;
280				columns[index] = array;
281			}
282			None => {
283				let index = fields
284					.iter()
285					.position(|other| {
286						SystemColumn::from_name(other.name())
287							.is_some_and(|other| other.rank() > column.rank())
288					})
289					.unwrap_or(fields.len());
290				fields.insert(index, field);
291				columns.insert(index, array);
292			}
293		}
294	}
295	RecordBatch::try_new_with_options(
296		Arc::new(Schema::new_with_metadata(fields, metadata)),
297		columns,
298		&RecordBatchOptions::new().with_row_count(Some(row_count)),
299	)
300	.map_err(arrow_error)
301}
302
303pub fn keep_system_columns(batch: &RecordBatch, keep: &[SystemColumn]) -> Result<RecordBatch> {
304	let indices: Vec<usize> = batch
305		.schema_ref()
306		.fields()
307		.iter()
308		.enumerate()
309		.filter(|(_, field)| SystemColumn::from_name(field.name()).is_none_or(|column| keep.contains(&column)))
310		.map(|(index, _)| index)
311		.collect();
312	batch.project(&indices).map_err(arrow_error)
313}
314
315pub fn is_system_field(field: &Field) -> bool {
316	field.name().starts_with('#')
317}
318
319pub fn user_columns(batch: &RecordBatch) -> impl Iterator<Item = (&FieldRef, &ArrayRef)> {
320	batch.schema_ref().fields().iter().zip(batch.columns()).filter(|(field, _)| !is_system_field(field))
321}
322
323pub fn column_view<'a>(batch: &'a RecordBatch, name: &str) -> Result<Option<ColumnView<'a>>> {
324	match position(batch, name) {
325		Some(index) => ColumnView::try_from((batch.column(index), batch.schema_ref().field(index))).map(Some),
326		None => Ok(None),
327	}
328}
329
330pub fn resolve_column(batch: &RecordBatch, name: &str) -> Option<usize> {
331	let bare = name.strip_prefix('#').unwrap_or(name);
332	SystemColumn::ALL
333		.into_iter()
334		.find(|column| &column.name()[1..] == bare)
335		.and_then(|column| position(batch, column.name()))
336		.or_else(|| position(batch, name))
337}
338
339pub fn check_user_columns(batch: &RecordBatch) -> Result<()> {
340	if batch.num_rows() > 0 && batch.num_columns() == 0 {
341		return Err(field_error(format!("a batch of {} rows has no columns", batch.num_rows())));
342	}
343	Ok(())
344}
345
346fn position(batch: &RecordBatch, name: &str) -> Option<usize> {
347	batch.schema_ref().fields().iter().position(|field| field.name() == name)
348}
349
350fn present<T: 'static>(batch: &RecordBatch, column: SystemColumn) -> Result<Option<&T>> {
351	let Some(array) = system_column(batch, column) else {
352		return Ok(None);
353	};
354	if array.logical_null_count() > 0 {
355		return Err(field_error(format!(
356			"system column {} holds {} none rows",
357			column.name(),
358			array.logical_null_count()
359		)));
360	}
361	array.as_any().downcast_ref::<T>().map(Some).ok_or_else(|| {
362		field_error(format!("system column {} holds arrow type {}", column.name(), array.data_type()))
363	})
364}
365
366fn require(batch: &RecordBatch, column: SystemColumn) -> Result<()> {
367	match system_column(batch, column) {
368		Some(_) => Ok(()),
369		None => Err(field_error(format!("system column {} is missing", column.name()))),
370	}
371}
372
373fn datetime_column(batch: &RecordBatch, column: SystemColumn) -> Result<&[DateTime]> {
374	match present::<TimestampNanosecondArray>(batch, column)? {
375		Some(array) => Ok(datetimes(array)),
376		None => Ok(&[]),
377	}
378}
379
380fn arrow_error(error: ArrowError) -> Error {
381	field_error(error.to_string())
382}
383
384#[cfg(test)]
385mod tests {
386	use arrow_array::{Int32Array, UInt64Array};
387	use arrow_schema::DataType;
388
389	use super::*;
390	use crate::value::{
391		Value,
392		container::{temporal_array::datetime_array, wide_int_array::wide_array},
393	};
394
395	fn user(name: &str, values: Vec<i32>) -> (FieldRef, ArrayRef) {
396		let field = to_field(
397			name,
398			&FieldType {
399				value_type: Some(ValueType::Int4),
400				..FieldType::default()
401			},
402		);
403		(Arc::new(field), Arc::new(Int32Array::from(values)))
404	}
405
406	fn batch(columns: Vec<(FieldRef, ArrayRef)>) -> RecordBatch {
407		let (fields, arrays): (Vec<FieldRef>, Vec<ArrayRef>) = columns.into_iter().unzip();
408		RecordBatch::try_new(Arc::new(Schema::new(fields)), arrays).unwrap()
409	}
410
411	fn names(batch: &RecordBatch) -> Vec<String> {
412		batch.schema_ref().fields().iter().map(|field| field.name().clone()).collect()
413	}
414
415	fn stamps(values: Vec<i64>) -> ArrayRef {
416		Arc::new(datetime_array(values.into_iter().map(DateTime::from_nanos)))
417	}
418
419	#[test]
420	fn row_numbers_read_the_rownum_column_and_are_empty_when_it_is_absent() {
421		// A getter that fails on an absent column would break every batch that never carried row numbers.
422		let plain = batch(vec![user("a", vec![1, 2])]);
423		assert!(row_numbers(&plain).unwrap().is_empty());
424		let numbered =
425			with_system_column(plain, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![7u64, 9])))
426				.unwrap();
427		assert_eq!(row_numbers(&numbered).unwrap(), &[RowNumber(7), RowNumber(9)]);
428	}
429
430	#[test]
431	fn partitions_decode_the_sixteen_byte_column() {
432		// Reading #partition as a u16 column would truncate every partition id above 65535.
433		let wide: ArrayRef = Arc::new(wide_array([1u128 << 100, 3]));
434		let with =
435			with_system_column(batch(vec![user("a", vec![1, 2])]), SystemColumn::Partitions, wide).unwrap();
436		assert_eq!(partitions(&with).unwrap(), vec![Partition(1u128 << 100), Partition(3)]);
437	}
438
439	#[test]
440	fn datetime_getters_read_their_own_column() {
441		// Mixing up #created_at, #updated_at and #time would stamp rows with the wrong instant.
442		let base = batch(vec![user("a", vec![1])]);
443		let base = with_system_column(base, SystemColumn::CreatedAt, stamps(vec![10])).unwrap();
444		let base = with_system_column(base, SystemColumn::UpdatedAt, stamps(vec![20])).unwrap();
445		let base = with_system_column(base, SystemColumn::Time, stamps(vec![30])).unwrap();
446		let base =
447			with_system_column(base, SystemColumn::CommitVersion, Arc::new(UInt64Array::from(vec![5u64])))
448				.unwrap();
449		assert_eq!(created_at(&base).unwrap(), &[DateTime::from_nanos(10)]);
450		assert_eq!(updated_at(&base).unwrap(), &[DateTime::from_nanos(20)]);
451		assert_eq!(time(&base).unwrap(), &[DateTime::from_nanos(30)]);
452		assert_eq!(commit_versions(&base).unwrap(), &[5]);
453	}
454
455	#[test]
456	fn a_getter_rejects_a_none_row() {
457		// A slice getter has no way to say none, so a none row read as a value would invent a timestamp.
458		let mixed: ArrayRef =
459			Arc::new(TimestampNanosecondArray::from(vec![Some(1), None]).with_timezone("+00:00"));
460		let with = with_system_column(batch(vec![user("a", vec![1, 2])]), SystemColumn::Time, mixed).unwrap();
461		assert!(with.schema_ref().field_with_name("#time").unwrap().is_nullable());
462		assert!(time(&with).is_err());
463	}
464
465	#[test]
466	fn a_getter_rejects_a_column_of_the_wrong_arrow_type() {
467		// Reinterpreting an int32 column as row numbers would read garbage rows.
468		let (_, wrong) = user("#rownum", vec![1]);
469		let field = Arc::new(Field::new("#rownum", wrong.data_type().clone(), false));
470		let bad = batch(vec![(field, wrong)]);
471		assert!(row_numbers(&bad).is_err());
472	}
473
474	#[test]
475	fn system_columns_are_inserted_after_user_columns_in_all_order() {
476		// Index based readers break if a system column lands between user columns or out of order.
477		let base = batch(vec![user("a", vec![1]), user("b", vec![2])]);
478		let base = with_system_column(base, SystemColumn::Time, stamps(vec![3])).unwrap();
479		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![1u64])))
480			.unwrap();
481		let base = with_system_column(base, SystemColumn::CreatedAt, stamps(vec![4])).unwrap();
482		assert_eq!(names(&base), ["a", "b", "#rownum", "#created_at", "#time"]);
483	}
484
485	#[test]
486	fn a_present_system_column_is_replaced_in_place() {
487		// Appending a second #rownum would leave two columns under one name.
488		let base = with_system_column(
489			batch(vec![user("a", vec![1])]),
490			SystemColumn::RowNumbers,
491			Arc::new(UInt64Array::from(vec![1u64])),
492		)
493		.unwrap();
494		let replaced =
495			with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![8u64])))
496				.unwrap();
497		assert_eq!(names(&replaced), ["a", "#rownum"]);
498		assert_eq!(row_numbers(&replaced).unwrap(), &[RowNumber(8)]);
499	}
500
501	#[test]
502	fn a_system_column_on_an_empty_batch_sets_the_row_count() {
503		// A named-only #rownum result must keep its rows, not collapse to the empty batch's zero rows.
504		let empty = RecordBatch::new_empty(Arc::new(Schema::empty()));
505		let with = with_system_column(
506			empty,
507			SystemColumn::RowNumbers,
508			Arc::new(UInt64Array::from(vec![1u64, 2, 3])),
509		)
510		.unwrap();
511		assert_eq!(with.num_rows(), 3);
512	}
513
514	#[test]
515	fn a_system_column_of_the_wrong_length_is_rejected() {
516		// A short #rownum would pair rows with the wrong row numbers.
517		let result = with_system_column(
518			batch(vec![user("a", vec![1, 2])]),
519			SystemColumn::RowNumbers,
520			Arc::new(UInt64Array::from(vec![1u64])),
521		);
522		assert!(result.is_err());
523	}
524
525	#[test]
526	fn keep_drops_only_unnamed_system_columns() {
527		// Dropping #rownum or a user column named with a hash, like #op, would lose data the caller needs.
528		let op = Arc::new(Field::new("#op", DataType::Int32, false));
529		let base = batch(vec![user("a", vec![1]), (op, Arc::new(Int32Array::from(vec![1])))]);
530		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![1u64])))
531			.unwrap();
532		let base = with_system_column(base, SystemColumn::CreatedAt, stamps(vec![1])).unwrap();
533		let base = with_system_column(base, SystemColumn::Time, stamps(vec![1])).unwrap();
534		let kept = keep_system_columns(&base, &[SystemColumn::RowNumbers, SystemColumn::Time]).unwrap();
535		assert_eq!(names(&kept), ["a", "#op", "#rownum", "#time"]);
536		assert_eq!(kept.num_rows(), 1);
537	}
538
539	#[test]
540	fn a_bare_name_resolves_to_the_system_column_first() {
541		// A user column named rownum must not shadow #rownum for a bare lookup, as today.
542		let base = batch(vec![user("rownum", vec![5]), user("x", vec![6])]);
543		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![1u64])))
544			.unwrap();
545		assert_eq!(resolve_column(&base, "rownum"), Some(2));
546		assert_eq!(resolve_column(&base, "#rownum"), Some(2));
547		assert_eq!(resolve_column(&base, "x"), Some(1));
548		assert_eq!(resolve_column(&base, "created_at"), None);
549		assert_eq!(resolve_column(&base, "missing"), None);
550	}
551
552	#[test]
553	fn rows_without_any_column_are_rejected() {
554		// A row count with no column to carry it is the hazard; a named-only #rownum result is valid.
555		let counted = RecordBatch::try_new_with_options(
556			Arc::new(Schema::empty()),
557			vec![],
558			&RecordBatchOptions::new().with_row_count(Some(2)),
559		)
560		.unwrap();
561		assert!(check_user_columns(&counted).is_err());
562		let named = with_system_column(
563			RecordBatch::new_empty(Arc::new(Schema::empty())),
564			SystemColumn::RowNumbers,
565			Arc::new(UInt64Array::from(vec![1u64, 2])),
566		)
567		.unwrap();
568		assert!(check_user_columns(&named).is_ok());
569		assert!(check_user_columns(&RecordBatch::new_empty(Arc::new(Schema::empty()))).is_ok());
570	}
571
572	#[test]
573	fn a_hash_prefix_marks_a_system_field() {
574		// Counting #op as a user column would give subscriptions a phantom column.
575		assert!(is_system_field(&Field::new("#rownum", DataType::UInt64, false)));
576		assert!(is_system_field(&Field::new("#op", DataType::Int32, false)));
577		assert!(!is_system_field(&Field::new("rownum", DataType::UInt64, false)));
578	}
579
580	#[test]
581	fn column_view_finds_a_column_by_its_exact_name() {
582		// A lookup that strips the hash would read the user column rownum for #rownum.
583		let base = batch(vec![user("rownum", vec![5])]);
584		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![9u64])))
585			.unwrap();
586		let user_view = column_view(&base, "rownum").unwrap().unwrap();
587		assert_eq!(user_view.get_value(0), Value::Int4(5));
588		let system_view = column_view(&base, "#rownum").unwrap().unwrap();
589		assert_eq!(system_view.get_value(0), Value::Uint8(9));
590		assert!(column_view(&base, "missing").unwrap().is_none());
591	}
592
593	#[test]
594	fn require_getters_fail_naming_the_absent_column() {
595		// An absent column read as empty would pair a non-empty batch with no row numbers or stamps.
596		let plain = batch(vec![user("a", vec![1])]);
597		for (result, name) in [
598			(require_row_numbers(&plain).map(|_| ()), "#rownum"),
599			(require_created_at(&plain).map(|_| ()), "#created_at"),
600			(require_updated_at(&plain).map(|_| ()), "#updated_at"),
601			(require_time(&plain).map(|_| ()), "#time"),
602		] {
603			let message = format!("{:?}", result.unwrap_err());
604			assert!(message.contains(name), "{name} missing from {message}");
605		}
606	}
607
608	#[test]
609	fn require_getters_read_a_present_column() {
610		// A require getter that read the wrong column would stamp rows with another column's values.
611		let base = batch(vec![user("a", vec![1])]);
612		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![4u64])))
613			.unwrap();
614		let base = with_system_column(base, SystemColumn::CreatedAt, stamps(vec![10])).unwrap();
615		let base = with_system_column(base, SystemColumn::UpdatedAt, stamps(vec![20])).unwrap();
616		let base = with_system_column(base, SystemColumn::Time, stamps(vec![30])).unwrap();
617		assert_eq!(require_row_numbers(&base).unwrap(), &[RowNumber(4)]);
618		assert_eq!(require_created_at(&base).unwrap(), &[DateTime::from_nanos(10)]);
619		assert_eq!(require_updated_at(&base).unwrap(), &[DateTime::from_nanos(20)]);
620		assert_eq!(require_time(&base).unwrap(), &[DateTime::from_nanos(30)]);
621	}
622
623	#[test]
624	fn require_getters_still_reject_a_none_row() {
625		// Presence alone must not let a none row through as an invented timestamp.
626		let mixed: ArrayRef =
627			Arc::new(TimestampNanosecondArray::from(vec![Some(1), None]).with_timezone("+00:00"));
628		let with = with_system_column(batch(vec![user("a", vec![1, 2])]), SystemColumn::Time, mixed).unwrap();
629		assert!(require_time(&with).is_err());
630	}
631
632	#[test]
633	fn user_columns_skip_every_hash_column_in_order() {
634		// Counting #rownum or #op as a user column would give callers a phantom column.
635		let op = Arc::new(Field::new("#op", DataType::Int32, false));
636		let base =
637			batch(vec![user("a", vec![1]), (op, Arc::new(Int32Array::from(vec![2]))), user("b", vec![3])]);
638		let base = with_system_column(base, SystemColumn::RowNumbers, Arc::new(UInt64Array::from(vec![1u64])))
639			.unwrap();
640		let names: Vec<&str> = user_columns(&base).map(|(field, _)| field.name().as_str()).collect();
641		assert_eq!(names, ["a", "b"]);
642		let (_, b) = user_columns(&base).nth(1).unwrap();
643		assert_eq!(b.as_any().downcast_ref::<Int32Array>().unwrap().value(0), 3);
644		assert_eq!(user_columns(&RecordBatch::new_empty(Arc::new(Schema::empty()))).count(), 0);
645	}
646}
reifydb_value/value/system_columns.rs

reifydb_value/value/
system_columns.rs
