use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use alopex_core::kv::{KVStore, KVTransaction};
use alopex_core::types::TxnMode;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use crate::ast::ddl::{IndexMethod, VectorMetric};
use crate::catalog::{
Catalog, ColumnMetadata, Compression, IndexMetadata, MemoryCatalog, RowIdMode,
};
use crate::catalog::{StorageOptions, StorageType, TableMetadata};
use crate::planner::PlannerError;
use crate::planner::types::ResolvedType;
pub const CATALOG_PREFIX: &[u8] = b"__catalog__/";
pub const CATALOGS_PREFIX: &[u8] = b"__catalog__/catalogs/";
pub const NAMESPACES_PREFIX: &[u8] = b"__catalog__/namespaces/";
pub const TABLES_PREFIX: &[u8] = b"__catalog__/tables/";
pub const INDEXES_PREFIX: &[u8] = b"__catalog__/indexes/";
pub const META_KEY: &[u8] = b"__catalog__/meta";
const CATALOG_VERSION: u32 = 2;
#[derive(Debug, Error)]
pub enum CatalogError {
#[error("kv error: {0}")]
Kv(#[from] alopex_core::Error),
#[error("serialize error: {0}")]
Serialize(#[from] bincode::Error),
#[error("invalid catalog key: {0}")]
InvalidKey(String),
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct CatalogState {
version: u32,
table_id_counter: u32,
index_id_counter: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedCatalogMeta {
pub name: String,
pub comment: Option<String>,
pub storage_root: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedNamespaceMeta {
pub name: String,
pub catalog_name: String,
pub comment: Option<String>,
pub storage_root: Option<String>,
}
pub type CatalogMeta = PersistedCatalogMeta;
pub type NamespaceMeta = PersistedNamespaceMeta;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TableFqn {
pub catalog: String,
pub namespace: String,
pub table: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct IndexFqn {
pub catalog: String,
pub namespace: String,
pub table: String,
pub index: String,
}
impl TableFqn {
pub fn new(catalog: &str, namespace: &str, table: &str) -> Self {
Self {
catalog: catalog.to_string(),
namespace: namespace.to_string(),
table: table.to_string(),
}
}
}
impl IndexFqn {
pub fn new(catalog: &str, namespace: &str, table: &str, index: &str) -> Self {
Self {
catalog: catalog.to_string(),
namespace: namespace.to_string(),
table: table.to_string(),
index: index.to_string(),
}
}
}
impl From<&TableMetadata> for TableFqn {
fn from(value: &TableMetadata) -> Self {
Self::new(&value.catalog_name, &value.namespace_name, &value.name)
}
}
impl From<&IndexMetadata> for IndexFqn {
fn from(value: &IndexMetadata) -> Self {
Self::new(
&value.catalog_name,
&value.namespace_name,
&value.table,
&value.name,
)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum TableType {
Managed,
External,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum DataSourceFormat {
#[default]
Alopex,
Parquet,
Delta,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedVectorMetric {
Cosine,
L2,
Inner,
}
impl From<VectorMetric> for PersistedVectorMetric {
fn from(value: VectorMetric) -> Self {
match value {
VectorMetric::Cosine => Self::Cosine,
VectorMetric::L2 => Self::L2,
VectorMetric::Inner => Self::Inner,
}
}
}
impl From<PersistedVectorMetric> for VectorMetric {
fn from(value: PersistedVectorMetric) -> Self {
match value {
PersistedVectorMetric::Cosine => Self::Cosine,
PersistedVectorMetric::L2 => Self::L2,
PersistedVectorMetric::Inner => Self::Inner,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedType {
Integer,
BigInt,
Float,
Double,
Text,
Blob,
Boolean,
Timestamp,
Vector {
dimension: u32,
metric: PersistedVectorMetric,
},
Null,
}
impl From<ResolvedType> for PersistedType {
fn from(value: ResolvedType) -> Self {
match value {
ResolvedType::Integer => Self::Integer,
ResolvedType::BigInt => Self::BigInt,
ResolvedType::Float => Self::Float,
ResolvedType::Double => Self::Double,
ResolvedType::Text => Self::Text,
ResolvedType::Blob => Self::Blob,
ResolvedType::Boolean => Self::Boolean,
ResolvedType::Timestamp => Self::Timestamp,
ResolvedType::Vector { dimension, metric } => Self::Vector {
dimension,
metric: metric.into(),
},
ResolvedType::Null => Self::Null,
}
}
}
impl From<PersistedType> for ResolvedType {
fn from(value: PersistedType) -> Self {
match value {
PersistedType::Integer => Self::Integer,
PersistedType::BigInt => Self::BigInt,
PersistedType::Float => Self::Float,
PersistedType::Double => Self::Double,
PersistedType::Text => Self::Text,
PersistedType::Blob => Self::Blob,
PersistedType::Boolean => Self::Boolean,
PersistedType::Timestamp => Self::Timestamp,
PersistedType::Vector { dimension, metric } => Self::Vector {
dimension,
metric: metric.into(),
},
PersistedType::Null => Self::Null,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedIndexType {
BTree,
Hnsw,
}
impl From<PersistedIndexType> for IndexMethod {
fn from(value: PersistedIndexType) -> Self {
match value {
PersistedIndexType::BTree => IndexMethod::BTree,
PersistedIndexType::Hnsw => IndexMethod::Hnsw,
}
}
}
impl TryFrom<IndexMethod> for PersistedIndexType {
type Error = ();
fn try_from(value: IndexMethod) -> Result<Self, Self::Error> {
match value {
IndexMethod::BTree => Ok(Self::BTree),
IndexMethod::Hnsw => Ok(Self::Hnsw),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedStorageType {
Row,
Columnar,
}
impl From<PersistedStorageType> for StorageType {
fn from(value: PersistedStorageType) -> Self {
match value {
PersistedStorageType::Row => Self::Row,
PersistedStorageType::Columnar => Self::Columnar,
}
}
}
impl From<StorageType> for PersistedStorageType {
fn from(value: StorageType) -> Self {
match value {
StorageType::Row => Self::Row,
StorageType::Columnar => Self::Columnar,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedCompression {
None,
Lz4,
Zstd,
}
impl From<PersistedCompression> for Compression {
fn from(value: PersistedCompression) -> Self {
match value {
PersistedCompression::None => Self::None,
PersistedCompression::Lz4 => Self::Lz4,
PersistedCompression::Zstd => Self::Zstd,
}
}
}
impl From<Compression> for PersistedCompression {
fn from(value: Compression) -> Self {
match value {
Compression::None => Self::None,
Compression::Lz4 => Self::Lz4,
Compression::Zstd => Self::Zstd,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum PersistedRowIdMode {
None,
Direct,
}
impl From<PersistedRowIdMode> for RowIdMode {
fn from(value: PersistedRowIdMode) -> Self {
match value {
PersistedRowIdMode::None => Self::None,
PersistedRowIdMode::Direct => Self::Direct,
}
}
}
impl From<RowIdMode> for PersistedRowIdMode {
fn from(value: RowIdMode) -> Self {
match value {
RowIdMode::None => Self::None,
RowIdMode::Direct => Self::Direct,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedStorageOptions {
pub storage_type: PersistedStorageType,
pub compression: PersistedCompression,
pub row_group_size: u32,
pub row_id_mode: PersistedRowIdMode,
}
impl From<StorageOptions> for PersistedStorageOptions {
fn from(value: StorageOptions) -> Self {
Self {
storage_type: value.storage_type.into(),
compression: value.compression.into(),
row_group_size: value.row_group_size,
row_id_mode: value.row_id_mode.into(),
}
}
}
impl From<PersistedStorageOptions> for StorageOptions {
fn from(value: PersistedStorageOptions) -> Self {
Self {
storage_type: value.storage_type.into(),
compression: value.compression.into(),
row_group_size: value.row_group_size,
row_id_mode: value.row_id_mode.into(),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedColumnMeta {
pub name: String,
pub data_type: PersistedType,
pub not_null: bool,
pub primary_key: bool,
pub unique: bool,
}
impl From<&ColumnMetadata> for PersistedColumnMeta {
fn from(value: &ColumnMetadata) -> Self {
Self {
name: value.name.clone(),
data_type: value.data_type.clone().into(),
not_null: value.not_null,
primary_key: value.primary_key,
unique: value.unique,
}
}
}
impl From<PersistedColumnMeta> for ColumnMetadata {
fn from(value: PersistedColumnMeta) -> Self {
ColumnMetadata::new(value.name, value.data_type.into())
.with_not_null(value.not_null)
.with_primary_key(value.primary_key)
.with_unique(value.unique)
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedTableMeta {
pub table_id: u32,
pub name: String,
pub catalog_name: String,
pub namespace_name: String,
pub table_type: TableType,
pub data_source_format: DataSourceFormat,
pub columns: Vec<PersistedColumnMeta>,
pub primary_key: Option<Vec<String>>,
pub storage_options: PersistedStorageOptions,
pub storage_location: Option<String>,
pub comment: Option<String>,
pub properties: HashMap<String, String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct PersistedTableMetaV1 {
table_id: u32,
name: String,
columns: Vec<PersistedColumnMeta>,
primary_key: Option<Vec<String>>,
storage_options: PersistedStorageOptions,
}
impl From<&TableMetadata> for PersistedTableMeta {
fn from(value: &TableMetadata) -> Self {
Self {
table_id: value.table_id,
name: value.name.clone(),
catalog_name: value.catalog_name.clone(),
namespace_name: value.namespace_name.clone(),
table_type: value.table_type,
data_source_format: value.data_source_format,
columns: value
.columns
.iter()
.map(PersistedColumnMeta::from)
.collect(),
primary_key: value.primary_key.clone(),
storage_options: value.storage_options.clone().into(),
storage_location: value.storage_location.clone(),
comment: value.comment.clone(),
properties: value.properties.clone(),
}
}
}
impl From<PersistedTableMeta> for TableMetadata {
fn from(value: PersistedTableMeta) -> Self {
let mut table = TableMetadata::new(
value.name,
value
.columns
.into_iter()
.map(ColumnMetadata::from)
.collect(),
)
.with_table_id(value.table_id);
table.primary_key = value.primary_key;
table.storage_options = value.storage_options.into();
table.catalog_name = value.catalog_name;
table.namespace_name = value.namespace_name;
table.table_type = value.table_type;
table.data_source_format = value.data_source_format;
table.storage_location = value.storage_location;
table.comment = value.comment;
table.properties = value.properties;
table
}
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedIndexMeta {
pub index_id: u32,
pub name: String,
pub table: String,
pub columns: Vec<String>,
pub column_indices: Vec<usize>,
pub unique: bool,
pub method: Option<PersistedIndexType>,
pub options: Vec<(String, String)>,
pub catalog_name: String,
pub namespace_name: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
struct PersistedIndexMetaV1 {
index_id: u32,
name: String,
table: String,
columns: Vec<String>,
column_indices: Vec<usize>,
unique: bool,
method: Option<PersistedIndexType>,
options: Vec<(String, String)>,
}
impl From<&IndexMetadata> for PersistedIndexMeta {
fn from(value: &IndexMetadata) -> Self {
Self {
index_id: value.index_id,
name: value.name.clone(),
table: value.table.clone(),
columns: value.columns.clone(),
column_indices: value.column_indices.clone(),
unique: value.unique,
method: value
.method
.and_then(|m| PersistedIndexType::try_from(m).ok()),
options: value.options.clone(),
catalog_name: value.catalog_name.clone(),
namespace_name: value.namespace_name.clone(),
}
}
}
impl From<PersistedIndexMeta> for IndexMetadata {
fn from(value: PersistedIndexMeta) -> Self {
let mut index = IndexMetadata::new(value.index_id, value.name, value.table, value.columns)
.with_column_indices(value.column_indices)
.with_unique(value.unique)
.with_options(value.options);
index.catalog_name = value.catalog_name;
index.namespace_name = value.namespace_name;
if let Some(method) = value.method {
index = index.with_method(method.into());
}
index
}
}
fn deserialize_table_meta(bytes: &[u8]) -> Result<PersistedTableMeta, CatalogError> {
match bincode::deserialize::<PersistedTableMeta>(bytes) {
Ok(meta) => Ok(meta),
Err(err) => {
let is_legacy = matches!(
err.as_ref(),
bincode::ErrorKind::Io(io)
if io.kind() == std::io::ErrorKind::UnexpectedEof
);
if !is_legacy {
return Err(err.into());
}
let legacy: PersistedTableMetaV1 = bincode::deserialize(bytes)?;
Ok(PersistedTableMeta {
table_id: legacy.table_id,
name: legacy.name,
catalog_name: "default".to_string(),
namespace_name: "default".to_string(),
table_type: TableType::Managed,
data_source_format: DataSourceFormat::Alopex,
columns: legacy.columns,
primary_key: legacy.primary_key,
storage_options: legacy.storage_options,
storage_location: None,
comment: None,
properties: HashMap::new(),
})
}
}
}
fn deserialize_index_meta(bytes: &[u8]) -> Result<PersistedIndexMeta, CatalogError> {
match bincode::deserialize::<PersistedIndexMeta>(bytes) {
Ok(meta) => Ok(meta),
Err(err) => {
let is_legacy = matches!(
err.as_ref(),
bincode::ErrorKind::Io(io)
if io.kind() == std::io::ErrorKind::UnexpectedEof
);
if !is_legacy {
return Err(err.into());
}
let legacy: PersistedIndexMetaV1 = bincode::deserialize(bytes)?;
Ok(PersistedIndexMeta {
index_id: legacy.index_id,
name: legacy.name,
table: legacy.table,
columns: legacy.columns,
column_indices: legacy.column_indices,
unique: legacy.unique,
method: legacy.method,
options: legacy.options,
catalog_name: "default".to_string(),
namespace_name: "default".to_string(),
})
}
}
}
fn table_key(catalog_name: &str, namespace_name: &str, table_name: &str) -> Vec<u8> {
let mut key = TABLES_PREFIX.to_vec();
key.extend_from_slice(catalog_name.as_bytes());
key.push(b'/');
key.extend_from_slice(namespace_name.as_bytes());
key.push(b'/');
key.extend_from_slice(table_name.as_bytes());
key
}
fn catalog_key(name: &str) -> Vec<u8> {
let mut key = CATALOGS_PREFIX.to_vec();
key.extend_from_slice(name.as_bytes());
key
}
fn namespace_key(catalog_name: &str, namespace_name: &str) -> Vec<u8> {
let mut key = NAMESPACES_PREFIX.to_vec();
key.extend_from_slice(catalog_name.as_bytes());
key.push(b'/');
key.extend_from_slice(namespace_name.as_bytes());
key
}
fn index_key(
catalog_name: &str,
namespace_name: &str,
table_name: &str,
index_name: &str,
) -> Vec<u8> {
let mut key = INDEXES_PREFIX.to_vec();
key.extend_from_slice(catalog_name.as_bytes());
key.push(b'/');
key.extend_from_slice(namespace_name.as_bytes());
key.push(b'/');
key.extend_from_slice(table_name.as_bytes());
key.push(b'/');
key.extend_from_slice(index_name.as_bytes());
key
}
fn index_prefix(catalog_name: &str, namespace_name: &str, table_name: &str) -> Vec<u8> {
let mut key = INDEXES_PREFIX.to_vec();
key.extend_from_slice(catalog_name.as_bytes());
key.push(b'/');
key.extend_from_slice(namespace_name.as_bytes());
key.push(b'/');
key.extend_from_slice(table_name.as_bytes());
key.push(b'/');
key
}
fn key_suffix(prefix: &[u8], key: &[u8]) -> Result<String, CatalogError> {
let suffix = key
.strip_prefix(prefix)
.ok_or_else(|| CatalogError::InvalidKey(format!("{key:?}")))?;
std::str::from_utf8(suffix)
.map(|s| s.to_string())
.map_err(|_| CatalogError::InvalidKey(format!("{key:?}")))
}
fn parse_table_key_suffix(suffix: &str) -> Result<TableFqn, CatalogError> {
let mut parts = suffix.splitn(3, '/');
let catalog = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
let namespace = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
let table = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
Ok(TableFqn::new(catalog, namespace, table))
}
fn parse_index_key_suffix(suffix: &str) -> Result<IndexFqn, CatalogError> {
let mut parts = suffix.splitn(4, '/');
let catalog = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
let namespace = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
let table = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
let index = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.to_string()))?;
Ok(IndexFqn::new(catalog, namespace, table, index))
}
#[derive(Debug, Clone, Default)]
pub struct CatalogOverlay {
added_catalogs: HashMap<String, CatalogMeta>,
dropped_catalogs: HashSet<String>,
added_namespaces: HashMap<(String, String), NamespaceMeta>,
dropped_namespaces: HashSet<(String, String)>,
added_tables: HashMap<TableFqn, TableMetadata>,
dropped_tables: HashSet<TableFqn>,
added_indexes: HashMap<IndexFqn, IndexMetadata>,
dropped_indexes: HashSet<IndexFqn>,
}
impl CatalogOverlay {
pub fn new() -> Self {
Self::default()
}
pub fn add_catalog(&mut self, meta: CatalogMeta) {
self.dropped_catalogs.remove(&meta.name);
self.added_catalogs.insert(meta.name.clone(), meta);
}
pub fn drop_catalog(&mut self, name: &str) {
self.added_catalogs.remove(name);
self.dropped_catalogs.insert(name.to_string());
}
pub fn add_namespace(&mut self, meta: NamespaceMeta) {
let key = (meta.catalog_name.clone(), meta.name.clone());
self.dropped_namespaces.remove(&key);
self.added_namespaces.insert(key, meta);
}
pub fn drop_namespace(&mut self, catalog_name: &str, namespace_name: &str) {
let key = (catalog_name.to_string(), namespace_name.to_string());
self.added_namespaces.remove(&key);
self.dropped_namespaces.insert(key);
}
pub fn add_table(&mut self, fqn: TableFqn, table: TableMetadata) {
self.dropped_tables.remove(&fqn);
self.added_tables.insert(fqn, table);
}
pub fn drop_table(&mut self, fqn: &TableFqn) {
self.added_tables.remove(fqn);
self.dropped_tables.insert(fqn.clone());
self.added_indexes.retain(|key, _| {
key.catalog != fqn.catalog || key.namespace != fqn.namespace || key.table != fqn.table
});
}
pub fn add_index(&mut self, fqn: IndexFqn, index: IndexMetadata) {
self.dropped_indexes.remove(&fqn);
self.added_indexes.insert(fqn, index);
}
pub fn drop_index(&mut self, fqn: &IndexFqn) {
self.added_indexes.remove(fqn);
self.dropped_indexes.insert(fqn.clone());
}
pub fn drop_cascade_catalog(&mut self, catalog: &str) {
self.drop_catalog(catalog);
let namespace_keys: Vec<(String, String)> = self
.added_namespaces
.keys()
.filter(|(cat, _)| cat == catalog)
.cloned()
.collect();
for (catalog_name, namespace_name) in namespace_keys {
self.drop_namespace(&catalog_name, &namespace_name);
}
let index_keys: Vec<IndexFqn> = self
.added_indexes
.keys()
.filter(|fqn| fqn.catalog == catalog)
.cloned()
.collect();
for fqn in index_keys {
self.drop_index(&fqn);
}
let table_keys: Vec<TableFqn> = self
.added_tables
.keys()
.filter(|fqn| fqn.catalog == catalog)
.cloned()
.collect();
for fqn in table_keys {
self.drop_table(&fqn);
}
}
pub fn drop_cascade_namespace(&mut self, catalog: &str, namespace: &str) {
self.drop_namespace(catalog, namespace);
let index_keys: Vec<IndexFqn> = self
.added_indexes
.keys()
.filter(|fqn| fqn.catalog == catalog && fqn.namespace == namespace)
.cloned()
.collect();
for fqn in index_keys {
self.drop_index(&fqn);
}
let table_keys: Vec<TableFqn> = self
.added_tables
.keys()
.filter(|fqn| fqn.catalog == catalog && fqn.namespace == namespace)
.cloned()
.collect();
for fqn in table_keys {
self.drop_table(&fqn);
}
}
}
pub struct TxnCatalogView<'a, S: KVStore> {
catalog: &'a PersistentCatalog<S>,
overlay: &'a CatalogOverlay,
}
impl<'a, S: KVStore> TxnCatalogView<'a, S> {
pub fn new(catalog: &'a PersistentCatalog<S>, overlay: &'a CatalogOverlay) -> Self {
Self { catalog, overlay }
}
}
impl<'a, S: KVStore> Catalog for TxnCatalogView<'a, S> {
fn create_table(&mut self, _table: TableMetadata) -> Result<(), PlannerError> {
unreachable!("TxnCatalogView は参照専用です")
}
fn get_table(&self, name: &str) -> Option<&TableMetadata> {
self.catalog.get_table_in_txn(name, self.overlay)
}
fn drop_table(&mut self, _name: &str) -> Result<(), PlannerError> {
unreachable!("TxnCatalogView は参照専用です")
}
fn create_index(&mut self, _index: IndexMetadata) -> Result<(), PlannerError> {
unreachable!("TxnCatalogView は参照専用です")
}
fn get_index(&self, name: &str) -> Option<&IndexMetadata> {
self.catalog.get_index_in_txn(name, self.overlay)
}
fn get_indexes_for_table(&self, table: &str) -> Vec<&IndexMetadata> {
let Some(table_meta) = self.catalog.get_table_in_txn(table, self.overlay) else {
return Vec::new();
};
let mut indexes: Vec<&IndexMetadata> = self
.catalog
.inner
.get_indexes_for_table(table)
.into_iter()
.filter(|idx| {
idx.catalog_name == table_meta.catalog_name
&& idx.namespace_name == table_meta.namespace_name
&& !self.catalog.index_hidden_by_overlay(idx, self.overlay)
})
.collect();
for idx in self.overlay.added_indexes.values() {
if idx.table == table
&& idx.catalog_name == table_meta.catalog_name
&& idx.namespace_name == table_meta.namespace_name
&& !self.catalog.index_hidden_by_overlay(idx, self.overlay)
{
indexes.push(idx);
}
}
indexes
}
fn drop_index(&mut self, _name: &str) -> Result<(), PlannerError> {
unreachable!("TxnCatalogView は参照専用です")
}
fn table_exists(&self, name: &str) -> bool {
self.catalog.table_exists_in_txn(name, self.overlay)
}
fn index_exists(&self, name: &str) -> bool {
self.catalog.index_exists_in_txn(name, self.overlay)
}
fn next_table_id(&mut self) -> u32 {
unreachable!("TxnCatalogView は参照専用です")
}
fn next_index_id(&mut self) -> u32 {
unreachable!("TxnCatalogView は参照専用です")
}
fn list_tables(&self) -> Vec<TableMetadata> {
let mut names = HashSet::new();
for name in self.catalog.inner.table_names() {
names.insert(name.to_string());
}
for fqn in self.overlay.added_tables.keys() {
names.insert(fqn.table.clone());
}
let mut tables = Vec::new();
for name in names {
if let Some(table) = self.catalog.get_table_in_txn(&name, self.overlay) {
tables.push(table.clone());
}
}
tables
}
}
#[derive(Debug)]
pub struct PersistentCatalog<S: KVStore> {
inner: MemoryCatalog,
store: Arc<S>,
catalogs: HashMap<String, CatalogMeta>,
namespaces: HashMap<(String, String), NamespaceMeta>,
}
impl<S: KVStore> PersistentCatalog<S> {
pub fn load(store: Arc<S>) -> Result<Self, CatalogError> {
let mut txn = store.begin(TxnMode::ReadOnly)?;
let meta_key = META_KEY.to_vec();
let mut meta_state: Option<CatalogState> = None;
if let Some(meta_bytes) = txn.get(&meta_key)? {
let meta: CatalogState = bincode::deserialize(&meta_bytes)?;
if meta.version > CATALOG_VERSION {
return Err(CatalogError::InvalidKey(format!(
"unsupported catalog version: {}",
meta.version
)));
}
meta_state = Some(meta);
}
let mut needs_migration = meta_state
.as_ref()
.is_some_and(|meta| meta.version < CATALOG_VERSION);
if !needs_migration && meta_state.is_none() {
for (key, _) in txn.scan_prefix(TABLES_PREFIX)? {
let suffix = key_suffix(TABLES_PREFIX, &key)?;
if !suffix.contains('/') {
needs_migration = true;
break;
}
}
if !needs_migration {
for (key, _) in txn.scan_prefix(INDEXES_PREFIX)? {
let suffix = key_suffix(INDEXES_PREFIX, &key)?;
if !suffix.contains('/') {
needs_migration = true;
break;
}
}
}
}
if needs_migration {
txn.rollback_self()?;
Self::migrate_v1_to_v2(&store)?;
return Self::load(store);
}
let mut inner = MemoryCatalog::new();
let mut catalogs = HashMap::new();
let mut namespaces = HashMap::new();
let mut max_table_id = 0u32;
let mut max_index_id = 0u32;
for (key, value) in txn.scan_prefix(CATALOGS_PREFIX)? {
let catalog_name = key_suffix(CATALOGS_PREFIX, &key)?;
let mut meta: CatalogMeta = bincode::deserialize(&value)?;
if meta.name != catalog_name {
meta.name = catalog_name.clone();
}
catalogs.insert(catalog_name, meta);
}
for (key, value) in txn.scan_prefix(NAMESPACES_PREFIX)? {
let suffix = key_suffix(NAMESPACES_PREFIX, &key)?;
let mut parts = suffix.splitn(2, '/');
let catalog_name = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.clone()))?;
let namespace_name = parts
.next()
.filter(|part| !part.is_empty())
.ok_or_else(|| CatalogError::InvalidKey(suffix.clone()))?;
let mut meta: NamespaceMeta = bincode::deserialize(&value)?;
if meta.catalog_name != catalog_name {
meta.catalog_name = catalog_name.to_string();
}
if meta.name != namespace_name {
meta.name = namespace_name.to_string();
}
namespaces.insert((meta.catalog_name.clone(), meta.name.clone()), meta);
}
for (key, value) in txn.scan_prefix(TABLES_PREFIX)? {
let suffix = key_suffix(TABLES_PREFIX, &key)?;
let fqn = parse_table_key_suffix(&suffix)?;
let mut persisted = deserialize_table_meta(&value)?;
if persisted.catalog_name != fqn.catalog {
persisted.catalog_name = fqn.catalog.clone();
}
if persisted.namespace_name != fqn.namespace {
persisted.namespace_name = fqn.namespace.clone();
}
if persisted.name != fqn.table {
persisted.name = fqn.table.clone();
}
max_table_id = max_table_id.max(persisted.table_id);
let table: TableMetadata = persisted.into();
inner.insert_table_unchecked(table);
}
for (key, value) in txn.scan_prefix(INDEXES_PREFIX)? {
let suffix = key_suffix(INDEXES_PREFIX, &key)?;
let fqn = parse_index_key_suffix(&suffix)?;
let mut persisted = deserialize_index_meta(&value)?;
if persisted.catalog_name != fqn.catalog {
persisted.catalog_name = fqn.catalog.clone();
}
if persisted.namespace_name != fqn.namespace {
persisted.namespace_name = fqn.namespace.clone();
}
if persisted.table != fqn.table {
persisted.table = fqn.table.clone();
}
if persisted.name != fqn.index {
persisted.name = fqn.index.clone();
}
max_index_id = max_index_id.max(persisted.index_id);
let mut index: IndexMetadata = persisted.into();
if let Some(table) = inner.get_table(&index.table) {
if index.catalog_name != table.catalog_name
|| index.namespace_name != table.namespace_name
{
index.catalog_name = table.catalog_name.clone();
index.namespace_name = table.namespace_name.clone();
}
inner.insert_index_unchecked(index);
}
}
let (mut table_id_counter, mut index_id_counter) = (max_table_id, max_index_id);
if let Some(meta) = meta_state
.as_ref()
.filter(|meta| meta.version == CATALOG_VERSION)
{
table_id_counter = table_id_counter.max(meta.table_id_counter);
index_id_counter = index_id_counter.max(meta.index_id_counter);
}
inner.set_counters(table_id_counter, index_id_counter);
txn.rollback_self()?;
Ok(Self {
inner,
store,
catalogs,
namespaces,
})
}
fn migrate_v1_to_v2(store: &Arc<S>) -> Result<(), CatalogError> {
let mut txn = store.begin(TxnMode::ReadWrite)?;
if txn.get(&catalog_key("default"))?.is_none() {
let meta = CatalogMeta {
name: "default".to_string(),
comment: None,
storage_root: None,
};
let value = bincode::serialize(&meta)?;
txn.put(catalog_key("default"), value)?;
}
if txn.get(&namespace_key("default", "default"))?.is_none() {
let meta = NamespaceMeta {
name: "default".to_string(),
catalog_name: "default".to_string(),
comment: None,
storage_root: None,
};
let value = bincode::serialize(&meta)?;
txn.put(namespace_key("default", "default"), value)?;
}
let mut table_updates = Vec::new();
let mut table_keys_to_delete = Vec::new();
let mut max_table_id = 0u32;
for (key, value) in txn.scan_prefix(TABLES_PREFIX)? {
let suffix = key_suffix(TABLES_PREFIX, &key)?;
if suffix.contains('/') {
continue;
}
let mut persisted = deserialize_table_meta(&value)?;
if persisted.catalog_name.is_empty() {
persisted.catalog_name = "default".to_string();
}
if persisted.namespace_name.is_empty() {
persisted.namespace_name = "default".to_string();
}
persisted.table_type = TableType::Managed;
persisted.data_source_format = DataSourceFormat::Alopex;
max_table_id = max_table_id.max(persisted.table_id);
let new_key = table_key(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.name,
);
let bytes = bincode::serialize(&persisted)?;
table_updates.push((new_key, bytes));
table_keys_to_delete.push(key);
}
for (key, value) in table_updates {
txn.put(key, value)?;
}
for key in table_keys_to_delete {
txn.delete(key)?;
}
let mut index_updates = Vec::new();
let mut index_keys_to_delete = Vec::new();
let mut max_index_id = 0u32;
for (key, value) in txn.scan_prefix(INDEXES_PREFIX)? {
let suffix = key_suffix(INDEXES_PREFIX, &key)?;
if suffix.contains('/') {
continue;
}
let mut persisted = deserialize_index_meta(&value)?;
if persisted.catalog_name.is_empty() {
persisted.catalog_name = "default".to_string();
}
if persisted.namespace_name.is_empty() {
persisted.namespace_name = "default".to_string();
}
max_index_id = max_index_id.max(persisted.index_id);
let new_key = index_key(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.table,
&persisted.name,
);
let bytes = bincode::serialize(&persisted)?;
index_updates.push((new_key, bytes));
index_keys_to_delete.push(key);
}
for (key, value) in index_updates {
txn.put(key, value)?;
}
for key in index_keys_to_delete {
txn.delete(key)?;
}
let mut table_id_counter = max_table_id;
let mut index_id_counter = max_index_id;
if let Some(meta_bytes) = txn.get(&META_KEY.to_vec())? {
let meta: CatalogState = bincode::deserialize(&meta_bytes)?;
table_id_counter = table_id_counter.max(meta.table_id_counter);
index_id_counter = index_id_counter.max(meta.index_id_counter);
}
let meta = CatalogState {
version: CATALOG_VERSION,
table_id_counter,
index_id_counter,
};
let meta_bytes = bincode::serialize(&meta)?;
txn.put(META_KEY.to_vec(), meta_bytes)?;
txn.commit_self()?;
Ok(())
}
pub fn new(store: Arc<S>) -> Self {
Self {
inner: MemoryCatalog::new(),
store,
catalogs: HashMap::new(),
namespaces: HashMap::new(),
}
}
pub fn store(&self) -> &Arc<S> {
&self.store
}
pub fn list_catalogs(&self) -> Vec<CatalogMeta> {
let mut catalogs: Vec<CatalogMeta> = self.catalogs.values().cloned().collect();
catalogs.sort_by(|a, b| a.name.cmp(&b.name));
catalogs
}
pub fn get_catalog(&self, name: &str) -> Option<CatalogMeta> {
self.catalogs.get(name).cloned()
}
pub fn create_catalog(&mut self, meta: CatalogMeta) -> Result<(), CatalogError> {
let mut txn = self.store.begin(TxnMode::ReadWrite)?;
let value = bincode::serialize(&meta)?;
txn.put(catalog_key(&meta.name), value)?;
txn.commit_self()?;
self.catalogs.insert(meta.name.clone(), meta);
Ok(())
}
pub fn delete_catalog(&mut self, name: &str) -> Result<(), CatalogError> {
let mut txn = self.store.begin(TxnMode::ReadWrite)?;
txn.delete(catalog_key(name))?;
let mut namespace_prefix = NAMESPACES_PREFIX.to_vec();
namespace_prefix.extend_from_slice(name.as_bytes());
namespace_prefix.push(b'/');
let mut namespace_keys = Vec::new();
for (key, _) in txn.scan_prefix(&namespace_prefix)? {
namespace_keys.push(key);
}
for key in namespace_keys {
txn.delete(key)?;
}
let mut table_keys = Vec::new();
let mut table_fqns = Vec::new();
for (key, value) in txn.scan_prefix(TABLES_PREFIX)? {
let persisted = deserialize_table_meta(&value)?;
if persisted.catalog_name == name {
table_fqns.push(TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.name,
));
table_keys.push(key);
}
}
let table_set: HashSet<TableFqn> = table_fqns.iter().cloned().collect();
for key in table_keys {
txn.delete(key)?;
}
if !table_set.is_empty() {
let mut index_keys = Vec::new();
for (key, value) in txn.scan_prefix(INDEXES_PREFIX)? {
let persisted = deserialize_index_meta(&value)?;
let fqn = TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.table,
);
if table_set.contains(&fqn) {
index_keys.push(key);
}
}
for key in index_keys {
txn.delete(key)?;
}
}
txn.commit_self()?;
self.catalogs.remove(name);
self.namespaces.retain(|(catalog, _), _| catalog != name);
for fqn in table_fqns {
self.inner.remove_table_unchecked(&fqn.table);
}
Ok(())
}
pub fn list_namespaces(&self, catalog_name: &str) -> Vec<NamespaceMeta> {
let mut namespaces: Vec<NamespaceMeta> = self
.namespaces
.values()
.filter(|meta| meta.catalog_name == catalog_name)
.cloned()
.collect();
namespaces.sort_by(|a, b| a.name.cmp(&b.name));
namespaces
}
pub fn get_namespace(&self, catalog_name: &str, namespace_name: &str) -> Option<NamespaceMeta> {
self.namespaces
.get(&(catalog_name.to_string(), namespace_name.to_string()))
.cloned()
}
pub fn create_namespace(&mut self, meta: NamespaceMeta) -> Result<(), CatalogError> {
if !self.catalogs.contains_key(&meta.catalog_name) {
return Err(CatalogError::InvalidKey(format!(
"catalog not found: {}",
meta.catalog_name
)));
}
let mut txn = self.store.begin(TxnMode::ReadWrite)?;
let value = bincode::serialize(&meta)?;
txn.put(namespace_key(&meta.catalog_name, &meta.name), value)?;
txn.commit_self()?;
self.namespaces
.insert((meta.catalog_name.clone(), meta.name.clone()), meta);
Ok(())
}
pub fn delete_namespace(
&mut self,
catalog_name: &str,
namespace_name: &str,
) -> Result<(), CatalogError> {
if !self.catalogs.contains_key(catalog_name) {
return Err(CatalogError::InvalidKey(format!(
"catalog not found: {}",
catalog_name
)));
}
let mut txn = self.store.begin(TxnMode::ReadWrite)?;
txn.delete(namespace_key(catalog_name, namespace_name))?;
txn.commit_self()?;
self.namespaces
.remove(&(catalog_name.to_string(), namespace_name.to_string()));
Ok(())
}
fn persist_create_catalog(
&mut self,
txn: &mut S::Transaction<'_>,
meta: &CatalogMeta,
) -> Result<(), CatalogError> {
let value = bincode::serialize(meta)?;
txn.put(catalog_key(&meta.name), value)?;
Ok(())
}
fn persist_drop_catalog(
&mut self,
txn: &mut S::Transaction<'_>,
name: &str,
) -> Result<(), CatalogError> {
txn.delete(catalog_key(name))?;
let mut namespace_prefix = NAMESPACES_PREFIX.to_vec();
namespace_prefix.extend_from_slice(name.as_bytes());
namespace_prefix.push(b'/');
let mut namespace_keys = Vec::new();
for (key, _) in txn.scan_prefix(&namespace_prefix)? {
namespace_keys.push(key);
}
for key in namespace_keys {
txn.delete(key)?;
}
let mut table_keys = Vec::new();
let mut table_fqns = Vec::new();
for (key, value) in txn.scan_prefix(TABLES_PREFIX)? {
let persisted = deserialize_table_meta(&value)?;
if persisted.catalog_name == name {
table_fqns.push(TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.name,
));
table_keys.push(key);
}
}
let table_set: HashSet<TableFqn> = table_fqns.iter().cloned().collect();
for key in table_keys {
txn.delete(key)?;
}
if !table_set.is_empty() {
let mut index_keys = Vec::new();
for (key, value) in txn.scan_prefix(INDEXES_PREFIX)? {
let persisted = deserialize_index_meta(&value)?;
let fqn = TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.table,
);
if table_set.contains(&fqn) {
index_keys.push(key);
}
}
for key in index_keys {
txn.delete(key)?;
}
}
Ok(())
}
fn persist_create_namespace(
&mut self,
txn: &mut S::Transaction<'_>,
meta: &NamespaceMeta,
) -> Result<(), CatalogError> {
let value = bincode::serialize(meta)?;
txn.put(namespace_key(&meta.catalog_name, &meta.name), value)?;
Ok(())
}
fn persist_drop_namespace(
&mut self,
txn: &mut S::Transaction<'_>,
catalog_name: &str,
namespace_name: &str,
) -> Result<(), CatalogError> {
txn.delete(namespace_key(catalog_name, namespace_name))?;
let mut table_keys = Vec::new();
let mut table_fqns = Vec::new();
for (key, value) in txn.scan_prefix(TABLES_PREFIX)? {
let persisted = deserialize_table_meta(&value)?;
if persisted.catalog_name == catalog_name && persisted.namespace_name == namespace_name
{
table_fqns.push(TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.name,
));
table_keys.push(key);
}
}
let table_set: HashSet<TableFqn> = table_fqns.iter().cloned().collect();
for key in table_keys {
txn.delete(key)?;
}
if !table_set.is_empty() {
let mut index_keys = Vec::new();
for (key, value) in txn.scan_prefix(INDEXES_PREFIX)? {
let persisted = deserialize_index_meta(&value)?;
let fqn = TableFqn::new(
&persisted.catalog_name,
&persisted.namespace_name,
&persisted.table,
);
if table_set.contains(&fqn) {
index_keys.push(key);
}
}
for key in index_keys {
txn.delete(key)?;
}
}
Ok(())
}
fn write_meta(&self, txn: &mut S::Transaction<'_>) -> Result<(), CatalogError> {
let (table_id_counter, index_id_counter) = self.inner.counters();
let meta = CatalogState {
version: CATALOG_VERSION,
table_id_counter,
index_id_counter,
};
let meta_bytes = bincode::serialize(&meta)?;
txn.put(META_KEY.to_vec(), meta_bytes)?;
Ok(())
}
pub fn persist_create_table(
&mut self,
txn: &mut S::Transaction<'_>,
table: &TableMetadata,
) -> Result<(), CatalogError> {
let persisted = PersistedTableMeta::from(table);
let value = bincode::serialize(&persisted)?;
txn.put(
table_key(&table.catalog_name, &table.namespace_name, &table.name),
value,
)?;
self.write_meta(txn)?;
Ok(())
}
pub fn persist_drop_table(
&mut self,
txn: &mut S::Transaction<'_>,
fqn: &TableFqn,
) -> Result<(), CatalogError> {
txn.delete(table_key(&fqn.catalog, &fqn.namespace, &fqn.table))?;
let mut to_delete: Vec<String> = Vec::new();
let prefix = index_prefix(&fqn.catalog, &fqn.namespace, &fqn.table);
for (key, _) in txn.scan_prefix(&prefix)? {
let index_name = key_suffix(&prefix, &key)?;
to_delete.push(index_name);
}
for index_name in to_delete {
txn.delete(index_key(
&fqn.catalog,
&fqn.namespace,
&fqn.table,
&index_name,
))?;
}
Ok(())
}
pub fn persist_create_index(
&mut self,
txn: &mut S::Transaction<'_>,
index: &IndexMetadata,
) -> Result<(), CatalogError> {
let persisted = PersistedIndexMeta::from(index);
let value = bincode::serialize(&persisted)?;
txn.put(
index_key(
&index.catalog_name,
&index.namespace_name,
&index.table,
&index.name,
),
value,
)?;
self.write_meta(txn)?;
Ok(())
}
pub fn persist_drop_index(
&mut self,
txn: &mut S::Transaction<'_>,
fqn: &IndexFqn,
) -> Result<(), CatalogError> {
txn.delete(index_key(
&fqn.catalog,
&fqn.namespace,
&fqn.table,
&fqn.index,
))?;
Ok(())
}
pub fn persist_overlay(
&mut self,
txn: &mut S::Transaction<'_>,
overlay: &CatalogOverlay,
) -> Result<(), CatalogError> {
self.ensure_overlay_name_uniqueness(overlay)?;
for catalog in overlay.dropped_catalogs.iter() {
self.persist_drop_catalog(txn, catalog)?;
}
for (catalog, namespace) in overlay.dropped_namespaces.iter() {
if overlay.dropped_catalogs.contains(catalog) {
continue;
}
self.persist_drop_namespace(txn, catalog, namespace)?;
}
for fqn in overlay.dropped_tables.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
{
continue;
}
self.persist_drop_table(txn, fqn)?;
}
for fqn in overlay.dropped_indexes.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
{
continue;
}
self.persist_drop_index(txn, fqn)?;
}
for meta in overlay.added_catalogs.values() {
if overlay.dropped_catalogs.contains(&meta.name) {
continue;
}
self.persist_create_catalog(txn, meta)?;
}
for meta in overlay.added_namespaces.values() {
if overlay.dropped_catalogs.contains(&meta.catalog_name)
|| overlay
.dropped_namespaces
.contains(&(meta.catalog_name.clone(), meta.name.clone()))
{
continue;
}
self.persist_create_namespace(txn, meta)?;
}
for (fqn, table) in overlay.added_tables.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_tables.contains(fqn)
{
continue;
}
self.persist_create_table(txn, table)?;
}
for (fqn, index) in overlay.added_indexes.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_indexes.contains(fqn)
{
continue;
}
self.persist_create_index(txn, index)?;
}
Ok(())
}
fn ensure_overlay_name_uniqueness(&self, overlay: &CatalogOverlay) -> Result<(), CatalogError> {
let mut table_names: HashMap<String, TableFqn> = HashMap::new();
for name in self.inner.table_names() {
let Some(table) = self.inner.get_table(name) else {
continue;
};
let fqn = TableFqn::from(table);
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_tables.contains(&fqn)
{
continue;
}
table_names.insert(table.name.clone(), fqn);
}
for (fqn, table) in overlay.added_tables.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_tables.contains(fqn)
{
continue;
}
if let Some(existing) = table_names.get(&table.name)
&& existing != fqn
{
return Err(CatalogError::InvalidKey(format!(
"table name '{}' conflicts across namespaces: {}.{} vs {}.{}",
table.name, existing.catalog, existing.namespace, fqn.catalog, fqn.namespace
)));
}
table_names.insert(table.name.clone(), fqn.clone());
}
let mut index_names: HashMap<String, IndexFqn> = HashMap::new();
for name in self.inner.index_names() {
let Some(index) = self.inner.get_index(name) else {
continue;
};
let fqn = IndexFqn::from(index);
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_indexes.contains(&fqn)
|| overlay.dropped_tables.contains(&TableFqn::new(
&fqn.catalog,
&fqn.namespace,
&fqn.table,
))
{
continue;
}
index_names.insert(index.name.clone(), fqn);
}
for (fqn, index) in overlay.added_indexes.iter() {
if overlay.dropped_catalogs.contains(&fqn.catalog)
|| overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
|| overlay.dropped_indexes.contains(fqn)
|| overlay.dropped_tables.contains(&TableFqn::new(
&fqn.catalog,
&fqn.namespace,
&fqn.table,
))
{
continue;
}
if let Some(existing) = index_names.get(&index.name)
&& existing != fqn
{
return Err(CatalogError::InvalidKey(format!(
"index name '{}' conflicts across namespaces: {}.{} vs {}.{}",
index.name, existing.catalog, existing.namespace, fqn.catalog, fqn.namespace
)));
}
index_names.insert(index.name.clone(), fqn.clone());
}
Ok(())
}
fn namespace_dropped(overlay: &CatalogOverlay, catalog: &str, namespace: &str) -> bool {
overlay.dropped_catalogs.contains(catalog)
|| overlay
.dropped_namespaces
.contains(&(catalog.to_string(), namespace.to_string()))
}
fn overlay_added_table_by_name<'a>(
overlay: &'a CatalogOverlay,
name: &str,
) -> Option<&'a TableMetadata> {
let mut iter = overlay
.added_tables
.values()
.filter(|table| table.name == name);
let first = iter.next()?;
if iter.next().is_some() {
return None;
}
Some(first)
}
fn overlay_added_index_by_name<'a>(
overlay: &'a CatalogOverlay,
name: &str,
) -> Option<&'a IndexMetadata> {
let mut iter = overlay
.added_indexes
.values()
.filter(|index| index.name == name);
let first = iter.next()?;
if iter.next().is_some() {
return None;
}
Some(first)
}
fn base_table_conflicts_with_overlay(
&self,
overlay: &CatalogOverlay,
table: &TableMetadata,
) -> bool {
let Some(base) = self.inner.get_table(&table.name) else {
return false;
};
if self.table_hidden_by_overlay(base, overlay) {
return false;
}
if overlay.dropped_tables.contains(&TableFqn::from(base)) {
return false;
}
TableFqn::from(base) != TableFqn::from(table)
}
fn base_index_conflicts_with_overlay(
&self,
overlay: &CatalogOverlay,
index: &IndexMetadata,
) -> bool {
let Some(base) = self.inner.get_index(&index.name) else {
return false;
};
if Self::namespace_dropped(overlay, &base.catalog_name, &base.namespace_name) {
return false;
}
if overlay.dropped_indexes.contains(&IndexFqn::from(base)) {
return false;
}
if self.dropped_table_matches_fqn(
&base.table,
&base.catalog_name,
&base.namespace_name,
overlay,
) {
return false;
}
IndexFqn::from(base) != IndexFqn::from(index)
}
fn table_hidden_by_overlay(&self, table: &TableMetadata, overlay: &CatalogOverlay) -> bool {
Self::namespace_dropped(overlay, &table.catalog_name, &table.namespace_name)
}
fn dropped_table_matches_fqn(
&self,
table_name: &str,
catalog: &str,
namespace: &str,
overlay: &CatalogOverlay,
) -> bool {
let fqn = TableFqn::new(catalog, namespace, table_name);
overlay.dropped_tables.contains(&fqn)
}
fn index_hidden_by_overlay(&self, index: &IndexMetadata, overlay: &CatalogOverlay) -> bool {
let index_fqn = IndexFqn::from(index);
if overlay.dropped_indexes.contains(&index_fqn) {
return true;
}
if Self::namespace_dropped(overlay, &index.catalog_name, &index.namespace_name) {
return true;
}
if self.dropped_table_matches_fqn(
&index.table,
&index.catalog_name,
&index.namespace_name,
overlay,
) {
return true;
}
match self.get_table_in_txn(&index.table, overlay) {
Some(table) => {
table.catalog_name != index.catalog_name
|| table.namespace_name != index.namespace_name
}
None => true,
}
}
pub fn get_catalog_in_txn<'a>(
&'a self,
name: &str,
overlay: &'a CatalogOverlay,
) -> Option<&'a CatalogMeta> {
if overlay.dropped_catalogs.contains(name) {
return None;
}
if let Some(catalog) = overlay.added_catalogs.get(name) {
return Some(catalog);
}
self.catalogs.get(name)
}
pub fn get_namespace_in_txn<'a>(
&'a self,
catalog_name: &str,
namespace_name: &str,
overlay: &'a CatalogOverlay,
) -> Option<&'a NamespaceMeta> {
if overlay.dropped_catalogs.contains(catalog_name) {
return None;
}
let key = (catalog_name.to_string(), namespace_name.to_string());
if overlay.dropped_namespaces.contains(&key) {
return None;
}
if let Some(namespace) = overlay.added_namespaces.get(&key) {
return Some(namespace);
}
self.namespaces.get(&key)
}
pub fn list_catalogs_in_txn(&self, overlay: &CatalogOverlay) -> Vec<CatalogMeta> {
let mut catalogs: HashMap<String, CatalogMeta> = HashMap::new();
for (name, meta) in &self.catalogs {
if !overlay.dropped_catalogs.contains(name) {
catalogs.insert(name.clone(), meta.clone());
}
}
for (name, meta) in &overlay.added_catalogs {
if !overlay.dropped_catalogs.contains(name) {
catalogs.insert(name.clone(), meta.clone());
}
}
let mut values: Vec<CatalogMeta> = catalogs.into_values().collect();
values.sort_by(|a, b| a.name.cmp(&b.name));
values
}
pub fn list_namespaces_in_txn(
&self,
catalog_name: &str,
overlay: &CatalogOverlay,
) -> Vec<NamespaceMeta> {
if overlay.dropped_catalogs.contains(catalog_name) {
return Vec::new();
}
let mut namespaces: HashMap<(String, String), NamespaceMeta> = HashMap::new();
for ((catalog, namespace), meta) in &self.namespaces {
if catalog != catalog_name {
continue;
}
let key = (catalog.clone(), namespace.clone());
if overlay.dropped_namespaces.contains(&key) {
continue;
}
namespaces.insert(key, meta.clone());
}
for ((catalog, namespace), meta) in &overlay.added_namespaces {
if catalog != catalog_name {
continue;
}
let key = (catalog.clone(), namespace.clone());
if overlay.dropped_namespaces.contains(&key) {
continue;
}
namespaces.insert(key, meta.clone());
}
let mut values: Vec<NamespaceMeta> = namespaces.into_values().collect();
values.sort_by(|a, b| a.name.cmp(&b.name));
values
}
pub fn table_exists_in_txn(&self, name: &str, overlay: &CatalogOverlay) -> bool {
if let Some(table) = Self::overlay_added_table_by_name(overlay, name) {
if self.table_hidden_by_overlay(table, overlay) {
return false;
}
if self.base_table_conflicts_with_overlay(overlay, table) {
return false;
}
return true;
}
match self.inner.get_table(name) {
Some(table) => {
!self.table_hidden_by_overlay(table, overlay)
&& !overlay.dropped_tables.contains(&TableFqn::from(table))
}
None => false,
}
}
pub fn get_table_in_txn<'a>(
&'a self,
name: &str,
overlay: &'a CatalogOverlay,
) -> Option<&'a TableMetadata> {
if let Some(table) = Self::overlay_added_table_by_name(overlay, name) {
if self.table_hidden_by_overlay(table, overlay) {
return None;
}
if self.base_table_conflicts_with_overlay(overlay, table) {
return None;
}
return Some(table);
}
self.inner.get_table(name).filter(|table| {
!self.table_hidden_by_overlay(table, overlay)
&& !overlay.dropped_tables.contains(&TableFqn::from(*table))
})
}
pub fn index_exists_in_txn(&self, name: &str, overlay: &CatalogOverlay) -> bool {
if let Some(index) = Self::overlay_added_index_by_name(overlay, name) {
if self.index_hidden_by_overlay(index, overlay) {
return false;
}
if self.base_index_conflicts_with_overlay(overlay, index) {
return false;
}
return true;
}
match self.inner.get_index(name) {
Some(index) => !self.index_hidden_by_overlay(index, overlay),
None => false,
}
}
pub fn get_index_in_txn<'a>(
&'a self,
name: &str,
overlay: &'a CatalogOverlay,
) -> Option<&'a IndexMetadata> {
if let Some(index) = Self::overlay_added_index_by_name(overlay, name) {
if self.index_hidden_by_overlay(index, overlay) {
return None;
}
if self.base_index_conflicts_with_overlay(overlay, index) {
return None;
}
return Some(index);
}
match self.inner.get_index(name) {
Some(index) if self.index_hidden_by_overlay(index, overlay) => None,
other => other,
}
}
pub fn list_tables_in_txn(
&self,
catalog_name: &str,
namespace_name: &str,
overlay: &CatalogOverlay,
) -> Vec<TableMetadata> {
if overlay.dropped_catalogs.contains(catalog_name) {
return Vec::new();
}
if overlay
.dropped_namespaces
.contains(&(catalog_name.to_string(), namespace_name.to_string()))
{
return Vec::new();
}
let mut tables: HashMap<TableFqn, TableMetadata> = HashMap::new();
for name in self.inner.table_names() {
if let Some(table) = self.inner.get_table(name)
&& table.catalog_name == catalog_name
&& table.namespace_name == namespace_name
{
let fqn = TableFqn::from(table);
if !overlay.dropped_tables.contains(&fqn) {
tables.insert(fqn, table.clone());
}
}
}
for table in overlay.added_tables.values() {
if table.catalog_name == catalog_name && table.namespace_name == namespace_name {
tables.insert(TableFqn::from(table), table.clone());
}
}
let mut values: Vec<TableMetadata> = tables.into_values().collect();
values.sort_by(|a, b| a.name.cmp(&b.name));
values
}
pub fn list_indexes_in_txn(
&self,
fqn: &TableFqn,
overlay: &CatalogOverlay,
) -> Vec<IndexMetadata> {
if overlay.dropped_catalogs.contains(&fqn.catalog) {
return Vec::new();
}
if overlay
.dropped_namespaces
.contains(&(fqn.catalog.clone(), fqn.namespace.clone()))
{
return Vec::new();
}
if overlay
.dropped_tables
.contains(&TableFqn::new(&fqn.catalog, &fqn.namespace, &fqn.table))
{
return Vec::new();
}
let mut indexes: HashMap<IndexFqn, IndexMetadata> = HashMap::new();
for index in self.inner.get_indexes_for_table(&fqn.table) {
if index.catalog_name == fqn.catalog && index.namespace_name == fqn.namespace {
let index_fqn = IndexFqn::from(index);
if !overlay.dropped_indexes.contains(&index_fqn) {
indexes.insert(index_fqn, index.clone());
}
}
}
for index in overlay.added_indexes.values() {
if index.table == fqn.table
&& index.catalog_name == fqn.catalog
&& index.namespace_name == fqn.namespace
{
indexes.insert(IndexFqn::from(index), index.clone());
}
}
let mut values: Vec<IndexMetadata> = indexes.into_values().collect();
values.sort_by(|a, b| a.name.cmp(&b.name));
values
}
pub fn apply_overlay(&mut self, overlay: CatalogOverlay) {
let CatalogOverlay {
added_catalogs,
dropped_catalogs,
added_namespaces,
dropped_namespaces,
added_tables,
dropped_tables,
added_indexes,
dropped_indexes,
} = overlay;
for (name, meta) in added_catalogs {
self.catalogs.insert(name, meta);
}
for (catalog_name, namespace_name) in dropped_namespaces.iter() {
self.namespaces
.remove(&(catalog_name.clone(), namespace_name.clone()));
}
for ((catalog_name, namespace_name), meta) in added_namespaces {
self.namespaces.insert((catalog_name, namespace_name), meta);
}
for name in dropped_catalogs {
self.catalogs.remove(&name);
self.namespaces.retain(|(catalog, _), _| catalog != &name);
}
for (_, table) in added_tables {
self.inner.insert_table_unchecked(table);
}
for fqn in dropped_tables {
self.inner.remove_table_unchecked(&fqn.table);
}
for (_, index) in added_indexes {
self.inner.insert_index_unchecked(index);
}
for fqn in dropped_indexes {
self.inner.remove_index_unchecked(&fqn.index);
}
}
pub fn discard_overlay(_overlay: CatalogOverlay) {}
}
impl<S: KVStore> Catalog for PersistentCatalog<S> {
fn create_table(&mut self, table: TableMetadata) -> Result<(), PlannerError> {
self.inner.create_table(table)
}
fn get_table(&self, name: &str) -> Option<&TableMetadata> {
self.inner.get_table(name)
}
fn drop_table(&mut self, name: &str) -> Result<(), PlannerError> {
self.inner.drop_table(name)
}
fn create_index(&mut self, index: IndexMetadata) -> Result<(), PlannerError> {
self.inner.create_index(index)
}
fn get_index(&self, name: &str) -> Option<&IndexMetadata> {
self.inner.get_index(name)
}
fn get_indexes_for_table(&self, table: &str) -> Vec<&IndexMetadata> {
self.inner.get_indexes_for_table(table)
}
fn drop_index(&mut self, name: &str) -> Result<(), PlannerError> {
self.inner.drop_index(name)
}
fn table_exists(&self, name: &str) -> bool {
self.inner.table_exists(name)
}
fn index_exists(&self, name: &str) -> bool {
self.inner.index_exists(name)
}
fn next_table_id(&mut self) -> u32 {
self.inner.next_table_id()
}
fn next_index_id(&mut self) -> u32 {
self.inner.next_index_id()
}
fn list_tables(&self) -> Vec<TableMetadata> {
let mut tables = Vec::new();
for name in self.inner.table_names() {
if let Some(table) = self.inner.get_table(name) {
tables.push(table.clone());
}
}
tables
}
fn persistence_enabled(&self) -> bool {
true
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::planner::types::ResolvedType;
use std::collections::HashSet;
fn test_table(name: &str, id: u32) -> TableMetadata {
TableMetadata::new(
name,
vec![ColumnMetadata::new("id", ResolvedType::Integer).with_primary_key(true)],
)
.with_table_id(id)
.with_primary_key(vec!["id".to_string()])
}
fn legacy_table_key(table_name: &str) -> Vec<u8> {
let mut key = TABLES_PREFIX.to_vec();
key.extend_from_slice(table_name.as_bytes());
key
}
fn legacy_index_key(index_name: &str) -> Vec<u8> {
let mut key = INDEXES_PREFIX.to_vec();
key.extend_from_slice(index_name.as_bytes());
key
}
fn seed_legacy_store(store: &Arc<alopex_core::kv::memory::MemoryKV>) {
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
let table = test_table("users", 7);
let legacy_table = PersistedTableMetaV1 {
table_id: table.table_id,
name: table.name.clone(),
columns: table
.columns
.iter()
.map(PersistedColumnMeta::from)
.collect(),
primary_key: table.primary_key.clone(),
storage_options: table.storage_options.clone().into(),
};
let table_bytes = bincode::serialize(&legacy_table).unwrap();
txn.put(legacy_table_key("users"), table_bytes).unwrap();
let legacy_index = PersistedIndexMetaV1 {
index_id: 3,
name: "idx_users_id".to_string(),
table: "users".to_string(),
columns: vec!["id".to_string()],
column_indices: vec![0],
unique: false,
method: Some(PersistedIndexType::BTree),
options: Vec::new(),
};
let index_bytes = bincode::serialize(&legacy_index).unwrap();
txn.put(legacy_index_key("idx_users_id"), index_bytes)
.unwrap();
let meta = CatalogState {
version: 1,
table_id_counter: 7,
index_id_counter: 3,
};
let meta_bytes = bincode::serialize(&meta).unwrap();
txn.put(META_KEY.to_vec(), meta_bytes).unwrap();
txn.commit_self().unwrap();
}
#[test]
fn load_empty_store() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let catalog = PersistentCatalog::load(store).unwrap();
assert_eq!(catalog.inner.table_count(), 0);
assert_eq!(catalog.inner.index_count(), 0);
}
#[test]
fn load_migrates_v1_keys_and_meta() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
seed_legacy_store(&store);
let reloaded = PersistentCatalog::load(store.clone()).unwrap();
assert!(reloaded.get_catalog("default").is_some());
assert!(reloaded.get_namespace("default", "default").is_some());
let table = reloaded.get_table("users").unwrap();
assert_eq!(table.catalog_name, "default");
assert_eq!(table.namespace_name, "default");
let index = reloaded.get_index("idx_users_id").unwrap();
assert_eq!(index.catalog_name, "default");
assert_eq!(index.namespace_name, "default");
assert_eq!(index.table, "users");
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
assert!(txn.get(&legacy_table_key("users")).unwrap().is_none());
assert!(
txn.get(&table_key("default", "default", "users"))
.unwrap()
.is_some()
);
assert!(
txn.get(&legacy_index_key("idx_users_id"))
.unwrap()
.is_none()
);
assert!(
txn.get(&index_key("default", "default", "users", "idx_users_id"))
.unwrap()
.is_some()
);
let meta_bytes = txn.get(&META_KEY.to_vec()).unwrap().unwrap();
let meta: CatalogState = bincode::deserialize(&meta_bytes).unwrap();
assert_eq!(meta.version, CATALOG_VERSION);
txn.rollback_self().unwrap();
}
#[test]
fn load_after_migration_keeps_v2_keys() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
seed_legacy_store(&store);
let _ = PersistentCatalog::load(store.clone()).unwrap();
let reloaded = PersistentCatalog::load(store.clone()).unwrap();
let table = reloaded.get_table("users").unwrap();
assert_eq!(table.catalog_name, "default");
assert_eq!(table.namespace_name, "default");
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
assert!(txn.get(&legacy_table_key("users")).unwrap().is_none());
assert!(
txn.get(&table_key("default", "default", "users"))
.unwrap()
.is_some()
);
let meta_bytes = txn.get(&META_KEY.to_vec()).unwrap().unwrap();
let meta: CatalogState = bincode::deserialize(&meta_bytes).unwrap();
assert_eq!(meta.version, CATALOG_VERSION);
txn.rollback_self().unwrap();
}
#[test]
fn create_table_persists() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog.inner.set_counters(1, 0);
let table = test_table("users", 1);
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
txn.commit_self().unwrap();
let reloaded = PersistentCatalog::load(store).unwrap();
assert!(reloaded.table_exists("users"));
assert_eq!(reloaded.get_table("users").unwrap().table_id, 1);
}
#[test]
fn drop_table_removes() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog.inner.set_counters(1, 0);
let table = test_table("users", 1);
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
txn.commit_self().unwrap();
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
let fqn = TableFqn::new("default", "default", "users");
catalog.persist_drop_table(&mut txn, &fqn).unwrap();
txn.commit_self().unwrap();
let reloaded = PersistentCatalog::load(store).unwrap();
assert!(!reloaded.table_exists("users"));
}
#[test]
fn reload_preserves_state() {
let temp_dir = tempfile::tempdir().unwrap();
let wal_path = temp_dir.path().join("catalog.wal");
let store = Arc::new(alopex_core::kv::memory::MemoryKV::open(&wal_path).unwrap());
let mut catalog = PersistentCatalog::new(store.clone());
catalog.inner.set_counters(1, 0);
let table = test_table("users", 1);
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
txn.commit_self().unwrap();
store.flush().unwrap();
drop(catalog);
drop(store);
let store = Arc::new(alopex_core::kv::memory::MemoryKV::open(&wal_path).unwrap());
let reloaded = PersistentCatalog::load(store).unwrap();
assert!(reloaded.table_exists("users"));
}
#[test]
fn overlay_applied_on_commit() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let users = test_table("users", 1);
catalog.inner.insert_table_unchecked(users.clone());
let mut overlay = CatalogOverlay::new();
overlay.drop_table(&TableFqn::from(&users));
let orders = test_table("orders", 2);
overlay.add_table(TableFqn::from(&orders), orders);
assert!(!catalog.table_exists_in_txn("users", &overlay));
assert!(catalog.table_exists_in_txn("orders", &overlay));
catalog.apply_overlay(overlay);
assert!(!catalog.table_exists("users"));
assert!(catalog.table_exists("orders"));
}
#[test]
fn overlay_discarded_on_rollback() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let users = test_table("users", 1);
catalog.inner.insert_table_unchecked(users.clone());
let mut overlay = CatalogOverlay::new();
overlay.drop_table(&TableFqn::from(&users));
PersistentCatalog::<alopex_core::kv::memory::MemoryKV>::discard_overlay(overlay);
assert!(catalog.table_exists("users"));
}
#[test]
fn catalog_crud_persists() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
let meta = CatalogMeta {
name: "main".to_string(),
comment: Some("primary".to_string()),
storage_root: Some("/tmp/alopex".to_string()),
};
catalog.create_catalog(meta.clone()).unwrap();
assert!(catalog.get_catalog("main").is_some());
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
let stored = txn.get(&catalog_key("main")).unwrap().unwrap();
let decoded: CatalogMeta = bincode::deserialize(&stored).unwrap();
txn.rollback_self().unwrap();
assert_eq!(decoded, meta);
catalog.delete_catalog("main").unwrap();
assert!(catalog.get_catalog("main").is_none());
}
#[test]
fn namespace_crud_persists_and_validates_catalog() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
let missing_catalog = NamespaceMeta {
name: "analytics".to_string(),
catalog_name: "missing".to_string(),
comment: None,
storage_root: None,
};
let err = catalog.create_namespace(missing_catalog).unwrap_err();
assert!(matches!(err, CatalogError::InvalidKey(_)));
catalog
.create_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
let namespace = NamespaceMeta {
name: "analytics".to_string(),
catalog_name: "main".to_string(),
comment: Some("warehouse".to_string()),
storage_root: None,
};
catalog.create_namespace(namespace.clone()).unwrap();
assert!(catalog.get_namespace("main", "analytics").is_some());
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
let stored = txn
.get(&namespace_key("main", "analytics"))
.unwrap()
.unwrap();
let decoded: NamespaceMeta = bincode::deserialize(&stored).unwrap();
txn.rollback_self().unwrap();
assert_eq!(decoded, namespace);
catalog.delete_namespace("main", "analytics").unwrap();
assert!(catalog.get_namespace("main", "analytics").is_none());
}
#[test]
fn delete_catalog_removes_namespaces_from_store() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog
.create_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
catalog
.create_namespace(NamespaceMeta {
name: "analytics".to_string(),
catalog_name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
catalog.delete_catalog("main").unwrap();
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
let mut prefix = NAMESPACES_PREFIX.to_vec();
prefix.extend_from_slice(b"main");
prefix.push(b'/');
let remaining: Vec<_> = txn.scan_prefix(&prefix).unwrap().collect();
txn.rollback_self().unwrap();
assert!(remaining.is_empty());
assert!(catalog.list_namespaces("main").is_empty());
}
#[test]
fn delete_catalog_removes_tables_and_indexes_from_store() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog
.create_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
let mut table = test_table("users", 1);
table.catalog_name = "main".to_string();
table.namespace_name = "default".to_string();
let mut index = IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
index.catalog_name = "main".to_string();
index.namespace_name = "default".to_string();
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
catalog.persist_create_index(&mut txn, &index).unwrap();
txn.commit_self().unwrap();
catalog.inner.insert_table_unchecked(table);
catalog.inner.insert_index_unchecked(index);
catalog.delete_catalog("main").unwrap();
assert!(catalog.inner.get_table("users").is_none());
assert!(catalog.inner.get_index("idx_users_id").is_none());
let mut txn = store.begin(TxnMode::ReadOnly).unwrap();
assert!(
txn.get(&table_key("main", "default", "users"))
.unwrap()
.is_none()
);
assert!(
txn.get(&index_key("main", "default", "users", "idx_users_id"))
.unwrap()
.is_none()
);
txn.rollback_self().unwrap();
}
#[test]
fn index_meta_loads_catalog_and_namespace_from_table() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog.inner.set_counters(1, 1);
let mut table = test_table("users", 1);
table.catalog_name = "main".to_string();
table.namespace_name = "analytics".to_string();
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
let mut index = IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0])
.with_method(IndexMethod::BTree);
index.catalog_name = "main".to_string();
index.namespace_name = "analytics".to_string();
catalog.persist_create_index(&mut txn, &index).unwrap();
txn.commit_self().unwrap();
let reloaded = PersistentCatalog::load(store).unwrap();
let index = reloaded.get_index("idx_users_id").unwrap();
assert_eq!(index.catalog_name, "main");
assert_eq!(index.namespace_name, "analytics");
}
#[test]
fn legacy_index_meta_loads_catalog_and_namespace_from_table() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
catalog.inner.set_counters(1, 1);
let mut table = test_table("users", 1);
table.catalog_name = "main".to_string();
table.namespace_name = "analytics".to_string();
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
catalog.persist_create_table(&mut txn, &table).unwrap();
let legacy = PersistedIndexMetaV1 {
index_id: 1,
name: "idx_users_id".to_string(),
table: "users".to_string(),
columns: vec!["id".to_string()],
column_indices: vec![0],
unique: false,
method: Some(PersistedIndexType::BTree),
options: Vec::new(),
};
let bytes = bincode::serialize(&legacy).unwrap();
txn.put(
index_key("main", "analytics", "users", "idx_users_id"),
bytes,
)
.unwrap();
txn.commit_self().unwrap();
let reloaded = PersistentCatalog::load(store).unwrap();
let index = reloaded.get_index("idx_users_id").unwrap();
assert_eq!(index.catalog_name, "main");
assert_eq!(index.namespace_name, "analytics");
}
#[test]
fn overlay_catalog_get_and_list() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
catalog
.create_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
let mut overlay = CatalogOverlay::new();
overlay.add_catalog(CatalogMeta {
name: "temp".to_string(),
comment: None,
storage_root: None,
});
overlay.drop_catalog("main");
assert!(catalog.get_catalog_in_txn("main", &overlay).is_none());
assert!(catalog.get_catalog_in_txn("temp", &overlay).is_some());
let names: Vec<String> = catalog
.list_catalogs_in_txn(&overlay)
.into_iter()
.map(|meta| meta.name)
.collect();
assert_eq!(names, vec!["temp".to_string()]);
}
#[test]
fn overlay_namespace_get_and_list() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
catalog
.create_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
catalog
.create_namespace(NamespaceMeta {
name: "default".to_string(),
catalog_name: "main".to_string(),
comment: None,
storage_root: None,
})
.unwrap();
let mut overlay = CatalogOverlay::new();
overlay.add_namespace(NamespaceMeta {
name: "analytics".to_string(),
catalog_name: "main".to_string(),
comment: None,
storage_root: None,
});
overlay.drop_namespace("main", "default");
assert!(
catalog
.get_namespace_in_txn("main", "default", &overlay)
.is_none()
);
assert!(
catalog
.get_namespace_in_txn("main", "analytics", &overlay)
.is_some()
);
let names: Vec<String> = catalog
.list_namespaces_in_txn("main", &overlay)
.into_iter()
.map(|meta| meta.name)
.collect();
assert_eq!(names, vec!["analytics".to_string()]);
}
#[test]
fn overlay_table_and_index_list() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let mut users = test_table("users", 1);
users.catalog_name = "main".to_string();
users.namespace_name = "default".to_string();
catalog.inner.insert_table_unchecked(users.clone());
let mut users_index =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
users_index.catalog_name = "main".to_string();
users_index.namespace_name = "default".to_string();
catalog.inner.insert_index_unchecked(users_index);
let mut overlay = CatalogOverlay::new();
let mut orders = test_table("orders", 2);
orders.catalog_name = "main".to_string();
orders.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&orders), orders.clone());
let mut orders_index =
IndexMetadata::new(2, "idx_orders_id", "orders", vec!["id".to_string()])
.with_column_indices(vec![0]);
orders_index.catalog_name = "main".to_string();
orders_index.namespace_name = "default".to_string();
overlay.add_index(IndexFqn::from(&orders_index), orders_index);
overlay.drop_table(&TableFqn::from(&users));
let table_names: Vec<String> = catalog
.list_tables_in_txn("main", "default", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert_eq!(table_names, vec!["orders".to_string()]);
let users_fqn = TableFqn::new("main", "default", "users");
assert!(catalog.list_indexes_in_txn(&users_fqn, &overlay).is_empty());
let orders_fqn = TableFqn::new("main", "default", "orders");
let index_names: Vec<String> = catalog
.list_indexes_in_txn(&orders_fqn, &overlay)
.into_iter()
.map(|index| index.name)
.collect();
assert_eq!(index_names, vec!["idx_orders_id".to_string()]);
}
#[test]
fn overlay_name_lookup_ambiguous_returns_none() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let catalog = PersistentCatalog::new(store);
let mut overlay = CatalogOverlay::new();
let mut users_default = test_table("users", 1);
users_default.catalog_name = "main".to_string();
users_default.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users_default), users_default);
let mut users_analytics = test_table("users", 2);
users_analytics.catalog_name = "main".to_string();
users_analytics.namespace_name = "analytics".to_string();
overlay.add_table(TableFqn::from(&users_analytics), users_analytics);
assert!(catalog.get_table_in_txn("users", &overlay).is_none());
assert!(!catalog.table_exists_in_txn("users", &overlay));
let mut idx_default =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_default.catalog_name = "main".to_string();
idx_default.namespace_name = "default".to_string();
overlay.add_index(IndexFqn::from(&idx_default), idx_default);
let mut idx_analytics =
IndexMetadata::new(2, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_analytics.catalog_name = "main".to_string();
idx_analytics.namespace_name = "analytics".to_string();
overlay.add_index(IndexFqn::from(&idx_analytics), idx_analytics);
assert!(catalog.get_index_in_txn("idx_users_id", &overlay).is_none());
assert!(!catalog.index_exists_in_txn("idx_users_id", &overlay));
}
#[test]
fn overlay_name_lookup_ambiguous_with_base_returns_none() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let mut overlay = CatalogOverlay::new();
let mut base_users = test_table("users", 1);
base_users.catalog_name = "main".to_string();
base_users.namespace_name = "default".to_string();
catalog.inner.insert_table_unchecked(base_users);
let mut overlay_users = test_table("users", 2);
overlay_users.catalog_name = "main".to_string();
overlay_users.namespace_name = "analytics".to_string();
overlay.add_table(TableFqn::from(&overlay_users), overlay_users);
assert!(catalog.get_table_in_txn("users", &overlay).is_none());
assert!(!catalog.table_exists_in_txn("users", &overlay));
let mut base_index = IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
base_index.catalog_name = "main".to_string();
base_index.namespace_name = "default".to_string();
catalog.inner.insert_index_unchecked(base_index);
let mut overlay_index =
IndexMetadata::new(2, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
overlay_index.catalog_name = "main".to_string();
overlay_index.namespace_name = "analytics".to_string();
overlay.add_index(IndexFqn::from(&overlay_index), overlay_index);
assert!(catalog.get_index_in_txn("idx_users_id", &overlay).is_none());
assert!(!catalog.index_exists_in_txn("idx_users_id", &overlay));
}
#[test]
fn overlay_fqn_tables_separate_namespaces() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let catalog = PersistentCatalog::new(store);
let mut overlay = CatalogOverlay::new();
let mut users_default = test_table("users", 1);
users_default.catalog_name = "main".to_string();
users_default.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users_default), users_default.clone());
let mut users_analytics = test_table("users", 2);
users_analytics.catalog_name = "main".to_string();
users_analytics.namespace_name = "analytics".to_string();
overlay.add_table(TableFqn::from(&users_analytics), users_analytics.clone());
let default_tables: Vec<String> = catalog
.list_tables_in_txn("main", "default", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert_eq!(default_tables, vec!["users".to_string()]);
let analytics_tables: Vec<String> = catalog
.list_tables_in_txn("main", "analytics", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert_eq!(analytics_tables, vec!["users".to_string()]);
overlay.drop_table(&TableFqn::from(&users_default));
let default_tables_after: Vec<String> = catalog
.list_tables_in_txn("main", "default", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert!(default_tables_after.is_empty());
let analytics_tables_after: Vec<String> = catalog
.list_tables_in_txn("main", "analytics", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert_eq!(analytics_tables_after, vec!["users".to_string()]);
}
#[test]
fn overlay_fqn_indexes_separate_namespaces() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let catalog = PersistentCatalog::new(store);
let mut overlay = CatalogOverlay::new();
let mut users_default = test_table("users", 1);
users_default.catalog_name = "main".to_string();
users_default.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users_default), users_default.clone());
let mut users_analytics = test_table("users", 2);
users_analytics.catalog_name = "main".to_string();
users_analytics.namespace_name = "analytics".to_string();
overlay.add_table(TableFqn::from(&users_analytics), users_analytics.clone());
let mut idx_default =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_default.catalog_name = "main".to_string();
idx_default.namespace_name = "default".to_string();
overlay.add_index(IndexFqn::from(&idx_default), idx_default);
let mut idx_analytics =
IndexMetadata::new(2, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_analytics.catalog_name = "main".to_string();
idx_analytics.namespace_name = "analytics".to_string();
overlay.add_index(IndexFqn::from(&idx_analytics), idx_analytics);
let default_fqn = TableFqn::new("main", "default", "users");
let analytics_fqn = TableFqn::new("main", "analytics", "users");
let default_indexes: Vec<String> = catalog
.list_indexes_in_txn(&default_fqn, &overlay)
.into_iter()
.map(|index| index.name)
.collect();
assert_eq!(default_indexes, vec!["idx_users_id".to_string()]);
let analytics_indexes: Vec<String> = catalog
.list_indexes_in_txn(&analytics_fqn, &overlay)
.into_iter()
.map(|index| index.name)
.collect();
assert_eq!(analytics_indexes, vec!["idx_users_id".to_string()]);
}
#[test]
fn persist_overlay_rejects_duplicate_table_names() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
let mut overlay = CatalogOverlay::new();
let mut users_default = test_table("users", 1);
users_default.catalog_name = "main".to_string();
users_default.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users_default), users_default);
let mut users_analytics = test_table("users", 2);
users_analytics.catalog_name = "main".to_string();
users_analytics.namespace_name = "analytics".to_string();
overlay.add_table(TableFqn::from(&users_analytics), users_analytics);
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
let err = catalog.persist_overlay(&mut txn, &overlay).unwrap_err();
assert!(matches!(err, CatalogError::InvalidKey(_)));
txn.rollback_self().unwrap();
}
#[test]
fn persist_overlay_rejects_duplicate_index_names() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store.clone());
let mut overlay = CatalogOverlay::new();
let mut idx_default =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_default.catalog_name = "main".to_string();
idx_default.namespace_name = "default".to_string();
overlay.add_index(IndexFqn::from(&idx_default), idx_default);
let mut idx_analytics =
IndexMetadata::new(2, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
idx_analytics.catalog_name = "main".to_string();
idx_analytics.namespace_name = "analytics".to_string();
overlay.add_index(IndexFqn::from(&idx_analytics), idx_analytics);
let mut txn = store.begin(TxnMode::ReadWrite).unwrap();
let err = catalog.persist_overlay(&mut txn, &overlay).unwrap_err();
assert!(matches!(err, CatalogError::InvalidKey(_)));
txn.rollback_self().unwrap();
}
#[test]
fn overlay_drop_cascade_namespace_removes_children() {
let mut overlay = CatalogOverlay::new();
overlay.add_namespace(NamespaceMeta {
name: "default".to_string(),
catalog_name: "main".to_string(),
comment: None,
storage_root: None,
});
let mut users = test_table("users", 1);
users.catalog_name = "main".to_string();
users.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users), users.clone());
let mut users_index =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
users_index.catalog_name = "main".to_string();
users_index.namespace_name = "default".to_string();
let users_index_fqn = IndexFqn::from(&users_index);
overlay.add_index(users_index_fqn.clone(), users_index);
overlay.drop_cascade_namespace("main", "default");
assert!(
overlay
.dropped_namespaces
.contains(&("main".to_string(), "default".to_string()))
);
assert!(overlay.dropped_tables.contains(&TableFqn::from(&users)));
assert!(overlay.dropped_indexes.contains(&users_index_fqn));
assert!(!overlay.added_tables.contains_key(&TableFqn::from(&users)));
assert!(!overlay.added_indexes.contains_key(&users_index_fqn));
}
#[test]
fn overlay_drop_cascade_catalog_removes_children() {
let mut overlay = CatalogOverlay::new();
overlay.add_catalog(CatalogMeta {
name: "main".to_string(),
comment: None,
storage_root: None,
});
overlay.add_namespace(NamespaceMeta {
name: "default".to_string(),
catalog_name: "main".to_string(),
comment: None,
storage_root: None,
});
let mut users = test_table("users", 1);
users.catalog_name = "main".to_string();
users.namespace_name = "default".to_string();
overlay.add_table(TableFqn::from(&users), users.clone());
let mut users_index =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
users_index.catalog_name = "main".to_string();
users_index.namespace_name = "default".to_string();
let users_index_fqn = IndexFqn::from(&users_index);
overlay.add_index(users_index_fqn.clone(), users_index);
overlay.drop_cascade_catalog("main");
assert!(overlay.dropped_catalogs.contains("main"));
assert!(overlay.dropped_tables.contains(&TableFqn::from(&users)));
assert!(overlay.dropped_indexes.contains(&users_index_fqn));
assert!(!overlay.added_tables.contains_key(&TableFqn::from(&users)));
assert!(!overlay.added_indexes.contains_key(&users_index_fqn));
}
#[test]
fn dropped_namespace_hides_tables_and_indexes() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let mut users = test_table("users", 1);
users.catalog_name = "main".to_string();
users.namespace_name = "default".to_string();
catalog.inner.insert_table_unchecked(users);
let mut users_index =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
users_index.catalog_name = "main".to_string();
users_index.namespace_name = "default".to_string();
catalog.inner.insert_index_unchecked(users_index);
let mut overlay = CatalogOverlay::new();
overlay.drop_namespace("main", "default");
let table_names: Vec<String> = catalog
.list_tables_in_txn("main", "default", &overlay)
.into_iter()
.map(|table| table.name)
.collect();
assert!(table_names.is_empty());
let fqn = TableFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
};
let index_names: Vec<String> = catalog
.list_indexes_in_txn(&fqn, &overlay)
.into_iter()
.map(|index| index.name)
.collect();
assert!(index_names.is_empty());
}
#[test]
fn dropped_namespace_hides_get_and_exists() {
let store = Arc::new(alopex_core::kv::memory::MemoryKV::new());
let mut catalog = PersistentCatalog::new(store);
let mut users = test_table("users", 1);
users.catalog_name = "main".to_string();
users.namespace_name = "default".to_string();
catalog.inner.insert_table_unchecked(users);
let mut users_index =
IndexMetadata::new(1, "idx_users_id", "users", vec!["id".to_string()])
.with_column_indices(vec![0]);
users_index.catalog_name = "main".to_string();
users_index.namespace_name = "default".to_string();
catalog.inner.insert_index_unchecked(users_index);
let mut overlay = CatalogOverlay::new();
overlay.drop_namespace("main", "default");
assert!(!catalog.table_exists_in_txn("users", &overlay));
assert!(catalog.get_table_in_txn("users", &overlay).is_none());
assert!(!catalog.index_exists_in_txn("idx_users_id", &overlay));
assert!(catalog.get_index_in_txn("idx_users_id", &overlay).is_none());
let view = TxnCatalogView::new(&catalog, &overlay);
assert!(view.get_indexes_for_table("users").is_empty());
}
#[test]
fn persisted_catalog_meta_roundtrip() {
let meta = PersistedCatalogMeta {
name: "main".to_string(),
comment: Some("primary catalog".to_string()),
storage_root: Some("/tmp/alopex".to_string()),
};
let bytes = bincode::serialize(&meta).unwrap();
let decoded: PersistedCatalogMeta = bincode::deserialize(&bytes).unwrap();
assert_eq!(meta, decoded);
}
#[test]
fn persisted_namespace_meta_roundtrip() {
let meta = PersistedNamespaceMeta {
name: "analytics".to_string(),
catalog_name: "main".to_string(),
comment: Some("warehouse".to_string()),
storage_root: Some("s3://bucket/ns".to_string()),
};
let bytes = bincode::serialize(&meta).unwrap();
let decoded: PersistedNamespaceMeta = bincode::deserialize(&bytes).unwrap();
assert_eq!(meta, decoded);
}
#[test]
fn table_fqn_hash_and_eq() {
let first = TableFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
};
let same = TableFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
};
let different = TableFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "orders".to_string(),
};
let mut set = HashSet::new();
set.insert(first);
assert!(set.contains(&same));
assert!(!set.contains(&different));
}
#[test]
fn index_fqn_hash_and_eq() {
let first = IndexFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
index: "idx_users_id".to_string(),
};
let same = IndexFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
index: "idx_users_id".to_string(),
};
let different = IndexFqn {
catalog: "main".to_string(),
namespace: "default".to_string(),
table: "users".to_string(),
index: "idx_users_email".to_string(),
};
let mut set = HashSet::new();
set.insert(first);
assert!(set.contains(&same));
assert!(!set.contains(&different));
}
#[test]
fn table_type_and_data_source_format_serde() {
let managed = serde_json::to_string(&TableType::Managed).unwrap();
let external = serde_json::to_string(&TableType::External).unwrap();
let alopex = serde_json::to_string(&DataSourceFormat::Alopex).unwrap();
let parquet = serde_json::to_string(&DataSourceFormat::Parquet).unwrap();
let delta = serde_json::to_string(&DataSourceFormat::Delta).unwrap();
assert_eq!(managed, "\"MANAGED\"");
assert_eq!(external, "\"EXTERNAL\"");
assert_eq!(alopex, "\"ALOPEX\"");
assert_eq!(parquet, "\"PARQUET\"");
assert_eq!(delta, "\"DELTA\"");
}
}