use super::{
context::{ViewCreationContext, ViewCreationTransactions},
publication, ViewRegistration,
};
use crate::catalog::view::{StoredView, StoredViewKind};
use crate::row_locks::{
binding::{bind_relation, RelationBinding},
RelationLockMode,
};
use uqa_core::RelationIdentity;
use uqa_sql::{
catalog::{
regrole_dependencies::StoredRegroleConstants,
view::{create_view_output_columns, named_view_schema, validate_view_column_types},
},
plan::{QueryPlan, UnifiedPlan},
schema::view_creation::{
replacement_is_view, temporary_view_notice, validate_replacement_schema,
view_becomes_temporary, view_creation_target,
},
SQLError,
};
pub fn register_view(
transactions: &dyn ViewCreationTransactions,
name: &str,
body: uqa_sql::ast::SelectStmt,
) -> Result<(), SQLError> {
transactions.with_view_creation(Box::new(move |context| {
let plan = UnifiedPlan::Query(Box::new(QueryPlan::lower_with(body, &|aggregate: &str| {
context
.routines
.has_registered_aggregate_function(aggregate)
})));
let UnifiedPlan::Query(plan) = plan else {
return Err(SQLError::Internal(
"view lowering produced a non-query plan".into(),
));
};
register_view_plan_inner(
context,
ViewRegistration {
name,
column_names: &[],
plan: *plan,
or_replace: true,
persistence: uqa_sql::ast::RelationPersistence::Permanent,
options: &[],
params: &[],
},
)
}))
}
pub fn register_view_plan(
transactions: &dyn ViewCreationTransactions,
registration: ViewRegistration<'_>,
) -> Result<(), SQLError> {
transactions.with_view_creation(Box::new(move |context| {
register_view_plan_inner(context, registration)
}))
}
fn replacement_view(
context: &ViewCreationContext<'_>,
name: &str,
relation: &RelationIdentity,
or_replace: bool,
replacement_schema: &uqa_sql::RowSchema,
) -> Result<Option<StoredView>, SQLError> {
let binding = bind_relation(
context.locks,
RelationLockMode::AccessExclusive,
false,
|| {
let kind = context.names.relation_kind_at(name).map_err(|error| {
SQLError::Internal(format!("resolve relation `{name}`: {error}"))
})?;
let view = if replacement_is_view(name, kind, or_replace)? {
Some(context.views.view(relation).ok_or_else(|| {
SQLError::Internal(format!(
"view `{name}` exists in the catalog but has no loaded definition"
))
})?)
} else {
None
};
Ok(view.map(|view| RelationBinding {
name: name.into(),
object_id: Some(view.object_id),
value: view,
}))
},
|binding| {
context.owners.ensure_owner(name, &binding.value)?;
context.namespace.ensure_create(name)
},
)?;
if let Some(binding) = &binding {
let existing = &binding.value;
let existing_schema = uqa_sql::semantics::view_rewrite::context::stored_view_schema(
context.rewrite,
&existing.rewrite_definition(),
)?;
validate_replacement_schema(&existing_schema, replacement_schema)?;
}
Ok(binding.map(|binding| binding.value))
}
pub(super) fn reject_regrole_constants(
context: &ViewCreationContext<'_>,
plan: &mut QueryPlan,
) -> Result<(), SQLError> {
let mut constants = StoredRegroleConstants::default();
constants.collect_query_plan(plan);
constants.reject_with(context.regroles)
}
fn register_view_plan_inner(
context: &ViewCreationContext<'_>,
registration: ViewRegistration<'_>,
) -> Result<(), SQLError> {
let ViewRegistration {
name,
column_names,
mut plan,
or_replace,
persistence,
options,
params,
} = registration;
context
.catalog
.synchronize()
.map_err(|err| SQLError::Internal(format!("refresh view catalog: {err}")))?;
let owner = context.namespace.bind_owner()?;
context.bindings.lock_relations(&plan)?;
let uses_temporary_relation = context.bindings.bind_relations(&mut plan)?;
if view_becomes_temporary(persistence, uses_temporary_relation) {
context.notices.push(temporary_view_notice(name)?);
}
let (name, persistence) = view_creation_target(
&context.namespace,
name,
persistence,
uses_temporary_relation,
)?;
let relation = RelationIdentity::from_legacy_name(&name)
.map_err(|err| SQLError::Internal(format!("invalid canonical view name: {err}")))?;
let query_schema = context.bindings.bind_routines(&mut plan, params)?;
context.bindings.bind_type_identities(&mut plan)?;
uqa_sql::schema::dependencies::oid_alias::read_oid_alias_constants_in_plan(
context.aliases,
&mut plan,
)?;
reject_regrole_constants(context, &mut plan)?;
let output_columns = create_view_output_columns(&query_schema, column_names)?;
for ty in query_schema.column_types().iter().flatten() {
context.routines.require_type_usage(ty)?;
}
validate_view_column_types(&query_schema, &output_columns)?;
let replacement_schema = named_view_schema(&query_schema, &output_columns)?;
let existing_view =
replacement_view(context, &name, &relation, or_replace, &replacement_schema)?;
if existing_view.is_none() {
context.namespace.retain_owner(&owner)?;
}
context.locks.prepare_definition_write()?;
context.namespace.ensure_create(&name)?;
let (object_id, catalog_oids) = if let Some(existing) = existing_view.as_ref() {
(existing.object_id, existing.catalog_oids)
} else {
let object_id = context.catalog.allocate_identity().map_err(|error| {
SQLError::Internal(format!("allocate view `{name}` identity: {error}"))
})?;
(object_id, Some(allocate_view_oids(context, &relation)?))
};
let row_type_array_name = existing_view
.as_ref()
.and_then(|view| view.row_type_array_name.clone());
let mut view = StoredView {
security: existing_view.as_ref().map_or_else(
|| uqa_sql::catalog::security::BoundTableSecurity::owner(owner.identity()),
|view| view.security.clone(),
),
definition: uqa_sql::catalog::stored_view::StoredViewDefinition {
object_id,
query: plan,
output_columns: Some(output_columns),
persistence,
options: options.to_vec(),
kind: StoredViewKind::View,
materialized_rows: Vec::new(),
materialized_column_types: Vec::new(),
populated: true,
catalog_oids,
row_type_array_name,
},
};
uqa_sql::semantics::view_rewrite::validate_view_definition_check_option(
context.rewrite,
&name,
&view.rewrite_definition(),
)?;
if existing_view.is_none() {
context.namespace.reserve_row_type_name(&name)?;
view.row_type_array_name = Some(crate::schema::types::arrays::reserve_array_name(
&context.namespace,
&relation.schema,
&relation.name,
)?);
}
publication::publish_regular_view(context.publication, context.changes, relation, view, &name)?;
Ok(())
}
pub(super) fn allocate_view_oids(
context: &ViewCreationContext<'_>,
relation: &RelationIdentity,
) -> Result<uqa_sql::catalog::relation_oids::RelationCatalogOids, SQLError> {
context
.identities
.allocator(crate::catalog::identity::allocate_catalog_object_id)
.allocate_relation_oids(
uqa_sql::catalog::relation_oids::RelationOidKind::View,
relation,
)
}