From ed47877723f21882bf674436eb48b3091b65234d Mon Sep 17 00:00:00 2001 From: lewis <32348572+wang-zhengxin@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:19:42 +0800 Subject: [PATCH] fix(opengauss): preserve sequence metadata semantics --- crates/dbx-core/src/db/postgres.rs | 144 +++++++++++++++++++++++------ crates/dbx-core/src/schema.rs | 106 +++++++++++++++++++++ 2 files changed, 221 insertions(+), 29 deletions(-) diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs index 2391add2b1..a66829eb0f 100644 --- a/crates/dbx-core/src/db/postgres.rs +++ b/crates/dbx-core/src/db/postgres.rs @@ -3476,28 +3476,59 @@ pub async fn list_functions(pool: &Pool, schema: &str) -> Result Result, String> { +fn postgres_sequences_sql() -> &'static 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'), \ + CASE WHEN s.seqcycle THEN 'YES' ELSE 'NO' END \ + FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + LEFT JOIN pg_sequence s ON s.seqrelid = c.oid \ + WHERE c.relkind = 'S' AND n.nspname = $1 \ + ORDER BY c.relname" +} + +fn opengauss_sequences_sql() -> &'static str { + "SELECT s.sequence_name, \ + COALESCE(s.data_type::text, 'bigint'), \ + COALESCE(s.start_value::text, '1'), \ + COALESCE(s.minimum_value::text, '1'), \ + COALESCE(s.maximum_value::text, '9223372036854775807'), \ + COALESCE(s.increment::text, '1'), \ + COALESCE(s.cycle_option::text, 'NO') \ + FROM information_schema.sequences s \ + JOIN pg_namespace n ON n.nspname = s.sequence_schema \ + JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = s.sequence_name \ + WHERE s.sequence_schema = $1 AND c.relkind IN ('S','L','z','Z') \ + ORDER BY s.sequence_name" +} + +fn postgres_sequence_last_values_sql() -> &'static str { + "SELECT c.relname, pg_sequence_last_value(c.oid)::text \ + FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE c.relkind = 'S' AND n.nspname = $1" +} + +fn opengauss_sequence_last_values_sql() -> &'static str { + "SELECT c.relname, (pg_sequence_last_value(c.oid)).last_value::text \ + FROM pg_class c \ + JOIN pg_namespace n ON n.oid = c.relnamespace \ + WHERE c.relkind IN ('S','L','z','Z') AND n.nspname = $1" +} + +async fn list_sequences_with_sql( + pool: &Pool, + schema: &str, + with_last_values: bool, + metadata_sql: &str, + last_values_sql: &str, +) -> Result, String> { let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?; - // Use pg_class + pg_sequence + pg_namespace instead of pg_sequences view - // for better compatibility and permission handling - let rows = postgres_query_cached( - &client, - "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'), \ - CASE WHEN s.seqcycle THEN 'YES' ELSE 'NO' END \ - FROM pg_class c \ - JOIN pg_namespace n ON n.oid = c.relnamespace \ - LEFT JOIN pg_sequence s ON s.seqrelid = c.oid \ - WHERE c.relkind = 'S' AND n.nspname = $1 \ - ORDER BY c.relname", - &[&schema], - ) - .await - .map_err(|e| e.to_string())?; + let rows = postgres_query_cached(&client, metadata_sql, &[&schema]).await.map_err(|e| e.to_string())?; let mut sequences: Vec = rows .iter() @@ -3514,17 +3545,12 @@ pub async fn list_sequences(pool: &Pool, schema: &str, with_last_values: bool) - .collect(); if with_last_values { - // Batch query: get last values for all sequences in one query - let sql = "SELECT c.relname, pg_sequence_last_value(c.oid) \ - FROM pg_class c \ - JOIN pg_namespace n ON n.oid = c.relnamespace \ - WHERE c.relkind = 'S' AND n.nspname = $1"; - if let Ok(rows) = postgres_query_cached(&client, sql, &[&schema]).await { + if let Ok(rows) = postgres_query_cached(&client, last_values_sql, &[&schema]).await { for row in rows { let name: String = pg_row_try_string(&row, 0); - if let Ok(val) = row.try_get::<_, i64>(1) { + if let Ok(Some(value)) = row.try_get::<_, Option>(1) { if let Some(seq) = sequences.iter_mut().find(|s| s.name == name) { - seq.last_value = Some(val.to_string()); + seq.last_value = Some(value); } } } @@ -3534,6 +3560,36 @@ pub async fn list_sequences(pool: &Pool, schema: &str, with_last_values: bool) - Ok(sequences) } +pub async fn list_sequences(pool: &Pool, schema: &str, with_last_values: bool) -> Result, String> { + // PostgreSQL 10+ stores sequence properties in pg_sequence. + list_sequences_with_sql( + pool, + schema, + with_last_values, + postgres_sequences_sql(), + postgres_sequence_last_values_sql(), + ) + .await +} + +pub async fn list_opengauss_sequences( + pool: &Pool, + schema: &str, + with_last_values: bool, +) -> Result, String> { + // openGauss does not expose PostgreSQL 10's pg_sequence catalog. Its + // information_schema view contains the portable sequence properties, while + // pg_sequence_last_value returns a record rather than a scalar. + list_sequences_with_sql( + pool, + schema, + with_last_values, + opengauss_sequences_sql(), + opengauss_sequence_last_values_sql(), + ) + .await +} + pub async fn list_rules(pool: &Pool, schema: &str) -> Result, String> { let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?; let rows = postgres_query_cached( @@ -4741,6 +4797,36 @@ mod tests { assert!(!POSTGRES_COLUMNS_INFORMATION_SCHEMA_SQL.contains("regclass")); } + #[test] + fn opengauss_sequence_metadata_uses_compatible_information_schema_view() { + let sql = opengauss_sequences_sql(); + + assert!(sql.contains("information_schema.sequences")); + assert!(sql.contains("s.sequence_schema = $1")); + assert!(sql.contains("c.relkind IN ('S','L','z','Z')")); + assert!(sql.contains("sequence_name")); + assert!(sql.contains("start_value")); + assert!(sql.contains("minimum_value")); + assert!(sql.contains("maximum_value")); + assert!(sql.contains("increment")); + assert!(sql.contains("cycle_option")); + assert!(!sql.contains("pg_sequence s")); + } + + #[test] + fn opengauss_sequence_last_values_extract_record_field_as_text() { + let sql = opengauss_sequence_last_values_sql(); + + assert!(sql.contains("(pg_sequence_last_value(c.oid)).last_value::text")); + assert!(sql.contains("c.relkind IN ('S','L','z','Z')")); + assert!(sql.contains("n.nspname = $1")); + } + + #[test] + fn postgres_sequence_last_values_are_read_as_text() { + assert!(postgres_sequence_last_values_sql().contains("pg_sequence_last_value(c.oid)::text")); + } + #[test] fn extension_member_query_filters_only_owned_relations_and_routines() { let sql = list_extension_member_objects_sql(); diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 133b7ea165..e37e726026 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -3011,6 +3011,20 @@ mod tests { assert!(!super::supports_agent_table_paging(&legacy_oracle)); } + #[test] + fn detects_opengauss_sequence_compatibility_profiles() { + assert!(super::is_opengauss_family_config(&test_connection_config(DatabaseType::OpenGauss))); + assert!(super::is_opengauss_family_config(&test_connection_config(DatabaseType::Gaussdb))); + assert!(!super::is_opengauss_family_config(&test_connection_config(DatabaseType::Postgres))); + + let mut profiled_postgres = test_connection_config(DatabaseType::Postgres); + profiled_postgres.driver_profile = Some("opengauss".to_string()); + assert!(super::is_opengauss_family_config(&profiled_postgres)); + + profiled_postgres.driver_profile = Some("gaussdb".to_string()); + assert!(super::is_opengauss_family_config(&profiled_postgres)); + } + #[test] fn agent_paging_detection_avoids_double_offset_only_when_page_sized() { assert!(super::agent_paging_likely_applied(true, Some(500), 500)); @@ -5472,10 +5486,14 @@ pub async fn list_sequences_core( ) -> Result, String> { retry_metadata_connection(state, connection_id, Some(database), || async { let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?; + let db_config = connection_config(state, connection_id).await; let connections = state.connections.read().await; let pool = connections.get(&pool_key).ok_or("Pool not found")?; match pool { + PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_opengauss_family_config) => { + db::postgres::list_opengauss_sequences(p, schema, with_last_values).await + } PoolKind::Postgres(p) => db::postgres::list_sequences(p, schema, with_last_values).await, _ => Ok(vec![]), } @@ -5943,6 +5961,46 @@ fn opengauss_object_source_sql( postgres_object_source_sql_inner(schema, name, kind, signature, true, true) } +fn opengauss_sequence_object_source_sql(schema: &str, name: &str, include_cache: bool) -> String { + let cache_clause = if include_cache { + "' cache ' || COALESCE((pg_sequence_last_value(c.oid)).cache_value::text, '1') || E'\\n' || " + } else { + "" + }; + format!( + "SELECT concat_ws(E'\\n\\n', \ + '-- auto-generated definition' || E'\\n' || \ + 'create ' || CASE WHEN c.relkind IN ('L','Z') THEN 'large ' ELSE '' END || \ + 'sequence ' || quote_ident(c.relname) || E'\\n' || \ + ' increment by ' || COALESCE(s.increment::text, '1') || E'\\n' || \ + ' minvalue ' || COALESCE(s.minimum_value::text, '1') || E'\\n' || \ + ' maxvalue ' || COALESCE(s.maximum_value::text, '9223372036854775807') || E'\\n' || \ + ' start with ' || COALESCE(s.start_value::text, '1') || E'\\n' || \ + {cache_clause} \ + CASE WHEN upper(COALESCE(s.cycle_option::text, 'NO')) = 'YES' \ + THEN ' cycle;' ELSE ' no cycle;' END, \ + 'alter ' || CASE WHEN c.relkind IN ('L','Z') THEN 'large ' ELSE '' END || \ + 'sequence ' || quote_ident(c.relname) || ' owner to ' || quote_ident(pg_get_userbyid(c.relowner)) || ';', \ + CASE WHEN owned.relname IS NOT NULL AND a.attname IS NOT NULL \ + THEN 'alter ' || CASE WHEN c.relkind IN ('L','Z') THEN 'large ' ELSE '' END || \ + 'sequence ' || quote_ident(c.relname) || ' owned by ' || quote_ident(owned.relname) || '.' || quote_ident(a.attname) || ';' \ + END \ + ) \ + FROM pg_catalog.pg_class c \ + JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \ + JOIN information_schema.sequences s \ + ON s.sequence_schema = n.nspname AND s.sequence_name = c.relname \ + LEFT JOIN pg_catalog.pg_depend d \ + ON d.classid = 'pg_class'::regclass AND d.objid = c.oid AND d.deptype = 'a' \ + LEFT JOIN pg_catalog.pg_class owned ON owned.oid = d.refobjid \ + LEFT JOIN pg_catalog.pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid \ + WHERE n.nspname = {schema} AND c.relname = {name} AND c.relkind IN ('S','L','z','Z') \ + ORDER BY c.oid LIMIT 1", + schema = sql_string(schema), + name = sql_string(name) + ) +} + fn postgres_object_source_sql_without_relispopulated( schema: &str, name: &str, @@ -6056,6 +6114,9 @@ fn postgres_object_source_sql_inner( ) } db::ObjectSourceKind::Sequence => { + if unwrap_opengauss_record { + return opengauss_sequence_object_source_sql(schema, name, true); + } format!( "SELECT concat_ws(E'\\n\\n', \ '-- auto-generated definition' || E'\\n' || \ @@ -6692,6 +6753,17 @@ async fn postgres_object_source( .and_then(first_string_cell) .map_err(|fallback_err| format!("{primary_err}; relispopulated fallback failed: {fallback_err}")) } + Err(primary_err) + if unwrap_opengauss_record + && matches!(object_type, db::ObjectSourceKind::Sequence) + && opengauss_sequence_cache_metadata_error(&primary_err) => + { + let fallback_sql = opengauss_sequence_object_source_sql(schema, name, false); + db::postgres::execute_query(pool, &fallback_sql) + .await + .and_then(first_string_cell) + .map_err(|fallback_err| format!("{primary_err}; sequence cache fallback failed: {fallback_err}")) + } Err(primary_err) if unwrap_opengauss_record && matches!(object_type, db::ObjectSourceKind::Procedure | db::ObjectSourceKind::Function) => @@ -6736,6 +6808,11 @@ fn postgres_missing_prokind_error(err: &str) -> bool { || lower.contains("column \"prokind\"")) } +fn opengauss_sequence_cache_metadata_error(err: &str) -> bool { + let lower = err.to_ascii_lowercase(); + lower.contains("pg_sequence_last_value") || lower.contains("cache_value") +} + fn postgres_missing_relispopulated_error(err: &str) -> bool { let lower = err.to_ascii_lowercase(); lower.contains("does not exist") @@ -6835,6 +6912,35 @@ mod object_source_tests { ); } + #[test] + fn builds_opengauss_sequence_source_without_pg_sequence_catalog() { + let sql = opengauss_object_source_sql("public", "order_id_seq", &ObjectSourceKind::Sequence, None); + + assert!(sql.contains("information_schema.sequences")); + assert!(sql.contains("s.sequence_schema = n.nspname")); + assert!(sql.contains("s.sequence_name = c.relname")); + assert!(sql.contains("c.relkind IN ('S','L','z','Z')")); + assert!(sql.contains("CASE WHEN c.relkind IN ('L','Z') THEN 'large '")); + assert!(sql.contains("increment by")); + assert!(sql.contains("start with")); + assert!(sql.contains("(pg_sequence_last_value(c.oid)).cache_value::text")); + assert!(sql.contains("cycle;")); + assert!(!sql.contains("pg_catalog.pg_sequence")); + + let fallback_sql = opengauss_sequence_object_source_sql("public", "order_id_seq", false); + assert!(fallback_sql.contains("c.relkind IN ('S','L','z','Z')")); + assert!(fallback_sql.contains("CASE WHEN c.relkind IN ('L','Z') THEN 'large '")); + assert!(!fallback_sql.contains("pg_sequence_last_value")); + assert!(!fallback_sql.contains("cache_value")); + } + + #[test] + fn detects_opengauss_sequence_cache_metadata_fallback_errors() { + assert!(opengauss_sequence_cache_metadata_error("cannot execute pg_sequence_last_value() on a standby node")); + assert!(opengauss_sequence_cache_metadata_error("column notation .cache_value applied to type text")); + assert!(!opengauss_sequence_cache_metadata_error("permission denied for sequence order_id_seq")); + } + #[test] fn composes_opengauss_text_return_and_missing_prokind_fallbacks() { let text_return = opengauss_routine_source_fallback_sqls(