reifydb_engine/transaction/operation/

table.rs

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

use reifydb_core::{
	common::CommitVersion,
	encoded::{row::EncodedRow, shape::RowShape},
	interface::{
		catalog::{shape::ShapeId, table::Table},
		change::{Change, ChangeOrigin, Diff},
	},
	key::row::RowKey,
	value::column::{ColumnWithName, buffer::ColumnBuffer, columns::Columns},
};
use reifydb_transaction::{
	change::{RowChange, TableRowInsertion},
	interceptor::table_row::TableRowInterceptor,
	transaction::{Transaction, admin::AdminTransaction, command::CommandTransaction},
};
use reifydb_type::{
	fragment::Fragment,
	util::cowvec::CowVec,
	value::{datetime::DateTime, row_number::RowNumber},
};
use smallvec::smallvec;

use crate::Result;

fn build_encoded_columns(shape: &RowShape, row_number: RowNumber, encoded: &EncodedRow) -> Columns {
	let fields = shape.fields();

	let mut columns_vec: Vec<ColumnWithName> = Vec::with_capacity(fields.len());
	for field in fields.iter() {
		columns_vec.push(ColumnWithName {
			name: Fragment::internal(&field.name),
			data: ColumnBuffer::with_capacity(field.constraint.get_type(), 1),
		});
	}

	for (i, _) in fields.iter().enumerate() {
		columns_vec[i].data.push_value(shape.get_value(encoded, i));
	}

	Columns::with_system_columns(
		columns_vec,
		vec![row_number],
		vec![DateTime::from_nanos(encoded.created_at_nanos())],
		vec![DateTime::from_nanos(encoded.updated_at_nanos())],
	)
}

fn build_table_insert_change(table: &Table, shape: &RowShape, row_number: RowNumber, encoded: &EncodedRow) -> Change {
	Change {
		origin: ChangeOrigin::Shape(ShapeId::Table(table.id)),
		version: CommitVersion(0),
		diffs: smallvec![Diff::insert(build_encoded_columns(shape, row_number, encoded))],
		changed_at: DateTime::default(),
	}
}

fn build_table_update_change(table: &Table, row_number: RowNumber, pre: &EncodedRow, post: &EncodedRow) -> Change {
	let shape: RowShape = (&table.columns).into();
	Change {
		origin: ChangeOrigin::Shape(ShapeId::Table(table.id)),
		version: CommitVersion(0),
		diffs: smallvec![Diff::update(
			build_encoded_columns(&shape, row_number, pre),
			build_encoded_columns(&shape, row_number, post),
		)],
		changed_at: DateTime::default(),
	}
}

fn build_table_remove_change(table: &Table, row_number: RowNumber, encoded: &EncodedRow) -> Change {
	let shape: RowShape = (&table.columns).into();
	Change {
		origin: ChangeOrigin::Shape(ShapeId::Table(table.id)),
		version: CommitVersion(0),
		diffs: smallvec![Diff::remove(build_encoded_columns(&shape, row_number, encoded))],
		changed_at: DateTime::default(),
	}
}

pub(crate) trait TableOperations {
	fn insert_table(
		&mut self,
		table: &Table,
		shape: &RowShape,
		row: EncodedRow,
		row_number: RowNumber,
	) -> Result<EncodedRow>;

	fn update_table(&mut self, table: Table, id: RowNumber, row: EncodedRow) -> Result<EncodedRow>;

	fn remove_from_table(&mut self, table: Table, id: RowNumber) -> Result<EncodedRow>;
}

impl TableOperations for CommandTransaction {
	fn insert_table(
		&mut self,
		table: &Table,
		shape: &RowShape,
		row: EncodedRow,
		row_number: RowNumber,
	) -> Result<EncodedRow> {
		let row = TableRowInterceptor::pre_insert(self, table, row_number, row)?;

		self.set(&RowKey::encoded(table.id, row_number), row.clone())?;

		TableRowInterceptor::post_insert(self, table, row_number, &row)?;

		// Track insertion for post-commit event emission
		self.track_row_change(RowChange::TableInsert(TableRowInsertion {
			table_id: table.id,
			row_number,
			encoded: row.clone(),
		}));

		// Track flow change for transactional view pre-commit processing
		self.track_flow_change(build_table_insert_change(table, shape, row_number, &row));

		Ok(row)
	}

	fn update_table(&mut self, table: Table, id: RowNumber, row: EncodedRow) -> Result<EncodedRow> {
		let key = RowKey::encoded(table.id, id);

		let pre = match self.get(&key)? {
			Some(v) => v.row,
			None => return Ok(row),
		};

		let row = TableRowInterceptor::pre_update(self, &table, id, row)?;

		if self.get_committed(&key)?.is_some() {
			self.mark_preexisting(&key)?;
		}
		self.set(&key, row.clone())?;

		TableRowInterceptor::post_update(self, &table, id, &row, &pre)?;

		self.track_flow_change(build_table_update_change(&table, id, &pre, &row));

		Ok(row)
	}

