reifydb-rql 0.4.6

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

use reifydb_catalog::{
	catalog::ringbuffer::RingBufferColumnToCreate,
	error::{CatalogError, CatalogObjectKind},
};
use reifydb_transaction::transaction::Transaction;
use reifydb_type::{
	fragment::Fragment,
	value::constraint::{Constraint, TypeConstraint},
};

use crate::{
	Result,
	ast::ast::{AstColumnProperty, AstCreateRingBuffer},
	convert_data_type_with_constraints,
	plan::logical::{Compiler, CreateRingBufferNode, LogicalPlan},
};

impl<'bump> Compiler<'bump> {
	pub(crate) fn compile_create_ringbuffer(
		&self,
		ast: AstCreateRingBuffer<'bump>,
		tx: &mut Transaction<'_>,
	) -> Result<LogicalPlan<'bump>> {
		let mut columns: Vec<RingBufferColumnToCreate> = vec![];

		let ringbuffer_ns_segments: Vec<&str> = ast.ringbuffer.namespace.iter().map(|n| n.text()).collect();

		for col in ast.columns.into_iter() {
			let column_name = col.name.text().to_string();
			let mut constraint = convert_data_type_with_constraints(&col.ty)?;
			let column_type = constraint.get_type();

			let name = col.name.to_owned();
			let ty_fragment = col.ty.name_fragment().to_owned();
			let fragment = Fragment::merge_all([name.clone(), ty_fragment]);

			let mut auto_increment = false;
			let mut dictionary_id = None;
			let properties = vec![];

			for property in &col.properties {
				match property {
					AstColumnProperty::AutoIncrement => auto_increment = true,
					AstColumnProperty::Dictionary(dict_ident) => {
						let dict_ns_segments: Vec<&str> = if dict_ident.namespace.is_empty() {
							ringbuffer_ns_segments.clone()
						} else {
							dict_ident.namespace.iter().map(|n| n.text()).collect()
						};
						let dict_name = dict_ident.name.text();

						let Some(namespace) = self
							.catalog
							.find_namespace_by_segments(tx, &dict_ns_segments)?
						else {
							return Err(CatalogError::NotFound {
								kind: CatalogObjectKind::Dictionary,
								namespace: dict_ns_segments.join("::"),
								name: dict_name.to_string(),
								fragment: dict_ident.name.to_owned(),
							}
							.into());
						};

						let Some(dictionary) = self.catalog.find_dictionary_by_name(
							tx,
							namespace.id(),
							dict_name,
						)?
						else {
							return Err(CatalogError::NotFound {
								kind: CatalogObjectKind::Dictionary,
								namespace: dict_ns_segments.join("::"),
								name: dict_name.to_string(),
								fragment: dict_ident.name.to_owned(),
							}
							.into());
						};

						if column_type != dictionary.value_type {
							return Err(CatalogError::DictionaryTypeMismatch {
								column: column_name.clone(),
								column_type,
								dictionary: dict_name.to_string(),
								dictionary_value_type: dictionary.value_type,
								fragment: col.name.to_owned(),
							}
							.into());
						}

						dictionary_id = Some(dictionary.id);
						// Embed dictionary constraint so the TypeConstraint carries id_type
						// info
						constraint = TypeConstraint::with_constraint(
							constraint.get_type(),
							Constraint::Dictionary(dictionary.id, dictionary.id_type),
						);
					}
					AstColumnProperty::Saturation(_) => {
						// TODO: inline saturation policy
					}
					AstColumnProperty::Default(_) => {
						// TODO: inline default policy
					}
				}
			}

			columns.push(RingBufferColumnToCreate {
				name,
				fragment,
				constraint,
				properties,
				auto_increment,
				dictionary_id,
			});
		}

		let partition_by: Vec<String> = ast.partition_by.iter().map(|s| s.to_string()).collect();

		// Validate that partition_by columns exist in the column list
		for pb_col in &partition_by {
			if !columns.iter().any(|c| c.name.text() == pb_col.as_str()) {
				return Err(CatalogError::NotFound {
					kind: CatalogObjectKind::Column,
					namespace: ringbuffer_ns_segments.join("::"),
					name: pb_col.clone(),
					fragment: Fragment::internal(pb_col.as_str()),
				}
				.into());
			}
		}

		let ringbuffer = ast.ringbuffer;

		Ok(LogicalPlan::CreateRingBuffer(CreateRingBufferNode {
			ringbuffer,
			if_not_exists: false,
			columns,
			capacity: ast.capacity,
			partition_by,
		}))
	}
}