use serde::{Deserialize, Serialize};
use crate::backend::{StorageBackendError, StorageBackendResult};
mod cache_revisions;
mod graph_access;
mod graph_snapshot;
mod relation;
mod schema;
pub use cache_revisions::CatalogCacheRevisions;
pub use graph_access::validate_graph_page;
pub use graph_access::{GraphEntityFilter, GraphEntityKind, MAX_GRAPH_ID_PAGE};
pub use relation::RelationIdentity;
mod table;
pub use schema::{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(default = "legacy_table_role_owner")]
pub role_owner: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub acl: Option<Vec<TableAclEntry>>,
#[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
pub column_acls: std::collections::BTreeMap<String, Vec<TableAclEntry>>,
#[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,
}
fn legacy_table_role_owner() -> String {
"uqa".into()
}
#[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 role_owner: String,
pub acl: Option<Vec<TableAclEntry>>,
pub column_acls: std::collections::BTreeMap<String, Vec<TableAclEntry>>,
pub server_name: String,
pub columns_json: String,
pub options_json: String,
}
#[derive(Debug, Clone)]
pub struct ViewRow {
pub relation: RelationIdentity,
pub role_owner: String,
pub acl: Option<Vec<TableAclEntry>>,
pub column_acls: std::collections::BTreeMap<String, Vec<TableAclEntry>>,
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(default = "default_sequence_role_owner")]
pub role_owner: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub acl: Option<Vec<SequenceAclEntry>>,
#[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,
}
fn default_sequence_role_owner() -> String {
"uqa".into()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SequenceValuePosition {
pub current: i64,
pub called: bool,
pub log_count: i64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SequenceValueReservation {
pub first_value: i64,
pub last_value: i64,
pub count: i64,
pub log_count: i64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SequenceReservationResult {
Missing,
DefinitionChanged,
Exhausted,
Reserved(SequenceValueReservation),
}
#[must_use]
pub fn sequence_value_reservation(
position: SequenceValuePosition,
increment: i64,
min_value: i64,
max_value: i64,
cycle: bool,
cache_size: i64,
) -> Option<SequenceValueReservation> {
let SequenceValuePosition {
current,
called,
log_count,
} = position;
debug_assert_ne!(increment, 0);
debug_assert!(cache_size > 0);
let first_value = if called {
match current
.checked_add(increment)
.filter(|value| (min_value..=max_value).contains(value))
{
Some(value) => value,
None if cycle && increment > 0 => min_value,
None if cycle => max_value,
None => return None,
}
} else {
current
};
let distance = if increment > 0 {
i128::from(max_value) - i128::from(first_value)
} else {
i128::from(first_value) - i128::from(min_value)
};
let step = i128::from(increment).abs();
let available = distance / step + 1;
let count = available.min(i128::from(cache_size));
let last_value = i128::from(first_value) + i128::from(increment) * (count - 1);
let initial_count = i128::from(!called);
let cache_fetch = i128::from(cache_size) - initial_count;
let mut fetch = cache_fetch;
let mut next_log_count = i128::from(log_count);
if i128::from(log_count) < cache_fetch || !called {
fetch += 32;
next_log_count = fetch;
}
let fetched = fetch.min(available - initial_count);
next_log_count -= fetched.min(cache_fetch);
next_log_count -= fetch - fetched;
Some(SequenceValueReservation {
first_value,
last_value: i64::try_from(last_value).expect("reserved sequence value stays in bounds"),
count: i64::try_from(count).expect("reservation count cannot exceed cache size"),
log_count: i64::try_from(next_log_count)
.expect("persisted sequence log count cannot exceed the cache request"),
})
}
pub trait CatalogFacade: Send + Sync {
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 get_metadata(&self, key: &str) -> StorageBackendResult<Option<String>>;
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 save_schema(&self, name: &str) -> StorageBackendResult<()> {
self.save_schema_row(&SchemaRow::legacy(name))
}
fn load_schemas(&self) -> StorageBackendResult<Vec<String>> {
Ok(self
.load_schema_rows()?
.into_iter()
.map(|schema| schema.name)
.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 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],
value: i64,
called: bool,
log_count: i64,
) -> StorageBackendResult<Option<i64>>;
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 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 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,
role_owner: &str,
acl: Option<&[TableAclEntry]>,
column_acls: &std::collections::BTreeMap<String, Vec<TableAclEntry>>,
) -> 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::{
sequence_value_reservation, RelationIdentity, SequenceValuePosition,
SequenceValueReservation,
};
const fn sequence_position(
current: i64,
called: bool,
log_count: i64,
) -> SequenceValuePosition {
SequenceValuePosition {
current,
called,
log_count,
}
}
#[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()
]
);
}
#[test]
fn sequence_reservations_track_postgresql_log_counts() {
assert_eq!(
sequence_value_reservation(sequence_position(1, false, 0), 1, 1, i64::MAX, false, 1),
Some(SequenceValueReservation {
first_value: 1,
last_value: 1,
count: 1,
log_count: 32,
})
);
assert_eq!(
sequence_value_reservation(sequence_position(1, true, 32), 1, 1, i64::MAX, false, 1),
Some(SequenceValueReservation {
first_value: 2,
last_value: 2,
count: 1,
log_count: 31,
})
);
assert_eq!(
sequence_value_reservation(sequence_position(1, false, 0), 1, 1, i64::MAX, false, 10),
Some(SequenceValueReservation {
first_value: 1,
last_value: 10,
count: 10,
log_count: 32,
})
);
assert_eq!(
sequence_value_reservation(sequence_position(5, false, 0), 2, 3, 9, true, 3),
Some(SequenceValueReservation {
first_value: 5,
last_value: 9,
count: 3,
log_count: 0,
})
);
assert_eq!(
sequence_value_reservation(
sequence_position(1, false, 0),
1,
1,
i64::MAX,
false,
i64::MAX,
),
Some(SequenceValueReservation {
first_value: 1,
last_value: i64::MAX,
count: i64::MAX,
log_count: 0,
})
);
}
}