reifydb-engine 0.10.0

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

use arrow_array::RecordBatch;
use reifydb_catalog::{
	catalog::table::{TableColumnToCreate, TableToCreate},
	store::row_settings::create::create_row_settings,
};
use reifydb_core::{
	interface::catalog::{change::CatalogTrackTableChangeOperations, storage::StorageId},
	row::RowSettings,
	value::batch::single_row,
};
use reifydb_rql::nodes::CreateTableNode;
use reifydb_transaction::transaction::{Transaction, admin::AdminTransaction};
use reifydb_value::{
	fragment::Fragment,
	value::{
		Value,
		constraint::{Constraint, TypeConstraint},
		value_type::ValueType,
	},
};

use crate::{Result, vm::services::Services};

pub(crate) fn create_table(
	services: &Services,
	txn: &mut AdminTransaction,
	plan: CreateTableNode,
) -> Result<RecordBatch> {
	if let Some(existing) = services.catalog.find_table_by_name(
		&mut Transaction::Admin(txn),
		plan.namespace.def().id(),
		plan.table.text(),
	)? && plan.if_not_exists
	{
		return single_row([
			("id", Value::Uint8(existing.id.0)),
			("namespace", Value::Utf8(plan.namespace.name().to_string())),
			("table", Value::Utf8(plan.table.text().to_string())),
			("created", Value::Boolean(false)),
		]);
	}

	let columns = expand_sumtype_columns(services, txn, plan.columns)?;

	let table = services.catalog.create_table(
		txn,
		TableToCreate {
			name: plan.table.clone(),
			namespace: plan.namespace.def().id(),
			columns,
			primary_key_columns: None,
			partition_by: plan.partition_by.clone(),
			time: plan.time.clone(),
		},
	)?;
	if let Some(ttl) = plan.ttl {
		create_row_settings(
			txn,
			StorageId::Table(table.id),
			&RowSettings {
				ttl: Some(ttl),
				persistent: plan.persistent,
			},
		)?;
	}

	txn.track_table_created(table.clone())?;

	single_row([
		("id", Value::Uint8(table.id.0)),
		("namespace", Value::Utf8(plan.namespace.name().to_string())),
		("table", Value::Utf8(plan.table.text().to_string())),
		("created", Value::Boolean(true)),
	])
}

fn expand_sumtype_columns(
	services: &Services,
	txn: &mut AdminTransaction,
	columns: Vec<TableColumnToCreate>,
) -> Result<Vec<TableColumnToCreate>> {
	let mut expanded = Vec::with_capacity(columns.len());

	for col in columns {
		match col.constraint.constraint() {
			Some(Constraint::SumType(id)) => {
				let def = services.catalog.get_sumtype(&mut Transaction::Admin(&mut *txn), *id)?;
				let col_name = col.name.text();

				expanded.push(TableColumnToCreate {
					name: Fragment::internal(format!("{col_name}_tag")),
					fragment: col.fragment.clone(),
					constraint: TypeConstraint::with_constraint(
						col.constraint.get_type(),
						Constraint::SumType(*id),
					),
					properties: vec![],
					auto_increment: false,
					dictionary_id: None,
				});

				for variant in &def.variants {
					for field in &variant.fields {
						let field_type = match field.field_type.get_type() {
							optional @ ValueType::Option(_) => optional,
							other => ValueType::Option(Box::new(other)),
						};
						expanded.push(TableColumnToCreate {
							name: Fragment::internal(format!(
								"{col_name}_{}_{}",
								variant.name, field.name
							)),
							fragment: col.fragment.clone(),
							constraint: TypeConstraint::unconstrained(field_type),
							properties: vec![],
							auto_increment: false,
							dictionary_id: None,
						});
					}
				}
			}
			_ => {
				expanded.push(col);
			}
		}
	}

	Ok(expanded)
}

#[cfg(test)]
pub mod tests {
	use reifydb_test_harness::engine::create_test_admin_transaction;
	use reifydb_value::{
		params::Params,
		value::{Value, column_view::ColumnView, frame::frame::Frame},
	};

	use crate::vm::{Admin, executor::Executor};

	#[test]
	fn test_create_table() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE test_namespace",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE test_namespace::test_table { id: Int4 }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 0), Value::Uint8(16385));
		assert_eq!(value_at(frame, 1), Value::Utf8("test_namespace".to_string()));
		assert_eq!(value_at(frame, 2), Value::Utf8("test_table".to_string()));
		assert_eq!(value_at(frame, 3), Value::Boolean(true));

