use datafusion::sql::sqlparser::ast::ObjectName;
use crate::LixError;
use super::super::catalog::{PublicCatalog, PublicSurfaceContract};
use super::expr::BoundColumnRef;
use super::write::BoundWriteOp;
#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct BoundTable {
pub(crate) name: String,
pub(crate) surface: PublicSurfaceContract,
}
pub(crate) fn bind_exact_table_name(name: &ObjectName) -> Result<String, LixError> {
if name.0.len() != 1 {
return Err(super::error::unsupported(
"qualified SQL table names are not supported",
));
}
name.0
.first()
.and_then(|part| part.as_ident())
.map(|ident| {
if ident.quote_style.is_some() {
ident.value.clone()
} else {
ident.value.to_ascii_lowercase()
}
})
.ok_or_else(|| super::error::unsupported("unsupported SQL table name"))
}
pub(crate) fn bind_public_table(
catalog: &PublicCatalog,
name: &ObjectName,
) -> Result<BoundTable, LixError> {
let table_name = bind_exact_table_name(name)?;
let surface = catalog.require_surface(&table_name)?.clone();
Ok(BoundTable {
name: table_name,
surface,
})
}
pub(crate) fn require_public_column<'a>(
table: &'a BoundTable,
column_name: &str,
) -> Result<&'a super::super::catalog::PublicColumn, LixError> {
if table.surface.public_column(column_name).is_some() {
return Ok(table
.surface
.public_column(column_name)
.expect("checked public column"));
}
if table.surface.column(column_name).is_some() {
return Err(LixError::new(
LixError::CODE_COLUMN_NOT_FOUND,
format!(
"column '{column_name}' is not part of public SQL surface '{}'",
table.name
),
));
}
Err(LixError::new(
LixError::CODE_COLUMN_NOT_FOUND,
format!(
"column '{column_name}' does not exist on SQL table '{}'",
table.name
),
))
}
pub(crate) fn require_writable_column(
table: &BoundTable,
column_name: &str,
op: BoundWriteOp,
) -> Result<BoundColumnRef, LixError> {
let column = require_public_column(table, column_name)?;
let allowed = match op {
BoundWriteOp::Insert => column.is_insertable(),
BoundWriteOp::Update => column.is_updatable(),
BoundWriteOp::Delete => false,
};
if !allowed {
if table.name == "lix_branch" && column_name == "id" && op == BoundWriteOp::Update {
return Err(super::error::unsupported(
"UPDATE lix_branch cannot change immutable column 'id'",
));
}
return Err(super::error::unsupported(format!(
"column '{column_name}' is not writable on SQL table '{}'",
table.name
)));
}
Ok(BoundColumnRef {
table: table.name.clone(),
column_id: column.id,
name: column.name.clone(),
})
}
pub(crate) fn bind_public_column_ref(
table: &BoundTable,
column_name: &str,
) -> Result<BoundColumnRef, LixError> {
let column = require_public_column(table, column_name)?;
Ok(BoundColumnRef {
table: table.name.clone(),
column_id: column.id,
name: column.name.clone(),
})
}
#[cfg(test)]
mod tests {
use datafusion::sql::sqlparser::ast::{SetExpr, Statement, TableFactor};
use datafusion::sql::sqlparser::parser::Parser;
use serde_json::json;
use super::*;
use crate::sql2::catalog::PublicSurfaceKind;
#[test]
fn rejects_qualified_table_name_even_when_leaf_exists() {
let catalog = catalog();
let error = bind_public_table(&catalog, &table_name("SELECT * FROM foo.unknown"))
.expect_err("qualified table should be rejected");
assert_eq!(error.code, LixError::CODE_UNSUPPORTED_SQL);
}
#[test]
fn rejects_unknown_table_name() {
let catalog = catalog();
let error = bind_public_table(&catalog, &table_name("SELECT * FROM missing"))
.expect_err("unknown table should be rejected");
assert_eq!(error.code, LixError::CODE_UNSUPPORTED_SQL);
assert!(error.message.contains("unknown SQL table 'missing'"));
}
#[test]
fn base_row_table_does_not_expose_branch_column() {
let catalog = catalog();
let table = bind_public_table(&catalog, &table_name("SELECT * FROM test_state_schema"))
.expect("base row table should bind");
assert!(matches!(
table.surface.kind,
PublicSurfaceKind::SchemaBase { .. }
));
assert!(require_public_column(&table, "name").is_ok());
let error = require_public_column(&table, "lixcol_branch_id")
.expect_err("base schema surface should not expose branch column");
assert!(error.message.contains("does not exist"));
}
#[test]
fn by_branch_row_exposes_lixcol_branch_id_without_branch_id_alias() {
let catalog = catalog();
let table = bind_public_table(
&catalog,
&table_name("SELECT * FROM test_state_schema_by_branch"),
)
.expect("by-branch row table should bind");
assert!(matches!(
table.surface.kind,
PublicSurfaceKind::SchemaByBranch { .. }
));
assert!(require_public_column(&table, "lixcol_branch_id").is_ok());
let error = require_public_column(&table, "branch_id")
.expect_err("by-branch schema surface should not alias branch_id");
assert!(error.message.contains("does not exist"));
}
#[test]
fn quoted_table_names_are_case_sensitive() {
let catalog = catalog();
bind_public_table(&catalog, &table_name("SELECT * FROM \"lix_file\""))
.expect("quoted exact case should bind");
let error = bind_public_table(&catalog, &table_name("SELECT * FROM \"LIX_FILE\""))
.expect_err("quoted mixed case should not be folded");
assert!(error.message.contains("unknown SQL table 'LIX_FILE'"));
}
#[test]
fn hidden_columns_cannot_bind_as_public_columns() {
let catalog = catalog();
let table = bind_public_table(&catalog, &table_name("SELECT * FROM lix_file"))
.expect("lix_file should bind");
let error = require_public_column(&table, "lixcol_schema_key")
.expect_err("hidden column should not bind");
assert!(error.message.contains("not part of public SQL surface"));
}
#[test]
fn catalog_rejects_runtime_schema_in_reserved_namespace_before_surface_collision() {
let error = PublicCatalog::from_visible_schemas(&[json!({
"$schema": "https://lix.dev/schema-v1.json",
"key": "lix_file",
"columns": [
{ "name": "id", "type": "text", "nullable": false },
],
"primary_key": ["id"],
})])
.expect_err("the complete lix_* runtime namespace should be rejected");
assert_eq!(error.code, LixError::CODE_RESERVED_SCHEMA_NAMESPACE);
assert!(error.message.contains("lix_file"));
}
#[test]
fn catalog_uses_validated_schema_surface_derivation() {
let error = PublicCatalog::from_visible_schemas(&[json!({
"$schema": "https://lix.dev/schema-v1.json",
"key": "bad_row",
"columns": [
{ "name": "value", "type": "jsonb", "nullable": false },
],
"primary_key": ["value"],
})])
.expect_err("Schema v1 rejects unsupported primary-key types");
assert_eq!(error.code, LixError::CODE_SCHEMA_DEFINITION);
}
#[test]
fn fixed_catalog_exposes_only_the_deliberate_lix_sql_contract() {
let actual = PublicCatalog::fixed_system()
.surfaces()
.map(|surface| surface.name.as_str())
.collect::<Vec<_>>();
let expected = vec![
"lix_account",
"lix_account_by_branch",
"lix_account_history",
"lix_apply",
"lix_branch",
"lix_branch_descriptor",
"lix_branch_descriptor_by_branch",
"lix_branch_descriptor_history",
"lix_branch_ref",
"lix_branch_ref_by_branch",
"lix_branch_ref_history",
"lix_change",
"lix_checkpoint",
"lix_checkpoint_history",
"lix_commit",
"lix_commit_by_branch",
"lix_commit_edge",
"lix_commit_edge_by_branch",
"lix_create_checkpoint",
"lix_directory",
"lix_directory_by_branch",
"lix_directory_history",
"lix_directory_working_diff",
"lix_directory_working_diff_by_branch",
"lix_file",
"lix_file_by_branch",
"lix_file_history",
"lix_file_working_diff",
"lix_file_working_diff_by_branch",
"lix_key_value",
"lix_key_value_by_branch",
"lix_key_value_history",
"lix_registered_schema",
"lix_registered_schema_by_branch",
"lix_registered_schema_history",
"lix_revert",
"lix_working_diff",
"lix_working_diff_by_branch",
];
assert_eq!(actual, expected);
}
#[test]
fn fixed_catalog_keeps_registry_surfaces_and_hides_storage_adapters() {
let catalog = PublicCatalog::fixed_system();
for surface_name in [
"lix_key_value",
"lix_key_value_by_branch",
"lix_key_value_history",
"lix_registered_schema",
"lix_registered_schema_by_branch",
"lix_registered_schema_history",
"lix_checkpoint",
"lix_checkpoint_history",
"lix_working_diff",
"lix_working_diff_by_branch",
] {
assert!(
catalog.surface(surface_name).is_some(),
"{surface_name} should remain public"
);
}
for surface_name in [
"lix_state",
"lix_state_by_branch",
"lix_state_history",
"lix_label",
"lix_label_by_branch",
"lix_label_history",
"lix_label_assignment",
"lix_label_assignment_by_branch",
"lix_label_assignment_history",
"lix_checkpoint_by_branch",
"lix_undo_redo_marker",
"lix_collection_generation",
"lix_binary_blob_ref",
"lix_binary_blob_ref_by_branch",
"lix_binary_blob_ref_history",
"lix_directory_descriptor",
"lix_directory_descriptor_by_branch",
"lix_directory_descriptor_history",
"lix_file_descriptor",
"lix_file_descriptor_by_branch",
"lix_file_descriptor_history",
] {
assert!(
catalog.surface(surface_name).is_none(),
"{surface_name} should not be public"
);
}
}
#[test]
fn runtime_schema_namespace_check_matches_unquoted_sql_normalization() {
for schema_key in [
"lix",
"LIX",
"lix_plugin_note",
"LIX_PLUGIN_NOTE",
"LiX_PlUgIn_NoTe",
] {
assert!(
PublicCatalog::runtime_schema_key_uses_reserved_namespace(schema_key),
"{schema_key} should normalize into the reserved namespace"
);
}
assert!(!PublicCatalog::runtime_schema_key_uses_reserved_namespace(
"acme_lix_note"
));
}
#[test]
fn dynamic_row_history_surface_uses_provider_history_column_names() {
let catalog = catalog();
let table = bind_public_table(
&catalog,
&table_name("SELECT * FROM test_state_schema_history()"),
)
.expect("schema history surface should bind");
assert!(matches!(
table.surface.kind,
PublicSurfaceKind::SchemaHistory { .. }
));
assert!(require_public_column(&table, "lixcol_row_pk").is_ok());
assert!(require_public_column(&table, "lixcol_snapshot_content").is_err());
}
#[test]
fn dynamic_row_file_id_is_public_and_insert_only() {
let catalog = catalog();
let table = bind_public_table(&catalog, &table_name("SELECT * FROM test_state_schema"))
.expect("schema surface should bind");
assert!(require_public_column(&table, "lixcol_file_id").is_ok());
assert!(require_writable_column(&table, "lixcol_file_id", BoundWriteOp::Insert).is_ok());
let error = require_writable_column(&table, "lixcol_file_id", BoundWriteOp::Update)
.expect_err("row file id should remain immutable after insert");
assert!(error.message.contains("is not writable"));
}
fn catalog() -> PublicCatalog {
PublicCatalog::from_visible_schemas(&[json!({
"$schema": "https://lix.dev/schema-v1.json",
"key": "test_state_schema",
"columns": [
{ "name": "id", "type": "text", "nullable": false },
{ "name": "name", "type": "text", "nullable": true },
{ "name": "lixcol_internal", "type": "text", "nullable": true },
],
"primary_key": ["id"],
})])
.expect("test catalog")
}
fn table_name(sql: &str) -> ObjectName {
let mut statements =
Parser::parse_sql(&crate::sql2::dialect::lix_sql_dialect(), sql).expect("parse SQL");
let Some(Statement::Query(query)) = statements.pop() else {
panic!("expected query");
};
let SetExpr::Select(select) = query.body.as_ref() else {
panic!("expected select");
};
let TableFactor::Table { name, .. } = &select.from[0].relation else {
panic!("expected table factor");
};
name.clone()
}
}