use crate::{
db::{DatabaseAdapter, quote_postgres_identifier},
error::{FraiseQLError, Result},
schema::{CompiledSchema, MutationOperation},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SourceKind {
Relation,
Function,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SourceProbe {
pub schema: Option<String>,
pub name: String,
pub kind: SourceKind,
}
impl SourceProbe {
#[must_use]
pub fn display_name(&self) -> String {
match &self.schema {
Some(s) => format!("{s}.{}", self.name),
None => self.name.clone(),
}
}
}
fn split_source(sql_source: &str) -> (Option<String>, String) {
match sql_source.split_once('.') {
Some((schema, name)) => (Some(schema.to_string()), name.to_string()),
None => (None, sql_source.to_string()),
}
}
const fn mutation_source(mutation: &crate::schema::MutationDefinition) -> Option<&str> {
if let Some(src) = &mutation.sql_source {
return Some(src.as_str());
}
match &mutation.operation {
MutationOperation::Insert { table }
| MutationOperation::Update { table }
| MutationOperation::Delete { table }
if !table.is_empty() =>
{
Some(table.as_str())
},
_ => None,
}
}
#[must_use]
pub fn sql_source_probes(schema: &CompiledSchema) -> Vec<SourceProbe> {
let mut probes = Vec::with_capacity(schema.queries.len() + schema.mutations.len());
for query in &schema.queries {
if let Some(source) = &query.sql_source {
let (schema_part, name) = split_source(source);
probes.push(SourceProbe {
schema: schema_part,
name,
kind: SourceKind::Relation,
});
}
}
for mutation in &schema.mutations {
if let Some(source) = mutation_source(mutation) {
let (schema_part, name) = split_source(source);
probes.push(SourceProbe {
schema: schema_part,
name,
kind: SourceKind::Function,
});
}
}
probes
}
pub async fn refuse_standby_unreadable_sources<A: DatabaseAdapter + ?Sized>(
adapter: &A,
schema: &CompiledSchema,
) -> Result<()> {
if !adapter.serves_reads_from_standbys() {
return Ok(());
}
let unreadable = standby_unreadable_sources(adapter, schema).await?;
if unreadable.is_empty() {
return Ok(());
}
Err(FraiseQLError::Configuration {
message: format!(
"Read replicas are configured, but {} source relation(s) of this schema depend \
on an UNLOGGED or temporary table, which a hot standby cannot read: every query \
over them would fail on every replica.\n - {}\nMake those tables LOGGED \
(`ALTER TABLE … SET LOGGED`; for pg_tviews, set \
`pg_tviews.unlogged_by_default = off` and recreate the TVIEWs) — at the cost \
of WAL for every refresh — or remove `read_replica_urls`.",
unreadable.len(),
unreadable.join("\n - ")
),
})
}
pub async fn standby_unreadable_sources<A: DatabaseAdapter + ?Sized>(
adapter: &A,
schema: &CompiledSchema,
) -> Result<Vec<String>> {
let mut unreadable: Vec<String> = Vec::new();
for probe in sql_source_probes(schema) {
if probe.kind != SourceKind::Relation {
continue;
}
let ident = match &probe.schema {
Some(s) => format!(
"{}.{}",
quote_postgres_identifier(s),
quote_postgres_identifier(&probe.name)
),
None => quote_postgres_identifier(&probe.name),
};
let literal = ident.replace('\'', "''");
let sql = format!(
"WITH RECURSIVE rel(oid, kind) AS ( \
SELECT c.oid, c.relkind FROM pg_class c WHERE c.oid = to_regclass('{literal}') \
UNION \
SELECT base.oid, base.relkind FROM rel \
JOIN pg_rewrite r ON r.ev_class = rel.oid \
JOIN pg_depend d ON d.classid = 'pg_rewrite'::regclass AND d.objid = r.oid \
AND d.refclassid = 'pg_class'::regclass \
JOIN pg_class base ON base.oid = d.refobjid AND base.oid <> rel.oid \
WHERE rel.kind = 'v' \
) \
SELECT c.oid::regclass::text AS relation FROM rel \
JOIN pg_class c ON c.oid = rel.oid WHERE c.relpersistence IN ('u', 't') \
ORDER BY 1"
);
for row in adapter.execute_raw_query(&sql).await? {
if let Some(relation) = row.get("relation").and_then(serde_json::Value::as_str) {
unreadable.push(format!("{} — reads {relation}", probe.display_name()));
}
}
}
Ok(unreadable)
}
#[cfg(test)]
#[path = "source_probe_tests.rs"]
mod source_probe_tests;