		// A duplicate name in the same namespace must fault rather than silently replace.
		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE test_namespace::test_table { id: Int4 }",
				params: Params::default(),
			},
		);
		assert!(r.is_err());
		assert_eq!(r.error.unwrap().diagnostic().code, "CA_003");
	}

	#[test]
	fn test_create_same_table_in_different_shape() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE test_namespace",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE another_shape",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE test_namespace::test_table { id: Int4 }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 0), Value::Uint8(16385));
		assert_eq!(value_at(frame, 1), Value::Utf8("test_namespace".to_string()));
		assert_eq!(value_at(frame, 2), Value::Utf8("test_table".to_string()));
		assert_eq!(value_at(frame, 3), Value::Boolean(true));

		// Uniqueness is per namespace, so the same name elsewhere must be accepted.
		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE another_shape::test_table { id: Int4 }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 0), Value::Uint8(16386));
		assert_eq!(value_at(frame, 1), Value::Utf8("another_shape".to_string()));
		assert_eq!(value_at(frame, 2), Value::Utf8("test_table".to_string()));
		assert_eq!(value_at(frame, 3), Value::Boolean(true));
	}

	#[test]
	fn test_create_table_with_sumtype_column() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE app",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE ENUM app::Object { Circle { radius: Float8 }, Rectangle { width: Float8, height: Float8 } }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE app::drawings { id: Int4, shape: app::Object }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 0), Value::Uint8(16386));
		assert_eq!(value_at(frame, 3), Value::Boolean(true));
	}

	#[test]
	fn test_create_table_with_unit_sumtype_column() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE app",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE ENUM app::Status { Active, Inactive, Pending }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE app::tasks { id: Int4, status: app::Status }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 0), Value::Uint8(16386));
		assert_eq!(value_at(frame, 3), Value::Boolean(true));
	}

	#[test]
	fn test_insert_with_sumtype_constructor() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE app",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE ENUM app::Object { Circle { radius: Float8 }, Rectangle { width: Float8, height: Float8 } }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE app::drawings { id: Int4, shape: app::Object }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "INSERT app::drawings [{ id: 1, shape: app::Object::Circle { radius: 5.0 } }]",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 2), Value::Uint8(1));
	}

	#[test]
	fn test_insert_with_unit_sumtype_constructor() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE app",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE ENUM app::Status { Active, Inactive, Pending }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE app::tasks { id: Int4, status: app::Status }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "INSERT app::tasks [{ id: 1, status: app::Status::Active }]",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}
		let frame = &r[0];
		assert_eq!(value_at(frame, 2), Value::Uint8(1));
	}

	#[test]
	fn test_filter_is_variant() {
		let instance = Executor::testing();
		let mut txn = create_test_admin_transaction();

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE NAMESPACE app",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE ENUM app::Object { Circle { radius: Float8 }, Rectangle { width: Float8, height: Float8 } }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "CREATE TABLE app::drawings { id: Int4, shape: app::Object }",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "INSERT app::drawings [{ id: 1, shape: app::Object::Circle { radius: 5.0 } }]",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "INSERT app::drawings [{ id: 2, shape: app::Object::Rectangle { width: 3.0, height: 4.0 } }]",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "INSERT app::drawings [{ id: 3, shape: app::Object::Circle { radius: 10.0 } }]",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		let r = instance.admin(
			&mut txn,
			Admin {
				rql: "FROM app::drawings | FILTER shape IS app::Object::Circle",
				params: Params::default(),
			},
		);
		if let Some(e) = r.error {
			panic!("{e:?}");
		}

		assert!(!r.is_empty());
		let frame = &r[0];
		let id_col = frame.try_column("id").expect("id column");
		assert_eq!(id_col.len(), 2);
		let mut ids: Vec<Value> = (0..2).map(|i| id_col.get_value(i)).collect();
		ids.sort_by_key(|v| match v {
			Value::Int4(n) => *n,
			_ => panic!("expected Int4"),
		});
		assert_eq!(ids, vec![Value::Int4(1), Value::Int4(3)]);
	}

	fn value_at(frame: &Frame, column: usize) -> Value {
		// Positional read: without it a reordered result column would still pass.
		ColumnView::try_from((frame.batch.column(column), frame.batch.schema_ref().field(column)))
			.unwrap()
			.get_value(0)
	}
}