reifydb-engine 0.4.12

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

pub mod any;
pub mod blob;
pub mod boolean;
pub mod number;
pub mod temporal;
pub mod text;
pub mod uuid;

use reifydb_core::value::column::data::ColumnData;
use reifydb_type::{
	error::TypeError, fragment::LazyFragment, storage::DataBitVec, util::bitvec::BitVec, value::r#type::Type,
};

use crate::{
	Result,
	expression::{cast::uuid::to_uuid, context::EvalContext},
};

pub fn cast_column_data(
	ctx: &EvalContext,
	data: &ColumnData,
	target: Type,
	lazy_fragment: impl LazyFragment + Clone,
) -> Result<ColumnData> {
	// Handle Option-wrapped data: cast the inner data, then re-wrap with the bitvec
	if let ColumnData::Option {
		inner,
		bitvec,
	} = data
	{
		let inner_target = match &target {
			Type::Option(t) => t.as_ref().clone(),
			other => other.clone(),
		};
		let total_len = inner.len();
		let defined_count = DataBitVec::count_ones(bitvec);

		if defined_count == 0 {
			return Ok(ColumnData::none_typed(inner_target, total_len));
		}

		if defined_count < total_len {
			// Compact: keep only defined positions (avoids parsing placeholders like "" for text)
			let mut compacted = inner.as_ref().clone();
			compacted.filter(bitvec)?;

			// Cast only real values
			let mut cast_compacted = cast_column_data(ctx, &compacted, inner_target, lazy_fragment)?;

			// Expand back to full length: defined positions → compacted index, None positions → sentinel
			// (gets type default)
			let sentinel = defined_count;
			let mut expand_indices = Vec::with_capacity(total_len);
			let mut src_idx = 0usize;
			for i in 0..total_len {
				if DataBitVec::get(bitvec, i) {
					expand_indices.push(src_idx);
					src_idx += 1;
				} else {
					expand_indices.push(sentinel);
				}
			}
			cast_compacted.reorder(&expand_indices);

			return Ok(match cast_compacted {
				already @ ColumnData::Option {
					..
				} => already,
				other => ColumnData::Option {
					inner: Box::new(other),
					bitvec: bitvec.clone(),
				},
			});
		}

		// All positions defined — cast directly (fast path)
		let cast_inner = cast_column_data(ctx, inner, inner_target, lazy_fragment)?;
		return Ok(match cast_inner {
			already @ ColumnData::Option {
				..
			} => already,
			other => ColumnData::Option {
				inner: Box::new(other),
				bitvec: bitvec.clone(),
			},
		});
	}
	// Handle bare data -> Option(T) target: cast to inner type, wrap with all-defined bitvec
	if let Type::Option(inner_target) = &target {
		let cast_inner = cast_column_data(ctx, data, *inner_target.clone(), lazy_fragment)?;
		return Ok(match cast_inner {
			already @ ColumnData::Option {
				..
			} => already,
			other => {
				let bitvec = BitVec::repeat(other.len(), true);
				ColumnData::Option {
					inner: Box::new(other),
					bitvec,
				}
			}
		});
	}

	let shape_type = data.get_type();
	if target == shape_type {
		return Ok(data.clone());
	}
	match (&shape_type, &target) {
		(Type::Any, _) => any::from_any(ctx, data, target, lazy_fragment),
		(_, t) if t.is_number() => number::to_number(ctx, data, target, lazy_fragment),
		(_, t) if t.is_blob() => blob::to_blob(data, lazy_fragment),
		(_, t) if t.is_bool() => boolean::to_boolean(data, lazy_fragment),
		(_, t) if t.is_utf8() => text::to_text(data, lazy_fragment),
		(_, t) if t.is_temporal() => temporal::to_temporal(data, target, lazy_fragment),
		(_, Type::IdentityId) => to_uuid(data, target, lazy_fragment),
		(Type::IdentityId, _) => to_uuid(data, target, lazy_fragment),
		(_, t) if t.is_uuid() => to_uuid(data, target, lazy_fragment),
		(source, t) if source.is_uuid() || t.is_uuid() => to_uuid(data, target, lazy_fragment),
		_ => Err(TypeError::UnsupportedCast {
			from: shape_type,
			to: target,
			fragment: lazy_fragment.fragment(),
		}
		.into()),
	}
}

#[cfg(test)]
pub mod tests {
	use reifydb_core::value::column::data::ColumnData;
	use reifydb_rql::expression::{
		CastExpression, ConstantExpression,
		ConstantExpression::Number,
		Expression::{Cast, Constant, Prefix},
		PrefixExpression, PrefixOperator, TypeExpression,
	};
	use reifydb_type::{fragment::Fragment, value::r#type::Type};

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

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

		assert_eq!(*result.data(), ColumnData::int4([42]));
	}

	#[test]
	fn test_cast_negative_integer() {
		let mut ctx = EvalContext::testing();
		let result = evaluate(
			&mut 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: Type::Int4,
				},
			}),
		)
		.unwrap();

		assert_eq!(*result.data(), ColumnData::int4([-42]));
	}

	#[test]
	fn test_cast_negative_min() {
		let mut ctx = EvalContext::testing();
		let result = evaluate(
			&mut 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: Type::Int1,
				},
			}),
		)
		.unwrap();

		assert_eq!(*result.data(), ColumnData::int1([-128]));
	}

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

		assert_eq!(*result.data(), ColumnData::float8([4.2]));
	}

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

		assert_eq!(*result.data(), ColumnData::float4([4.2]));
	}

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

		assert_eq!(*result.data(), ColumnData::float4([-1.1]));
	}

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

		assert_eq!(*result.data(), ColumnData::float8([-1.1]));
	}

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

		assert_eq!(*result.data(), ColumnData::bool([false]));
	}

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

		assert!(result.is_err());

		// Check that the error is the expected CAST_004
		// (invalid_boolean) error
		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 mut ctx = EvalContext::testing();
		let result = evaluate(
			&mut ctx,
			&Cast(CastExpression {
				fragment: Fragment::testing_empty(),
				expression: Box::new(Constant(ConstantExpression::Bool {
					fragment: Fragment::internal("true"),
				})),
				to: TypeExpression {
					fragment: Fragment::testing_empty(),
					ty: Type::Date,
				},
			}),
		);

		assert!(result.is_err());

		// Check that the error is the expected CAST_001
		// (unsupported_cast) error
		let err = result.unwrap_err();
		let diagnostic = err.0;
		assert_eq!(diagnostic.code, "CAST_001");
	}

	#[test]
	fn test_cast_text_to_decimal() {
		let mut ctx = EvalContext::testing();
		let result = evaluate(
			&mut 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: Type::Decimal,
				},
			}),
		)
		.unwrap();

		if let ColumnData::Decimal {
			container,
			..
		} = result.data()
		{
			assert_eq!(container.len(), 1);
			assert!(container.is_defined(0));
			let value = &container[0];
			assert_eq!(value.to_string(), "123.456789");
		} else {
			panic!("Expected Decimal column data");
		}
	}
}