reifydb-engine 0.10.1

Query execution and processing engine for ReifyDB
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

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

use arrow_array::ArrayRef;
use arrow_schema::FieldRef;
use reifydb_core::{
	interface::catalog::column::Column,
	value::{
		batch::empty_batch,
		column::{
			cast::cast_column_data,
			factory::from_one,
			write::{check_digest_write, check_digest_write_type},
		},
	},
};
use reifydb_evaluate::{expression::context::EvalContext, stack::SymbolTable};
use reifydb_routine_abi::registry::Routines;
use reifydb_runtime::context::{RuntimeContext, clock::Clock};
use reifydb_value::{
	error::Error,
	fragment::Fragment,
	params::Params,
	value::{Value, column_view::ColumnView, identity::IdentityId},
};

use crate::Result;

pub(super) struct RowCoercer {
	runtime_context: RuntimeContext,
	routines: Routines,
	symbols: SymbolTable,
	identity: IdentityId,
}

impl RowCoercer {
	pub(super) fn new(identity: IdentityId) -> Self {
		Self {
			runtime_context: RuntimeContext::with_clock(Clock::Real),
			routines: Routines::empty(),
			symbols: SymbolTable::new(),
			identity,
		}
	}

	pub(super) fn coerce(&self, value: Value, column: &Column, source_name: &str, row_idx: usize) -> Result<Value> {
		if matches!(value, Value::None { .. }) {
			return Ok(value);
		}
		self.cast(value, column).map_err(|e| with_row_note(e, source_name, row_idx))
	}

	pub(super) fn cast_column(&self, view: &ColumnView<'_>, column: &Column) -> Result<(FieldRef, ArrayRef)> {
		let target = column.constraint.get_type();
		let fragment = || Fragment::internal(&column.name);
		check_digest_write(view, &target, fragment)?;
		cast_column_data(&self.context(view.len()), view, target.inner_type().clone(), fragment)
	}

	fn cast(&self, value: Value, column: &Column) -> Result<Value> {
		let target = column.constraint.get_type();
		let fragment = || Fragment::internal(&column.name);
		check_digest_write_type(&value.get_type(), &target, fragment)?;
		let cast_target = target.inner_type().clone();
		if value.get_type() == cast_target {
			return Ok(value);
		}
		let column = from_one("value", value);
		let casted =
			cast_column_data(&self.context(1), &ColumnView::try_from(&column)?, cast_target, fragment)?;
		Ok(ColumnView::try_from(&casted)?.get_value(0))
	}

	fn context(&self, row_count: usize) -> EvalContext<'_> {
		EvalContext {
			params: &Params::None,
			symbols: &self.symbols,
			routines: &self.routines,
			runtime_context: &self.runtime_context,
			identity: self.identity,
			batch: empty_batch(),
			row_count,
			target: None,
			take: None,
		}
	}
}

pub(crate) fn with_row_note(mut error: Error, source_name: &str, row_idx: usize) -> Error {
	error.0.notes.push(format!("row {} of the bulk insert into `{}`", row_idx + 1, source_name));
	error
}