diff --git a/crates/dbx-core/src/data/database_export.rs b/crates/dbx-core/src/data/database_export.rs index fd6ababe73..697a9d797b 100644 --- a/crates/dbx-core/src/data/database_export.rs +++ b/crates/dbx-core/src/data/database_export.rs @@ -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, +} + +/// Pre-10 servers store sequence parameters in the sequence relation itself +/// (`SELECT ... FROM `), 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, 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(), diff --git a/crates/dbx-driver-postgres/src/postgres.rs b/crates/dbx-driver-postgres/src/postgres.rs index f932259d37..27c4d0717d 100644 --- a/crates/dbx-driver-postgres/src/postgres.rs +++ b/crates/dbx-driver-postgres/src/postgres.rs @@ -5204,9 +5204,11 @@ fn postgres_check_constraints_sql() -> &'static str { pub async fn list_constraints(pool: &Pool, schema: &str, table: &str) -> Result, 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, ¶ms)) + .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);