#![allow(clippy::result_large_err)]
use std::collections::BTreeMap;
use crate::catalog::CatalogQuery;
use crate::catalog::assemble::reparse_expression_text;
use crate::catalog::error::CatalogError;
use crate::catalog::rows::Row;
use crate::identifier::{Identifier, QualifiedName};
use crate::ir::publication::{PublicationScope, PublishKinds, PublishedTable};
const Q_PUB: CatalogQuery = CatalogQuery::Publications;
const Q_REL: CatalogQuery = CatalogQuery::PublicationRel;
const Q_NS: CatalogQuery = CatalogQuery::PublicationNamespace;
const Q_ATTR: CatalogQuery = CatalogQuery::PublicationAttributes;
pub struct PartialPublication {
pub oid: i64,
pub name: Identifier,
pub owner: Option<Identifier>,
pub all_tables: bool,
pub publish: PublishKinds,
pub publish_via_partition_root: bool,
pub comment: Option<String>,
}
pub fn decode_publication_row(row: &Row) -> Result<PartialPublication, CatalogError> {
let name_str = row.get_text(Q_PUB, "name")?;
let owner_str = row.get_text(Q_PUB, "owner")?;
let comment_str = row.get_text(Q_PUB, "comment")?;
let name = Identifier::from_unquoted(&name_str).map_err(|e| CatalogError::BadColumnType {
query: Q_PUB,
column: "name".to_string(),
message: format!("invalid publication name {name_str:?}: {e}"),
})?;
let owner = if owner_str.is_empty() {
None
} else {
Some(
Identifier::from_unquoted(&owner_str).map_err(|e| CatalogError::BadColumnType {
query: Q_PUB,
column: "owner".to_string(),
message: format!("invalid owner name {owner_str:?}: {e}"),
})?,
)
};
let comment = if comment_str.is_empty() {
None
} else {
Some(comment_str)
};
Ok(PartialPublication {
oid: row.get_int(Q_PUB, "oid")?,
name,
owner,
all_tables: row.get_bool(Q_PUB, "all_tables")?,
publish: PublishKinds {
insert: row.get_bool(Q_PUB, "pub_insert")?,
update: row.get_bool(Q_PUB, "pub_update")?,
delete: row.get_bool(Q_PUB, "pub_delete")?,
truncate: row.get_bool(Q_PUB, "pub_truncate")?,
},
publish_via_partition_root: row.get_bool(Q_PUB, "publish_via_partition_root")?,
comment,
})
}
pub struct PartialPublicationRel {
pub pub_oid: i64,
pub qname: QualifiedName,
pub row_filter_sql: Option<String>,
pub col_attnums: Option<Vec<i64>>,
pub rel_oid: i64,
}
pub fn decode_publication_rel_row(row: &Row) -> Result<PartialPublicationRel, CatalogError> {
let schema_str = row.get_text(Q_REL, "schema")?;
let table_str = row.get_text(Q_REL, "table_name")?;
let schema =
Identifier::from_unquoted(&schema_str).map_err(|e| CatalogError::BadColumnType {
query: Q_REL,
column: "schema".to_string(),
message: format!("invalid schema {schema_str:?}: {e}"),
})?;
let table = Identifier::from_unquoted(&table_str).map_err(|e| CatalogError::BadColumnType {
query: Q_REL,
column: "table_name".to_string(),
message: format!("invalid table {table_str:?}: {e}"),
})?;
let col_attnums = if row.is_null("col_attnums") {
None
} else {
Some(row.get_int_array(Q_REL, "col_attnums")?)
};
Ok(PartialPublicationRel {
pub_oid: row.get_int(Q_REL, "pub_oid")?,
qname: QualifiedName::new(schema, table),
row_filter_sql: row.get_opt_text(Q_REL, "row_filter")?,
col_attnums,
rel_oid: row.get_int(Q_REL, "rel_oid")?,
})
}
pub struct PartialPublicationNamespace {
pub pub_oid: i64,
pub schema: Identifier,
}
pub fn decode_publication_namespace_row(
row: &Row,
) -> Result<PartialPublicationNamespace, CatalogError> {
let schema_str = row.get_text(Q_NS, "schema")?;
let schema =
Identifier::from_unquoted(&schema_str).map_err(|e| CatalogError::BadColumnType {
query: Q_NS,
column: "schema".to_string(),
message: format!("invalid schema {schema_str:?}: {e}"),
})?;
Ok(PartialPublicationNamespace {
pub_oid: row.get_int(Q_NS, "pub_oid")?,
schema,
})
}
pub struct PublicationAttribute {
pub rel_oid: i64,
pub attnum: i64,
pub attname: String,
}
pub fn decode_publication_attribute_row(row: &Row) -> Result<PublicationAttribute, CatalogError> {
Ok(PublicationAttribute {
rel_oid: row.get_int(Q_ATTR, "rel_oid")?,
attnum: row.get_int(Q_ATTR, "attnum")?,
attname: row.get_text(Q_ATTR, "attname")?,
})
}
pub fn resolve_column_names(
attnums: &[i64],
attname_by_attnum: &BTreeMap<i64, String>,
) -> Result<Vec<Identifier>, CatalogError> {
attnums
.iter()
.map(|n| {
let name = attname_by_attnum
.get(n)
.ok_or_else(|| CatalogError::DanglingReference {
kind: "publication column attnum",
what: format!("attnum {n}"),
})?;
Identifier::from_unquoted(name).map_err(|e| CatalogError::BadColumnType {
query: Q_ATTR,
column: "attname".to_string(),
message: format!("invalid column name {name:?}: {e}"),
})
})
.collect()
}
pub fn assemble_published_table(
rel: PartialPublicationRel,
columns: Option<Vec<Identifier>>,
) -> Result<PublishedTable, CatalogError> {
let row_filter = rel
.row_filter_sql
.map(|sql| reparse_expression_text(&sql))
.transpose()?;
Ok(PublishedTable {
qname: rel.qname,
row_filter,
columns,
})
}
pub fn build_scope(
all_tables: bool,
rels: Vec<PartialPublicationRel>,
column_resolver: impl Fn(&PartialPublicationRel) -> Result<Option<Vec<Identifier>>, CatalogError>,
namespaces: Vec<PartialPublicationNamespace>,
) -> Result<PublicationScope, CatalogError> {
if all_tables {
return Ok(PublicationScope::AllTables);
}
let mut tables = Vec::with_capacity(rels.len());
for r in rels {
let cols = column_resolver(&r)?;
tables.push(assemble_published_table(r, cols)?);
}
let schemas = namespaces.into_iter().map(|n| n.schema).collect();
Ok(PublicationScope::Selective { schemas, tables })
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn resolve_columns_handles_simple_case() {
let map = BTreeMap::from([(1_i64, "id".to_string()), (2_i64, "name".to_string())]);
let attnums = vec![2, 1];
let cols = resolve_column_names(&attnums, &map).unwrap();
assert_eq!(
cols.iter().map(Identifier::as_str).collect::<Vec<_>>(),
vec!["name", "id"]
);
}
#[test]
fn resolve_columns_fails_on_missing_attnum() {
let map = BTreeMap::from([(1_i64, "id".to_string())]);
let err = resolve_column_names(&[3], &map).unwrap_err();
assert!(
format!("{err}").contains("attnum 3"),
"error should mention attnum 3: {err}"
);
}
#[test]
fn resolve_columns_empty_input_returns_empty() {
let map = BTreeMap::from([(1_i64, "id".to_string())]);
let cols = resolve_column_names(&[], &map).unwrap();
assert!(cols.is_empty());
}
#[test]
fn build_scope_all_tables_ignores_rels_and_namespaces() {
let scope = build_scope(true, vec![], |_| Ok(None), vec![]).unwrap();
assert!(matches!(scope, PublicationScope::AllTables));
}
}