	fn remove_from_table(&mut self, table: Table, id: RowNumber) -> Result<EncodedRow> {
		let key = RowKey::encoded(table.id, id);

		let displayed = match self.get(&key)? {
			Some(v) => v.row,
			None => return Ok(EncodedRow(CowVec::new(vec![]))),
		};
		let committed = self.get_committed(&key)?.map(|v| v.row);

		TableRowInterceptor::pre_delete(self, &table, id)?;

		let pre_for_cdc = committed.clone().unwrap_or_else(|| displayed.clone());

		if committed.is_some() {
			self.mark_preexisting(&key)?;
		}
		self.unset(&key, pre_for_cdc.clone())?;

		TableRowInterceptor::post_delete(self, &table, id, &pre_for_cdc)?;

		self.track_flow_change(build_table_remove_change(&table, id, &pre_for_cdc));

		Ok(displayed)
	}
}

impl TableOperations for AdminTransaction {
	fn insert_table(
		&mut self,
		table: &Table,
		shape: &RowShape,
		row: EncodedRow,
		row_number: RowNumber,
	) -> Result<EncodedRow> {
		let row = TableRowInterceptor::pre_insert(self, table, row_number, row)?;

		self.set(&RowKey::encoded(table.id, row_number), row.clone())?;

		TableRowInterceptor::post_insert(self, table, row_number, &row)?;

		// Track insertion for post-commit event emission
		self.track_row_change(RowChange::TableInsert(TableRowInsertion {
			table_id: table.id,
			row_number,
			encoded: row.clone(),
		}));

		// Track flow change for transactional view pre-commit processing
		self.track_flow_change(build_table_insert_change(table, shape, row_number, &row));

		Ok(row)
	}

	fn update_table(&mut self, table: Table, id: RowNumber, row: EncodedRow) -> Result<EncodedRow> {
		let key = RowKey::encoded(table.id, id);

		let pre = match self.get(&key)? {
			Some(v) => v.row,
			None => return Ok(row),
		};

		let row = TableRowInterceptor::pre_update(self, &table, id, row)?;

		if self.get_committed(&key)?.is_some() {
			self.mark_preexisting(&key)?;
		}
		self.set(&key, row.clone())?;

		TableRowInterceptor::post_update(self, &table, id, &row, &pre)?;

		self.track_flow_change(build_table_update_change(&table, id, &pre, &row));

		Ok(row)
	}

	fn remove_from_table(&mut self, table: Table, id: RowNumber) -> Result<EncodedRow> {
		let key = RowKey::encoded(table.id, id);

		let displayed = match self.get(&key)? {
			Some(v) => v.row,
			None => return Ok(EncodedRow(CowVec::new(vec![]))),
		};
		let committed = self.get_committed(&key)?.map(|v| v.row);

		TableRowInterceptor::pre_delete(self, &table, id)?;

		let pre_for_cdc = committed.clone().unwrap_or_else(|| displayed.clone());

		if committed.is_some() {
			self.mark_preexisting(&key)?;
		}
		self.unset(&key, pre_for_cdc.clone())?;

		TableRowInterceptor::post_delete(self, &table, id, &pre_for_cdc)?;

		self.track_flow_change(build_table_remove_change(&table, id, &pre_for_cdc));

		Ok(displayed)
	}
}

impl TableOperations for Transaction<'_> {
	fn insert_table(
		&mut self,
		table: &Table,
		shape: &RowShape,
		row: EncodedRow,
		row_number: RowNumber,
	) -> Result<EncodedRow> {
		match self {
			Transaction::Command(txn) => txn.insert_table(table, shape, row, row_number),
			Transaction::Admin(txn) => txn.insert_table(table, shape, row, row_number),
			Transaction::Test(t) => t.inner.insert_table(table, shape, row, row_number),
			Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
			Transaction::Replica(_) => panic!("Write operations not supported on Replica transaction"),
		}
	}

	fn update_table(&mut self, table: Table, id: RowNumber, row: EncodedRow) -> Result<EncodedRow> {
		match self {
			Transaction::Command(txn) => txn.update_table(table, id, row),
			Transaction::Admin(txn) => txn.update_table(table, id, row),
			Transaction::Test(t) => t.inner.update_table(table, id, row),
			Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
			Transaction::Replica(_) => panic!("Write operations not supported on Replica transaction"),
		}
	}

	fn remove_from_table(&mut self, table: Table, id: RowNumber) -> Result<EncodedRow> {
		match self {
			Transaction::Command(txn) => txn.remove_from_table(table, id),
			Transaction::Admin(txn) => txn.remove_from_table(table, id),
			Transaction::Test(t) => t.inner.remove_from_table(table, id),
			Transaction::Query(_) => panic!("Write operations not supported on Query transaction"),
			Transaction::Replica(_) => panic!("Write operations not supported on Replica transaction"),
		}
	}
}