reifydb_core/value/column/
columns.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2025 ReifyDB
3
4use std::{
5	collections::HashMap,
6	hash::Hash,
7	mem,
8	ops::{Deref, Index, IndexMut},
9};
10
11use indexmap::IndexMap;
12use reifydb_type::{
13	Result,
14	fragment::Fragment,
15	util::cowvec::CowVec,
16	value::{Value, constraint::Constraint, row_number::RowNumber, r#type::Type},
17};
18
19use crate::{
20	encoded::schema::{RowSchema, RowSchemaField},
21	interface::{
22		catalog::{table::Table, view::View},
23		resolved::{ResolvedRingBuffer, ResolvedTable, ResolvedView},
24	},
25	row::Row,
26	value::column::{Column, ColumnData, headers::ColumnHeaders},
27};
28
29#[derive(Debug, Clone)]
30pub struct Columns {
31	pub row_numbers: CowVec<RowNumber>,
32	pub columns: CowVec<Column>,
33}
34
35impl Deref for Columns {
36	type Target = [Column];
37
38	fn deref(&self) -> &Self::Target {
39		self.columns.deref()
40	}
41}
42
43impl Index<usize> for Columns {
44	type Output = Column;
45
46	fn index(&self, index: usize) -> &Self::Output {
47		self.columns.index(index)
48	}
49}
50
51impl IndexMut<usize> for Columns {
52	fn index_mut(&mut self, index: usize) -> &mut Self::Output {
53		&mut self.columns.make_mut()[index]
54	}
55}
56
57impl Columns {
58	/// Create a 1-column, 1-row Columns from a single Value.
59	/// Used to store scalar values inside `Variable::Scalar(Columns)`.
60	pub fn scalar(value: Value) -> Self {
61		let data = match value {
62			Value::None {
63				..
64			} => ColumnData::none_typed(Type::Boolean, 1),
65			Value::Boolean(v) => ColumnData::bool([v]),
66			Value::Float4(v) => ColumnData::float4([v.into()]),
67			Value::Float8(v) => ColumnData::float8([v.into()]),
68			Value::Int1(v) => ColumnData::int1([v]),
69			Value::Int2(v) => ColumnData::int2([v]),
70			Value::Int4(v) => ColumnData::int4([v]),
71			Value::Int8(v) => ColumnData::int8([v]),
72			Value::Int16(v) => ColumnData::int16([v]),
73			Value::Utf8(v) => ColumnData::utf8([v]),
74			Value::Uint1(v) => ColumnData::uint1([v]),
75			Value::Uint2(v) => ColumnData::uint2([v]),
76			Value::Uint4(v) => ColumnData::uint4([v]),
77			Value::Uint8(v) => ColumnData::uint8([v]),
78			Value::Uint16(v) => ColumnData::uint16([v]),
79			Value::Date(v) => ColumnData::date([v]),
80			Value::DateTime(v) => ColumnData::datetime([v]),
81			Value::Time(v) => ColumnData::time([v]),
82			Value::Duration(v) => ColumnData::duration([v]),
83			Value::IdentityId(v) => ColumnData::identity_id([v]),
84			Value::Uuid4(v) => ColumnData::uuid4([v]),
85			Value::Uuid7(v) => ColumnData::uuid7([v]),
86			Value::Blob(v) => ColumnData::blob([v]),
87			Value::Int(v) => ColumnData::int(vec![v]),
88			Value::Uint(v) => ColumnData::uint(vec![v]),
89			Value::Decimal(v) => ColumnData::decimal(vec![v]),
90			Value::DictionaryId(v) => ColumnData::dictionary_id(vec![v]),
91			Value::Any(v) => ColumnData::any(vec![v]),
92			Value::Type(v) => ColumnData::any(vec![Box::new(Value::Type(v))]),
93			Value::List(v) => ColumnData::any(vec![Box::new(Value::List(v))]),
94			Value::Record(v) => ColumnData::any(vec![Box::new(Value::Record(v))]),
95			Value::Tuple(v) => ColumnData::any(vec![Box::new(Value::Tuple(v))]),
96		};
97		let column = Column {
98			name: Fragment::internal("value"),
99			data,
100		};
101		Self {
102			row_numbers: CowVec::new(Vec::new()),
103			columns: CowVec::new(vec![column]),
104		}
105	}
106
107	/// Extract the single value from a 1-column, 1-row Columns.
108	/// Panics if the Columns does not have exactly 1 column and 1 row.
109	pub fn scalar_value(&self) -> Value {
110		debug_assert_eq!(self.len(), 1, "scalar_value() requires exactly 1 column, got {}", self.len());
111		debug_assert_eq!(
112			self.row_count(),
113			1,
114			"scalar_value() requires exactly 1 row, got {}",
115			self.row_count()
116		);
117		self.columns[0].data().get_value(0)
118	}
119
120	pub fn new(columns: Vec<Column>) -> Self {
121		let n = columns.first().map_or(0, |c| c.data().len());
122		assert!(columns.iter().all(|c| c.data().len() == n));
123
124		Self {
125			row_numbers: CowVec::new(Vec::new()),
126			columns: CowVec::new(columns),
127		}
128	}
129
130	pub fn with_row_numbers(columns: Vec<Column>, row_numbers: Vec<RowNumber>) -> Self {
131		let n = columns.first().map_or(0, |c| c.data().len());
132		assert!(columns.iter().all(|c| c.data().len() == n));
133		assert_eq!(row_numbers.len(), n, "row_numbers length must match column data length");
134
135		Self {
136			row_numbers: CowVec::new(row_numbers),
137			columns: CowVec::new(columns),
138		}
139	}
140
141	pub fn single_row<'b>(rows: impl IntoIterator<Item = (&'b str, Value)>) -> Columns {
142		let mut columns = Vec::new();
143		let mut index = HashMap::new();
144
145		for (idx, (name, value)) in rows.into_iter().enumerate() {
146			let data = match value {
147				Value::None {
148					..
149				} => ColumnData::none_typed(Type::Boolean, 1),
150				Value::Boolean(v) => ColumnData::bool([v]),
151				Value::Float4(v) => ColumnData::float4([v.into()]),
152				Value::Float8(v) => ColumnData::float8([v.into()]),
153				Value::Int1(v) => ColumnData::int1([v]),
154				Value::Int2(v) => ColumnData::int2([v]),
155				Value::Int4(v) => ColumnData::int4([v]),
156				Value::Int8(v) => ColumnData::int8([v]),
157				Value::Int16(v) => ColumnData::int16([v]),
158				Value::Utf8(v) => ColumnData::utf8([v.clone()]),
159				Value::Uint1(v) => ColumnData::uint1([v]),
160				Value::Uint2(v) => ColumnData::uint2([v]),
161				Value::Uint4(v) => ColumnData::uint4([v]),
162				Value::Uint8(v) => ColumnData::uint8([v]),
163				Value::Uint16(v) => ColumnData::uint16([v]),
164				Value::Date(v) => ColumnData::date([v.clone()]),
165				Value::DateTime(v) => ColumnData::datetime([v.clone()]),
166				Value::Time(v) => ColumnData::time([v.clone()]),
167				Value::Duration(v) => ColumnData::duration([v.clone()]),
168				Value::IdentityId(v) => ColumnData::identity_id([v]),
169				Value::Uuid4(v) => ColumnData::uuid4([v]),
170				Value::Uuid7(v) => ColumnData::uuid7([v]),
171				Value::Blob(v) => ColumnData::blob([v.clone()]),
172				Value::Int(v) => ColumnData::int(vec![v]),
173				Value::Uint(v) => ColumnData::uint(vec![v]),
174				Value::Decimal(v) => ColumnData::decimal(vec![v]),
175				Value::DictionaryId(v) => ColumnData::dictionary_id(vec![v]),
176				Value::Type(t) => ColumnData::any(vec![Box::new(Value::Type(t))]),
177				Value::Any(v) => ColumnData::any(vec![v]),
178				Value::List(v) => ColumnData::any(vec![Box::new(Value::List(v))]),
179				Value::Record(v) => ColumnData::any(vec![Box::new(Value::Record(v))]),
180				Value::Tuple(v) => ColumnData::any(vec![Box::new(Value::Tuple(v))]),
181			};
182
183			let column = Column {
184				name: Fragment::internal(name.to_string()),
185				data,
186			};
187			index.insert(name, idx);
188			columns.push(column);
189		}
190
191		Self {
192			row_numbers: CowVec::new(Vec::new()),
193			columns: CowVec::new(columns),
194		}
195	}
196
197	pub fn apply_headers(&mut self, headers: &ColumnHeaders) {
198		// Apply the column names from headers to this Columns instance
199		for (i, name) in headers.columns.iter().enumerate() {
200			if i < self.len() {
201				let column = &mut self[i];
202				let data = mem::replace(column.data_mut(), ColumnData::none_typed(Type::Boolean, 0));
203
204				*column = Column {
205					name: name.clone(),
206					data,
207				};
208			}
209		}
210	}
211}
212
213impl Columns {
214	/// Get the row number (for single-row Columns). Panics if Columns has 0 or multiple rows.
215	pub fn number(&self) -> RowNumber {
216		assert_eq!(self.row_count(), 1, "number() requires exactly 1 row, got {}", self.row_count());
217		if self.row_numbers.is_empty() {
218			RowNumber(0)
219		} else {
220			self.row_numbers[0]
221		}
222	}
223
224	pub fn shape(&self) -> (usize, usize) {
225		let row_count = if !self.row_numbers.is_empty() {
226			self.row_numbers.len()
227		} else {
228			self.get(0).map(|c| c.data().len()).unwrap_or(0)
229		};
230		(row_count, self.len())
231	}
232
233	pub fn into_iter(self) -> impl Iterator<Item = Column> {
234		self.columns.into_iter()
235	}
236
237	pub fn is_empty(&self) -> bool {
238		self.shape().0 == 0
239	}
240
241	pub fn row(&self, i: usize) -> Vec<Value> {
242		self.iter().map(|c| c.data().get_value(i)).collect()
243	}
244
245	pub fn column(&self, name: &str) -> Option<&Column> {
246		self.iter().find(|col| col.name().text() == name)
247	}
248
249	pub fn row_count(&self) -> usize {
250		if !self.row_numbers.is_empty() {
251			self.row_numbers.len()
252		} else {
253			self.first().map_or(0, |col| col.data().len())
254		}
255	}
256
257	pub fn get_row(&self, index: usize) -> Vec<Value> {
258		self.iter().map(|col| col.data().get_value(index)).collect()
259	}
260}
261
262impl Column {
263	pub fn extend(&mut self, other: Column) -> Result<()> {
264		self.data_mut().extend(other.data().clone())
265	}
266}
267
268impl Columns {
269	pub fn from_rows(names: &[&str], result_rows: &[Vec<Value>]) -> Self {
270		let column_count = names.len();
271
272		let mut columns: Vec<Column> = names
273			.iter()
274			.map(|name| Column {
275				name: Fragment::internal(name.to_string()),
276				data: ColumnData::none_typed(Type::Boolean, 0),
277			})
278			.collect();
279
280		for row in result_rows {
281			assert_eq!(row.len(), column_count, "row length does not match column count");
282			for (i, value) in row.iter().enumerate() {
283				columns[i].data_mut().push_value(value.clone());
284			}
285		}
286
287		Columns::new(columns)
288	}
289
290	pub fn from_rows_with_row_numbers(
291		names: &[&str],
292		result_rows: &[Vec<Value>],
293		row_numbers: Vec<RowNumber>,
294	) -> Self {
295		let column_count = names.len();
296
297		let mut columns: Vec<Column> = names
298			.iter()
299			.map(|name| Column {
300				name: Fragment::internal(name.to_string()),
301				data: ColumnData::none_typed(Type::Boolean, 0),
302			})
303			.collect();
304
305		for row in result_rows {
306			assert_eq!(row.len(), column_count, "row length does not match column count");
307			for (i, value) in row.iter().enumerate() {
308				columns[i].data_mut().push_value(value.clone());
309			}
310		}
311
312		Columns::with_row_numbers(columns, row_numbers)
313	}
314}
315
316impl Columns {
317	pub fn empty() -> Self {
318		Self {
319			row_numbers: CowVec::new(vec![]),
320			columns: CowVec::new(vec![]),
321		}
322	}
323
324	pub fn from_resolved_table(table: &ResolvedTable) -> Self {
325		Self::from_table(table.def())
326	}
327
328	/// Create empty Columns (0 rows) with schema from a Table
329	pub fn from_table(table: &Table) -> Self {
330		let columns: Vec<Column> = table
331			.columns
332			.iter()
333			.map(|col| Column {
334				name: Fragment::internal(&col.name),
335				data: ColumnData::with_capacity(col.constraint.get_type(), 0),
336			})
337			.collect();
338
339		Self {
340			row_numbers: CowVec::new(Vec::new()),
341			columns: CowVec::new(columns),
342		}
343	}
344
345	/// Create empty Columns (0 rows) with schema from a View
346	pub fn from_view(view: &View) -> Self {
347		let columns: Vec<Column> = view
348			.columns()
349			.iter()
350			.map(|col| Column {
351				name: Fragment::internal(&col.name),
352				data: ColumnData::with_capacity(col.constraint.get_type(), 0),
353			})
354			.collect();
355
356		Self {
357			row_numbers: CowVec::new(Vec::new()),
358			columns: CowVec::new(columns),
359		}
360	}
361
362	pub fn from_ringbuffer(ringbuffer: &ResolvedRingBuffer) -> Self {
363		let _source = ringbuffer.clone();
364
365		let columns: Vec<Column> = ringbuffer
366			.columns()
367			.iter()
368			.map(|col| {
369				let column_ident = Fragment::internal(&col.name);
370				Column {
371					name: column_ident,
372					data: ColumnData::with_capacity(col.constraint.get_type(), 0),
373				}
374			})
375			.collect();
376
377		Self {
378			row_numbers: CowVec::new(Vec::new()),
379			columns: CowVec::new(columns),
380		}
381	}
382
383	pub fn from_resolved_view(view: &ResolvedView) -> Self {
384		Self::from_view(view.def())
385	}
386}
387
388impl Columns {
389	/// Extract a subset of rows by indices, returning a new Columns
390	pub fn extract_by_indices(&self, indices: &[usize]) -> Columns {
391		if indices.is_empty() {
392			return Columns::empty();
393		}
394
395		let new_columns: Vec<Column> = self
396			.columns
397			.iter()
398			.map(|col| {
399				let mut new_data = ColumnData::with_capacity(col.data().get_type(), indices.len());
400				for &idx in indices {
401					new_data.push_value(col.data().get_value(idx));
402				}
403				Column {
404					name: col.name.clone(),
405					data: new_data,
406				}
407			})
408			.collect();
409
410		if self.row_numbers.is_empty() {
411			Columns::new(new_columns)
412		} else {
413			let new_row_numbers: Vec<RowNumber> = indices.iter().map(|&i| self.row_numbers[i]).collect();
414			Columns::with_row_numbers(new_columns, new_row_numbers)
415		}
416	}
417
418	/// Extract a single row by index, returning a new Columns with 1 row
419	pub fn extract_row(&self, index: usize) -> Columns {
420		self.extract_by_indices(&[index])
421	}
422
423	/// Project to a subset of columns by name, preserving the order of the provided names.
424	/// Columns not found in self are silently skipped.
425	pub fn project_by_names(&self, names: &[String]) -> Columns {
426		let new_columns: Vec<Column> = names
427			.iter()
428			.filter_map(|name| self.columns.iter().find(|c| c.name().text() == name.as_str()).cloned())
429			.collect();
430
431		if new_columns.is_empty() {
432			return Columns::empty();
433		}
434
435		Columns {
436			row_numbers: self.row_numbers.clone(),
437			columns: CowVec::new(new_columns),
438		}
439	}
440
441	/// Partition Columns into groups based on keys (one key per row).
442	/// Returns an IndexMap preserving insertion order of first occurrence.
443	pub fn partition_by_keys<K: Hash + Eq + Clone>(&self, keys: &[K]) -> IndexMap<K, Columns> {
444		assert_eq!(keys.len(), self.row_count(), "keys length must match row count");
445
446		// Group indices by key
447		let mut key_to_indices: IndexMap<K, Vec<usize>> = IndexMap::new();
448		for (idx, key) in keys.iter().enumerate() {
449			key_to_indices.entry(key.clone()).or_default().push(idx);
450		}
451
452		// Convert to Columns
453		key_to_indices.into_iter().map(|(key, indices)| (key, self.extract_by_indices(&indices))).collect()
454	}
455
456	/// Create Columns from a Row by decoding its encoded values
457	pub fn from_row(row: &Row) -> Self {
458		let mut columns = Vec::new();
459
460		for (idx, field) in row.schema.fields().iter().enumerate() {
461			let value = row.schema.get_value(&row.encoded, idx);
462
463			// Use the field type for the column data, handling undefined values
464			let column_type = if matches!(value, Value::None { .. }) {
465				field.constraint.get_type()
466			} else {
467				value.get_type()
468			};
469
470			let mut data = if column_type.is_option() {
471				ColumnData::none_typed(column_type.clone(), 0)
472			} else {
473				ColumnData::with_capacity(column_type.clone(), 1)
474			};
475			data.push_value(value);
476
477			if column_type == Type::DictionaryId {
478				if let ColumnData::DictionaryId(container) = &mut data {
479					if let Some(Constraint::Dictionary(dict_id, _)) = field.constraint.constraint()
480					{
481						container.set_dictionary_id(*dict_id);
482					}
483				}
484			}
485
486			let name = row.schema.get_field_name(idx).expect("RowSchema missing name for field");
487
488			columns.push(Column {
489				name: Fragment::internal(name),
490				data,
491			});
492		}
493
494		Self {
495			row_numbers: CowVec::new(vec![row.number]),
496			columns: CowVec::new(columns),
497		}
498	}
499
500	/// Convert Columns back to a Row (assumes single row)
501	/// Panics if Columns contains more than 1 row
502	pub fn to_single_row(&self) -> Row {
503		assert_eq!(self.row_count(), 1, "to_row() requires exactly 1 row, got {}", self.row_count());
504		assert_eq!(
505			self.row_numbers.len(),
506			1,
507			"to_row() requires exactly 1 row number, got {}",
508			self.row_numbers.len()
509		);
510
511		let row_number = self.row_numbers.first().unwrap().clone();
512
513		// Build schema fields for the layout
514		let fields: Vec<RowSchemaField> = self
515			.columns
516			.iter()
517			.map(|col| RowSchemaField::unconstrained(col.name().text().to_string(), col.data().get_type()))
518			.collect();
519
520		let layout = RowSchema::new(fields);
521		let mut encoded = layout.allocate();
522
523		// Get values and set them
524		let values: Vec<Value> = self.columns.iter().map(|col| col.data().get_value(0)).collect();
525		layout.set_values(&mut encoded, &values);
526
527		Row {
528			number: row_number,
529			encoded,
530			schema: layout,
531		}
532	}
533}
534
535#[cfg(test)]
536pub mod tests {
537	use reifydb_type::value::{date::Date, datetime::DateTime, duration::Duration, time::Time};
538
539	use super::*;
540
541	#[test]
542	fn test_single_row_temporal_types() {
543		let date = Date::from_ymd(2025, 1, 15).unwrap();
544		let datetime = DateTime::from_timestamp(1642694400).unwrap();
545		let time = Time::from_hms(14, 30, 45).unwrap();
546		let duration = Duration::from_days(30).unwrap();
547
548		let columns = Columns::single_row([
549			("date_col", Value::Date(date.clone())),
550			("datetime_col", Value::DateTime(datetime.clone())),
551			("time_col", Value::Time(time.clone())),
552			("interval_col", Value::Duration(duration.clone())),
553		]);
554
555		assert_eq!(columns.len(), 4);
556		assert_eq!(columns.shape(), (1, 4));
557
558		// Check that the values are correctly stored
559		assert_eq!(columns.column("date_col").unwrap().data().get_value(0), Value::Date(date));
560		assert_eq!(columns.column("datetime_col").unwrap().data().get_value(0), Value::DateTime(datetime));
561		assert_eq!(columns.column("time_col").unwrap().data().get_value(0), Value::Time(time));
562		assert_eq!(columns.column("interval_col").unwrap().data().get_value(0), Value::Duration(duration));
563	}
564
565	#[test]
566	fn test_single_row_mixed_types() {
567		let date = Date::from_ymd(2025, 7, 15).unwrap();
568		let time = Time::from_hms(9, 15, 30).unwrap();
569
570		let columns = Columns::single_row([
571			("bool_col", Value::Boolean(true)),
572			("int_col", Value::Int4(42)),
573			("str_col", Value::Utf8("hello".to_string())),
574			("date_col", Value::Date(date.clone())),
575			("time_col", Value::Time(time.clone())),
576			("none_col", Value::none()),
577		]);
578
579		assert_eq!(columns.len(), 6);
580		assert_eq!(columns.shape(), (1, 6));
581
582		// Check all values are correctly stored
583		assert_eq!(columns.column("bool_col").unwrap().data().get_value(0), Value::Boolean(true));
584		assert_eq!(columns.column("int_col").unwrap().data().get_value(0), Value::Int4(42));
585		assert_eq!(columns.column("str_col").unwrap().data().get_value(0), Value::Utf8("hello".to_string()));
586		assert_eq!(columns.column("date_col").unwrap().data().get_value(0), Value::Date(date));
587		assert_eq!(columns.column("time_col").unwrap().data().get_value(0), Value::Time(time));
588		assert_eq!(columns.column("none_col").unwrap().data().get_value(0), Value::none());
589	}
590
591	#[test]
592	fn test_single_row_normal_column_names_work() {
593		let columns = Columns::single_row([("normal_column", Value::Int4(42))]);
594		assert_eq!(columns.len(), 1);
595		assert_eq!(columns.column("normal_column").unwrap().data().get_value(0), Value::Int4(42));
596	}
597}
reifydb_core/value/column/columns.rs

reifydb_core/value/column/
columns.rs
