feat(opengauss): support A-compatibility packages

crates/dbx-core/src/connection.rs

@@ -9830,6 +9830,308 @@ for line in sys.stdin:
         .await;
     }
 
+    #[tokio::test]
+    #[ignore = "requires a reachable openGauss A-compatibility instance via environment variables"]
+    async fn live_opengauss_package_feature() {
+        let host = std::env::var("DBX_TEST_OPENGAUSS_HOST").expect("DBX_TEST_OPENGAUSS_HOST not set");
+        let port = std::env::var("DBX_TEST_OPENGAUSS_PORT")
+            .expect("DBX_TEST_OPENGAUSS_PORT not set")
+            .parse::<u16>()
+            .expect("DBX_TEST_OPENGAUSS_PORT should be a u16");
+        let username = std::env::var("DBX_TEST_OPENGAUSS_USER").expect("DBX_TEST_OPENGAUSS_USER not set");
+        let password = std::env::var("DBX_TEST_OPENGAUSS_PASSWORD").expect("DBX_TEST_OPENGAUSS_PASSWORD not set");
+        let database = std::env::var("DBX_TEST_OPENGAUSS_DATABASE").unwrap_or_else(|_| "postgres".to_string());
+
+        let mut config = live_postgres_like_config(DatabaseType::OpenGauss, &host, port, &username, &password, None);
+        config.id = "opengauss-live".to_string();
+        config.database = Some(database.clone());
+
+        let (state, dir) = test_app_state().await;
+        state.configs.write().await.insert(config.id.clone(), config);
+        let pool_key = state.get_or_create_pool("opengauss-live", Some(&database)).await.unwrap();
+        let pool = match state.pool_handle(&pool_key).await.expect("openGauss pool should be created") {
+            PoolKind::Postgres(pool) => pool,
+            _ => panic!("openGauss should use the PostgreSQL pool path"),
+        };
+
+        let mode = db::postgres::opengauss_compatibility_mode(&pool).await.unwrap();
+        println!("[live] current database = {database}, compatibility mode = {mode:?}");
+        let is_a_mode = mode.as_deref().map(str::trim).map(|value| value.eq_ignore_ascii_case("A")) == Some(true);
+
+        let databases = schema::list_databases_core(&state, "opengauss-live").await.unwrap();
+        for item in &databases {
+            println!("[live] database {} -> {:?}", item.name, item.compatibility_mode);
+        }
+        assert!(databases.iter().any(|item| item.name == database), "expected {database} in the database list");
+        if is_a_mode {
+            assert_eq!(
+                databases.iter().find(|item| item.name == database).and_then(|item| item.compatibility_mode.as_deref()),
+                mode.as_deref(),
+                "listed compatibility mode should match the probed mode"
+            );
+        }
+
+        // Statement splitting must keep an A-mode package intact.
+        let script = "CREATE OR REPLACE PACKAGE p AS\n  FUNCTION f RETURN INT;\nEND p;\n/\nSELECT 1;";
+        let a_split = crate::sql::split_sql_statements_for_database_with_compatibility(
+            script,
+            DatabaseType::OpenGauss,
+            Some("A"),
+        );
+        let pg_split = crate::sql::split_sql_statements_for_database_with_compatibility(
+            script,
+            DatabaseType::OpenGauss,
+            Some("PG"),
+        );
+        println!("[live] A-mode split = {a_split:?}");
+        println!("[live] PG-mode split = {pg_split:?}");
+        assert_eq!(a_split.len(), 2, "A-mode package must stay one statement");
+        assert!(pg_split.len() > 2, "PG-mode must split the same script at inner semicolons");
+
+        // A non-A database on the same instance must keep the package gate closed.
+        if let Some(other) = databases.iter().find(|item| {
+            item.name != database
+                && item.compatibility_mode.as_deref().map(str::trim).is_some_and(|mode| !mode.eq_ignore_ascii_case("A"))
+        }) {
+            let other_pool_key = state.get_or_create_pool("opengauss-live", Some(&other.name)).await.unwrap();
+            if let Some(PoolKind::Postgres(other_pool)) = state.pool_handle(&other_pool_key).await {
+                let other_mode = db::postgres::opengauss_compatibility_mode(&other_pool).await.unwrap();
+                let other_packages =
+                    db::postgres::list_opengauss_packages(&other_pool, "public", true, true).await.unwrap();
+                println!(
+                    "[live] non-A database {} mode = {:?}, listed packages = {}",
+                    other.name,
+                    other_mode,
+                    other_packages.len()
+                );
+                assert!(!db::postgres::opengauss_is_oracle_compatible(&other_pool).await.unwrap());
+                assert!(other_packages.is_empty(), "non-A database must not list packages");
+                other_pool.close();
+            }
+        }
+
+        if !is_a_mode {
+            println!("[live] database is not A-compatible; skipping package catalog assertions");
+            pool.close();
+            let _ = std::fs::remove_dir_all(dir);
+            return;
+        }
+
+        let schema_name = "public";
+        let pkg = "dbx_live_pkg";
+        let _ = db::postgres::execute_query(&pool, &format!("DROP PACKAGE BODY IF EXISTS {schema_name}.{pkg}")).await;
+        let _ = db::postgres::execute_query(&pool, &format!("DROP PACKAGE IF EXISTS {schema_name}.{pkg}")).await;
+
+        let spec = format!(
+            "CREATE OR REPLACE PACKAGE {schema_name}.{pkg} AS\n  g_version VARCHAR2(20) := '1.0';\n  PROCEDURE log_message(p_message IN VARCHAR2);\n  FUNCTION add_numbers(p_left IN INTEGER, p_right IN INTEGER) RETURN INTEGER;\nEND {pkg};"
+        );
+        db::postgres::execute_query(&pool, &spec)
+            .await
+            .unwrap_or_else(|error| panic!("create package spec failed: {error}"));
+        let body = format!(
+            "CREATE OR REPLACE PACKAGE BODY {schema_name}.{pkg} AS\n  PROCEDURE log_message(p_message IN VARCHAR2) IS\n  BEGIN\n    NULL;\n  END;\n  FUNCTION add_numbers(p_left IN INTEGER, p_right IN INTEGER) RETURN INTEGER IS\n  BEGIN\n    RETURN p_left + p_right;\n  END;\nBEGIN\n  g_version := '1.1';\nEND {pkg};"
+        );
+        db::postgres::execute_query(&pool, &body)
+            .await
+            .unwrap_or_else(|error| panic!("create package body failed: {error}"));
+
+        let package_types = vec!["PACKAGE".to_string(), "PACKAGE_BODY".to_string()];
+        let package_objects = schema::list_objects_core(
+            &state,
+            "opengauss-live",
+            &database,
+            schema_name,
+            None,
+            None,
+            None,
+            Some(package_types.as_slice()),
+            None,
+        )
+        .await
+        .unwrap();
+        println!(
+            "[live] package objects = {:?}",
+            package_objects
+                .iter()
+                .filter(|object| object.name == pkg)
+                .map(|object| object.object_type.as_str())
+                .collect::<Vec<_>>()
+        );
+        assert!(package_objects.iter().any(|object| object.name == pkg && object.object_type == "PACKAGE"));
+        assert!(package_objects.iter().any(|object| object.name == pkg && object.object_type == "PACKAGE_BODY"));
+
+        let routine_types = vec!["PROCEDURE".to_string(), "FUNCTION".to_string()];
+        let routine_objects = schema::list_objects_core(
+            &state,
+            "opengauss-live",
+            &database,
+            schema_name,
+            None,
+            None,
+            None,
+            Some(routine_types.as_slice()),
+            None,
+        )
+        .await
+        .unwrap();
+        println!(
+            "[live] routine objects (top-level) = {:?}",
+            routine_objects.iter().map(|object| object.name.as_str()).collect::<Vec<_>>()
+        );
+        assert!(
+            !routine_objects.iter().any(|object| object.name == "log_message" || object.name == "add_numbers"),
+            "package members must not surface as top-level routines"
+        );
+
+        let package_search = schema::completion_assistant_search_core(
+            &state,
+            db::CompletionAssistantRequest {
+                connection_id: "opengauss-live".to_string(),
+                database: database.clone(),
+                schema: Some(schema_name.to_string()),
+                object_kinds: vec![db::CompletionAssistantObjectKind::Routine],
+                mask: pkg.to_string(),
+                case_sensitive: false,
+                global_search: false,
+                max_results: Some(50),
+                search_in_comments: false,
+                search_in_definitions: false,
+                parent_schema: None,
+                parent_name: None,
+                match_mode: Some(db::CompletionAssistantMatchMode::Prefix),
+            },
+        )
+        .await
+        .unwrap();
+        println!(
+            "[live] unqualified package candidates = {:?}",
+            package_search
+                .candidates
+                .iter()
+                .map(|candidate| (&candidate.name, &candidate.kind, &candidate.data_type))
+                .collect::<Vec<_>>()
+        );
+        assert!(
+            package_search.candidates.iter().any(|candidate| {
+                candidate.name == pkg
+                    && candidate.kind == db::CompletionAssistantCandidateKind::Object
+                    && candidate.data_type.as_deref() == Some("PACKAGE")
+            }),
+            "package name should be offered as an unqualified completion candidate"
+        );
+
+        let member_search = schema::completion_assistant_search_core(
+            &state,
+            db::CompletionAssistantRequest {
+                connection_id: "opengauss-live".to_string(),
+                database: database.clone(),
+                schema: Some(schema_name.to_string()),
+                object_kinds: vec![db::CompletionAssistantObjectKind::Routine],
+                mask: String::new(),
+                case_sensitive: false,
+                global_search: false,
+                max_results: Some(50),
+                search_in_comments: false,
+                search_in_definitions: false,
+                parent_schema: Some(schema_name.to_string()),
+                parent_name: Some(pkg.to_string()),
+                match_mode: Some(db::CompletionAssistantMatchMode::Prefix),
+            },
+        )
+        .await
+        .unwrap();
+        println!(
+            "[live] package member candidates = {:?}",
+            member_search
+                .candidates
+                .iter()
+                .map(|candidate| (&candidate.name, &candidate.kind, &candidate.signature))
+                .collect::<Vec<_>>()
+        );
+        assert!(!member_search.fallback_used, "package members should not fall back to top-level routines");
+        assert!(member_search.candidates.iter().any(|candidate| {
+            candidate.name == "log_message" && candidate.kind == db::CompletionAssistantCandidateKind::Procedure
+        }));
+        assert!(member_search.candidates.iter().any(|candidate| {
+            candidate.name == "add_numbers" && candidate.kind == db::CompletionAssistantCandidateKind::Function
+        }));
+
+        let spec_source = schema::get_object_source_core(
+            &state,
+            "opengauss-live",
+            &database,
+            schema_name,
+            pkg,
+            db::ObjectSourceKind::Package,
+            None,
+            None,
+        )
+        .await
+        .unwrap();
+        let body_source = schema::get_object_source_core(
+            &state,
+            "opengauss-live",
+            &database,
+            schema_name,
+            pkg,
+            db::ObjectSourceKind::PackageBody,
+            None,
+            None,
+        )
+        .await
+        .unwrap();
+        println!("[live] spec source length = {}", spec_source.source.len());
+        println!("[live] body source length = {}", body_source.source.len());
+        println!("[live] spec source =\n{}", spec_source.source);
+        println!("[live] body source =\n{}", body_source.source);
+        assert!(spec_source.source.to_uppercase().contains("PACKAGE"));
+        assert!(body_source.source.to_uppercase().contains("PACKAGE BODY"));
+        assert_eq!(spec_source.editable, Some(false));
+        assert_eq!(body_source.editable, Some(false));
+
+        // Catalog and gs_source lookups are case-insensitive for unquoted names.
+        let upper = schema::get_object_source_core(
+            &state,
+            "opengauss-live",
+            &database,
+            schema_name,
+            &pkg.to_uppercase(),
+            db::ObjectSourceKind::Package,
+            None,
+            None,
+        )
+        .await;
+        println!("[live] uppercase lookup ok = {}", upper.is_ok());
+        assert!(upper.is_ok(), "case-insensitive package source lookup failed: {upper:?}");
+
+        // Force the catalog-rebuild fallback by removing the gs_source rows, then
+        // prove the rebuilt spec/body are valid DDL by re-executing them.
+        let _ =
+            db::postgres::execute_query(&pool, &format!("DELETE FROM dbe_pldeveloper.gs_source WHERE name = '{pkg}'"))
+                .await;
+        let fallback_spec = db::postgres::opengauss_package_source(&pool, schema_name, pkg, false).await.unwrap();
+        let fallback_body = db::postgres::opengauss_package_source(&pool, schema_name, pkg, true).await.unwrap();
+        println!("[live] fallback spec =\n{fallback_spec}");
+        println!("[live] fallback body =\n{fallback_body}");
+        assert!(fallback_spec.contains("AUTHID"), "rebuilt spec must preserve the authid clause");
+        assert!(fallback_body.contains("g_version := '1.1'"), "rebuilt body must keep the initialization section");
+        db::postgres::execute_query(&pool, &fallback_spec)
+            .await
+            .unwrap_or_else(|error| panic!("rebuilt spec is not valid DDL: {error}\n{fallback_spec}"));
+        db::postgres::execute_query(&pool, &fallback_body)
+            .await
+            .unwrap_or_else(|error| panic!("rebuilt body is not valid DDL: {error}\n{fallback_body}"));
+        println!("[live] rebuilt fallback DDL re-executed successfully");
+
+        let _ = db::postgres::execute_query(&pool, &format!("DROP PACKAGE BODY IF EXISTS {schema_name}.{pkg}")).await;
+        let _ = db::postgres::execute_query(&pool, &format!("DROP PACKAGE IF EXISTS {schema_name}.{pkg}")).await;
+        let remaining = db::postgres::list_opengauss_packages(&pool, schema_name, true, true).await.unwrap();
+        assert!(!remaining.iter().any(|object| object.name == pkg), "cleanup left the test package behind");
+        println!("[live] cleanup verified: test package removed");
+        pool.close();
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[tokio::test]
     #[ignore = "requires a reachable KWDB instance via environment variables"]
     async fn live_kwdb_native_connection_succeeds() {

crates/dbx-core/src/db/influxdb3_driver.rs

@@ -244,6 +244,7 @@ pub async fn list_databases(client: &Influxdb3Client) -> Result<Vec<DatabaseInfo
                     comment: None,
                     default_charset: None,
                     default_collation: None,
+                    compatibility_mode: None,
                 })
         })
         .collect())

