reifydb-evaluate 0.10.0

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

use arrow_array::ArrayRef;
use arrow_schema::FieldRef;
use reifydb_core::{
	expression::Expression,
	value::column::{cast::cast_column_data, write::check_digest_write},
};
use reifydb_value::{
	fragment::LazyFragment,
	value::{
		Value,
		column_view::ColumnView,
		constraint::{TypeConstraint, precision::Precision},
		value_type::ValueType,
	},
};

use crate::{Result, expression::context::EvalContext, lower::LoweredExpr};

pub fn evaluate(ctx: &EvalContext, expr: &Expression) -> Result<(FieldRef, ArrayRef)> {
	let column = LoweredExpr::new(expr.clone(), "evaluate").evaluate(ctx)?;

	if let Some(ty) = ctx.target.as_ref().map(|c| c.column_type()) {
		let view = ColumnView::try_from(&column)?;
		check_digest_write(&view, &ty, expr.lazy_fragment())?;
		cast_for_write(ctx, &view, ty, &expr.lazy_fragment())
	} else {
		Ok(column)
	}
}

pub fn cast_for_write(
	ctx: &EvalContext,
	view: &ColumnView<'_>,
	target: ValueType,
	fragment: impl LazyFragment + Clone,
) -> Result<(FieldRef, ArrayRef)> {
	refuse_lost_scale(view, &target, fragment.clone())?;
	cast_column_data(ctx, view, target, fragment)
}

pub fn loses_scale(value: &Value, target: &ValueType) -> bool {
	matches!(
		(value, target.inner_type()),
		(Value::Decimal(decimal), ValueType::Decimal { scale, .. })
			if decimal.scale() > scale.value() && decimal.rescale(scale.value()).is_none()
	)
}

fn refuse_lost_scale(data: &ColumnView, target: &ValueType, fragment: impl LazyFragment) -> Result<()> {
	match (data.get_type().inner_type(), target.inner_type()) {
		(
			ValueType::Decimal {
				scale: from,
				..
			},
			ValueType::Decimal {
				scale: to,
				..
			},
		) if from > to => data.iter().filter(|value| loses_scale(value, target)).try_for_each(|mut value| {
			let exact = TypeConstraint::unconstrained(ValueType::decimal(Precision::MAX, *to));
			exact.coerce(&mut value).map_err(|mut e| {
				e.0.fragment = fragment.fragment();
				e
			})
		}),
		_ => Ok(()),
	}
}

#[cfg(test)]
pub mod tests {
	use reifydb_core::{
		expression::{
			CastExpression, ConstantExpression,
			ConstantExpression::Number,
			Expression::{Cast, Constant, Prefix},
			PrefixExpression, PrefixOperator, TypeExpression,
		},
		value::column::factory::{bool, float4, float8, int1, int4},
	};
	use reifydb_value::{
		fragment::Fragment,
		value::{
			column_view::{ColumnView, ViewData},
			container::decimal_array::{decimal_at, decimals},
			value_type::ValueType,
		},
	};

	use crate::expression::{context::EvalContext, eval::evaluate};

	#[test]
	fn test_cast_integer() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(Number {
					fragment: Fragment::internal("42"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Int4,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), int4("", [42]).1.as_ref());
	}

	#[test]
	fn test_cast_negative_integer() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Prefix(PrefixExpression {
					operator: PrefixOperator::Minus(Fragment::testing_empty()),
					expression: Box::new(Constant(Number {
						fragment: Fragment::internal("42"),
					})),
					fragment: Fragment::testing_empty(),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Int4,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), int4("", [-42]).1.as_ref());
	}

	#[test]
	fn test_cast_negative_min() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Prefix(PrefixExpression {
					operator: PrefixOperator::Minus(Fragment::testing_empty()),
					expression: Box::new(Constant(Number {
						fragment: Fragment::internal("128"),
					})),
					fragment: Fragment::testing_empty(),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Int1,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), int1("", [-128]).1.as_ref());
	}

	#[test]
	fn test_cast_float_8() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(Number {
					fragment: Fragment::internal("4.2"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Float8,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), float8("", [4.2]).1.as_ref());
	}

	#[test]
	fn test_cast_float_4() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(Number {
					fragment: Fragment::internal("4.2"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Float4,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), float4("", [4.2]).1.as_ref());
	}

	#[test]
	fn test_cast_negative_float_4() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(Number {
					fragment: Fragment::internal("-1.1"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Float4,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), float4("", [-1.1]).1.as_ref());
	}

	#[test]
	fn test_cast_negative_float_8() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(Number {
					fragment: Fragment::internal("-1.1"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Float8,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), float8("", [-1.1]).1.as_ref());
	}

	#[test]
	fn test_cast_string_to_bool() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(ConstantExpression::Text {
					fragment: Fragment::internal("0"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Boolean,
					constraint: None,
				},
			}),
		)
		.unwrap();

		assert_eq!(result.1.as_ref(), bool("", [false]).1.as_ref());
	}

	#[test]
	fn test_cast_string_neg_one_to_bool_should_fail() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(ConstantExpression::Text {
					fragment: Fragment::internal("-1"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Boolean,
					constraint: None,
				},
			}),
		);

		assert!(result.is_err());

		// The cast code must carry the underlying reason as its cause, not flatten to one code.
		let err = result.unwrap_err();
		let diagnostic = err.0;
		assert_eq!(diagnostic.code, "CAST_004");
		assert!(diagnostic.cause.is_some());
		let cause = diagnostic.cause.unwrap();
		assert_eq!(cause.code, "BOOLEAN_003"); // invalid_number_boolean
	}

	#[test]
	fn test_cast_boolean_to_date_should_fail() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(ConstantExpression::Bool {
					fragment: Fragment::internal("true"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::Date,
					constraint: None,
				},
			}),
		);

		assert!(result.is_err());

		// An impossible cast pair is unsupported (CAST_001), not a value-level failure.
		let err = result.unwrap_err();
		let diagnostic = err.0;
		assert_eq!(diagnostic.code, "CAST_001");
	}

	#[test]
	fn test_cast_text_to_decimal() {
		let ctx = EvalContext::testing();
		let result = evaluate(
			&ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(ConstantExpression::Text {
					fragment: Fragment::internal("123.456789"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: ValueType::DECIMAL,
					constraint: None,
				},
			}),
		)
		.unwrap();

		if let ViewData::Decimal(container) = &ColumnView::try_from(&result).unwrap().data {
			assert_eq!(container.len(), 1);
			assert!(decimal_at(container, 0).is_some());
			let value = &decimals(container)[0];
			assert_eq!(value.to_string(), "123.4567890000");
		} else {
			panic!("Expected Decimal column data");
		}
	}
}