use serde::{Deserialize, Serialize};
use crate::backend::{StorageBackendError, StorageBackendResult};
mod cache_revisions;
mod graph_access;
pub mod graph_guards;
pub mod graph_identifiers;
pub mod graph_observations;
pub(crate) mod graph_snapshot;
mod identity;
mod relation;
pub mod relation_acl;
mod relation_security;
mod schema;
mod sequence_security;
mod sequence_values;
pub use cache_revisions::CatalogCacheRevisions;
pub use graph_access::validate_graph_page;
pub use graph_access::{
GraphEntityFilter, GraphEntityKind, GraphEntitySelector, MAX_GRAPH_ID_PAGE,
};
pub use identity::new_nonzero_catalog_identity;
pub use relation::RelationIdentity;
pub use relation_security::{BoundRelationSecurity, LegacyRelationSecurity, RelationSecurityRow};
pub use sequence_security::{BoundSequenceSecurity, LegacySequenceSecurity, SequenceSecurityRow};
pub use sequence_values::{
sequence_value_allocation, sequence_value_reservation, SequenceLogResult,
SequenceReservationResult, SequenceSetValueResult, SequenceValueAllocation,
SequenceValuePosition, SequenceValueReservation,
};
mod table;
pub use schema::{BoundSchemaRow, SchemaAclEntry, SchemaPrivileges, SchemaRow};
pub use table::{TableAclEntry, TablePrivileges};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RelationKind {
Table,
View,
Sequence,
ForeignTable,
Index,
}
impl RelationKind {
pub fn as_str(self) -> &'static str {
match self {
Self::Table => "table",
Self::View => "view",
Self::Sequence => "sequence",
Self::ForeignTable => "foreign_table",
Self::Index => "index",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TableSchema {
pub relation: RelationIdentity,
#[serde(flatten)]
pub security: RelationSecurityRow,
#[serde(default)]
pub object_id: [u8; 16],
#[serde(default)]
pub storage_generation: [u8; 16],
pub analyzer_json: String,
pub fts_fields: Vec<String>,
pub vector_fields: Vec<VectorFieldSchema>,
#[serde(default)]
pub columns_json: String,
#[serde(default)]
pub constraints_json: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VectorFieldSchema {
pub field: String,
pub dimensions: u32,
}
#[derive(Debug, Clone)]
pub struct EdgeRow {
pub edge_id: u64,
pub source_id: u64,
pub target_id: u64,
pub label: String,
pub properties_json: String,
}
#[derive(Debug, Clone)]
pub struct GraphVertexRow {
pub vertex_id: u64,
pub label: String,
pub properties_json: String,
}
#[derive(Debug, Clone)]
pub struct GraphSnapshot {
pub vertices: Vec<GraphVertexRow>,
pub edges: Vec<EdgeRow>,
pub label_registry_json: String,
}
#[derive(Debug, Clone)]
pub struct ForeignTableRow {
pub relation: RelationIdentity,
pub security: RelationSecurityRow,
pub server_name: String,
pub columns_json: String,
pub options_json: String,
}
#[derive(Debug, Clone)]
pub struct ViewRow {
pub relation: RelationIdentity,
pub security: RelationSecurityRow,
pub definition_json: String,
}
pub use uqa_core::catalog_index::CatalogIndexRow;
#[derive(Debug, Clone, Copy)]
pub struct ColumnStatsInput<'a> {
pub table_name: &'a str,
pub column_name: &'a str,
pub distinct_count: i64,
pub null_count: i64,
pub min_value: Option<&'a str>,
pub max_value: Option<&'a str>,
pub row_count: i64,
pub histogram_json: &'a str,
pub mcv_values_json: &'a str,
pub mcv_frequencies_json: &'a str,
}
impl<'a> ColumnStatsInput<'a> {
pub fn basic(
table_name: &'a str,
column_name: &'a str,
distinct_count: i64,
null_count: i64,
min_value: Option<&'a str>,
max_value: Option<&'a str>,
row_count: i64,
) -> Self {
Self {
table_name,
column_name,
distinct_count,
null_count,
min_value,
max_value,
row_count,
histogram_json: "[]",
mcv_values_json: "[]",
mcv_frequencies_json: "[]",
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct ColumnStatsRow {
pub column_name: String,
pub distinct_count: i64,
pub null_count: i64,
pub min_value: Option<String>,
pub max_value: Option<String>,
pub row_count: i64,
pub histogram_json: String,
pub mcv_values_json: String,
pub mcv_frequencies_json: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SequenceOptions {
pub data_type: String,
pub min_value: Option<i64>,
pub max_value: Option<i64>,
pub cycle: bool,
#[serde(default = "default_sequence_cache_size")]
pub cache_size: i64,
}
const fn default_sequence_cache_size() -> i64 {
1
}
impl Default for SequenceOptions {
fn default() -> Self {
Self {
data_type: "bigint".into(),
min_value: None,
max_value: None,
cycle: false,
cache_size: default_sequence_cache_size(),
}
}
}
pub use uqa_core::catalog_sequence::{
SequenceAclEntry, SequenceOwner, SequenceOwnerDependency, SequencePrivileges,
};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SequenceRow {
pub relation: RelationIdentity,
#[serde(flatten)]
pub security: SequenceSecurityRow,
#[serde(default)]
pub object_id: [u8; 16],
#[serde(default)]
pub definition_generation: [u8; 16],
pub start: i64,
pub increment: i64,
pub current: i64,
pub called: bool,
#[serde(default)]
pub log_count: i64,
pub persistence: String,
#[serde(default)]
pub owner: Option<SequenceOwner>,
#[serde(default)]
pub options: SequenceOptions,
}
pub trait CatalogFacade: Send + Sync {
fn transaction_model(&self) -> crate::StorageTransactionModel {
crate::StorageTransactionModel::ProviderSerialized
}
fn transaction_affinity(&self) -> Option<crate::StorageSessionAffinity> {
None
}
fn initialize_storage(&self) -> StorageBackendResult<()> {
Ok(())
}
fn cache_revisions(&self) -> StorageBackendResult<Option<CatalogCacheRevisions>> {
Ok(None)
}
fn set_metadata(&self, key: &str, value: &str) -> StorageBackendResult<()>;
fn delete_metadata(&self, key: &str) -> StorageBackendResult<()>;
fn get_metadata(&self, key: &str) -> StorageBackendResult<Option<String>>;
fn metadata_with_prefix(&self, prefix: &str) -> StorageBackendResult<Vec<(String, String)>>;
fn save_relation_acl(
&self,
relation: &RelationIdentity,
column: Option<&str>,
entry: &relation_acl::RelationAclTuple,
) -> StorageBackendResult<()> {
if !self.transaction_model().is_versioned() {
return relation_acl::save_serialized(self, relation, column, entry);
}
self.set_metadata(&relation_acl::key(relation, column), &entry.encode(column)?)
}
fn metadata_has_private_changes(&self, _key: &str) -> StorageBackendResult<bool> {
if self.transaction_model().is_versioned() {
return Err(StorageBackendError::Other(
"private metadata provenance is not supported by this catalog".into(),
));
}
Ok(false)
}
fn save_statistics_maintenance(
&self,
table: &str,
state: &crate::statistics_maintenance::StatisticsMaintenance,
) -> StorageBackendResult<()> {
self.set_metadata(
&crate::statistics_maintenance::StatisticsMaintenance::key(table),
&serde_json::to_string(state)?,
)
}
fn fts_storage_was_reset(&self) -> bool {
false
}
fn migrate_relation_namespace(&self) -> StorageBackendResult<()>;
fn save_schema_row(&self, schema: &SchemaRow) -> StorageBackendResult<()>;
fn drop_schema(&self, name: &str) -> StorageBackendResult<()>;
fn load_schema_rows(&self) -> StorageBackendResult<Vec<SchemaRow>>;
fn schema_has_private_changes(&self, _name: &str) -> StorageBackendResult<bool> {
if self.transaction_model().is_versioned() {
return Err(StorageBackendError::Other(
"private schema provenance is not supported by this catalog".into(),
));
}
Ok(false)
}
fn save_schema(&self, name: &str) -> StorageBackendResult<()> {
self.save_schema_row(&SchemaRow::bootstrap(name))
}
fn load_schemas(&self) -> StorageBackendResult<Vec<String>> {
Ok(self
.load_schema_rows()?
.into_iter()
.map(|schema| schema.name().to_owned())
.collect())
}
fn save_table(&self, schema: &TableSchema) -> StorageBackendResult<()>;
fn load_tables(&self) -> StorageBackendResult<Vec<TableSchema>>;
fn drop_table(&self, name: &str) -> StorageBackendResult<()>;
fn drop_table_and_data(&self, name: &str) -> StorageBackendResult<()>;
fn purge_table_data(&self, name: &str) -> StorageBackendResult<()>;
fn rename_table_data(&self, from: &str, to: &str) -> StorageBackendResult<()>;
fn drop_column_data(&self, table_name: &str, column_name: &str) -> StorageBackendResult<()>;
fn rename_column_data(
&self,
table_name: &str,
from: &str,
to: &str,
) -> StorageBackendResult<()>;
fn save_model(&self, name: &str, json: &str) -> StorageBackendResult<()>;
fn load_models(&self) -> StorageBackendResult<Vec<(String, String)>>;
fn load_model(&self, name: &str) -> StorageBackendResult<Option<String>>;
fn drop_model(&self, name: &str) -> StorageBackendResult<()>;
fn save_scoring_params(&self, name: &str, params_json: &str) -> StorageBackendResult<()>;
fn load_scoring_params(&self, name: &str) -> StorageBackendResult<Option<String>>;
fn load_all_scoring_params(&self) -> StorageBackendResult<Vec<(String, String)>>;
fn drop_scoring_params(&self, name: &str) -> StorageBackendResult<()>;
fn create_sequence_row(&self, sequence: &SequenceRow) -> StorageBackendResult<bool>;
fn replace_sequence_row(&self, sequence: &SequenceRow) -> StorageBackendResult<bool>;
fn rename_sequence_row(&self, from: &str, to: &str) -> StorageBackendResult<bool>;
fn drop_sequence_row(&self, name: &str) -> StorageBackendResult<bool>;
fn load_sequence_rows(&self) -> StorageBackendResult<Vec<SequenceRow>>;
fn sequence_has_private_changes(
&self,
_relation: &RelationIdentity,
_object_id: [u8; 16],
) -> StorageBackendResult<bool> {
if self.transaction_model().is_versioned() {
return Err(StorageBackendError::Other(
"private sequence provenance is not supported by this catalog".into(),
));
}
Ok(false)
}
fn reserve_sequence_values(
&self,
name: &str,
object_id: [u8; 16],
definition_generation: [u8; 16],
) -> StorageBackendResult<SequenceReservationResult>;
fn next_sequence_value(
&self,
name: &str,
object_id: [u8; 16],
) -> StorageBackendResult<Option<i64>> {
loop {
let relation =
RelationIdentity::from_legacy_name(name).map_err(StorageBackendError::Other)?;
let Some(row) = self
.load_sequence_rows()?
.into_iter()
.find(|row| row.relation == relation && row.object_id == object_id)
else {
return Ok(None);
};
match self.reserve_sequence_values(name, object_id, row.definition_generation)? {
SequenceReservationResult::Reserved(reservation) => {
return Ok(Some(reservation.first_value));
}
SequenceReservationResult::DefinitionChanged => {}
SequenceReservationResult::Missing => return Ok(None),
SequenceReservationResult::Exhausted => {
return Err(StorageBackendError::Other(format!(
"sequence `{name}` exhausted"
)));
}
}
}
}
fn set_sequence_value(
&self,
name: &str,
object_id: [u8; 16],
definition_generation: [u8; 16],
value: i64,
called: bool,
log_count: i64,
) -> StorageBackendResult<SequenceSetValueResult>;
fn log_sequence_values(
&self,
name: &str,
object_id: [u8; 16],
definition_generation: [u8; 16],
expected: (i64, bool),
logged: SequenceValuePosition,
) -> StorageBackendResult<SequenceLogResult>;
fn save_view(&self, view: &ViewRow) -> StorageBackendResult<()>;
fn rename_view(
&self,
from: &RelationIdentity,
to: &RelationIdentity,
) -> StorageBackendResult<bool>;
fn drop_view(&self, relation: &RelationIdentity) -> StorageBackendResult<bool>;
fn load_views(&self) -> StorageBackendResult<Vec<ViewRow>>;
fn save_named_graph(&self, name: &str) -> StorageBackendResult<()>;
fn guard_graph_definition(&self, _graph: Option<&str>) -> StorageBackendResult<()> {
Err(crate::StorageBackendError::Other(
"graph definition guards are not supported".into(),
))
}
fn drop_named_graph(&self, name: &str) -> StorageBackendResult<()>;
fn load_named_graphs(&self) -> StorageBackendResult<Vec<String>>;
fn named_graph_exists(&self, name: &str) -> StorageBackendResult<bool>;
fn graph_vertex(&self, id: u64) -> StorageBackendResult<Option<GraphVertexRow>>;
fn for_each_graph_vertex_borrowed(
&self,
_ids: &[u64],
_visit: &mut dyn FnMut(u64, Option<&GraphVertexRow>) -> bool,
) -> StorageBackendResult<Option<usize>> {
Ok(None)
}
fn graph_edge(&self, id: u64) -> StorageBackendResult<Option<EdgeRow>>;
fn graph_entity_ids(
&self,
filter: GraphEntityFilter<'_>,
after: Option<u64>,
limit: usize,
) -> StorageBackendResult<Vec<u64>>;
fn graph_entity_count(&self, filter: GraphEntityFilter<'_>) -> StorageBackendResult<u64>;
fn graph_entity_max_id(&self, kind: GraphEntityKind) -> StorageBackendResult<Option<u64>>;
fn graph_entity_memberships(
&self,
kind: GraphEntityKind,
id: u64,
) -> StorageBackendResult<Vec<String>>;
fn graph_has_membership(
&self,
kind: GraphEntityKind,
id: u64,
graph: &str,
) -> StorageBackendResult<bool>;
fn load_named_graph_snapshot(&self, name: &str) -> StorageBackendResult<Option<GraphSnapshot>> {
graph_snapshot::load(self, name)
}
fn save_vertex(
&self,
vertex_id: u64,
label: &str,
properties_json: &str,
) -> StorageBackendResult<()>;
fn delete_vertex(&self, vertex_id: u64) -> StorageBackendResult<()>;
fn load_vertices(&self) -> StorageBackendResult<Vec<(u64, String, String)>>;
fn save_edge(
&self,
edge_id: u64,
source_id: u64,
target_id: u64,
label: &str,
properties_json: &str,
) -> StorageBackendResult<()>;
fn delete_edge(&self, edge_id: u64) -> StorageBackendResult<()>;
fn load_edges(&self) -> StorageBackendResult<Vec<EdgeRow>>;
fn save_graph_membership(
&self,
entity_type: &str,
entity_id: u64,
graph_name: &str,
) -> StorageBackendResult<()>;
fn delete_graph_membership(
&self,
entity_type: &str,
entity_id: u64,
graph_name: &str,
) -> StorageBackendResult<()>;
fn delete_graph_membership_for_graph(&self, graph_name: &str) -> StorageBackendResult<()>;
fn load_graph_memberships(&self) -> StorageBackendResult<Vec<(String, u64, String)>>;
fn purge_orphan_graph_entities(&self) -> StorageBackendResult<()>;
fn replace_named_graph(
&self,
graph_name: &str,
snapshot: &GraphSnapshot,
) -> StorageBackendResult<()>;
fn drop_named_graph_data(&self, graph_name: &str) -> StorageBackendResult<()>;
fn save_analyzer(&self, name: &str, config_json: &str) -> StorageBackendResult<()>;
fn drop_analyzer(&self, name: &str) -> StorageBackendResult<()>;
fn load_analyzers(&self) -> StorageBackendResult<Vec<(String, String)>>;
fn save_analyzer_revision(
&self,
_name: &str,
_config_json: &str,
_descriptor_json: &str,
) -> StorageBackendResult<()> {
Err(StorageBackendError::Other(
"durable analyzer descriptors are not supported by this catalog".into(),
))
}
fn load_analyzer_descriptors(&self) -> StorageBackendResult<Vec<(String, String)>> {
Ok(Vec::new())
}
fn replace_table_field_analyzer_binding(
&self,
_table: &str,
_field: &str,
_phase: &str,
_name: &str,
_binding_json: &str,
) -> StorageBackendResult<()> {
Err(StorageBackendError::Other(
"durable analyzer bindings are not supported by this catalog".into(),
))
}
fn load_table_field_analyzer_bindings(
&self,
) -> StorageBackendResult<Vec<(String, String, String)>> {
Ok(Vec::new())
}
fn save_table_field_analyzer(
&self,
table_name: &str,
field: &str,
phase: &str,
analyzer_name: &str,
) -> StorageBackendResult<()>;
fn replace_table_field_analyzer(
&self,
table_name: &str,
field: &str,
phase: &str,
analyzer_name: &str,
) -> StorageBackendResult<()>;
fn drop_table_field_analyzer_field(
&self,
table_name: &str,
field: &str,
) -> StorageBackendResult<()>;
fn drop_table_field_analyzers(&self, table_name: &str) -> StorageBackendResult<()>;
fn load_table_field_analyzers(
&self,
) -> StorageBackendResult<Vec<(String, String, String, String)>>;
fn save_foreign_server(
&self,
name: &str,
fdw_type: &str,
options_json: &str,
) -> StorageBackendResult<()>;
fn drop_foreign_server(&self, name: &str) -> StorageBackendResult<()>;
fn load_foreign_servers(&self) -> StorageBackendResult<Vec<(String, String, String)>>;
fn save_foreign_table(&self, row: &ForeignTableRow) -> StorageBackendResult<()>;
fn rename_foreign_table(
&self,
from: &RelationIdentity,
to: &RelationIdentity,
) -> StorageBackendResult<bool>;
fn update_foreign_table_security(
&self,
relation: &RelationIdentity,
security: &RelationSecurityRow,
) -> StorageBackendResult<bool>;
fn drop_foreign_table(&self, relation: &RelationIdentity) -> StorageBackendResult<()>;
fn load_foreign_tables(&self) -> StorageBackendResult<Vec<ForeignTableRow>>;
fn save_catalog_index(
&self,
relation: &RelationIdentity,
index_type: &str,
table_name: &str,
columns_json: &str,
parameters_json: &str,
) -> StorageBackendResult<()> {
self.save_catalog_index_row(&CatalogIndexRow {
relation: relation.clone(),
index_type: index_type.to_string(),
table_name: table_name.to_string(),
columns_json: columns_json.to_string(),
parameters_json: parameters_json.to_string(),
definition_json: None,
})
}
fn save_catalog_index_row(&self, index: &CatalogIndexRow) -> StorageBackendResult<()>;
fn drop_catalog_index(&self, relation: &RelationIdentity) -> StorageBackendResult<()>;
fn drop_catalog_indexes_for_table(&self, table_name: &str) -> StorageBackendResult<()>;
fn load_catalog_indexes(&self) -> StorageBackendResult<Vec<CatalogIndexRow>>;
fn save_path_index(
&self,
graph_name: &str,
label_sequences_json: &str,
) -> StorageBackendResult<()>;
fn drop_path_index(&self, graph_name: &str) -> StorageBackendResult<()>;
fn load_path_indexes(&self) -> StorageBackendResult<Vec<(String, String)>>;
fn clear_path_index_data(&self, index: &str) -> StorageBackendResult<()>;
fn save_path_index_pairs(
&self,
index: &str,
sequence: &str,
pairs: &[(u64, u64)],
) -> StorageBackendResult<()>;
fn finish_path_index_data(
&self,
index: &str,
graph: &str,
definition: &str,
) -> StorageBackendResult<()>;
fn path_index_data_is_current(
&self,
index: &str,
definition: &str,
) -> StorageBackendResult<bool>;
fn path_index_pairs(
&self,
index: &str,
sequence: &str,
after: Option<(u64, u64)>,
limit: usize,
) -> StorageBackendResult<Vec<(u64, u64)>>;
fn save_column_stats(&self, stats: ColumnStatsInput<'_>) -> StorageBackendResult<()>;
fn replace_column_stats(
&self,
table_name: &str,
stats: &[ColumnStatsInput<'_>],
) -> StorageBackendResult<()>;
fn load_column_stats(&self, table_name: &str) -> StorageBackendResult<Vec<ColumnStatsRow>>;
fn delete_column_stats(&self, table_name: &str) -> StorageBackendResult<()>;
}
#[cfg(test)]
mod tests {
use super::RelationIdentity;
#[test]
fn relation_identity_rendering_is_reversible_and_collision_free() {
let left = RelationIdentity::new("a.b", "c");
let right = RelationIdentity::new("a", "b.c");
assert_eq!(left.qualified_name(), "\"a.b\".c");
assert_eq!(right.qualified_name(), "a.\"b.c\"");
assert_ne!(left.qualified_name(), right.qualified_name());
assert_eq!(
RelationIdentity::from_legacy_name(&left.qualified_name()).unwrap(),
left
);
assert_eq!(
RelationIdentity::from_legacy_name(&right.qualified_name()).unwrap(),
right
);
}
#[test]
fn relation_identity_preserves_quotes_and_unqualified_public_alias() {
let quoted = RelationIdentity::new("public", "a\"b.c");
assert_eq!(quoted.qualified_name(), "public.\"a\"\"b.c\"");
assert_eq!(
quoted.canonical_and_legacy_public_names(),
vec![
"public.\"a\"\"b.c\"".to_string(),
"\"a\"\"b.c\"".to_string()
]
);
assert_eq!(
RelationIdentity::from_legacy_name("ed.qualified_name()).unwrap(),
quoted
);
assert_eq!(
RelationIdentity::from_legacy_name("plain").unwrap(),
RelationIdentity::new("public", "plain")
);
assert_eq!(
RelationIdentity::new("app", "plain").canonical_and_legacy_public_names(),
vec!["app.plain".to_string()]
);
assert_eq!(
RelationIdentity::new("public", "Upper").canonical_and_legacy_public_names(),
vec![
"public.\"Upper\"".to_string(),
"\"Upper\"".to_string(),
"Upper".to_string()
]
);
}
}