crates/dbx-core/src/db/postgres.rs

@@ -59,6 +59,16 @@ pub(crate) fn gaussdb_identifier_quote_for_compatibility_mode(compatibility_mode
     }
 }
 
+pub async fn opengauss_compatibility_mode(pool: &Pool) -> Result<Option<String>, String> {
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let row = postgres_query_one_cached(&client, GAUSSDB_COMPATIBILITY_SQL, &[]).await.map_err(|e| e.to_string())?;
+    Ok(row.try_get::<_, Option<String>>(0).ok().flatten().filter(|value| !value.trim().is_empty()))
+}
+
+pub async fn opengauss_is_oracle_compatible(pool: &Pool) -> Result<bool, String> {
+    Ok(opengauss_compatibility_mode(pool).await?.is_some_and(|mode| mode.trim().eq_ignore_ascii_case("A")))
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PostgresTablePrivilegeInfo {
     pub grantor: String,
@@ -3290,6 +3300,13 @@ fn list_databases_sql() -> &'static str {
      ORDER BY datname"
 }
 
+fn list_opengauss_databases_sql() -> &'static str {
+    "SELECT datname, datcompatibility::text \
+     FROM pg_catalog.pg_database \
+     WHERE datallowconn = true \
+     ORDER BY datname"
+}
+
 fn database_metadata_sql() -> &'static str {
     "SELECT d.datname, \
             CASE \
@@ -3334,6 +3351,40 @@ pub async fn list_databases(pool: &Pool) -> Result<Vec<DatabaseInfo>, String> {
     Ok(rows.iter().map(|row| DatabaseInfo { name: pg_row_try_string(row, 0), ..Default::default() }).collect())
 }
 
