use super::Storage;
use super::ViewColumns;
use super::index_name;
use super::indexes::{create_tview_indexes, storage_clause};
use crate::error::TViewError;
use crate::error::TViewResult;
use crate::utils::log_debug;
use crate::utils::quote_identifier;
use pgrx::pg_sys;
use pgrx::pg_sys::Oid;
use pgrx::prelude::Spi;
use pgrx::prelude::notice;
pub(crate) fn key_table_on_identity(
schema_name: &str,
tview_name: &str,
identity: &str,
) -> TViewResult<()> {
let table = relation_oid(schema_name, tview_name)?;
let qualified = crate::utils::qualified_relname_from_oid(table)?;
let catalog = |e: pgrx::spi::Error| TViewError::CatalogError {
operation: format!("Read the keys of {qualified}"),
pg_error: e.to_string(),
};
let pk_unique = index_name(tview_name, "pk_unique");
let leftover = Spi::get_one_with_args::<bool>(
"SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_index i \
JOIN pg_catalog.pg_class c ON c.oid = i.indexrelid \
WHERE i.indrelid = $1 AND c.relname = $2)",
&[
crate::utils::spi::oid(table),
crate::utils::spi::text(pk_unique.as_str()),
],
)
.map_err(catalog)?;
if leftover == Some(true) {
let sql = format!(
"DROP INDEX {}.{}",
quote_identifier(schema_name),
quote_identifier(&pk_unique)
);
crate::utils::spi_run_ddl(&sql).map_err(|e| TViewError::SpiError {
query: sql,
error: e,
})?;
}
let key = Spi::get_one_with_args::<Vec<String>>(
"SELECT pg_catalog.array_agg(a.attname::text ORDER BY a.attnum) \
FROM pg_catalog.pg_index i \
JOIN pg_catalog.pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY (i.indkey) \
WHERE i.indrelid = $1 AND i.indisprimary",
&[crate::utils::spi::oid(table)],
)
.map_err(catalog)?
.unwrap_or_default();
match key.as_slice() {
[column] if column == identity => Ok(()),
[] => {
let sql = format!(
"ALTER TABLE {qualified} ADD PRIMARY KEY ({})",
quote_identifier(identity)
);
crate::utils::spi_run_ddl(&sql).map_err(|e| TViewError::SpiError {
query: sql,
error: e,
})
}
_ => Err(TViewError::DefinitionRefused {
reason: format!(
"{qualified} is keyed on ({}), but its rows are named by {identity}: \
pg_tviews_create_or_replace() with the same query rebuilds it",
key.join(", ")
),
}),
}
}
pub(crate) fn relation_oid(schema: &str, name: &str) -> TViewResult<pg_sys::Oid> {
find_relation(schema, name)?.ok_or_else(|| TViewError::CatalogError {
operation: format!("Look up {schema}.{name}"),
pg_error: "relation not found".to_string(),
})
}
pub(crate) fn relation_exists(schema: &str, name: &str) -> TViewResult<bool> {
Ok(find_relation(schema, name)?.is_some())
}
pub(crate) fn find_relation(schema: &str, name: &str) -> TViewResult<Option<pg_sys::Oid>> {
Spi::get_one_with_args::<pg_sys::Oid>(
"SELECT (SELECT c.oid FROM pg_catalog.pg_class c \
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
WHERE n.nspname = $1 AND c.relname = $2)",
&[
crate::utils::spi::text(schema),
crate::utils::spi::text(name),
],
)
.map_err(|e| TViewError::CatalogError {
operation: format!("Look up {schema}.{name}"),
pg_error: e.to_string(),
})
}
pub(crate) fn view_source_columns(view_oid: Oid, source_oid: Oid) -> Vec<String> {
match crate::catalog::reads::view_columns_read(view_oid, source_oid) {
Ok(columns) => columns.into_iter().map(|(name, _)| name).collect(),
Err(e) => {
notice!(
"view_source_columns({view_oid:?}, {source_oid:?}): {e} — cascade will always refresh"
);
Vec::new()
}
}
}
pub(crate) fn tview_exists(tview_name: &str) -> TViewResult<bool> {
let entity_name = tview_name.trim_start_matches("tv_");
let args = vec![crate::utils::spi::text(entity_name)];
Spi::get_one_with_args::<bool>(
&format!(
"SELECT COUNT(*) > 0 FROM {} WHERE entity = $1",
crate::utils::meta_table()
),
&args,
)
.map_err(|e| TViewError::CatalogError {
operation: format!("Check TVIEW exists: {tview_name}"),
pg_error: e.to_string(),
})
.map(|opt| opt.unwrap_or(false))
}
pub(crate) fn create_backing_view(
view_name: &str,
select_sql: &str,
schema_name: &str,
) -> TViewResult<()> {
let qi_schema = quote_identifier(schema_name);
let qi_view = quote_identifier(view_name);
let create_view_sql = format!("CREATE VIEW {qi_schema}.{qi_view} AS {select_sql}");
log_debug!(
"create_backing_view START - schema='{}', view='{}', sql_len={}",
schema_name,
view_name,
create_view_sql.len()
);
match crate::utils::spi_run_ddl(&create_view_sql) {
Ok(()) => {
log_debug!("spi_run_ddl SUCCEEDED for {}.{}", schema_name, view_name);
}
Err(e) => {
log_debug!(
"spi_run_ddl FAILED - {}.{} - error: {}",
schema_name,
view_name,
e
);
return Err(TViewError::SpiError {
query: create_view_sql.clone(),
error: e,
});
}
}
log_debug!(
"checking if view exists - schema='{}', view='{}' in pg_class",
schema_name,
view_name
);
let check_args = vec![
crate::utils::spi::text(view_name),
crate::utils::spi::text(schema_name),
];
let exists = match Spi::get_one_with_args::<i32>(
"SELECT 1 FROM pg_class c \
JOIN pg_namespace n ON c.relnamespace = n.oid \
WHERE c.relname = $1 AND n.nspname = $2 AND c.relkind = 'v'",
&check_args,
) {
Ok(result) => {
if result.is_some() {
log_debug!(
"VERIFIED - backing view {}.{} exists in pg_class",
schema_name,
view_name
);
true
} else {
log_debug!(
"VERIFICATION FAILED - backing view {}.{} not found in pg_class after spi_run_ddl",
schema_name,
view_name
);
false
}
}
Err(e) => {
log_debug!(
"verification query FAILED - could not check pg_class: {}",
e
);
return Err(TViewError::SpiError {
query: format!("Check view {schema_name}.{view_name} exists"),
error: e.to_string(),
});
}
};
if !exists {
return Err(TViewError::CatalogError {
operation: format!("Create view {schema_name}.{view_name}"),
pg_error: "View was not created (CREATE VIEW succeeded but view missing from pg_class)"
.to_string(),
});
}
Ok(())
}
pub(crate) fn create_materialized_table(
tview_name: &str,
schema: &ViewColumns,
schema_name: &str,
identity: &str,
storage: Storage,
view_oid: pg_sys::Oid,
) -> TViewResult<Vec<String>> {
let qi_schema = quote_identifier(schema_name);
let qi_tview = quote_identifier(tview_name);
let key = |name: &str| if name == identity { " PRIMARY KEY" } else { "" };
let mut columns = Vec::new();
if let Some(pk) = &schema.pk {
columns.push(format!("{} BIGINT{}", quote_identifier(pk), key(pk)));
}
if let Some(id) = &schema.id {
let not_null = if id == identity { key(id) } else { " NOT NULL" };
columns.push(format!("{} UUID{not_null}", quote_identifier(id)));
}
let view_types: std::collections::HashMap<String, String> =
crate::utils::column_types(view_oid)?.into_iter().collect();
let view_type = |col: &str, fallback: &str| {
view_types
.get(col)
.cloned()
.unwrap_or_else(|| fallback.to_string())
};
if let Some(identifier) = &schema.identifier {
columns.push(format!(
"{} {}{}",
quote_identifier(identifier),
view_type(identifier, "TEXT"),
key(identifier)
));
}
if let Some(data) = &schema.data {
columns.push(format!("{} JSONB", quote_identifier(data)));
}
for fk in &schema.fk {
columns.push(format!(
"{} {}{}",
quote_identifier(fk),
view_type(fk, "BIGINT"),
key(fk)
));
}
for uuid_fk in &schema.uuid_fk {
columns.push(format!(
"{} {}{}",
quote_identifier(uuid_fk),
view_type(uuid_fk, "UUID"),
key(uuid_fk)
));
}
for (col_name, col_type) in &schema.additional {
columns.push(format!(
"{} {}{}",
quote_identifier(col_name),
view_type(col_name, col_type),
key(col_name)
));
}
if !columns.iter().any(|c| c.ends_with(" PRIMARY KEY")) {
return Err(TViewError::DefinitionRefused {
reason: format!("{tview_name} has no column {identity} to key its rows on"),
});
}
columns.push("created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()".to_string());
columns.push("updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()".to_string());
let columns_sql = columns.join(",\n ");
let unlogged_keyword = if storage.logged { "" } else { "UNLOGGED " };
let with = storage_clause(storage.fillfactor);
let create_table_sql = format!(
"CREATE {unlogged_keyword}TABLE {qi_schema}.{qi_tview} (\n {columns_sql}\n){with}"
);
crate::utils::spi_run_ddl(&create_table_sql).map_err(|e| TViewError::SpiError {
query: create_table_sql,
error: e,
})?;
create_tview_indexes(tview_name, schema, schema_name, storage.data_gin_index)
}
pub(crate) fn populate_initial_data(
tview_name: &str,
schema_name: &str,
view_oid: pg_sys::Oid,
) -> TViewResult<u64> {
let view = crate::utils::qualified_relname_from_oid(view_oid)?;
let view_columns = crate::utils::get_view_columns_by_oid(view_oid)?;
if view_columns.is_empty() {
return Err(TViewError::CatalogError {
operation: format!("Get columns for view {view}"),
pg_error: "View has no selectable columns".to_string(),
});
}
let insert_columns = view_columns;
let qi_schema = quote_identifier(schema_name);
let qi_tview = quote_identifier(tview_name);
let col_list = insert_columns
.iter()
.map(|c| quote_identifier(c))
.collect::<Vec<_>>()
.join(", ");
let insert_sql = format!(
"INSERT INTO {qi_schema}.{qi_tview} ({col_list}) \
SELECT {col_list} FROM {view}"
);
let _pin = crate::owner::RenderPin::new();
let rows = Spi::connect_mut(|client| client.update(&insert_sql, None, &[]).map(|t| t.len()))
.map_err(|e| TViewError::SpiError {
query: insert_sql,
error: e.to_string(),
})?;
Ok(rows as u64)
}