reifydb-core 0.10.0

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

use arrow_array::ArrayRef;
use arrow_schema::FieldRef;
use reifydb_value::{
	Result,
	error::{BinaryOp, TypeError},
	fragment::LazyFragment,
	reifydb_assertions,
	value::{
		column_view::{ColumnView, ViewData},
		container::{decimal_array::decimals, wide_int_array::wides},
		is::IsNumber,
		number::{promote::Promote, safe::remainder::SafeRemainder},
		value_type::{ValueType, get::GetType},
	},
};

use crate::value::column::{
	arith::{
		ArithOp, arith_target, context::ArithContext, option::arith_op_unwrap_option, output_len, row_of,
		scalar::FitFamily,
	},
	builder::ColumnBuilder,
	push::Push,
};

pub fn rem_columns(
	ctx: &ArithContext,
	left: &(FieldRef, ArrayRef),
	right: &(FieldRef, ArrayRef),
	fragment: impl LazyFragment + Copy,
) -> Result<(FieldRef, ArrayRef)> {
	arith_op_unwrap_option(left, right, fragment.fragment(), |left, right| {
		let (left, right) = (ColumnView::try_from(left)?, ColumnView::try_from(right)?);
		let target = arith_target(ArithOp::Rem, left.get_type(), right.get_type());

		dispatch_arith!(
			&left.data, &right.data;
			fixed: rem_numeric, arb: rem_numeric_clone (ctx, target, fragment);

			_ => Err(TypeError::BinaryOperatorNotApplicable {
				operator: BinaryOp::Rem,
				left: left.get_type(),
				right: right.get_type(),
				fragment: fragment.fragment(),
			}.into()),
		)
	})
}

fn rem_numeric<L, R>(
	ctx: &ArithContext,
	l: &[L],
	r: &[R],
	target: ValueType,
	fragment: impl LazyFragment + Copy,
) -> Result<(FieldRef, ArrayRef)>
where
	L: GetType + Promote<R> + IsNumber,
	R: GetType + IsNumber,
	<L as Promote<R>>::Output: IsNumber,
	<L as Promote<R>>::Output: SafeRemainder,
	ColumnBuilder: Push<<L as Promote<R>>::Output>,
{
	reifydb_assertions! {
		assert!(l.len() == r.len() || l.len() == 1 || r.len() == 1);
	}

	let mut data = ColumnBuilder::with_capacity(target, output_len(l.len(), r.len()));
	for i in 0..output_len(l.len(), r.len()) {
		if let Some(value) = ctx.remainder(&l[row_of(l.len(), i)], &r[row_of(r.len(), i)], fragment)? {
			data.push(value);
		} else {
			data.push_none()
		}
	}
	Ok(data.finish(fragment.fragment().text()))
}

fn rem_numeric_clone<L, R>(
	ctx: &ArithContext,
	l: &[L],
	r: &[R],
	target: ValueType,
	fragment: impl LazyFragment + Copy,
) -> Result<(FieldRef, ArrayRef)>
where
	L: Clone + GetType + Promote<R> + IsNumber,
	R: Clone + GetType + IsNumber,
	<L as Promote<R>>::Output: IsNumber + FitFamily,
	<L as Promote<R>>::Output: SafeRemainder,
	ColumnBuilder: Push<<L as Promote<R>>::Output>,
{
	reifydb_assertions! {
		assert!(l.len() == r.len() || l.len() == 1 || r.len() == 1);
	}

	let mut data = ColumnBuilder::with_capacity(target.clone(), output_len(l.len(), r.len()));
	for i in 0..output_len(l.len(), r.len()) {
		let l_clone = l[row_of(l.len(), i)].clone();
		let r_clone = r[row_of(r.len(), i)].clone();
		match ctx.remainder(&l_clone, &r_clone, fragment)? {
			Some(value) => match ctx.fit_family(value, &target, fragment)? {
				Some(value) => data.push(value),
				None => data.push_none(),
			},
			None => data.push_none(),
		}
	}
	Ok(data.finish(fragment.fragment().text()))
}