+pub async fn list_opengauss_databases(pool: &Pool) -> Result<Vec<DatabaseInfo>, String> {
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let rows = match postgres_query_cached(&client, list_opengauss_databases_sql(), &[]).await {
+        Ok(rows) => rows,
+        Err(primary_error) => {
+            let primary_error = primary_error.to_string();
+            return postgres_query_cached(&client, list_databases_sql(), &[])
+                .await
+                .map(|rows| {
+                    rows.iter()
+                        .map(|row| DatabaseInfo { name: pg_row_try_string(row, 0), ..Default::default() })
+                        .collect()
+                })
+                .map_err(|fallback_error| {
+                    format!(
+                        "openGauss database compatibility lookup failed: {primary_error}; database list fallback failed: {fallback_error}"
+                    )
+                });
+        }
+    };
+    Ok(rows
+        .iter()
+        .map(|row| DatabaseInfo {
+            name: pg_row_try_string(row, 0),
+            compatibility_mode: row
+                .try_get::<_, Option<String>>(1)
+                .ok()
+                .flatten()
+                .filter(|value| !value.trim().is_empty()),
+            ..Default::default()
+        })
+        .collect())
+}
+
 pub async fn list_database_metadata(pool: &Pool) -> Result<Vec<DatabaseInfo>, String> {
     let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
     let rows = postgres_query_cached(&client, database_metadata_sql(), &[]).await.map_err(|e| e.to_string())?;
@@ -3437,6 +3488,25 @@ async fn list_tables_filtered_by_kind(
 pub async fn completion_assistant_search(
     pool: &Pool,
     request: &CompletionAssistantRequest,
+) -> Result<CompletionAssistantResponse, String> {
+    completion_assistant_search_inner(pool, request, false).await
+}
+
+/// openGauss variant of [`completion_assistant_search`]: top-level routines
+/// exclude package members (`propackage = true`) and private members
+/// (`proisprivate = true`), which are surfaced under their PACKAGE nodes by
+/// [`opengauss_package_members`] instead.
+pub async fn opengauss_completion_assistant_search(
+    pool: &Pool,
+    request: &CompletionAssistantRequest,
+) -> Result<CompletionAssistantResponse, String> {
+    completion_assistant_search_inner(pool, request, true).await
+}
+
+async fn completion_assistant_search_inner(
+    pool: &Pool,
+    request: &CompletionAssistantRequest,
+    exclude_package_members: bool,
 ) -> Result<CompletionAssistantResponse, String> {
     let schema = request.schema.as_deref().or(request.parent_schema.as_deref());
     let routine_schema = schema.unwrap_or("public");
@@ -3506,13 +3576,49 @@ pub async fn completion_assistant_search(
         }
     }
 
+    if candidates.len() < limit
+        && exclude_package_members
+        && request.parent_name.as_deref().is_none_or(|name| name.trim().is_empty())
+        && kinds.iter().any(CompletionAssistantObjectKind::is_routine_like)
+    {
+        let rows = match postgres_query_cached(
+            &client,
+            opengauss_completion_packages_sql(),
+            &[&routine_schema, &pattern, &((limit - candidates.len()) as i64)],
+        )
+        .await
+        {
+            Ok(rows) => rows,
+            Err(error) if opengauss_optional_package_catalog_error(&error.to_string()) => Vec::new(),
+            Err(error) => return Err(error.to_string()),
+        };
+        for row in rows {
+            candidates.push(CompletionAssistantCandidate {
+                name: pg_row_try_string(&row, 0),
+                kind: CompletionAssistantCandidateKind::Object,
+                database: Some(request.database.clone()),
+                schema: Some(pg_row_try_string(&row, 1)),
+                parent_schema: None,
+                parent_name: None,
+                comment: None,
+                data_type: Some("PACKAGE".to_string()),
+                signature: None,
+            });
+        }
+    }
+
     if candidates.len() < limit && kinds.iter().any(CompletionAssistantObjectKind::is_routine_like) {
         let has_proc_prokind = postgres_proc_has_prokind(&client).await?;
         let rows = if has_proc_prokind {
             let prokinds = postgres_completion_prokinds(&kinds);
+            let sql = if exclude_package_members {
+                opengauss_completion_routines_sql(true)
+            } else {
+                postgres_completion_routines_sql(true).to_string()
+            };
             postgres_query_cached(
                 &client,
-                postgres_completion_routines_sql(true),
+                &sql,
                 &[&routine_schema, &pattern, &prokinds, &((limit - candidates.len()) as i64)],
             )
             .await
@@ -3520,13 +3626,14 @@ pub async fn completion_assistant_search(
         } else if kinds.iter().any(|kind| {
             matches!(kind, CompletionAssistantObjectKind::Function | CompletionAssistantObjectKind::Routine)
         }) {
-            postgres_query_cached(
-                &client,
-                postgres_completion_routines_sql(false),
-                &[&routine_schema, &pattern, &((limit - candidates.len()) as i64)],
-            )
-            .await
-            .map_err(|e| e.to_string())?
+            let sql = if exclude_package_members {
+                opengauss_completion_routines_sql(false)
+            } else {
+                postgres_completion_routines_sql(false).to_string()
+            };
+            postgres_query_cached(&client, &sql, &[&routine_schema, &pattern, &((limit - candidates.len()) as i64)])
+                .await
+                .map_err(|e| e.to_string())?
         } else {
             Vec::new()
         };
@@ -3655,6 +3762,30 @@ fn postgres_completion_routines_sql(has_proc_prokind: bool) -> &'static str {
      ORDER BY p.proname LIMIT $3"
 }
 
+/// openGauss variant of [`postgres_completion_routines_sql`]: package members
+/// (`propackage = true`) and private members (`proisprivate = true`) are stored
+/// in the same `pg_proc` rows, but must not surface as top-level routines.
+/// The package/private columns only exist on openGauss kernels; plain
+/// PostgreSQL and the Redshift completion path never use this variant, so the
+/// extra predicates can be injected unconditionally here.
+fn opengauss_completion_routines_sql(has_proc_prokind: bool) -> String {
+    postgres_completion_routines_sql(has_proc_prokind).replace(
+        "WHERE n.nspname = $1",
+        "WHERE n.nspname = $1 AND COALESCE(p.propackage, false) = false \
+       AND COALESCE(p.proisprivate, false) = false",
+    )
+}
+
+fn opengauss_completion_packages_sql() -> &'static str {
+    "SELECT p.pkgname::text, n.nspname::text \
+     FROM pg_catalog.gs_package p \
+     JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+     WHERE n.nspname = $1 \
+       AND EXISTS (SELECT 1 FROM pg_catalog.pg_database d WHERE d.datname = current_database() AND d.datcompatibility::text = 'A') \
+       AND ($2 = '%%' OR p.pkgname ILIKE $2 ESCAPE '~') \
+     ORDER BY p.pkgname LIMIT $3"
+}
+
 fn postgres_completion_sequences_sql() -> &'static str {
     "SELECT c.relname, n.nspname, obj_description(c.oid) AS sequence_comment \
      FROM pg_catalog.pg_class c \
@@ -5344,13 +5475,24 @@ fn list_objects_sql(
     include_relations: bool,
     include_routines: bool,
     include_custom_types: bool,
+    exclude_package_members: bool,
 ) -> String {
     let mut parts: Vec<String> = Vec::new();
     if include_relations {
         parts.push(list_object_relations_sql(include_timestamps).to_string());
     }
     if include_routines {
-        parts.push(list_object_routines_sql(include_timestamps, has_proc_prokind, has_proc_prosp).to_string());
+        let mut routine = list_object_routines_sql(include_timestamps, has_proc_prokind, has_proc_prosp).to_string();
+        // openGauss stores package members in pg_proc (propackage/propackageid).
+        // They are listed under their PACKAGE nodes by list_opengauss_packages, so
+        // exclude them from the top-level routine list to avoid duplicate entries.
+        // The column only exists on openGauss kernels; plain PostgreSQL never
+        // passes exclude_package_members=true.
+        if exclude_package_members {
+            routine = routine
+                .replace("WHERE n.nspname = $1", "WHERE n.nspname = $1 AND COALESCE(p.propackage, false) = false");
+        }
+        parts.push(routine);
     }
     if include_custom_types {
         parts.push(list_object_custom_types_sql().to_string());
@@ -5448,6 +5590,7 @@ async fn list_objects_rows(
     include_relations: bool,
     include_routines: bool,
     include_custom_types: bool,
+    exclude_package_members: bool,
 ) -> Result<Vec<Row>, String> {
     let sql = list_objects_sql(
         include_timestamps,
@@ -5457,6 +5600,7 @@ async fn list_objects_rows(
         include_relations,
         include_routines,
         include_custom_types,
+        exclude_package_members,
     );
     if sql.is_empty() {
         // No branch was selected (e.g. an object_types filter the catalog
@@ -5501,6 +5645,7 @@ pub async fn list_objects(
         include_relations,
         include_routines,
         include_custom_types,
+        false,
     )
     .await
     {
@@ -5517,6 +5662,7 @@ pub async fn list_objects(
                 include_relations,
                 include_routines,
                 include_custom_types,
+                false,
             )
             .await
             {
@@ -5531,6 +5677,424 @@ pub async fn list_objects(
     Ok(object_rows_to_infos(&rows, schema))
 }
 
+/// openGauss variant of [`list_objects`]: top-level routines exclude package
+/// members (`propackage = true`), which are surfaced under their PACKAGE nodes
+/// by [`list_opengauss_packages`] instead.
+pub async fn list_opengauss_objects(
+    pool: &Pool,
+    schema: &str,
+    include_relations: bool,
+    include_routines: bool,
+    include_custom_types: bool,
+) -> Result<Vec<ObjectInfo>, String> {
+    let include_custom_types = include_custom_types && !is_postgres_system_schema(schema);
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let (has_proc_prokind, has_proc_prosp, has_function_identity_arguments) = if include_routines {
+        let has_proc_prokind = postgres_proc_has_prokind(&client).await?;
+        let has_proc_prosp = postgres_proc_has_prosp(&client).await?;
+        let has_function_identity_arguments = postgres_has_function_identity_arguments(&client).await?;
+        (has_proc_prokind, has_proc_prosp, has_function_identity_arguments)
+    } else {
+        (false, false, false)
+    };
+    let rows = match list_objects_rows(
+        &client,
+        schema,
+        true,
+        has_proc_prokind,
+        has_proc_prosp,
+        has_function_identity_arguments,
+        include_relations,
+        include_routines,
+        include_custom_types,
+        true,
+    )
+    .await
+    {
+        Ok(rows) => rows,
+        Err(primary_error) => {
+            log::debug!("[postgres][list_opengauss_objects:timestamp-fallback] primary_error={}", primary_error);
+            match list_objects_rows(
+                &client,
+                schema,
+                false,
+                has_proc_prokind,
+                has_proc_prosp,
+                has_function_identity_arguments,
+                include_relations,
+                include_routines,
+                include_custom_types,
+                true,
+            )
+            .await
+            {
+                Ok(rows) => rows,
+                Err(fallback_error) => {
+                    return Err(format!("{primary_error}; timestamp fallback failed: {fallback_error}"));
+                }
+            }
+        }
+    };
+
+    Ok(object_rows_to_infos(&rows, schema))
+}
+
+pub async fn list_opengauss_packages(
+    pool: &Pool,
+    schema: &str,
+    include_spec: bool,
+    include_body: bool,
+) -> Result<Vec<ObjectInfo>, String> {
+    if !include_spec && !include_body {
+        return Ok(Vec::new());
+    }
+    match opengauss_is_oracle_compatible(pool).await {
+        Ok(true) => {}
+        Ok(false) | Err(_) => return Ok(Vec::new()),
+    }
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let rows = match postgres_query_cached(
+        &client,
+        "SELECT p.pkgname::text, \
+                (p.pkgbodydeclsrc IS NOT NULL OR p.pkgbodyinitsrc IS NOT NULL) AS has_body \
+         FROM pg_catalog.gs_package p \
+         JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+         WHERE n.nspname = $1 \
+         ORDER BY p.pkgname",
+        &[&schema],
+    )
+    .await
+    {
+        Ok(rows) => rows,
+        Err(error) if opengauss_optional_package_catalog_error(&error.to_string()) => return Ok(Vec::new()),
+        Err(error) => return Err(error.to_string()),
+    };
+    let mut objects = Vec::with_capacity(rows.len() * 2);
+    for row in rows {
+        let name = pg_row_try_string(&row, 0);
+        let has_body = pg_row_try_bool(&row, 1).unwrap_or(false);
+        let object = |object_type: &str| ObjectInfo {
+            name: name.clone(),
+            object_type: object_type.to_string(),
+            schema: Some(schema.to_string()),
+            valid: None,
+            signature: None,
+            custom_type_kind: None,
+            has_members: None,
+            comment: None,
+            created_at: None,
+            updated_at: None,
+            parent_schema: None,
+            parent_name: None,
+            trigger: None,
+            xugu_type_members_expandable: None,
+        };
+        if include_spec {
+            objects.push(object("PACKAGE"));
+        }
+        if include_body && has_body {
+            objects.push(object("PACKAGE_BODY"));
+        }
+    }
+    Ok(objects)
+}
+
+fn opengauss_optional_package_catalog_error(error: &str) -> bool {
+    let lower = error.to_ascii_lowercase();
+    lower.contains("permission denied")
+        || lower.contains("insufficient privilege")
+        || lower.contains("does not exist")
+        || lower.contains("undefined table")
+        || lower.contains("undefined column")
+        || lower.contains("undefined schema")
+}
+
+/// Whether an openGauss gs_source lookup error means the catalog is not usable
+/// (missing relation/schema, or insufficient privileges) so the gs_package
+/// fallback should be attempted instead of surfacing the error. Connection
+/// failures and protocol errors are not swallowed.
+fn opengauss_gs_source_lookup_should_fallback(error: &str) -> bool {
+    opengauss_optional_package_catalog_error(error)
+}
+
+pub async fn opengauss_package_source(
+    pool: &Pool,
+    schema: &str,
+    name: &str,
+    package_body: bool,
+) -> Result<String, String> {
+    match opengauss_is_oracle_compatible(pool).await {
+        Ok(true) => {}
+        Ok(false) => {
+            return Err("openGauss packages are only available in A compatibility mode".to_string());
+        }
+        Err(_) => {}
+    }
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let source_type = if package_body { "package body" } else { "package" };
+    let rows = match postgres_query_cached(
+        &client,
+        "SELECT s.src::text \
+         FROM dbe_pldeveloper.gs_source s \
+         JOIN pg_catalog.gs_package p ON p.oid = s.id \
+         JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+         WHERE lower(n.nspname) = lower($1) \
+           AND lower(p.pkgname::text) = lower($2) \
+           AND lower(s.type) = $3 \
+           AND s.status = true \
+         ORDER BY (p.pkgname::text = $2) DESC, p.oid \
+         LIMIT 1",
+        &[&schema, &name, &source_type],
+    )
+    .await
+    {
+        Ok(rows) => rows,
+        Err(error) => {
+            let error = error.to_string();
+            if opengauss_gs_source_lookup_should_fallback(&error) {
+                Vec::new()
+            } else {
+                return Err(error);
+            }
+        }
+    };
+    if let Some(source) = rows.first().map(|row| pg_row_try_string(row, 0)).filter(|source| !source.trim().is_empty()) {
+        return Ok(source);
+    }
+
+    // dbe_pldeveloper.gs_source is written asynchronously and can be absent when
+    // SKIP_GS_SOURCE is set, autonomous-transaction limits are hit, a compile
+    // failed, or the current user lacks privileges on the schema. Rebuild the
+    // source from the gs_package fragment columns so an existing package can
+    // still be viewed and exported. When the stored shape is unrecognised we
+    // fail loudly instead of emitting a broken CREATE statement.
+    let meta_rows = postgres_query_cached(
+        &client,
+        "SELECT p.pkgname::text, n.nspname::text, p.pkgsecdef, \
+                p.pkgspecsrc::text, p.pkgbodydeclsrc::text, p.pkgbodyinitsrc::text \
+         FROM pg_catalog.gs_package p \
+         JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+         WHERE lower(n.nspname) = lower($1) AND lower(p.pkgname::text) = lower($2) \
+         ORDER BY (p.pkgname::text = $2) DESC, (n.nspname::text = $1) DESC, p.oid \
+         LIMIT 1",
+        &[&schema, &name],
+    )
+    .await
+    .map_err(|e| e.to_string())?;
+    let Some(row) = meta_rows.first() else {
+        return Err(format!("openGauss {} source not found: {schema}.{name}", source_type));
+    };
+    let catalog_name = row
+        .try_get::<_, Option<String>>(0)
+        .ok()
+        .flatten()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or_else(|| name.to_string());
+    let catalog_schema = row
+        .try_get::<_, Option<String>>(1)
+        .ok()
+        .flatten()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or_else(|| schema.to_string());
+    let pkgsecdef = row.try_get::<_, bool>(2).map_err(|error| {
+        format!("Cannot read openGauss package security mode for {catalog_schema}.{catalog_name}: {error}")
+    })?;
+    let spec = row.try_get::<_, Option<String>>(3).ok().flatten().unwrap_or_default();
+    let body_decl = row.try_get::<_, Option<String>>(4).ok().flatten().unwrap_or_default();
+    let body_init = row.try_get::<_, Option<String>>(5).ok().flatten().unwrap_or_default();
+    if package_body {
+        opengauss_package_body_source(&catalog_schema, &catalog_name, &body_decl, &body_init).ok_or_else(|| {
+            format!("Cannot rebuild openGauss package body source from catalog for {catalog_schema}.{catalog_name}")
+        })
+    } else {
+        opengauss_package_spec_source(&catalog_schema, &catalog_name, &spec, pkgsecdef).ok_or_else(|| {
+            format!("Cannot rebuild openGauss package source from catalog for {catalog_schema}.{catalog_name}")
+        })
+    }
+}
+
+/// Marker openGauss stores at the start of the declaration fragment
+/// (`AS` is rewritten to this during CREATE processing).
+const OPENGAUSS_PACKAGE_DECL_MARKER: &str = "PACKAGE  DECLARE";
+/// Marker at the start of the package initialization fragment.
+const OPENGAUSS_PACKAGE_INIT_MARKER: &str = "INSTANTIATION";
+
+/// Strip the openGauss declaration-fragment wrapper: the ` PACKAGE  DECLARE `
+/// prefix and the trailing bare `END` marker that closes the fragment. Member
+/// terminators are always `END;` (with a semicolon) and are therefore kept.
+fn trim_opengauss_package_decl_fragment(fragment: &str) -> Option<String> {
+    let trimmed = fragment.trim();
+    let body = trimmed.strip_prefix(OPENGAUSS_PACKAGE_DECL_MARKER)?.trim();
+    let body = body.strip_suffix("END").unwrap_or(body).trim_end();
+    Some(body.trim().to_string())
+}
+
+/// Rebuild a full `CREATE OR REPLACE PACKAGE` statement from the gs_package
+/// declaration fragment. Used as the fallback when `dbe_pldeveloper.gs_source`
+/// has no row for the package.
+#[doc(hidden)]
+pub fn opengauss_package_spec_source(schema: &str, name: &str, spec_fragment: &str, pkgsecdef: bool) -> Option<String> {
+    let decl = trim_opengauss_package_decl_fragment(spec_fragment)?;
+    let authid = if pkgsecdef { " AUTHID DEFINER" } else { " AUTHID CURRENT_USER" };
+    Some(format!(
+        "CREATE OR REPLACE PACKAGE {}.{}{} AS\n{}\nEND {};",
+        pg_quote_ident(schema),
+        pg_quote_ident(name),
+        authid,
+        decl,
+        pg_quote_ident(name)
+    ))
+}
+
+/// Rebuild a full `CREATE OR REPLACE PACKAGE BODY` statement from the
+/// gs_package declaration and initialization fragments.
+#[doc(hidden)]
+pub fn opengauss_package_body_source(
+    schema: &str,
+    name: &str,
+    decl_fragment: &str,
+    init_fragment: &str,
+) -> Option<String> {
+    let decl = trim_opengauss_package_decl_fragment(decl_fragment)?;
+    let mut out = format!("CREATE OR REPLACE PACKAGE BODY {}.{} AS\n", pg_quote_ident(schema), pg_quote_ident(name));
+    if !decl.is_empty() {
+        out.push_str(&decl);
+        out.push('\n');
+    }
+
+    let init = init_fragment.trim();
+    if init.is_empty() {
+        // No initialization section: the declaration fragment's own trailing
+        // END marker was already removed, so close the body here.
+        out.push_str(&format!("END {};", pg_quote_ident(name)));
+        return Some(out);
+    }
+
+    // Initialization section is stored as `INSTANTIATION \nBEGIN ... END`.
+    let init = init.strip_prefix(OPENGAUSS_PACKAGE_INIT_MARKER)?.trim_start();
+    // The trailing `END` of the init fragment is the body terminator; give it
+    // the package name so the rebuilt DDL stays self-contained.
+    let core = init.strip_suffix("END").unwrap_or(init).trim_end();
+    out.push_str(core);
+    out.push('\n');
+    out.push_str(&format!("END {};", pg_quote_ident(name)));
+    Some(out)
+}
+
+/// SQL for the package identity lookup used by [`opengauss_package_members`].
+/// Prefer an exact spelling, while retaining a case-insensitive fallback for
+/// ordinary unquoted identifiers typed with a different case.
+fn opengauss_package_identity_sql() -> &'static str {
+    "SELECT p.oid::oid, p.pkgname::text, n.nspname::text \
+     FROM pg_catalog.gs_package p \
+     JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+     WHERE lower(n.nspname) = lower($1) AND lower(p.pkgname::text) = lower($2) \
+     ORDER BY (p.pkgname::text = $2) DESC, (n.nspname::text = $1) DESC, p.oid \
+     LIMIT 1"
+}
+
+/// SQL for [`opengauss_package_members`]: members selected by the resolved
+/// package OID, preventing same-lowercase quoted packages from being merged.
+fn opengauss_package_members_sql() -> &'static str {
+    "SELECT f.proname::text, f.prokind::text, \
+            pg_catalog.pg_get_function_identity_arguments(f.oid)::text, \
+            pg_catalog.format_type(f.prorettype, NULL)::text \
+     FROM pg_catalog.pg_proc f \
+     JOIN pg_catalog.gs_package p ON p.oid = f.propackageid \
+     JOIN pg_catalog.pg_namespace n ON n.oid = p.pkgnamespace \
+     WHERE f.propackageid = $1 \
+       AND f.prokind IN ('f','p','v','b','i') \
+       AND COALESCE(f.proisprivate, false) = false \
+       AND ($2 = '%%' OR f.proname ILIKE $2 ESCAPE '~') \
+     ORDER BY f.proname, f.oid LIMIT $3"
+}
+
+pub async fn opengauss_package_members(
+    pool: &Pool,
+    request: &CompletionAssistantRequest,
+) -> Result<CompletionAssistantResponse, String> {
+    match opengauss_is_oracle_compatible(pool).await {
+        Ok(true) => {}
+        Ok(false) | Err(_) => {
+            return Ok(CompletionAssistantResponse { candidates: Vec::new(), incomplete: false, fallback_used: true });
+        }
+    }
+    let package_name = request.parent_name.as_deref().unwrap_or("").trim();
+    if package_name.is_empty() {
+        return Ok(CompletionAssistantResponse { candidates: Vec::new(), incomplete: false, fallback_used: true });
+    }
+    let schema = request.parent_schema.as_deref().or(request.schema.as_deref()).unwrap_or("public");
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    // Resolve one package identity before looking up members. Lowercase matching
+    // remains useful for ordinary unquoted names, while the exact-name ordering
+    // preserves quoted mixed-case package identity when it is available.
+    let identity_rows =
+        match postgres_query_cached(&client, opengauss_package_identity_sql(), &[&schema, &package_name]).await {
+            Ok(rows) => rows,
+            Err(error) if opengauss_optional_package_catalog_error(&error.to_string()) => {
+                return Ok(CompletionAssistantResponse {
+                    candidates: Vec::new(),
+                    incomplete: false,
+                    fallback_used: true,
+                });
+            }
+            Err(error) => return Err(error.to_string()),
+        };
+    let Some(identity_row) = identity_rows.first() else {
+        return Ok(CompletionAssistantResponse { candidates: Vec::new(), incomplete: false, fallback_used: true });
+    };
+    let Some(package_oid) = pg_row_try_u32(identity_row, 0) else {
+        return Ok(CompletionAssistantResponse { candidates: Vec::new(), incomplete: false, fallback_used: true });
+    };
+    let catalog_name = pg_row_try_string(identity_row, 1);
+    let catalog_schema = pg_row_try_string(identity_row, 2);
+    let limit = request.max_results.unwrap_or(100).clamp(1, 1000);
+    // The member prefix mask is applied on the server before the LIMIT so a
+    // matching member is never shadowed by the first N rows of the full
+    // alphabetically-sorted member list (the editor only requests a limited
+    // slice and then filters locally).
+    let pattern = postgres_completion_like_pattern(&request.mask, request.match_mode.as_ref());
+    let rows = match postgres_query_cached(
+        &client,
+        opengauss_package_members_sql(),
+        &[&package_oid, &pattern, &(limit as i64)],
+    )
+    .await
+    {
+        Ok(rows) => rows,
+        Err(error) if opengauss_optional_package_catalog_error(&error.to_string()) => {
+            return Ok(CompletionAssistantResponse { candidates: Vec::new(), incomplete: false, fallback_used: true });
+        }
+        Err(error) => return Err(error.to_string()),
+    };
+    let candidates = rows
+        .iter()
+        .map(|row| {
+            let prokind = pg_row_try_string(row, 1);
+            // openGauss catalog reports ordinary f/p members on current releases.
+            // v (subfunction) / b (subprocedure) are kept defensive for
+            // variants that tag package members with subkind markers.
+            let kind = if matches!(prokind.as_str(), "p" | "b") {
+                CompletionAssistantCandidateKind::Procedure
+            } else {
+                CompletionAssistantCandidateKind::Function
+            };
+            CompletionAssistantCandidate {
+                name: pg_row_try_string(row, 0),
+                kind,
+                database: Some(request.database.clone()),
+                schema: Some(catalog_schema.clone()),
+                parent_schema: Some(catalog_schema.clone()),
+                parent_name: Some(catalog_name.clone()),
+                comment: None,
+                data_type: row.try_get::<_, Option<String>>(3).ok().flatten(),
+                signature: row.try_get::<_, Option<String>>(2).ok().flatten(),
+            }
+        })
+        .collect::<Vec<_>>();
+    Ok(CompletionAssistantResponse { incomplete: candidates.len() >= limit, candidates, fallback_used: false })
+}
+
 pub async fn list_redshift_objects(
     pool: &Pool,
     schema: &str,
@@ -5543,7 +6107,7 @@ pub async fn list_redshift_objects(
     if include_relations {
         // Redshift does not support PostgreSQL's generic file, object-location,
         // or transaction-ID helpers, so skip the timestamp variant entirely.
-        rows = list_objects_rows(&client, schema, false, false, false, false, true, false, false).await?;
+        rows = list_objects_rows(&client, schema, false, false, false, false, true, false, false, false).await?;
     }
 
     if include_routines {
@@ -9357,6 +9921,79 @@ mod tests {
         assert_eq!(gaussdb_identifier_quote_for_compatibility_mode(""), None);
     }
 
+    #[test]
+    fn opengauss_package_catalog_fallback_rebuilds_spec_ddl() {
+        // Real shape captured from openGauss-lite 5.0.1 gs_package.pkgspecsrc.
+        let fragment = " PACKAGE  DECLARE  g_version VARCHAR2(20) := '1.0';\n  PROCEDURE log_message(p_message IN VARCHAR2);\n  FUNCTION add_numbers(p_left IN INTEGER, p_right IN INTEGER) RETURN INTEGER;\nEND ";
+        let ddl = opengauss_package_spec_source("public", "dbx_test_pkg_math", fragment, false).unwrap();
+        assert!(ddl.starts_with("CREATE OR REPLACE PACKAGE \"public\".\"dbx_test_pkg_math\" AUTHID CURRENT_USER AS\n"));
+        assert!(ddl.contains("AUTHID CURRENT_USER"));
+        assert!(ddl.contains("g_version VARCHAR2(20) := '1.0';"));
+        assert!(ddl.contains("PROCEDURE log_message(p_message IN VARCHAR2);"));
+        assert!(ddl.ends_with("END \"dbx_test_pkg_math\";"));
+        assert!(!ddl.contains("END  "));
+        assert_eq!(ddl.matches("END").count(), 1);
+    }
+
+    #[test]
+    fn opengauss_package_catalog_fallback_preserves_definer_security() {
+        let fragment = " PACKAGE  DECLARE  PROCEDURE ping;\nEND ";
+        let ddl = opengauss_package_spec_source("public", "secure_pkg", fragment, true).unwrap();
+        assert!(ddl.contains("AUTHID DEFINER"));
+        assert!(!ddl.contains("AUTHID CURRENT_USER"));
+    }
+
+    #[test]
+    fn opengauss_package_catalog_fallback_rebuilds_body_with_init_ddl() {
+        // Real shapes from gs_package.pkgbodydeclsrc / pkgbodyinitsrc.
+        let decl = " PACKAGE  DECLARE  PROCEDURE log_message(p_message IN VARCHAR2) IS\n  BEGIN\n    NULL;\n  END;\n\n  FUNCTION add_numbers(p_left IN INTEGER, p_right IN INTEGER) RETURN INTEGER IS\n  BEGIN\n    RETURN p_left + p_right;\n  END;\n\nEND\n";
+        let init = " INSTANTIATION \nBEGIN\n  g_version := '1.1';\nEND\n";
+        let ddl = opengauss_package_body_source("public", "dbx_test_pkg_math", decl, init).unwrap();
+        assert!(ddl.starts_with("CREATE OR REPLACE PACKAGE BODY \"public\".\"dbx_test_pkg_math\" AS\n"));
+        assert!(ddl.contains("PROCEDURE log_message(p_message IN VARCHAR2) IS"));
+        assert!(ddl.contains("FUNCTION add_numbers(p_left IN INTEGER, p_right IN INTEGER) RETURN INTEGER IS"));
+        assert!(ddl.contains("g_version := '1.1';"));
+        assert!(ddl.ends_with("END \"dbx_test_pkg_math\";"));
+    }
+
+    #[test]
+    fn opengauss_package_catalog_fallback_rebuilds_body_without_init_ddl() {
+        let decl = " PACKAGE  DECLARE  FUNCTION format_value(p_value IN INTEGER) RETURN VARCHAR2 IS\n  BEGIN\n    RETURN 'x';\n  END;\nEND ";
+        let ddl = opengauss_package_body_source("public", "dbx_test_pkg_overload", decl, "").unwrap();
+        assert!(ddl.starts_with("CREATE OR REPLACE PACKAGE BODY \"public\".\"dbx_test_pkg_overload\" AS\n"));
+        assert!(ddl.contains("FUNCTION format_value(p_value IN INTEGER) RETURN VARCHAR2 IS"));
+        assert!(ddl.ends_with("END \"dbx_test_pkg_overload\";"));
+        assert!(!ddl.contains("INSTANTIATION"));
+    }
+
+    #[test]
+    fn opengauss_package_catalog_fallback_rejects_unrecognized_shape() {
+        assert!(opengauss_package_spec_source("public", "pkg", "CREATE OR REPLACE PACKAGE pkg AS\nEND pkg;", true)
+            .is_none());
+        assert!(opengauss_package_spec_source("public", "pkg", "", true).is_none());
+    }
+
+    #[test]
+    fn opengauss_gs_source_error_classification_falls_back_only_for_catalog_issues() {
+        for message in [
+            "ERROR: permission denied for schema dbe_pldeveloper",
+            "ERROR: insufficient privilege to query table gs_source",
+            "ERROR: relation \"gs_source\" does not exist",
+            "ERROR: schema \"dbe_pldeveloper\" does not exist",
+        ] {
+            assert!(opengauss_gs_source_lookup_should_fallback(message), "expected fallback for: {message}");
+        }
+        for message in [
+            "connection refused",
+            "server closed the connection unexpectedly",
+            "timeout expired",
+            "canceling statement due to statement timeout",
+            "syntax error at or near \"SELECT\"",
+        ] {
+            assert!(!opengauss_gs_source_lookup_should_fallback(message), "expected no fallback for: {message}");
+        }
+    }
+
     fn test_query_message(message: &str) -> QueryMessage {
         QueryMessage {
             severity: "NOTICE".to_string(),
@@ -13706,7 +14343,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_includes_routines() {
-        let sql = list_objects_sql(true, true, false, true, true, true, false);
+        let sql = list_objects_sql(true, true, false, true, true, true, false, false);
         assert!(sql.contains("pg_catalog.pg_class"));
         assert!(sql.contains("pg_catalog.pg_proc"));
         assert!(sql.contains("pg_catalog.pg_inherits"));
@@ -13724,7 +14361,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_includes_custom_types_when_enabled() {
-        let sql = list_objects_sql(true, true, false, true, true, true, true);
+        let sql = list_objects_sql(true, true, false, true, true, true, true, false);
         assert!(sql.contains("pg_catalog.pg_type"));
         assert!(sql.contains("pg_catalog.pg_class"));
         assert!(sql.contains("pg_catalog.pg_description"));
@@ -13755,14 +14392,14 @@ mod tests {
 
     #[test]
     fn list_objects_sql_omits_custom_types_when_disabled() {
-        let sql = list_objects_sql(true, true, false, true, true, true, false);
+        let sql = list_objects_sql(true, true, false, true, true, true, false, false);
         assert!(!sql.contains("pg_type"));
         assert!(!sql.contains("t.typtype"));
     }
 
     #[test]
     fn list_objects_sql_type_only_skips_relations_and_routines() {
-        let sql = list_objects_sql(true, true, false, true, false, false, true);
+        let sql = list_objects_sql(true, true, false, true, false, false, true, false);
         assert!(sql.contains("pg_catalog.pg_type"));
         assert!(!sql.contains("FROM pg_catalog.pg_class"));
         assert!(!sql.contains("pg_catalog.pg_proc"));
@@ -13772,7 +14409,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_table_only_skips_routines_and_custom_types() {
-        let sql = list_objects_sql(true, true, false, true, true, false, false);
+        let sql = list_objects_sql(true, true, false, true, true, false, false, false);
         assert!(sql.contains("pg_catalog.pg_class"));
         assert!(!sql.contains("pg_catalog.pg_proc"));
         assert!(!sql.contains("pg_catalog.pg_type"));
@@ -13780,7 +14417,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_routine_only_skips_relations_and_custom_types() {
-        let sql = list_objects_sql(true, true, false, true, false, true, false);
+        let sql = list_objects_sql(true, true, false, true, false, true, false, false);
         assert!(sql.contains("pg_catalog.pg_proc"));
         assert!(!sql.contains("pg_catalog.pg_class"));
         assert!(!sql.contains("pg_catalog.pg_type"));
@@ -13788,7 +14425,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_empty_scope_produces_no_sql() {
-        let sql = list_objects_sql(true, true, false, true, false, false, false);
+        let sql = list_objects_sql(true, true, false, true, false, false, false, false);
         assert!(sql.is_empty());
     }
 
@@ -13797,8 +14434,8 @@ mod tests {
         // The pg_type branch carries the same filters whether the timestamp
         // query or the timestamp fallback query is used.
         for sql in [
-            list_objects_sql(true, true, false, true, false, false, true),
-            list_objects_sql(false, true, false, true, false, false, true),
+            list_objects_sql(true, true, false, true, false, false, true, false),
+            list_objects_sql(false, true, false, true, false, false, true, false),
         ] {
             assert!(sql.contains("pg_catalog.pg_type"));
             assert!(sql.contains("t.typtype IN ('b', 'c', 'd', 'e', 'r', 'm')"));
@@ -13810,7 +14447,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_custom_types_keeps_comment_join_semantics() {
-        let sql = list_objects_sql(true, true, false, true, false, false, true);
+        let sql = list_objects_sql(true, true, false, true, false, false, true, false);
         assert!(sql.contains("d.objoid = t.oid"));
         assert!(sql.contains("d.classoid = 'pg_catalog.pg_type'::regclass"));
         assert!(sql.contains("d.objsubid = 0"));
@@ -13819,7 +14456,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_custom_types_uses_null_timestamps() {
-        let sql = list_objects_sql(true, true, false, true, false, false, true);
+        let sql = list_objects_sql(true, true, false, true, false, false, true, false);
         assert!(sql.contains("NULL::text AS created_at"));
         assert!(sql.contains("NULL::text AS updated_at"));
         assert!(sql.contains("NULL::text AS parent_schema"));
@@ -13830,7 +14467,7 @@ mod tests {
 
     #[test]
     fn list_objects_sql_without_timestamps_omits_stat_file() {
-        let sql = list_objects_sql(false, true, false, true, true, true, false);
+        let sql = list_objects_sql(false, true, false, true, true, true, false, false);
         assert!(!sql.contains("pg_stat_file"));
         assert!(sql.contains("NULL::text AS created_at"));
         assert!(sql.contains("NULL::text AS updated_at"));
@@ -13838,7 +14475,7 @@ mod tests {
 
     #[test]
     fn redshift_compatible_list_objects_sql_uses_legacy_argument_formatter() {
-        let sql = list_objects_sql(false, false, false, false, true, true, false);
+        let sql = list_objects_sql(false, false, false, false, true, true, false, false);
         assert!(sql.contains("pg_get_function_arguments(p.oid) AS signature"));
         assert!(!sql.contains("pg_get_function_identity_arguments"));
     }
@@ -13858,7 +14495,7 @@ mod tests {
 
     #[test]
     fn redshift_relation_objects_sql_avoids_unsupported_postgres_helpers() {
-        let sql = list_objects_sql(false, false, false, false, true, false, false);
+        let sql = list_objects_sql(false, false, false, false, true, false, false, false);
         assert!(sql.contains("pg_catalog.pg_class"));
         assert!(!sql.contains("pg_catalog.pg_proc"));
         assert!(!sql.contains("pg_stat_file"));
@@ -13932,31 +14569,31 @@ mod tests {
 
     #[test]
     fn both_list_objects_sql_variants_use_parameter() {
-        assert!(list_objects_sql(true, true, true, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(false, true, true, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(true, true, false, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(false, true, false, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(true, false, true, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(false, false, true, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(true, false, false, true, true, true, false).contains("$1"));
-        assert!(list_objects_sql(false, false, false, true, true, true, false).contains("$1"));
+        assert!(list_objects_sql(true, true, true, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(false, true, true, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(true, true, false, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(false, true, false, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(true, false, true, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(false, false, true, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(true, false, false, true, true, true, false, false).contains("$1"));
+        assert!(list_objects_sql(false, false, false, true, true, true, false, false).contains("$1"));
     }
 
     #[test]
     fn both_list_objects_sql_variants_include_pg_proc() {
-        assert!(list_objects_sql(true, true, true, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(false, true, true, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(true, true, false, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(false, true, false, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(true, false, true, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(false, false, true, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(true, false, false, true, true, true, false).contains("pg_catalog.pg_proc"));
-        assert!(list_objects_sql(false, false, false, true, true, true, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(true, true, true, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(false, true, true, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(true, true, false, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(false, true, false, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(true, false, true, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(false, false, true, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(true, false, false, true, true, true, false, false).contains("pg_catalog.pg_proc"));
+        assert!(list_objects_sql(false, false, false, true, true, true, false, false).contains("pg_catalog.pg_proc"));
     }
 
     #[test]
     fn legacy_list_objects_sql_avoids_pg11_proc_kind_column() {
-        let sql = list_objects_sql(true, false, false, true, true, true, false);
+        let sql = list_objects_sql(true, false, false, true, true, true, false, false);
         assert!(!sql.contains("p.prokind"));
         assert!(!sql.contains("p.prosp"));
         assert!(sql.contains("NOT p.proisagg"));
@@ -13968,7 +14605,7 @@ mod tests {
 
     #[test]
     fn gaussdb_compatible_list_objects_sql_uses_prosp_when_prokind_is_missing() {
-        let sql = list_objects_sql(true, false, true, true, true, true, false);
+        let sql = list_objects_sql(true, false, true, true, true, true, false, false);
         assert!(!sql.contains("p.prokind"));
         assert!(sql.contains("CASE WHEN p.prosp THEN 'PROCEDURE' ELSE 'FUNCTION' END AS object_type"));
         assert!(sql.contains("CASE WHEN p.prosp THEN 2 ELSE 3 END AS sort_order"));
@@ -13979,7 +14616,7 @@ mod tests {
 
     #[test]
     fn gaussdb_compatible_list_objects_sql_uses_prosp_with_prokind_when_available() {
-        let sql = list_objects_sql(true, true, true, true, true, true, false);
+        let sql = list_objects_sql(true, true, true, true, true, true, false, false);
         assert!(
             sql.contains("CASE WHEN p.prokind = 'p' OR p.prosp THEN 'PROCEDURE' ELSE 'FUNCTION' END AS object_type")
         );
@@ -14611,6 +15248,48 @@ mod tests {
         assert!(postgres_visible_table_schema_sql().contains("pg_catalog.pg_table_is_visible(c.oid)"));
     }
 
+    #[test]
+    fn opengauss_completion_packages_sql_filters_a_mode_and_prefix_before_limit() {
+        let sql = opengauss_completion_packages_sql();
+        assert!(sql.contains("FROM pg_catalog.gs_package"));
+        assert!(sql.contains("d.datcompatibility::text = 'A'"));
+        assert!(sql.contains("p.pkgname ILIKE $2 ESCAPE '~'"));
+        assert!(sql.contains("ORDER BY p.pkgname LIMIT $3"));
+    }
+
+    #[test]
+    fn opengauss_completion_routines_sql_excludes_package_and_private_members() {
+        // openGauss stores package members (propackage=true) and private members
+        // (proisprivate=true) in pg_proc; the completion variant must not surface
+        // them as top-level routines. See review P1 #1.
+        for has_proc_prokind in [true, false] {
+            let sql = opengauss_completion_routines_sql(has_proc_prokind);
+            assert!(sql.contains("COALESCE(p.propackage, false) = false"));
+            assert!(sql.contains("COALESCE(p.proisprivate, false) = false"));
+            assert!(sql.contains("WHERE n.nspname = $1"));
+            // The prefix mask still applies before the LIMIT.
+            assert!(sql.contains("$2 = '%%' OR p.proname ILIKE $2 ESCAPE '~'"));
+        }
+        assert!(opengauss_completion_routines_sql(true).contains("p.prokind::text = ANY($3::text[])"));
+        assert!(!opengauss_completion_routines_sql(false).contains("p.prokind"));
+    }
+
+    #[test]
+    fn opengauss_package_members_sql_uses_case_insensitive_names_and_prefix_mask() {
+        // Package names are normalized by the parser, so catalog and
+        // user-typed spellings must match case-insensitively.
+        // #3: the member prefix must filter on the server before LIMIT so a
+        // matching member is not shadowed by the first N alphabetically-sorted
+        // rows.
+        let sql = opengauss_package_members_sql();
+        assert!(sql.contains("WHERE f.propackageid = $1"));
+        assert!(sql.contains("AND f.prokind IN ('f','p','v','b','i')"));
+        assert!(sql.contains("COALESCE(f.proisprivate, false) = false"));
+        let identity = opengauss_package_identity_sql();
+        assert!(identity.contains("SELECT p.oid::oid, p.pkgname::text, n.nspname::text"));
+        assert!(identity.contains("ORDER BY (p.pkgname::text = $2) DESC"));
+    }
+
     #[test]
     fn postgres_schema_info_sql_only_filters_system_schemas_when_disabled() {
         let hidden_sql = postgres_schema_infos_sql(false);

crates/dbx-core/src/query.rs

@@ -793,6 +793,21 @@ async fn connection_database_type(state: &AppState, connection_id: &str) -> Opti
     configs.get(connection_id).map(|config| config.db_type)
 }
 
+async fn connection_sql_compatibility_mode(
+    state: &AppState,
+    pool_key: &str,
+    db_type: Option<DatabaseType>,
+) -> Option<String> {
+    if db_type != Some(DatabaseType::OpenGauss) {
+        return None;
+    }
+    let pool = match state.pool_handle(pool_key).await {
+        Some(PoolKind::Postgres(pool)) => pool,
+        _ => return None,
+    };
+    db::postgres::opengauss_compatibility_mode(&pool).await.ok().flatten()
+}
+
 async fn connection_mysql_query_dialect(state: &AppState, connection_id: &str) -> db::mysql::MySqlQueryDialect {
     let configs = state.configs.read().await;
     configs
@@ -3112,7 +3127,9 @@ pub async fn execute_multi_core_with_options_for_client_and_progress_typed(
         }
     };
 
-    let execution_plan = query_execution_plan(sql, db_type, is_sqlserver_agent);
+    let compatibility_mode = connection_sql_compatibility_mode(state, &pool_key, db_type).await;
+    let execution_plan =
+        query_execution_plan_with_compatibility(sql, db_type, is_sqlserver_agent, compatibility_mode.as_deref());
     let continue_on_error = options.continue_on_error && !execution_plan.stop_on_error;
     let statements = execution_plan.statements;
     if statements.is_empty() {
@@ -3304,6 +3321,19 @@ pub fn query_execution_plan(
     sql: &str,
     db_type: Option<DatabaseType>,
     preserve_sqlserver_batches: bool,
+) -> crate::sql::SqlExecutionPlan {
+    query_execution_plan_with_compatibility(sql, db_type, preserve_sqlserver_batches, None)
+}
+
+/// Same as [`query_execution_plan`], but lets openGauss callers pass the
+/// database compatibility mode so A-mode PL/SQL (package) bodies are never split
+/// on inner semicolons. `None` keeps the conservative PL/SQL-capable openGauss
+/// profile while the probe is cold or unavailable.
+pub fn query_execution_plan_with_compatibility(
+    sql: &str,
+    db_type: Option<DatabaseType>,
+    preserve_sqlserver_batches: bool,
+    compatibility_mode: Option<&str>,
 ) -> crate::sql::SqlExecutionPlan {
     if preserve_sqlserver_batches && db_type == Some(DatabaseType::SqlServer) {
         return crate::sql::SqlExecutionPlan { statements: split_sql_batches(sql), stop_on_error: false };
@@ -3311,7 +3341,7 @@ pub fn query_execution_plan(
 
     db_type.map_or_else(
         || crate::sql::SqlExecutionPlan { statements: split_sql_statements(sql), stop_on_error: false },
-        |db_type| crate::sql::sql_execution_plan_for_database(sql, db_type),
+        |db_type| crate::sql::sql_execution_plan_for_database_with_compatibility(sql, db_type, compatibility_mode),
     )
 }
 
@@ -4191,10 +4221,41 @@ pub async fn execute_schema_diff_deploy(
     let now = chrono::Utc::now().to_rfc3339();
     let db_type = connection_database_type(state, connection_id).await;
 
+    // openGauss A-mode deploy scripts may contain CREATE PACKAGE … / blocks
+    // that must stay intact. Probe the database compatibility mode when a pool
+    // is available; on failure fall back to the plain openGauss splitter.
+    let compatibility_mode = if db_type == Some(DatabaseType::OpenGauss) {
+        let pool = match state
+            .get_or_create_pool(connection_id, if database.is_empty() { None } else { Some(database) })
+            .await
+        {
+            Ok(pool_key) => match state.pool_handle(&pool_key).await {
+                Some(PoolKind::Postgres(pool)) => Some(pool),
+                _ => None,
+            },
+            Err(_) => None,
+        };
+        match pool {
+            Some(pool) => db::postgres::opengauss_compatibility_mode(&pool).await.ok().flatten(),
+            None => None,
+        }
+    } else {
+        None
+    };
+
     let parsed: Vec<String> = statements
         .iter()
         .flat_map(|s| {
-            db_type.map_or_else(|| split_sql_statements(s), |dt| crate::sql::split_sql_statements_for_database(s, dt))
+            db_type.map_or_else(
+                || split_sql_statements(s),
+                |dt| {
+                    crate::sql::split_sql_statements_for_database_with_compatibility(
+                        s,
+                        dt,
+                        compatibility_mode.as_deref(),
+                    )
+                },
+            )
         })
         .map(|s| s.trim().to_string())
         .filter(|s| {
@@ -5351,9 +5412,16 @@ pub async fn execute_in_manual_transaction_with_options(
     };
 
     let db_type = connection_database_type(state, &connection_id).await;
+    let compatibility_mode = connection_sql_compatibility_mode(state, &pool_key, db_type).await;
     let statements = db_type.map_or_else(
         || split_sql_statements(sql),
-        |db_type| crate::sql::split_sql_statements_for_database(sql, db_type),
+        |db_type| {
+            crate::sql::split_sql_statements_for_database_with_compatibility(
+                sql,
+                db_type,
+                compatibility_mode.as_deref(),
+            )
+        },
     );
     if statements.is_empty() {
         // Oracle-only UX marker: the no-op is Core's decision that the script

crates/dbx-core/src/schema.rs

@@ -766,6 +766,9 @@ async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Ve
         }
         PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => db::ob_oracle::list_databases(p).await,
         PoolKind::Mysql(p, _) => db::mysql::list_databases_with_timeout(p, mysql_database_list_timeout).await,
+        PoolKind::Postgres(p) if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss) => {
+            db::postgres::list_opengauss_databases(p).await
+        }
         PoolKind::Postgres(p) => db::postgres::list_databases(p).await,
         PoolKind::Sqlite(p) => db::sqlite::list_databases(p).await,
         PoolKind::Rqlite(client) => db::rqlite_driver::list_databases(client).await,
@@ -5816,7 +5819,26 @@ pub async fn completion_assistant_search_core(
                 PoolKind::Postgres(pool) => Some(pool.clone()),
                 _ => None,
             }) {
-                return db::postgres::completion_assistant_search(&pool, &request).await;
+                let db_config = connection_config(state, &request.connection_id).await;
+                if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss)
+                    && request.parent_name.as_deref().is_some_and(|name| !name.trim().is_empty())
+                    && request.object_kinds.iter().any(db::CompletionAssistantObjectKind::is_routine_like)
+                {
+                    let response = db::postgres::opengauss_package_members(&pool, &request).await?;
+                    // fallback_used marks "parent is not a package": continue with
+                    // the ordinary routine completion instead of returning nothing.
+                    if !response.fallback_used {
+                        return Ok(response);
+                    }
+                }
+                return if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss) {
+                    // openGauss variant excludes package/private members from the
+                    // ordinary routine lookup, so the fallback stays consistent
+                    // with the package-aware path above.
+                    db::postgres::opengauss_completion_assistant_search(&pool, &request).await
+                } else {
+                    db::postgres::completion_assistant_search(&pool, &request).await
+                };
             }
         }
 
@@ -6327,9 +6349,29 @@ async fn list_objects_once(
             let include_routines = object_types_include_routines(object_types);
             let include_custom_types = db_config.as_ref().is_some_and(supports_pg_custom_type_objects)
                 && object_types_include_custom_types(object_types);
-            db::postgres::list_objects(p, schema, include_relations, include_routines, include_custom_types)
-                .await
-                .map(unpaged_object_list)
+            let mut objects = if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss) {
+                // openGauss variant excludes package members (propackage=true)
+                // from the top-level routine list; they are surfaced under
+                // their PACKAGE nodes below.
+                db::postgres::list_opengauss_objects(
+                    p,
+                    schema,
+                    include_relations,
+                    include_routines,
+                    include_custom_types,
+                )
+                .await?
+            } else {
+                db::postgres::list_objects(p, schema, include_relations, include_routines, include_custom_types).await?
+            };
+            if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss) {
+                let (include_package_spec, include_package_body) = object_types_include_packages(object_types);
+                objects.extend(
+                    db::postgres::list_opengauss_packages(p, schema, include_package_spec, include_package_body)
+                        .await?,
+                );
+            }
+            Ok(unpaged_object_list(objects))
         }
         _ => Ok(unpaged_object_list(
             list_tables_core(state, connection_id, database, schema, None, None, None, None, None)
@@ -6456,6 +6498,11 @@ async fn list_completion_objects_once(
         PoolKind::Postgres(p) if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::Redshift) => {
             db::postgres::list_redshift_objects(p, schema, false, true).await.map(filter_completion_objects)
         }
+        PoolKind::Postgres(p) if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss) => {
+            let mut objects = db::postgres::list_opengauss_objects(p, schema, false, true, false).await?;
+            objects.extend(db::postgres::list_opengauss_packages(p, schema, true, true).await?);
+            Ok(filter_completion_objects_with_packages(objects))
+        }
         PoolKind::Postgres(p) => {
             db::postgres::list_objects(p, schema, true, true, false).await.map(filter_completion_objects)
         }
@@ -6478,6 +6525,19 @@ fn filter_completion_objects(objects: Vec<db::ObjectInfo>) -> Vec<db::ObjectInfo
         .collect()
 }
 
+fn filter_completion_objects_with_packages(objects: Vec<db::ObjectInfo>) -> Vec<db::ObjectInfo> {
+    objects
+        .into_iter()
+        .filter(|object| {
+            let object_type = object.object_type.to_ascii_uppercase();
+            object_type.contains("PROCEDURE")
+                || object_type.contains("FUNCTION")
+                || object_type.contains("TRIGGER")
+                || object_type == "PACKAGE"
+        })
+        .collect()
+}
+
 fn is_agent_postgres_metadata_fallback_config(config: &ConnectionConfig) -> bool {
     // HighGo and Vastbase can use the native PostgreSQL metadata path when their
     // agent returns no rows. UXDB is JDBC-only; opening a PostgreSQL fallback
@@ -8294,6 +8354,16 @@ fn object_types_include_custom_types(object_types: Option<&[String]>) -> bool {
         .is_none_or(|types| types.iter().any(|t| t.eq_ignore_ascii_case("TYPE") || t.eq_ignore_ascii_case("TYPE_BODY")))
 }
 
+fn object_types_include_packages(object_types: Option<&[String]>) -> (bool, bool) {
+    match object_types {
+        None => (true, true),
+        Some(types) => (
+            types.iter().any(|value| value.eq_ignore_ascii_case("PACKAGE")),
+            types.iter().any(|value| value.eq_ignore_ascii_case("PACKAGE_BODY")),
+        ),
+    }
+}
+
 /// Whether the object-type filter exclusively asks for user-defined types.
 ///
 /// Used to keep agent errors visible: the native PostgreSQL fallback never
@@ -9473,6 +9543,18 @@ async fn get_object_source_once(
                     mysql_object_source(pool, mysql_table_metadata_catalog(database, schema), name, &object_type)
                         .await?
                 }
+                PoolKind::Postgres(pool)
+                    if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss)
+                        && matches!(object_type, db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody) =>
+                {
+                    db::postgres::opengauss_package_source(
+                        pool,
+                        schema,
+                        name,
+                        matches!(object_type, db::ObjectSourceKind::PackageBody),
+                    )
+                    .await?
+                }
                 PoolKind::Postgres(pool) if db_config.as_ref().is_some_and(is_questdb_config) => {
                     // only view
                     db::questdb::questdb_object_source(pool, name).await?
@@ -9527,7 +9609,14 @@ async fn get_object_source_once(
         }
     };
 
-    let editable = if matches!(object_type, db::ObjectSourceKind::Trigger)
+    let editable = if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::OpenGauss)
+        && matches!(object_type, db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody)
+    {
+        // gs_source returns the original CREATE text. Re-executing CREATE for an
+        // existing package is not a safe edit operation unless the user changes
+        // it to CREATE OR REPLACE explicitly, so keep the initial implementation read-only.
+        Some(false)
+    } else if matches!(object_type, db::ObjectSourceKind::Trigger)
         && db_config.as_ref().is_some_and(|config| {
             matches!(
                 config.db_type,
@@ -9542,7 +9631,8 @@ async fn get_object_source_once(
                     | DatabaseType::Uxdb
                     | DatabaseType::Vastbase
             )
-        }) {
+        })
+    {
         Some(false)
     } else {
         None

crates/dbx-core/src/sql.rs

@@ -246,6 +246,26 @@ impl SqlParsingOptions {
         Self::from_profile(SqlDialectProfile::for_database_type(db_type))
     }
 
+    pub fn for_database_type_and_compatibility(db_type: DatabaseType, compatibility_mode: Option<&str>) -> Self {
+        if db_type == DatabaseType::OpenGauss {
+            return match compatibility_mode.map(str::trim) {
+                // openGauss stores package specs/bodies as PL/SQL regardless of
+                // mode, and A mode is the only mode where the catalog reports
+                // them as packages. Any mode other than A keeps the PostgreSQL
+                // statement splitter.
+                Some(mode) if mode.eq_ignore_ascii_case("A") => Self::from_profile(SqlDialectProfile::gaussdb()),
+                Some(_) => Self::for_database_type(db_type),
+                // Unknown mode: the compatibility probe failed or the pool was
+                // unavailable. Falling back to the plain PostgreSQL profile would
+                // split an A-mode package body on its inner semicolons into
+                // fragments that the caller may then execute individually, so the
+                // conservative PL/SQL-capable profile is used instead.
+                None => Self::from_profile(SqlDialectProfile::gaussdb()),
+            };
+        }
+        Self::for_database_type(db_type)
+    }
+
     pub fn mysql_compatible() -> Self {
         Self::from_profile(SqlDialectProfile::mysql_compatible())
     }
@@ -677,6 +697,14 @@ pub fn split_sql_statements_for_database(sql: &str, db_type: DatabaseType) -> Ve
     sql_execution_plan_for_database(sql, db_type).statements
 }
 
+pub fn split_sql_statements_for_database_with_compatibility(
+    sql: &str,
+    db_type: DatabaseType,
+    compatibility_mode: Option<&str>,
+) -> Vec<String> {
+    sql_execution_plan_for_database_with_compatibility(sql, db_type, compatibility_mode).statements
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SqlExecutionPlan {
     pub statements: Vec<String>,
@@ -684,7 +712,21 @@ pub struct SqlExecutionPlan {
 }
 
 pub fn sql_execution_plan_for_database(sql: &str, db_type: DatabaseType) -> SqlExecutionPlan {
-    let options = SqlParsingOptions::for_database_type(db_type);
+    sql_execution_plan_with_options(sql, SqlParsingOptions::for_database_type(db_type))
+}
+
+pub fn sql_execution_plan_for_database_with_compatibility(
+    sql: &str,
+    db_type: DatabaseType,
+    compatibility_mode: Option<&str>,
+) -> SqlExecutionPlan {
+    sql_execution_plan_with_options(
+        sql,
+        SqlParsingOptions::for_database_type_and_compatibility(db_type, compatibility_mode),
+    )
+}
+
+fn sql_execution_plan_with_options(sql: &str, options: SqlParsingOptions) -> SqlExecutionPlan {
     if !options.profile.supports_psql_control_commands {
         return SqlExecutionPlan { statements: split_sql_statements_with_options(sql, options), stop_on_error: false };
     }
@@ -4338,6 +4380,23 @@ SELECT 1;";
         );
     }
 
+    #[test]
+    fn opengauss_a_mode_split_keeps_create_package_together() {
+        let sql = "CREATE OR REPLACE PACKAGE pkg_utils AS\n    FUNCTION get_version RETURN VARCHAR2;\n    PROCEDURE log_message(msg VARCHAR2);\nEND pkg_utils;\n/\nSELECT 1;";
+
+        assert_eq!(
+            super::split_sql_statements_for_database_with_compatibility(sql, DatabaseType::OpenGauss, Some("A")),
+            vec![
+                "CREATE OR REPLACE PACKAGE pkg_utils AS\n    FUNCTION get_version RETURN VARCHAR2;\n    PROCEDURE log_message(msg VARCHAR2);\nEND pkg_utils;",
+                "SELECT 1"
+            ]
+        );
+        assert_ne!(
+            super::split_sql_statements_for_database_with_compatibility(sql, DatabaseType::OpenGauss, Some("PG")),
+            super::split_sql_statements_for_database_with_compatibility(sql, DatabaseType::OpenGauss, Some("A"))
+        );
+    }
+
     #[test]
     fn xugu_split_keeps_create_package_body_together() {
         let sql = "\

crates/dbx-core/src/sql_file_import.rs

@@ -31,10 +31,43 @@ mod table_restore;
 pub use table_restore::SqlFileTable;
 use table_restore::TableRestoreFilter;
 
+/// How the database compatibility mode used for SQL statement splitting was
+/// resolved. `Option<String>` cannot express the difference between "this
+/// database has no compatibility mode concept" and "the probe failed", and
+/// conflating those silently downgraded openGauss A-mode PL/SQL splitting to
+/// the plain PostgreSQL splitter.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum SqlCompatibilityMode {
+    /// The database type has no compatibility-mode concept (non-openGauss).
+    NotApplicable,
+    /// The database is openGauss but the probe failed or the pool was
+    /// unavailable. The parser must not fall back to PostgreSQL semantics;
+    /// it uses the PL/SQL-capable openGauss profile so an A-mode package body
+    /// is never split on its inner semicolons.
+    Unknown,
+    /// Probe succeeded; carries the catalog-reported mode (may be B/C/PG/...).
+    Resolved(String),
+}
+
+impl SqlCompatibilityMode {
+    /// The mode string for [`SqlParsingOptions::for_database_type_and_compatibility`],
+    /// or `None` when the database has no mode concept or the mode is unknown.
+    ///
+    /// `None` is interpreted as "unknown openGauss" by the parser (see that
+    /// function), which selects the conservative PL/SQL-capable profile.
+    fn as_mode_str(&self) -> Option<&str> {
+        match self {
+            Self::Resolved(mode) => Some(mode.as_str()),
+            Self::Unknown | Self::NotApplicable => None,
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 struct SqlFileImportTarget {
     db_type: DatabaseType,
     driver_profile: Option<String>,
+    compatibility_mode: SqlCompatibilityMode,
 }
 
 #[derive(Debug)]
@@ -343,11 +376,12 @@ pub async fn execute_sql_file_content(
     started_at: Instant,
     mut emit: impl FnMut(SqlFileProgress),
 ) -> Result<(), String> {
-    let import_target = sql_file_import_target(state, &request.connection_id).await;
+    let import_target = sql_file_import_target(state, &request.connection_id, &request.database).await;
     validate_table_restore_target(request, import_target.as_ref(), 1)?;
     let mut statements = split_sql_file_import_statements_with_control(
         file_content,
         import_target.as_ref().map(|target| target.db_type),
+        import_target.as_ref().and_then(|target| target.compatibility_mode.as_mode_str()),
     );
 
     if let Some(selected) = &request.selected_tables {
@@ -413,9 +447,16 @@ pub async fn execute_sql_file_paths(
         return Err(error);
     }
 
-    let import_target = sql_file_import_target(state, &request.connection_id).await;
-    let options =
-        import_target.as_ref().map(|target| SqlParsingOptions::for_database_type(target.db_type)).unwrap_or_default();
+    let import_target = sql_file_import_target(state, &request.connection_id, &request.database).await;
+    let options = import_target
+        .as_ref()
+        .map(|target| {
+            SqlParsingOptions::for_database_type_and_compatibility(
+                target.db_type,
+                target.compatibility_mode.as_mode_str(),
+            )
+        })
+        .unwrap_or_default();
     let database_type = import_target.as_ref().map(|target| target.db_type);
     let mut progress = SqlFileExecutionProgress::new();
     // Validate the entire dump before any SQL reaches the database. This also
@@ -1300,7 +1341,7 @@ fn emit_sql_file_terminal_progress(
 
 #[cfg(test)]
 fn split_sql_file_import_statements(file_content: &str, db_type: Option<DatabaseType>) -> Vec<String> {
-    split_sql_file_import_statements_with_control(file_content, db_type)
+    split_sql_file_import_statements_with_control(file_content, db_type, None)
         .into_iter()
         .map(|statement| statement.sql)
         .collect()
@@ -1309,6 +1350,7 @@ fn split_sql_file_import_statements(file_content: &str, db_type: Option<Database
 fn split_sql_file_import_statements_with_control(
     file_content: &str,
     db_type: Option<DatabaseType>,
+    compatibility_mode: Option<&str>,
 ) -> Vec<SqlStatementWithControl> {
     if db_type == Some(DatabaseType::SqlServer) {
         // GO is a client-side batch delimiter, not T-SQL. SQL Server module DDL
@@ -1319,7 +1361,9 @@ fn split_sql_file_import_statements_with_control(
             .collect();
     }
 
-    let options = db_type.map(SqlParsingOptions::for_database_type).unwrap_or_default();
+    let options = db_type
+        .map(|db_type| SqlParsingOptions::for_database_type_and_compatibility(db_type, compatibility_mode))
+        .unwrap_or_default();
     let mut splitter = SqlStatementSplitter::with_options(options);
     let mut statements = splitter.push_chunk_with_control(file_content);
     statements.extend(splitter.finish_with_control());
@@ -1400,11 +1444,39 @@ fn sql_file_execution_error_progress(
     )
 }
 
-async fn sql_file_import_target(state: &AppState, connection_id: &str) -> Option<SqlFileImportTarget> {
-    let configs = state.configs.read().await;
-    configs
-        .get(connection_id)
-        .map(|config| SqlFileImportTarget { db_type: config.db_type, driver_profile: config.driver_profile.clone() })
+async fn sql_file_import_target(state: &AppState, connection_id: &str, database: &str) -> Option<SqlFileImportTarget> {
+    let config = state.configs.read().await.get(connection_id).cloned()?;
+    // Probing must never drop db_type: even when the pool is unavailable the file
+    // still uses the openGauss splitter. A failed probe is recorded as `Unknown`
+    // rather than `None` so the splitter keeps PL/SQL package bodies intact
+    // instead of silently falling back to PostgreSQL statement semantics.
+    let compatibility_mode = if config.db_type == DatabaseType::OpenGauss {
+        let pool = match state.get_or_create_pool(connection_id, Some(database)).await {
+            Ok(pool_key) => match state.pool_handle(&pool_key).await {
+                Some(PoolKind::Postgres(pool)) => Some(pool),
+                _ => None,
+            },
+            Err(_) => None,
+        };
+        match pool {
+            Some(pool) => match db::postgres::opengauss_compatibility_mode(&pool).await {
+                Ok(Some(mode)) => SqlCompatibilityMode::Resolved(mode),
+                // Catalog returned no row, or the probe errored (permissions,
+                // timeout, older kernel without pg_database.datcompatibility).
+                Ok(None) | Err(_) => {
+                    log::warn!(
+                        "[sql_file_import] openGauss compatibility mode probe failed for connection {connection_id}; \
+                         splitting with the conservative PL/SQL profile"
+                    );
+                    SqlCompatibilityMode::Unknown
+                }
+            },
+            None => SqlCompatibilityMode::Unknown,
+        }
+    } else {
+        SqlCompatibilityMode::NotApplicable
+    };
+    Some(SqlFileImportTarget { db_type: config.db_type, driver_profile: config.driver_profile, compatibility_mode })
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -2109,13 +2181,21 @@ mod tests {
         request.selected_tables = Some(vec![SqlFileTable { database: None, name: "a".to_string() }]);
         assert!(validate_table_restore_target(
             &request,
-            Some(&SqlFileImportTarget { db_type: DatabaseType::Postgres, driver_profile: None }),
+            Some(&SqlFileImportTarget {
+                db_type: DatabaseType::Postgres,
+                driver_profile: None,
+                compatibility_mode: SqlCompatibilityMode::NotApplicable
+            }),
             1
         )
         .is_err());
         assert!(validate_table_restore_target(
             &request,
-            Some(&SqlFileImportTarget { db_type: DatabaseType::Mysql, driver_profile: None }),
+            Some(&SqlFileImportTarget {
+                db_type: DatabaseType::Mysql,
+                driver_profile: None,
+                compatibility_mode: SqlCompatibilityMode::NotApplicable
+            }),
             2
         )
         .is_err());

crates/dbx-core/src/types.rs

@@ -17,6 +17,10 @@ pub struct DatabaseInfo {
     pub default_charset: Option<String>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub default_collation: Option<String>,
+    /// Database-level compatibility mode reported by compatible engines.
+    /// openGauss uses values such as A, B, C, and PG.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub compatibility_mode: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]