fix(opengauss): preserve sequence metadata semantics

This commit is contained in:
lewis
2026-07-27 15:19:42 +08:00
committed by GitHub
parent 7f20b03177
commit ed47877723
2 changed files with 221 additions and 29 deletions
+115 -29
View File
@@ -3476,28 +3476,59 @@ pub async fn list_functions(pool: &Pool, schema: &str) -> Result<Vec<FunctionInf
.collect())
}
pub async fn list_sequences(pool: &Pool, schema: &str, with_last_values: bool) -> Result<Vec<SequenceInfo>, 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<Vec<SequenceInfo>, 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<SequenceInfo> = 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<String>>(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<Vec<SequenceInfo>, 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<Vec<SequenceInfo>, 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<Vec<RuleInfo>, 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();
+106
View File
@@ -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<Vec<db::SequenceInfo>, 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(