use super::*;
use crate::{
catalog::{
foreign::StoredForeignTable,
view::{ViewIdentityAllocation, ViewRegistryRead, ViewRegistryState, ViewRegistryWrite},
},
schema::view_creation::context::ViewPlanBinding,
};
use context::*;
use std::{
cell::{Cell, RefCell},
sync::Arc,
};
use uqa_graph::GraphStoreHandle;
use uqa_sql::{
ast::{ColumnType, RelationPersistence},
catalog::roles::{
guards::{RoleCatalogGuards, RoleDefinitionRead, RoleMembershipRead},
RoleDefinition,
},
plan::{QueryPlan, UnifiedPlan},
RowSchema, SQLError, SQLParam,
};
#[derive(Clone, Copy, PartialEq, Eq)]
enum Failure {
None,
Binding,
Schema,
Persistence,
}
struct Fixture {
registry: RefCell<BTreeMap<RelationIdentity, StoredView>>,
rows: Vec<ViewRow>,
saved: RefCell<Vec<ViewRow>>,
failure: Failure,
bindings: Cell<usize>,
schemas: Cell<usize>,
identities: Cell<u8>,
writes: Cell<usize>,
roles: BTreeMap<String, RoleDefinition>,
foreign: BTreeMap<RelationIdentity, StoredForeignTable>,
graphs: BTreeMap<String, Arc<GraphStoreHandle>>,
}
fn view(object_id: u8, persistence: RelationPersistence) -> StoredView {
let UnifiedPlan::Query(query) =
UnifiedPlan::lower(uqa_sql::compile("SELECT 1 AS value").unwrap().remove(0))
else {
panic!("query fixture");
};
StoredView {
security: uqa_sql::catalog::security::BoundTableSecurity::owner(
uqa_sql::catalog::roles::RoleIdentity {
oid: 42,
object_id: [42; 16],
},
),
definition: uqa_sql::catalog::stored_view::StoredViewDefinition {
object_id: [object_id; 16],
catalog_oids: None,
row_type_array_name: None,
query: *query,
output_columns: Some(vec!["value".into()]),
persistence,
options: Vec::new(),
kind: StoredViewKind::View,
materialized_rows: Vec::new(),
materialized_column_types: Vec::new(),
populated: true,
},
}
}
impl Fixture {
fn new(failure: Failure) -> Self {
let rows = ["first", "second"]
.into_iter()
.map(|name| ViewRow {
relation: RelationIdentity::new("public", name),
security: uqa_storage::RelationSecurityRow::legacy("owner"),
definition_json: serde_json::to_string(
&view(0, RelationPersistence::Permanent).query,
)
.unwrap(),
})
.collect();
Self {
registry: RefCell::new(BTreeMap::from([
(
RelationIdentity::new("public", "previous"),
view(8, RelationPersistence::Permanent),
),
(
RelationIdentity::new("pg_temp_7", "session_view"),
view(9, RelationPersistence::Temporary),
),
])),
rows,
saved: RefCell::new(Vec::new()),
failure,
bindings: Cell::new(0),
schemas: Cell::new(0),
identities: Cell::new(0),
writes: Cell::new(0),
roles: BTreeMap::from([(
"owner".into(),
RoleDefinition {
oid: 42,
object_id: [42; 16],
revision: 1,
name: "owner".into(),
attributes: BTreeSet::new(),
connection_limit: -1,
},
)]),
foreign: BTreeMap::new(),
graphs: BTreeMap::new(),
}
}
fn context(&self) -> ViewRestoreContext<'_> {
ViewRestoreContext {
registry: self,
namespace: self,
roles: self,
identities: self,
bindings: self,
schemas: self,
sequences: self,
types: &crate::catalog::test_support::NoRoutines,
}
}
fn snapshot(
&self,
) -> Vec<(
RelationIdentity,
serde_json::Value,
uqa_sql::catalog::security::BoundTableSecurity,
)> {
self.registry
.borrow()
.iter()
.map(|(name, view)| {
(
name.clone(),
serde_json::to_value(view).unwrap(),
view.security(),
)
})
.collect()
}
fn assert_provisional_namespace(&self) {
assert!(
self.registry.try_borrow_mut().is_ok(),
"analysis runs after releasing the publication guard"
);
let views = self.registry.borrow();
assert_eq!(views.len(), 3);
assert!(views.contains_key(&RelationIdentity::new("public", "first")));
assert!(views.contains_key(&RelationIdentity::new("public", "second")));
assert!(views.contains_key(&RelationIdentity::new("pg_temp_7", "session_view")));
}
}
impl ViewRowsStorage for Fixture {
fn load_views(&self) -> StorageBackendResult<Vec<ViewRow>> {
Ok(self.rows.clone())
}
fn save_view(&self, row: &ViewRow) -> StorageBackendResult<()> {
if self.failure == Failure::Persistence && !self.saved.borrow().is_empty() {
return Err(StorageBackendError::Other(
"injected persistence failure".into(),
));
}
self.saved.borrow_mut().push(row.clone());
Ok(())
}
}
impl ViewRegistryState for Fixture {
fn views_read(&self) -> ViewRegistryRead<'_> {
Box::new(self.registry.borrow())
}
fn views_write(&self) -> ViewRegistryWrite<'_> {
self.writes.set(self.writes.get() + 1);
Box::new(self.registry.borrow_mut())
}
}
impl ViewIdentityAllocation for Fixture {
fn allocate_identity(&self) -> StorageBackendResult<[u8; 16]> {
let next = self.identities.get() + 1;
self.identities.set(next);
Ok([next; 16])
}
}
struct NoTables;
impl ViewRestoreTableNames for NoTables {
fn names(&self) -> Box<dyn Iterator<Item = &RelationIdentity> + '_> {
Box::new(std::iter::empty())
}
}
impl ViewRestoreNamespace for Fixture {
fn tables(&self) -> Box<dyn ViewRestoreTableNames + '_> {
Box::new(NoTables)
}
fn foreign_tables(&self) -> ViewRestoreForeignRead<'_> {
Box::new(&self.foreign)
}
fn graphs(&self) -> ViewRestoreGraphsRead<'_> {
Box::new(&self.graphs)
}
}
impl RoleCatalogGuards for Fixture {
fn role_definitions(&self) -> RoleDefinitionRead<'_> {
Box::new(&self.roles)
}
fn role_memberships(&self) -> RoleMembershipRead<'_> {
panic!("view metadata validation does not read memberships");
}
}
impl ViewPlanBinding for Fixture {
fn lock_relations(&self, _: &QueryPlan) -> Result<(), SQLError> {
panic!("restoration uses its explicit persisted namespace");
}
fn bind_relations(&self, _: &mut QueryPlan) -> Result<bool, SQLError> {
panic!("restoration uses its explicit persisted namespace");
}
fn bind_routines(
&self,
plan: &mut QueryPlan,
params: &[SQLParam],
) -> Result<RowSchema, SQLError> {
assert!(params.is_empty());
self.assert_provisional_namespace();
assert!(plan.relations_bound);
if self.bindings.get() == 1 {
assert!(
!self.registry.borrow()[&RelationIdentity::new("public", "first")]
.query
.relations_bound,
"the first bound definition is published before binding the next view"
);
}
plan.relations_bound = false;
self.bindings.set(self.bindings.get() + 1);
if self.failure == Failure::Binding && self.bindings.get() == 2 {
return Err(SQLError::Internal("injected binding failure".into()));
}
Ok(RowSchema::with_types(
vec!["value".into()],
vec![Some(ColumnType::Integer)],
))
}
fn bind_type_identities(&self, _: &mut QueryPlan) -> Result<(), SQLError> {
panic!("restoration keeps the recorded type identities");
}
}
impl ViewRestoreSchemas for Fixture {
fn stored_schema(&self, _: &StoredView) -> Result<RowSchema, SQLError> {
self.assert_provisional_namespace();
assert_eq!(self.bindings.get(), 2);
self.schemas.set(self.schemas.get() + 1);
if self.failure == Failure::Schema && self.schemas.get() == 2 {
return Err(SQLError::Internal("injected schema failure".into()));
}
Ok(RowSchema::with_types(
vec!["value".into()],
vec![Some(ColumnType::Integer)],
))
}
}
impl ViewRestoreSequences for Fixture {
fn resolve_loaded(&self, _: &str) -> StorageBackendResult<String> {
panic!("query fixture has no sequence reference");
}
}
#[test]
fn view_migration_publishes_the_complete_namespace_before_binding_and_keeps_temporary_views() {
let fixture = Fixture::new(Failure::None);
restore_views_from_catalog(&fixture.context(), &fixture, true).unwrap();
assert_eq!(fixture.bindings.get(), 2);
assert_eq!(fixture.schemas.get(), 2);
assert_eq!(fixture.identities.get(), 2);
assert_eq!(fixture.saved.borrow().len(), 2);
fixture.assert_provisional_namespace();
let views = fixture.registry.borrow();
assert_eq!(
views[&RelationIdentity::new("pg_temp_7", "session_view")].object_id,
[9; 16]
);
for name in ["first", "second"] {
let restored = &views[&RelationIdentity::new("public", name)];
assert_ne!(restored.object_id, [0; 16]);
assert_eq!(restored.output_columns, Some(vec!["value".into()]));
assert_eq!(
restored.security.role_owner,
fixture.roles["owner"].identity()
);
}
}
#[test]
fn view_migration_failures_restore_the_previous_registry_including_security_and_temporary_views() {
for failure in [Failure::Binding, Failure::Schema, Failure::Persistence] {
let fixture = Fixture::new(failure);
let before = fixture.snapshot();
let error = restore_views_from_catalog(&fixture.context(), &fixture, true).unwrap_err();
assert!(error.to_string().contains("injected"));
assert_eq!(fixture.snapshot(), before);
assert!(fixture.registry.try_borrow_mut().is_ok());
}
}
#[test]
fn load_only_view_restore_rejects_migration_before_registry_or_storage_publication() {
let mut fixture = Fixture::new(Failure::None);
for row in &mut fixture.rows {
row.security = view(0, RelationPersistence::Permanent)
.security
.row()
.into();
}
let before = fixture.snapshot();
let error = restore_views_from_catalog(&fixture.context(), &fixture, false).unwrap_err();
assert!(error
.to_string()
.contains("requires an initial-open metadata migration"));
assert_eq!(fixture.identities.get(), 2);
assert_eq!(fixture.writes.get(), 0);
assert_eq!(fixture.bindings.get(), 0);
assert!(fixture.saved.borrow().is_empty());
assert_eq!(fixture.snapshot(), before);
}
#[test]
fn load_only_view_restore_rejects_legacy_security_before_allocating_or_publishing() {
let mut fixture = Fixture::new(Failure::None);
for (identity, row) in (1..).zip(&mut fixture.rows) {
row.definition_json =
serde_json::to_string(&view(identity, RelationPersistence::Permanent)).unwrap();
}
let before = fixture.snapshot();
let error = restore_views_from_catalog(&fixture.context(), &fixture, false).unwrap_err();
assert!(error.to_string().contains("initial catalog migration"));
assert_eq!(fixture.identities.get(), 0);
assert_eq!(fixture.writes.get(), 0);
assert_eq!(fixture.bindings.get(), 0);
assert!(fixture.saved.borrow().is_empty());
assert_eq!(fixture.snapshot(), before);
}
fn retained_ranges(failure: Failure) -> Fixture {
let mut fixture = Fixture::new(failure);
for (identity, row) in (1..).zip(&mut fixture.rows) {
let mut stored = view(identity, RelationPersistence::Permanent);
let uqa_sql::plan::RelationalPlan::QueryBlock(block) = &mut stored.query.root else {
panic!("query fixture");
};
block.projections[0].expr = uqa_sql::ScalarExpr::Between {
expr: Box::new(uqa_sql::ScalarExpr::Literal(uqa_core::Value::Int(1))),
low: Box::new(uqa_sql::ScalarExpr::Literal(uqa_core::Value::Int(0))),
high: Box::new(uqa_sql::ScalarExpr::Literal(uqa_core::Value::Int(2))),
};
row.security = stored.security.row().into();
row.definition_json = serde_json::to_string(&stored.definition).unwrap();
}
fixture
}
#[test]
fn retained_range_migration_persists_without_reallocating_identities_or_rebinding_routines() {
let fixture = retained_ranges(Failure::None);
restore_views_from_catalog(&fixture.context(), &fixture, true).unwrap();
assert_eq!(fixture.identities.get(), 0);
assert_eq!(fixture.bindings.get(), 0);
assert_eq!(fixture.schemas.get(), 0);
assert_eq!(fixture.saved.borrow().len(), 2);
for row in fixture.saved.borrow().iter() {
let definition: uqa_sql::catalog::stored_view::StoredViewDefinition =
serde_json::from_str(&row.definition_json).unwrap();
let uqa_sql::plan::RelationalPlan::QueryBlock(block) = &definition.query.root else {
panic!("query fixture");
};
assert!(matches!(
block.projections[0].expr,
uqa_sql::ScalarExpr::And(_)
));
assert_eq!(
definition.object_id,
fixture.registry.borrow()[&row.relation].object_id
);
}
}
#[test]
fn retained_range_migration_rejects_load_only_restore_before_publication() {
let fixture = retained_ranges(Failure::None);
let before = fixture.snapshot();
let error = restore_views_from_catalog(&fixture.context(), &fixture, false).unwrap_err();
assert!(error
.to_string()
.contains("initial-open metadata migration"));
assert_eq!(fixture.identities.get(), 0);
assert_eq!(fixture.writes.get(), 0);
assert!(fixture.saved.borrow().is_empty());
assert_eq!(fixture.snapshot(), before);
}
#[test]
fn failed_retained_range_persistence_keeps_the_previous_registry() {
let fixture = retained_ranges(Failure::Persistence);
let before = fixture.snapshot();
let error = restore_views_from_catalog(&fixture.context(), &fixture, true).unwrap_err();
assert!(error.to_string().contains("injected persistence failure"));
assert_eq!(fixture.saved.borrow().len(), 1);
assert_eq!(fixture.writes.get(), 0);
assert_eq!(fixture.snapshot(), before);
}