github/dbx
4
0
Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-10-19 16:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(postgres): restore 9.x table DDL and sequence export

Browse Source
This commit is contained in:
zipg
2026-09-25 23:57:36 +08:00
committed by GitHub
parent 6db2589e37
commit e91340ee15
2 changed files with 256 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/dbx-core/src/data/database_export.rs
+165 -29
View File
@@ -1648,6 +1648,101 @@ fn is_postgres_extension_member_routine(object: &crate::types::ObjectInfo, membe
members.function_keys.contains(&(object.name.clone(), object.signature.clone().unwrap_or_default()))
}
const POSTGRES_EXPORT_SEQUENCES_SQL: &str = "SELECT c.relname, \
COALESCE(format_type(s.seqtypid, NULL), 'bigint'), \
COALESCE(s.seqstart::text, '1'), \
COALESCE(s.seqmin::text, '1'), \
COALESCE(s.seqmax::text, '9223372036854775807'), \
COALESCE(s.seqincrement::text, '1'), \
COALESCE(s.seqcycle, false), \
COALESCE(s.seqcache::text, '1'), \
t.relname, \
a.attname \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_sequence s ON s.seqrelid = c.oid \
LEFT JOIN pg_depend d ON d.classid = 'pg_class'::regclass \
AND d.objid = c.oid \
AND d.refclassid = 'pg_class'::regclass \
AND d.deptype IN ('a', 'i') \
LEFT JOIN pg_class t ON t.oid = d.refobjid \
LEFT JOIN pg_namespace tn ON tn.oid = t.relnamespace AND tn.nspname = n.nspname \
LEFT JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = d.refobjsubid \
WHERE c.relkind = 'S' AND n.nspname = $1 \
ORDER BY c.relname";
/// PostgreSQL 9.x sibling of [`POSTGRES_EXPORT_SEQUENCES_SQL`]: the `pg_sequence`
/// catalog only exists from PostgreSQL 10 on, so that query aborts the whole
/// structure export with `db error` on older servers (#10079). Sequence
/// parameters are filled in per sequence by
/// [`postgres_export_sequence_parameters_legacy`] instead.
const POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL: &str = "SELECT c.relname, \
'bigint', \
'1', \
'1', \
'9223372036854775807', \
'1', \
false, \
'1', \
t.relname, \
a.attname \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_depend d ON d.classid = 'pg_class'::regclass \
AND d.objid = c.oid \
AND d.refclassid = 'pg_class'::regclass \
AND d.deptype IN ('a', 'i') \
LEFT JOIN pg_class t ON t.oid = d.refobjid \
LEFT JOIN pg_namespace tn ON tn.oid = t.relnamespace AND tn.nspname = n.nspname \
LEFT JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = d.refobjsubid \
WHERE c.relkind = 'S' AND n.nspname = $1 \
ORDER BY c.relname";
#[derive(Debug)]
struct PostgresExportSequenceParameters {
start_value: String,
min_value: String,
max_value: String,
increment: String,
cycle: bool,
cache_value: String,
last_value: Option<String>,
}
/// Pre-10 servers store sequence parameters in the sequence relation itself
/// (`SELECT ... FROM <sequence>`), which is also the only way to read a
/// sequence's current value there. `last_value` is reported only once the
/// sequence has been called, matching `pg_sequence_last_value()`'s NULL on an
/// untouched sequence.
async fn postgres_export_sequence_parameters_legacy(
client: &deadpool_postgres::Client,
schema: &str,
sequence: &str,
) -> Result<Option<PostgresExportSequenceParameters>, tokio_postgres::Error> {
let qualified = format!(
"{}.{}",
quote_identifier(schema, &DatabaseType::Postgres),
quote_identifier(sequence, &DatabaseType::Postgres)
);
let sql = format!(
"SELECT start_value::text, min_value::text, max_value::text, increment_by::text, is_cycled, \
cache_value::text, CASE WHEN is_called THEN last_value::text END FROM {qualified}"
);
let rows = client.query(sql.as_str(), &[]).await?;
let Some(row) = rows.first() else {
return Ok(None);
};
Ok(Some(PostgresExportSequenceParameters {
start_value: row.get(0),
min_value: row.get(1),
max_value: row.get(2),
increment: row.get(3),
cycle: row.get(4),
cache_value: row.get(5),
last_value: row.get(6),
}))
}
async fn list_postgres_export_sequences(
state: &crate::connection::AppState,
pool_key: &str,
@@ -1665,34 +1760,23 @@ async fn list_postgres_export_sequences(
}
};
let client = pool.get().await.map_err(|e| e.to_string())?;
let rows = client
.query(
"SELECT c.relname, \
COALESCE(format_type(s.seqtypid, NULL), 'bigint'), \
COALESCE(s.seqstart::text, '1'), \
COALESCE(s.seqmin::text, '1'), \
COALESCE(s.seqmax::text, '9223372036854775807'), \
COALESCE(s.seqincrement::text, '1'), \
COALESCE(s.seqcycle, false), \
COALESCE(s.seqcache::text, '1'), \
t.relname, \
a.attname \
FROM pg_class c \
JOIN pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_sequence s ON s.seqrelid = c.oid \
LEFT JOIN pg_depend d ON d.classid = 'pg_class'::regclass \
AND d.objid = c.oid \
AND d.refclassid = 'pg_class'::regclass \
AND d.deptype IN ('a', 'i') \
LEFT JOIN pg_class t ON t.oid = d.refobjid \
LEFT JOIN pg_namespace tn ON tn.oid = t.relnamespace AND tn.nspname = n.nspname \
LEFT JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = d.refobjsubid \
WHERE c.relkind = 'S' AND n.nspname = $1 \
ORDER BY c.relname",
&[&schema],
)
.await
.map_err(|e| e.to_string())?;
// PostgreSQL 10+ keeps sequence parameters in `pg_sequence`; older servers
// have neither that catalog nor `pg_sequence_last_value()`, so the compat
// tier lists sequences (and owners) without it and reads each sequence
// relation directly afterwards.
let (rows, legacy_parameters) = match client.query(POSTGRES_EXPORT_SEQUENCES_SQL, &[&schema]).await {
Ok(rows) => (rows, false),
Err(primary_error) => match client.query(POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL, &[&schema]).await {
Ok(rows) => {
log::debug!(
"[postgres][database-export:sequences-compat-used] pg_sequence catalog unavailable ({}); reading sequence parameters from each sequence relation",
primary_error
);
(rows, true)
}
Err(_) => return Err(primary_error.to_string()),
},
};
let selected: HashSet<&str> = selected_tables.iter().map(String::as_str).collect();
let excluded: HashSet<&str> = excluded_tables.iter().map(String::as_str).collect();
@@ -1723,6 +1807,30 @@ async fn list_postgres_export_sequences(
return Ok(sequences);
}
if legacy_parameters {
for sequence in sequences.iter_mut() {
match postgres_export_sequence_parameters_legacy(&client, schema, &sequence.name).await {
Ok(Some(parameters)) => {
sequence.start_value = parameters.start_value;
sequence.min_value = parameters.min_value;
sequence.max_value = parameters.max_value;
sequence.increment = parameters.increment;
sequence.cycle = parameters.cycle;
sequence.cache_value = parameters.cache_value;
sequence.last_value = parameters.last_value;
}
Ok(None) => {}
Err(error) => log::debug!(
"[postgres][database-export:sequences-compat-read-failed] sequence={} schema={} error={}",
sequence.name,
schema,
error
),
}
}
return Ok(sequences);
}
let last_values = client
.query(
"SELECT c.relname, pg_sequence_last_value(c.oid)::text \
@@ -4088,7 +4196,7 @@ mod tests {
BuildDatabaseSqlExportOptions, BuildExportInsertStatementsOptions, BuildExportSqlInsertOptions,
DatabaseExportObjectCounts, DatabaseExportRequest, DatabaseExportWriter, DdlNormalizeOptions, ExportedTableSql,
PostgresExportExtension, PostgresExportSequence, PostgresExtensionMembers, DATABASE_EXPORT_INSERT_BATCH_SIZE,
DATABASE_EXPORT_ROW_LIMIT,
DATABASE_EXPORT_ROW_LIMIT, POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL, POSTGRES_EXPORT_SEQUENCES_SQL,
};
use super::{ExportProgress, LenientExportErrors};
use crate::connection::AppState;
@@ -6400,6 +6508,34 @@ mod tests {
);
}
#[test]
fn postgres_export_sequence_sql_tiers_skip_pg10_only_catalogs() {
// PostgreSQL 10 added `pg_sequence`; PostgreSQL 9.x rejects the whole
// query, which used to abort the structure export (#10079).
assert!(POSTGRES_EXPORT_SEQUENCES_SQL.contains("pg_sequence"));
assert!(!POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL.contains("pg_sequence"));
assert!(!POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL.contains("pg_sequence_last_value"));
// Both tiers must describe the same ten columns in the same order so
// the row decoding below stays identical.
for sql in [POSTGRES_EXPORT_SEQUENCES_SQL, POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL] {
for fragment in [
"SELECT c.relname",
"'bigint'",
"t.relname",
"a.attname",
"WHERE c.relkind = 'S' AND n.nspname = $1",
"ORDER BY c.relname",
] {
assert!(sql.contains(fragment), "missing {fragment} in {sql}");
}
}
// The compat tier substitutes literals for the pg_sequence columns.
for fallback in ["'1'", "'9223372036854775807'", "false"] {
assert!(POSTGRES_EXPORT_SEQUENCES_COMPAT_SQL.contains(fallback));
}
}
fn postgres_sequence(name: &str) -> PostgresExportSequence {
PostgresExportSequence {
name: name.to_string(),
crates/dbx-driver-postgres/src/postgres.rs
+91 -3
View File
@@ -5204,9 +5204,11 @@ fn postgres_check_constraints_sql() -> &'static str {
pub async fn list_constraints(pool: &Pool, schema: &str, table: &str) -> Result<Vec<ConstraintInfo>, String> {
let schema = if schema.is_empty() { "public" } else { schema };
let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
let rows = postgres_query_cached(&client, postgres_constraints_sql(), &[&schema, &table])
.await
.map_err(|e| e.to_string())?;
let tiers = postgres_constraint_query_tiers();
let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&schema, &table];
let rows =
query_with_compat_fallback("list_constraints", &tiers, |sql| postgres_query_cached(&client, sql, &params))
.await?;
Ok(rows
.iter()
@@ -5427,6 +5429,48 @@ fn postgres_constraints_sql() -> &'static str {
ORDER BY con.conname"
}
fn postgres_constraint_query_tiers() -> [&'static str; 2] {
[postgres_constraints_sql(), postgres_constraints_compat_sql()]
}
/// PostgreSQL 9.2/9.3 sibling of [`postgres_constraints_sql`]: those servers
/// have neither `LATERAL` (9.3+) nor `unnest(...) WITH ORDINALITY` (9.4+), so
/// the whole-table DDL/metadata path used to fail with a bare `db error`
/// (#10079). `generate_subscripts` plus a correlated scalar subquery produce
/// the same column arrays on every supported server.
fn postgres_constraints_compat_sql() -> &'static str {
"SELECT con.conname, \
con.contype::text, \
pg_catalog.pg_get_constraintdef(con.oid, true) AS definition, \
COALESCE(( \
SELECT array_agg(a.attname::text ORDER BY key_order.ord) \
FROM generate_subscripts(con.conkey, 1) AS key_order(ord) \
JOIN pg_catalog.pg_attribute a \
ON a.attrelid = con.conrelid AND a.attnum = con.conkey[key_order.ord] AND NOT a.attisdropped \
), ARRAY[]::text[]) AS columns, \
refn.nspname AS ref_schema, \
refc.relname AS ref_table, \
COALESCE(( \
SELECT array_agg(a.attname::text ORDER BY key_order.ord) \
FROM generate_subscripts(con.confkey, 1) AS key_order(ord) \
JOIN pg_catalog.pg_attribute a \
ON a.attrelid = con.confrelid AND a.attnum = con.confkey[key_order.ord] AND NOT a.attisdropped \
), ARRAY[]::text[]) AS ref_columns, \
CASE WHEN con.contype = 'f' THEN con.confmatchtype::text END AS match_type, \
CASE WHEN con.contype = 'f' THEN con.confupdtype::text END AS on_update, \
CASE WHEN con.contype = 'f' THEN con.confdeltype::text END AS on_delete, \
con.condeferrable, \
con.condeferred, \
con.convalidated \
FROM pg_catalog.pg_constraint con \
JOIN pg_catalog.pg_class c ON c.oid = con.conrelid \
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
LEFT JOIN pg_catalog.pg_class refc ON refc.oid = con.confrelid \
LEFT JOIN pg_catalog.pg_namespace refn ON refn.oid = refc.relnamespace \
WHERE n.nspname = $1 AND c.relname = $2 AND c.relkind IN ('r','p','f') \
ORDER BY con.conname"
}
/// Expand a `pg_constraint.contype` letter to a human-readable label.
fn postgres_constraint_type_label(contype: &str) -> String {
match contype {
@@ -13224,6 +13268,50 @@ mod tests {
assert!(sql.contains("ORDER BY con.conname"));
}
#[test]
fn postgres_constraint_tiers_keep_the_compat_sibling_first_usable() {
let tiers = postgres_constraint_query_tiers();
assert_eq!(tiers[0], postgres_constraints_sql());
assert_eq!(tiers[1], postgres_constraints_compat_sql());
}
#[test]
fn postgres_constraints_compat_sql_avoids_lateral_and_ordinality() {
let sql = postgres_constraints_compat_sql();
// PostgreSQL 9.2/9.3 fail on the primary query's `LEFT JOIN LATERAL`
// (9.3+) and `unnest(...) WITH ORDINALITY` (9.4+); the compat tier must
// not reference either, or table DDL export keeps reporting `db error`.
assert!(!sql.contains("LATERAL"));
assert!(!sql.contains("WITH ORDINALITY"));
assert!(!sql.contains("unnest("));
assert!(sql.contains("generate_subscripts(con.conkey, 1)"));
assert!(sql.contains("generate_subscripts(con.confkey, 1)"));
// Same projection, in the same order, as the primary query so
// `list_constraints`' positional row mapping stays valid.
for column in [
"con.conname",
"con.contype::text",
"pg_catalog.pg_get_constraintdef(con.oid, true) AS definition",
"AS columns",
"refn.nspname AS ref_schema",
"refc.relname AS ref_table",
"AS ref_columns",
"AS match_type",
"AS on_update",
"AS on_delete",
"con.condeferrable",
"con.condeferred",
"con.convalidated",
] {
assert!(sql.contains(column), "compat constraints SQL is missing {column}");
}
assert_eq!(sql.matches("attname::text").count(), 2);
assert!(sql.contains("a.attisdropped"));
assert!(sql.contains("n.nspname = $1 AND c.relname = $2"));
assert!(sql.contains("c.relkind IN ('r','p','f')"));
assert!(sql.contains("ORDER BY con.conname"));
}
#[test]
fn opengauss_constraint_sql_avoids_array_and_ordinality_dependencies() {
let sql = opengauss_constraints_sql(true);