reifydb-engine 0.10.0

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

use std::sync::Arc;

use arrow_array::RecordBatch;
use reifydb_core::internal_error;
use reifydb_evaluate::stack::{SymbolTable, Variable};
use reifydb_transaction::transaction::Transaction;
use reifydb_value::params::Params;

use crate::{
	Result,
	flow_sync::{load_flow_diff_need, sync_transactional_views},
	vm::{services::Services, vm::Vm},
};

fn reject_query_txn(tx: &Transaction<'_>) -> Result<()> {
	if matches!(tx, Transaction::Query(_)) {
		return Err(internal_error!("Mutation operations cannot be executed in a query transaction"));
	}
	Ok(())
}

impl<'a> Vm<'a> {
	pub(crate) fn exec_dml_with_params<F>(
		&mut self,
		services: &Arc<Services>,
		tx: &mut Transaction<'_>,
		params: &Params,
		handler: F,
	) -> Result<()>
	where
		F: FnOnce(&Arc<Services>, &mut Transaction<'_>, Params, &SymbolTable) -> Result<RecordBatch>,
	{
		reject_query_txn(tx)?;
		let mut txn = tx.reborrow();
		load_flow_diff_need(&services.catalog, &mut txn)?;
		let columns = handler(services, &mut txn, params.clone(), &self.symbols)?;
		sync_transactional_views(services, txn)?;
		self.stack.push(Variable::columns(columns));
		Ok(())
	}

	pub(crate) fn exec_dml_with_mut_symbols<F>(
		&mut self,
		services: &Arc<Services>,
		tx: &mut Transaction<'_>,
		handler: F,
	) -> Result<()>
	where
		F: FnOnce(&Arc<Services>, &mut Transaction<'_>, &mut SymbolTable) -> Result<RecordBatch>,
	{
		reject_query_txn(tx)?;
		let mut txn = tx.reborrow();
		load_flow_diff_need(&services.catalog, &mut txn)?;
		let columns = handler(services, &mut txn, &mut self.symbols)?;
		sync_transactional_views(services, txn)?;
		self.stack.push(Variable::columns(columns));
		Ok(())
	}
}