reifydb-catalog 0.4.13

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

use reifydb_core::{
	interface::catalog::{
		column::Column,
		id::{ColumnId, NamespaceId},
		shape::ShapeId,
	},
	key::column::ColumnKey,
};
use reifydb_transaction::transaction::Transaction;

use crate::{CatalogStore, Result, store::column::shape::primitive_column};

/// Extended column information for system catalogs
pub struct ColumnInfo {
	pub column: Column,
	pub shape_id: ShapeId,
	pub is_view: bool,
	pub entity_kind: &'static str,
	pub entity_name: String,
	pub namespace: NamespaceId,
}

impl CatalogStore {
	pub(crate) fn list_columns(rx: &mut Transaction<'_>, shape: impl Into<ShapeId>) -> Result<Vec<Column>> {
		let shape = shape.into();
		let mut result = vec![];

		// Collect column IDs first to avoid holding stream borrow
		let mut ids = Vec::new();
		{
			let stream = rx.range(ColumnKey::full_scan(shape), 1024)?;
			for entry in stream {
				let multi = entry?;
				let row = multi.row;
				ids.push(ColumnId(primitive_column::SHAPE.get_u64(&row, primitive_column::ID)));
			}
		}

		for id in ids {
			result.push(Self::get_column(rx, id)?);
		}

		result.sort_by_key(|c| c.index);

		Ok(result)
	}

	pub(crate) fn list_columns_all(rx: &mut Transaction<'_>) -> Result<Vec<ColumnInfo>> {
		let mut result = Vec::new();

		// Get all tables
		let tables = CatalogStore::list_tables_all(rx)?;
		for table in tables {
			let columns = CatalogStore::list_columns(rx, table.id)?;
			for column in columns {
				result.push(ColumnInfo {
					column,
					shape_id: table.id.into(),
					is_view: false,
					entity_kind: "table",
					entity_name: table.name.clone(),
					namespace: table.namespace,
				});
			}
		}

		// Get all views
		let views = CatalogStore::list_views_all(rx)?;
		for view in views {
			let columns = CatalogStore::list_columns(rx, view.id())?;
			for column in columns {
				result.push(ColumnInfo {
					column,
					shape_id: view.id().into(),
					is_view: true,
					entity_kind: "view",
					entity_name: view.name().to_string(),
					namespace: view.namespace(),
				});
			}
		}

		// Get all ring buffers
		let ringbuffers = CatalogStore::list_ringbuffers_all(rx)?;
		for ringbuffer in ringbuffers {
			let columns = CatalogStore::list_columns(rx, ringbuffer.id)?;
			for column in columns {
				result.push(ColumnInfo {
					column,
					shape_id: ringbuffer.id.into(),
					is_view: false,
					entity_kind: "ring buffer",
					entity_name: ringbuffer.name.clone(),
					namespace: ringbuffer.namespace,
				});
			}
		}

		Ok(result)
	}
}

#[cfg(test)]
pub mod tests {
	use reifydb_core::interface::catalog::{column::ColumnIndex, id::TableId};
	use reifydb_engine::test_harness::create_test_admin_transaction;
	use reifydb_transaction::transaction::Transaction;
	use reifydb_type::value::{constraint::TypeConstraint, r#type::Type};

	use crate::{CatalogStore, store::column::create::ColumnToCreate, test_utils::ensure_test_table};

	#[test]
	fn test_ok() {
		let mut txn = create_test_admin_transaction();
		ensure_test_table(&mut txn);

		// Create columns out of order
		CatalogStore::create_column(
			&mut txn,
			TableId(1),
			ColumnToCreate {
				fragment: None,
				namespace_name: "test_namespace".to_string(),
				shape_name: "test_table".to_string(),
				column: "b_col".to_string(),
				constraint: TypeConstraint::unconstrained(Type::Int4),
				properties: vec![],
				index: ColumnIndex(1),
				auto_increment: true,
				dictionary_id: None,
			},
		)
		.unwrap();

		CatalogStore::create_column(
			&mut txn,
			TableId(1),
			ColumnToCreate {
				fragment: None,
				namespace_name: "test_namespace".to_string(),
				shape_name: "test_table".to_string(),
				column: "a_col".to_string(),
				constraint: TypeConstraint::unconstrained(Type::Boolean),
				properties: vec![],
				index: ColumnIndex(0),
				auto_increment: false,
				dictionary_id: None,
			},
		)
		.unwrap();

		let columns = CatalogStore::list_columns(&mut Transaction::Admin(&mut txn), TableId(1)).unwrap();
		assert_eq!(columns.len(), 2);

		assert_eq!(columns[0].name, "a_col"); // index 0
		assert_eq!(columns[1].name, "b_col"); // index 1

		assert_eq!(columns[0].index, ColumnIndex(0));
		assert_eq!(columns[1].index, ColumnIndex(1));

		assert_eq!(columns[0].auto_increment, false);
		assert_eq!(columns[1].auto_increment, true);
	}

	#[test]
	fn test_empty() {
		let mut txn = create_test_admin_transaction();
		ensure_test_table(&mut txn);

		let columns = CatalogStore::list_columns(&mut Transaction::Admin(&mut txn), TableId(1)).unwrap();
		assert!(columns.is_empty());
	}

	#[test]
	fn test_table_does_not_exist() {
		let mut txn = create_test_admin_transaction();

		let columns = CatalogStore::list_columns(&mut Transaction::Admin(&mut txn), TableId(1)).unwrap();
		assert!(columns.is_empty());
	}
}