reifydb-core 0.4.5

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

use reifydb_type::value::constraint::{Constraint, TypeConstraint};

use crate::{
	encoded::schema::{RowSchema, RowSchemaField},
	interface::catalog::{column::Column, subscription::SubscriptionColumn},
};

impl From<&Vec<Column>> for RowSchema {
	fn from(value: &Vec<Column>) -> Self {
		RowSchema::from(value.as_slice())
	}
}
impl From<&[Column]> for RowSchema {
	fn from(value: &[Column]) -> Self {
		let fields = value
			.iter()
			.map(|col| {
				// For dictionary columns, the constraint carries Dictionary info with id_type.
				// Convert to TypeConstraint::dictionary() so the schema uses DictionaryId as base_type,
				// matching the schema used to encode the data.
				let constraint = match col.constraint.constraint() {
					Some(Constraint::Dictionary(dict_id, id_type)) => {
						TypeConstraint::dictionary(*dict_id, id_type.clone())
					}
					_ => col.constraint.clone(),
				};
				RowSchemaField::new(col.name.clone(), constraint)
			})
			.collect();
		RowSchema::new(fields)
	}
}

impl From<&Vec<SubscriptionColumn>> for RowSchema {
	fn from(value: &Vec<SubscriptionColumn>) -> Self {
		RowSchema::from(value.as_slice())
	}
}
impl From<&[SubscriptionColumn]> for RowSchema {
	fn from(value: &[SubscriptionColumn]) -> Self {
		let fields = value
			.iter()
			.map(|col| RowSchemaField::new(col.name.clone(), TypeConstraint::unconstrained(col.ty.clone())))
			.collect();
		RowSchema::new(fields)
	}
}

#[cfg(test)]
mod tests {
	mod from_schema {
		// Tests removed as From<&RowSchema> for the old layout type has been removed
		// RowSchema is now the canonical layout descriptor
	}

	mod from_column {
		use reifydb_type::value::{constraint::TypeConstraint, r#type::Type};

		use crate::{
			encoded::schema::{RowSchema, RowSchemaField},
			interface::catalog::{
				column::{Column, ColumnIndex},
				id::ColumnId,
			},
		};

		fn make_column(id: u64, name: &str, ty: Type, index: u8) -> Column {
			Column {
				id: ColumnId(id),
				name: name.to_string(),
				constraint: TypeConstraint::unconstrained(ty),
				properties: vec![],
				index: ColumnIndex(index),
				auto_increment: false,
				dictionary_id: None,
			}
		}

		#[test]
		fn test_from_column_single_field() {
			let columns = vec![make_column(1, "id", Type::Int8, 0)];

			let schema = RowSchema::from(columns.as_slice());

			assert_eq!(schema.fields().len(), 1);
			assert_eq!(schema.fields()[0].name, "id");
			assert_eq!(schema.fields()[0].constraint.get_type(), Type::Int8);
		}

		#[test]
		fn test_from_column_multiple_fields() {
			let columns = vec![
				make_column(1, "a", Type::Int1, 0),
				make_column(2, "b", Type::Int2, 1),
				make_column(3, "c", Type::Int4, 2),
			];

			let schema = RowSchema::from(columns.as_slice());

			assert_eq!(schema.fields().len(), 3);
			assert_eq!(schema.fields()[0].name, "a");
			assert_eq!(schema.fields()[0].constraint.get_type(), Type::Int1);
			assert_eq!(schema.fields()[1].name, "b");
			assert_eq!(schema.fields()[1].constraint.get_type(), Type::Int2);
			assert_eq!(schema.fields()[2].name, "c");
			assert_eq!(schema.fields()[2].constraint.get_type(), Type::Int4);
		}

		#[test]
		fn test_from_column_preserves_field_order() {
			let columns = vec![
				make_column(1, "first", Type::Utf8, 0),
				make_column(2, "second", Type::Int4, 1),
				make_column(3, "third", Type::Boolean, 2),
			];

			let schema = RowSchema::from(columns.as_slice());

			assert_eq!(schema.fields()[0].name, "first");
			assert_eq!(schema.fields()[0].constraint.get_type(), Type::Utf8);
			assert_eq!(schema.fields()[1].name, "second");
			assert_eq!(schema.fields()[1].constraint.get_type(), Type::Int4);
			assert_eq!(schema.fields()[2].name, "third");
			assert_eq!(schema.fields()[2].constraint.get_type(), Type::Boolean);
		}

		#[test]
		fn test_from_column_equivalence_with_direct_construction() {
			let columns = vec![
				make_column(1, "f0", Type::Uint1, 0),
				make_column(2, "f1", Type::Uint2, 1),
				make_column(3, "f2", Type::Uint4, 2),
				make_column(4, "f3", Type::Uint8, 3),
				make_column(5, "f4", Type::Uint16, 4),
			];

			let schema_from_columns = RowSchema::from(columns.as_slice());
			let schema_direct = RowSchema::new(vec![
				RowSchemaField::unconstrained("f0", Type::Uint1),
				RowSchemaField::unconstrained("f1", Type::Uint2),
				RowSchemaField::unconstrained("f2", Type::Uint4),
				RowSchemaField::unconstrained("f3", Type::Uint8),
				RowSchemaField::unconstrained("f4", Type::Uint16),
			]);

			// Full equivalence check
			assert_eq!(schema_from_columns.fields().len(), schema_direct.fields().len());
			assert_eq!(schema_from_columns.fingerprint(), schema_direct.fingerprint());

			for (i, (from_columns, direct)) in
				schema_from_columns.fields().iter().zip(schema_direct.fields().iter()).enumerate()
			{
				assert_eq!(from_columns.name, direct.name, "name mismatch at field {}", i);
				assert_eq!(
					from_columns.constraint, direct.constraint,
					"constraint mismatch at field {}",
					i
				);
				assert_eq!(from_columns.offset, direct.offset, "offset mismatch at field {}", i);
				assert_eq!(from_columns.size, direct.size, "size mismatch at field {}", i);
				assert_eq!(from_columns.align, direct.align, "align mismatch at field {}", i);
			}
		}

		#[test]
		fn test_from_column_empty() {
			let columns: Vec<Column> = vec![];

			let schema = RowSchema::from(columns.as_slice());

			assert_eq!(schema.fields().len(), 0);
		}

		#[test]
		fn test_from_column_nine_fields() {
			let columns = vec![
				make_column(1, "f0", Type::Boolean, 0),
				make_column(2, "f1", Type::Int1, 1),
				make_column(3, "f2", Type::Int2, 2),
				make_column(4, "f3", Type::Int4, 3),
				make_column(5, "f4", Type::Int8, 4),
				make_column(6, "f5", Type::Uint1, 5),
				make_column(7, "f6", Type::Uint2, 6),
				make_column(8, "f7", Type::Uint4, 7),
				make_column(9, "f8", Type::Uint8, 8),
			];

			let schema = RowSchema::from(columns.as_slice());

			assert_eq!(schema.fields().len(), 9);
			for (i, field) in schema.fields().iter().enumerate() {
				assert_eq!(field.name, format!("f{}", i));
			}
		}
	}
}