diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs
index abdd6ad7cd..1e69ede234 100644
--- a/crates/dbx-core/src/connection.rs
+++ b/crates/dbx-core/src/connection.rs
@@ -193,6 +193,15 @@ macro_rules! agent_connection_pool_database_type {
     };
 }
 
+#[derive(Clone)]
+enum GaussdbReservedKeywordsCacheEntry {
+    Available(Arc<HashSet<String>>),
+    /// The probe ran and deterministically found nothing usable (SQL-level
+    /// error, or an empty result) — cached because retrying it produces the
+    /// identical outcome every time for this server.
+    NotSupported,
+}
+
 pub struct AppState {
     pub connections: Arc<RwLock<HashMap<String, PoolKind>>>,
     task_supervisor: TaskSupervisor,
@@ -213,6 +222,13 @@ pub struct AppState {
     /// PostgreSQL TLS cancel context, keyed by pool_key.
     /// Used to reconstruct a TLS connector compatible with the original connection when cancelling.
     postgres_cancel_contexts: Arc<RwLock<HashMap<String, db::postgres::PostgresCancelContext>>>,
+    /// Live GaussDB/openGauss `pg_get_keywords()` reserved-word catalog, keyed
+    /// by pool_key. Populated lazily on first use per target connection
+    /// (t8y2/dbx#6283 follow-up) so identifier quoting during data transfer
+    /// reflects the actual server version instead of a hand-diffed static
+    /// list that can drift across GaussDB/openGauss releases (e.g. `maxvalue`
+    /// is reserved on openGauss 5.0 but not on current openGauss).
+    gaussdb_reserved_keywords_cache: Arc<RwLock<HashMap<String, GaussdbReservedKeywordsCacheEntry>>>,
     pub transaction_sessions: Arc<RwLock<HashMap<String, TransactionSession>>>,
     /// `save_password=false` 连接本次运行期的临时密码（内存，进程退出即丢，
     /// 绝不落盘）。键为 `(owner_scope, connection_id)`：桌面端 owner 为空串，
@@ -301,6 +317,7 @@ struct PoolRoutingControl {
     connections: Arc<RwLock<HashMap<String, PoolKind>>>,
     pool_activity: Arc<RwLock<HashMap<String, PoolActivity>>>,
     postgres_cancel_contexts: Arc<RwLock<HashMap<String, db::postgres::PostgresCancelContext>>>,
+    gaussdb_reserved_keywords_cache: Arc<RwLock<HashMap<String, GaussdbReservedKeywordsCacheEntry>>>,
     task_supervisor: TaskSupervisor,
 }
 
@@ -426,9 +443,11 @@ impl PoolRoutingControl {
         {
             let mut activity = self.pool_activity.write().await;
             let mut cancel_contexts = self.postgres_cancel_contexts.write().await;
+            let mut gaussdb_keywords = self.gaussdb_reserved_keywords_cache.write().await;
             for (key, _) in &removed {
                 activity.remove(key);
                 cancel_contexts.remove(key);
+                gaussdb_keywords.remove(key);
             }
         }
         self.close_removed_in_background(removed);
@@ -962,6 +981,7 @@ impl AppState {
             connections: self.connections.clone(),
             pool_activity: self.pool_activity.clone(),
             postgres_cancel_contexts: self.postgres_cancel_contexts.clone(),
+            gaussdb_reserved_keywords_cache: self.gaussdb_reserved_keywords_cache.clone(),
             task_supervisor: self.task_supervisor.clone(),
         }
     }
@@ -1083,6 +1103,7 @@ impl AppState {
             duckdb_worker_process_isolation: AtomicBool::new(false),
             duckdb_worker_max_processes: AtomicUsize::new(DUCKDB_WORKER_MAX_PROCESSES_DEFAULT),
             postgres_cancel_contexts: Arc::new(RwLock::new(HashMap::new())),
+            gaussdb_reserved_keywords_cache: Arc::new(RwLock::new(HashMap::new())),
             transaction_sessions: Arc::new(RwLock::new(HashMap::new())),
             session_credentials: SessionCredentialStore::new(),
             #[cfg(feature = "mq-admin")]
@@ -3278,6 +3299,7 @@ impl AppState {
         self.stop_keepalive_task(pool_key).await;
         self.pool_activity.write().await.remove(pool_key);
         self.postgres_cancel_contexts.write().await.remove(pool_key);
+        self.gaussdb_reserved_keywords_cache.write().await.remove(pool_key);
         let removed = self.connections.write().await.remove(pool_key);
         if let Some(pool) = removed {
             self.pool_routing_control().close_pool_with_timeout(pool_key.to_string(), pool).await;
@@ -3356,6 +3378,7 @@ impl AppState {
             self.stop_keepalive_task(&pool_key).await;
             self.pool_activity.write().await.remove(&pool_key);
             self.postgres_cancel_contexts.write().await.remove(&pool_key);
+            self.gaussdb_reserved_keywords_cache.write().await.remove(&pool_key);
             let removed = self.connections.write().await.remove(&pool_key);
             if let Some(pool) = removed {
                 self.pool_routing_control().close_pool_with_timeout(pool_key.clone(), pool).await;
@@ -3546,6 +3569,7 @@ impl AppState {
         self.stop_keepalive_task(&pool_key).await;
         self.pool_activity.write().await.remove(&pool_key);
         self.postgres_cancel_contexts.write().await.remove(&pool_key);
+        self.gaussdb_reserved_keywords_cache.write().await.remove(&pool_key);
         let removed = self.connections.write().await.remove(&pool_key);
         Ok(removed.map(|pool| (pool_key, pool)))
     }
@@ -3554,6 +3578,7 @@ impl AppState {
         self.stop_keepalive_task(pool_key).await;
         self.pool_activity.write().await.remove(pool_key);
         self.postgres_cancel_contexts.write().await.remove(pool_key);
+        self.gaussdb_reserved_keywords_cache.write().await.remove(pool_key);
         let removed = self.connections.write().await.remove(pool_key);
         if let Some(pool) = removed {
             self.pool_routing_control().close_pool_with_timeout(pool_key.to_string(), pool).await;
@@ -3622,6 +3647,7 @@ impl AppState {
 
         self.pool_activity.write().await.remove(pool_key);
         self.postgres_cancel_contexts.write().await.remove(pool_key);
+        self.gaussdb_reserved_keywords_cache.write().await.remove(pool_key);
         match close_reclaimed_agent_pool(pool).await {
             Ok(()) => true,
             Err((PoolKind::Agent(client), error)) if should_replace_agent_runtime(&error) => {
@@ -3700,9 +3726,11 @@ impl AppState {
         {
             let mut activity = self.pool_activity.write().await;
             let mut cancel_contexts = self.postgres_cancel_contexts.write().await;
+            let mut gaussdb_keywords = self.gaussdb_reserved_keywords_cache.write().await;
             for key in &keys_to_remove {
                 activity.remove(key);
                 cancel_contexts.remove(key);
+                gaussdb_keywords.remove(key);
             }
         }
         let mut conns = self.connections.write().await;
@@ -3833,6 +3861,55 @@ impl AppState {
         }
     }
 
+    /// Live GaussDB/openGauss reserved-keyword catalog for `pool_key`, queried
+    /// once via `pg_get_keywords()` and cached thereafter (t8y2/dbx#6283
+    /// follow-up), so data-transfer identifier quoting reflects the actual
+    /// target server version instead of a static list that can drift across
+    /// releases. Callers must only invoke this for targets already known to
+    /// be `DatabaseType::Gaussdb | DatabaseType::OpenGauss` — this method
+    /// itself only requires `pool_key` to resolve to `PoolKind::Postgres`
+    /// (the native pg-wire driver those types use); it does not re-check
+    /// db_type. JDBC/Agent/ExternalDriver GaussDB pool kinds are out of scope
+    /// and always fall back to `None` (the caller's static-list fallback).
+    ///
+    /// A deterministic outcome (the query ran and either failed or came back
+    /// empty — e.g. permission denied on `pg_get_keywords()`, or an engine
+    /// that doesn't have it) is cached too, so a multi-table transfer job
+    /// against a server that will never answer doesn't re-pay a connection
+    /// checkout + query round trip per table. A merely transient failure
+    /// (couldn't check out a connection, or the probe timed out) is NOT
+    /// cached and is retried on the next call.
+    pub async fn gaussdb_reserved_keywords(&self, pool_key: &str) -> Option<Arc<HashSet<String>>> {
+        if let Some(entry) = self.gaussdb_reserved_keywords_cache.read().await.get(pool_key) {
+            return match entry {
+                GaussdbReservedKeywordsCacheEntry::Available(words) => Some(words.clone()),
+                GaussdbReservedKeywordsCacheEntry::NotSupported => None,
+            };
+        }
+        let pool = match self.connections.read().await.get(pool_key) {
+            Some(PoolKind::Postgres(pool)) => pool.clone(),
+            _ => return None,
+        };
+        match db::postgres::gaussdb_reserved_keywords(&pool).await {
+            db::postgres::GaussdbReservedKeywordsProbe::Available(words) => {
+                let words = Arc::new(words);
+                self.gaussdb_reserved_keywords_cache
+                    .write()
+                    .await
+                    .insert(pool_key.to_string(), GaussdbReservedKeywordsCacheEntry::Available(words.clone()));
+                Some(words)
+            }
+            db::postgres::GaussdbReservedKeywordsProbe::NotSupported => {
+                self.gaussdb_reserved_keywords_cache
+                    .write()
+                    .await
+                    .insert(pool_key.to_string(), GaussdbReservedKeywordsCacheEntry::NotSupported);
+                None
+            }
+            db::postgres::GaussdbReservedKeywordsProbe::Retryable => None,
+        }
+    }
+
     pub async fn connection_database_info(
         &self,
         connection_id: &str,
@@ -4267,6 +4344,7 @@ impl AppState {
         self.pool_activity.write().await.clear();
         self.session_credentials.clear_pool_owners();
         self.postgres_cancel_contexts.write().await.clear();
+        self.gaussdb_reserved_keywords_cache.write().await.clear();
         self.draining_pools.lock().unwrap_or_else(|error| error.into_inner()).clear();
         self.connections.write().await.drain().collect()
     }
@@ -4299,9 +4377,11 @@ impl AppState {
         {
             let mut activity = self.pool_activity.write().await;
             let mut cancel_contexts = self.postgres_cancel_contexts.write().await;
+            let mut gaussdb_keywords = self.gaussdb_reserved_keywords_cache.write().await;
             for key in &keys_to_remove {
                 activity.remove(key);
                 cancel_contexts.remove(key);
+                gaussdb_keywords.remove(key);
             }
         }
         self.session_credentials.remove_pool_owners(&keys_to_remove);
@@ -4332,9 +4412,11 @@ impl AppState {
         {
             let mut activity = self.pool_activity.write().await;
             let mut cancel_contexts = self.postgres_cancel_contexts.write().await;
+            let mut gaussdb_keywords = self.gaussdb_reserved_keywords_cache.write().await;
             for key in &keys_to_remove {
                 activity.remove(key);
                 cancel_contexts.remove(key);
+                gaussdb_keywords.remove(key);
             }
         }
         let mut conns = self.connections.write().await;
diff --git a/crates/dbx-core/src/db/postgres.rs b/crates/dbx-core/src/db/postgres.rs
index ff7396ef4c..eb402198af 100644
--- a/crates/dbx-core/src/db/postgres.rs
+++ b/crates/dbx-core/src/db/postgres.rs
@@ -11,7 +11,7 @@ use rustls::server::ParsedCertificate;
 use sqlparser::ast::Statement;
 use sqlparser::dialect::PostgreSqlDialect;
 use sqlparser::parser::Parser;
-use std::collections::{BTreeSet, HashMap};
+use std::collections::{BTreeSet, HashMap, HashSet};
 use std::fs::File;
 use std::future::Future;
 use std::io::BufReader;
@@ -58,6 +58,59 @@ pub(crate) fn gaussdb_identifier_quote_for_compatibility_mode(compatibility_mode
     }
 }
 
+/// `pg_get_keywords()` catcode `R` (reserved) or `T` (reserved, can be
+/// function or type name) — the same criterion used to hand-derive the
+/// static GaussDB-only reserved-word list this function supersedes with a
+/// live, per-server-version result (t8y2/dbx#6283 follow-up).
+pub(crate) const GAUSSDB_RESERVED_KEYWORDS_SQL: &str =
+    "SELECT word FROM pg_get_keywords() WHERE catcode IN ('R','T')";
+
+/// Outcome of a live `pg_get_keywords()` probe. Distinguishes failures that
+/// are worth caching (the outcome will be identical on retry — e.g. the
+/// server-side query itself failed or came back empty) from failures that
+/// are likely transient (couldn't even check out a connection, or the probe
+/// timed out) and should be retried rather than pinned to "unavailable" for
+/// the rest of the connection's lifetime (t8y2/dbx#6283 follow-up).
+pub enum GaussdbReservedKeywordsProbe {
+    Available(HashSet<String>),
+    /// The query ran and either failed (e.g. permission denied on
+    /// `pg_get_keywords()`, or an engine that doesn't have the function) or
+    /// returned zero rows — a deterministic outcome for this server, safe to
+    /// cache so a multi-table transfer job doesn't re-pay a checkout + query
+    /// round trip per table for a server that will never answer.
+    NotSupported,
+    /// Couldn't check out a connection, or the probe timed out before the
+    /// server responded — likely transient, must NOT be cached.
+    Retryable,
+}
+
+/// Live catalog of GaussDB/openGauss reserved keywords for this exact server,
+/// so identifier quoting reflects the actual target version instead of a
+/// hand-diffed static list that can drift across releases (e.g. `maxvalue` is
+/// reserved on openGauss 5.0 but not on current openGauss).
+pub async fn gaussdb_reserved_keywords(pool: &Pool) -> GaussdbReservedKeywordsProbe {
+    let timeout = super::connection_timeout();
+    let client = match checkout_postgres_client(pool, None, timeout).await {
+        Ok(client) => client,
+        Err(_) => return GaussdbReservedKeywordsProbe::Retryable,
+    };
+    let query_result = match tokio::time::timeout(timeout, client.query(GAUSSDB_RESERVED_KEYWORDS_SQL, &[])).await {
+        Ok(result) => result,
+        Err(_) => return GaussdbReservedKeywordsProbe::Retryable,
+    };
+    let rows = match query_result {
+        Ok(rows) => rows,
+        Err(_) => return GaussdbReservedKeywordsProbe::NotSupported,
+    };
+    let words: HashSet<String> =
+        rows.iter().filter_map(|row| row.try_get::<_, String>(0).ok()).map(|w| w.to_ascii_lowercase()).collect();
+    if words.is_empty() {
+        GaussdbReservedKeywordsProbe::NotSupported
+    } else {
+        GaussdbReservedKeywordsProbe::Available(words)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PostgresTablePrivilegeInfo {
     pub grantor: String,
diff --git a/crates/dbx-core/src/sql_dialect.rs b/crates/dbx-core/src/sql_dialect.rs
index 062c7e380c..d4d9207613 100644
--- a/crates/dbx-core/src/sql_dialect.rs
+++ b/crates/dbx-core/src/sql_dialect.rs
@@ -37,7 +37,10 @@ pub use descriptor::{
 pub use identifiers::{
     normalize_where_input, qualified_table_name, qualified_table_name_with_catalog, quote_table_identifier,
 };
-pub(crate) use identifiers::{parse_sqlserver_linked_schema_ref, qualified_transfer_table, quote_transfer_identifier};
+pub(crate) use identifiers::{
+    parse_sqlserver_linked_schema_ref, qualified_transfer_table, qualified_transfer_table_with_gaussdb_keywords,
+    quote_transfer_identifier, quote_transfer_identifier_with_gaussdb_keywords,
+};
 pub use table_select::{
     build_count_table_sql, build_table_data_select_sql, build_table_select_sql, DBX_LARGE_VALUE_BYTES_COLUMN_PREFIX,
 };
diff --git a/crates/dbx-core/src/sql_dialect/identifiers.rs b/crates/dbx-core/src/sql_dialect/identifiers.rs
index 6ed24beb31..b6e4ac3787 100644
--- a/crates/dbx-core/src/sql_dialect/identifiers.rs
+++ b/crates/dbx-core/src/sql_dialect/identifiers.rs
@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use crate::models::connection::DatabaseType;
 use percent_encoding::percent_decode_str;
 
@@ -147,7 +149,11 @@ pub fn quote_table_identifier(database_type: Option<DatabaseType>, name: &str) -
     }
 }
 
-pub(crate) fn quote_gaussdb_jdbc_identifier(name: &str, identifier_quote: &str) -> String {
+pub(crate) fn quote_gaussdb_jdbc_identifier(
+    name: &str,
+    identifier_quote: &str,
+    database_type: Option<DatabaseType>,
+) -> String {
     if is_explicitly_quoted_identifier(name) {
         return name.to_string();
     }
@@ -157,6 +163,8 @@ pub(crate) fn quote_gaussdb_jdbc_identifier(name: &str, identifier_quote: &str)
     }
     let requires_quote = !is_simple_lower_identifier(name)
         || is_postgres_reserved_identifier(name)
+        || (matches!(database_type, Some(DatabaseType::Gaussdb | DatabaseType::OpenGauss))
+            && is_gaussdb_only_reserved_identifier(name))
         || (quote == "`" && is_mysql_only_reserved_identifier(name));
     if !requires_quote {
         return name.to_string();
@@ -287,6 +295,60 @@ fn is_postgres_reserved_identifier(name: &str) -> bool {
     )
 }
 
+/// Words GaussDB/openGauss reserve that plain PostgreSQL does not, so they
+/// are missed by [`is_postgres_reserved_identifier`] alone (t8y2/dbx#6283).
+/// For example `compact` is a bare lowercase identifier that passes
+/// [`is_simple_lower_identifier`] and isn't a Postgres keyword, but GaussDB
+/// reserves it and rejects it unquoted in DDL.
+///
+/// An initial cut of this list was diffed from Huawei's GaussDB(DWS)
+/// keyword reference, but DWS is the MPP/columnar variant and isn't a
+/// reliable proxy for the core engine's catalog. It was cross-checked
+/// against a writable openGauss 5.0.0 instance's own `pg_get_keywords()`,
+/// which is the authoritative source per openGauss's docs. That check
+/// found 9 of the original words are not actually reserved on core
+/// GaussDB/openGauss — `hot`, `nlssort`, and `warmup` aren't keywords at
+/// all, and `fenced`, `internal`, `plan`, `tsfield`, `tstag`, and `tstime`
+/// are `unreserved` — so quoting them would have reintroduced the
+/// case-locking bug this PR fixes. They've been dropped. `compact` was
+/// additionally confirmed by running `CREATE TABLE ... (compact int)`
+/// unquoted, which fails with `ERROR: syntax error at or near "compact"`.
+///
+/// A follow-up review on #6283 diffed this list against the *full*
+/// `pg_get_keywords()` result set from that same instance (653 rows) rather
+/// than spot-checking individual words, and found 7 more reserved words
+/// this hand-picked list had missed: `csn`, `excluded`, `groupparent`,
+/// `nocycle`, `rownum`, `shrink`, `verify`. All 22 words below have
+/// `catcode` `R` (`reserved`) or `T` (`reserved, can be function or type
+/// name`) on that instance.
+fn is_gaussdb_only_reserved_identifier(name: &str) -> bool {
+    matches!(
+        name,
+        "authid"
+            | "buckets"
+            | "compact"
+            | "csn"
+            | "deltamerge"
+            | "excluded"
+            | "groupparent"
+            | "hdfsdirectory"
+            | "less"
+            | "maxvalue"
+            | "minus"
+            | "modify"
+            | "nocycle"
+            | "performance"
+            | "procedure"
+            | "recyclebin"
+            | "reject"
+            | "rownum"
+            | "shrink"
+            | "sysdate"
+            | "timecapsule"
+            | "verify"
+    )
+}
+
 fn is_mysql_only_reserved_identifier(name: &str) -> bool {
     matches!(
         name,
@@ -374,6 +436,24 @@ pub fn normalize_where_input(where_input: Option<&str>) -> String {
 }
 
 pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType) -> String {
+    quote_transfer_identifier_with_gaussdb_keywords(name, database_type, None)
+}
+
+/// Like [`quote_transfer_identifier`], but for GaussDB/OpenGauss targets
+/// accepts a live, server-version-specific reserved-keyword catalog
+/// (queried once per connection via `pg_get_keywords()`, see
+/// `AppState::gaussdb_reserved_keywords`) to use INSTEAD of the static
+/// [`is_gaussdb_only_reserved_identifier`] list — never unioned with it,
+/// since the whole point is that the live catalog is authoritative for that
+/// exact server and must not force-quote a word (e.g. `maxvalue`) that isn't
+/// actually reserved on this version (t8y2/dbx#6283 follow-up). When `None`
+/// (no live probe available), falls back to the static list unchanged.
+/// `is_postgres_reserved_identifier` always applies regardless.
+pub(crate) fn quote_transfer_identifier_with_gaussdb_keywords(
+    name: &str,
+    database_type: &DatabaseType,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
     match database_type {
         DatabaseType::Mysql
         | DatabaseType::ClickHouse
@@ -385,10 +465,39 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType
         | DatabaseType::Spark
         | DatabaseType::Questdb => format!("`{}`", name.replace('`', "``")),
         DatabaseType::SqlServer => format!("[{}]", name.replace(']', "]]")),
+        _ if needs_conditional_quoting_for_gaussdb_family(database_type) => {
+            let is_target_only_reserved = match gaussdb_keywords {
+                Some(live) => live.contains(name),
+                None => is_gaussdb_only_reserved_identifier(name),
+            };
+            if is_simple_lower_identifier(name) && !is_postgres_reserved_identifier(name) && !is_target_only_reserved
+            {
+                name.to_string()
+            } else {
+                format!("\"{}\"", name.replace('\"', "\"\""))
+            }
+        }
         _ => format!("\"{}\"", name.replace('\"', "\"\"")),
     }
 }
 
+/// GaussDB/OpenGauss can run in an Oracle-compatible mode where *unquoted*
+/// identifiers fold to uppercase instead of Postgres' lowercase. Transfer
+/// always quoting every identifier therefore locks in the source's exact
+/// (often lowercase) case; a later unquoted reference from the user's own
+/// tooling then folds to a different case and the target reports "column
+/// does not exist" even though the migrated column is right there
+/// (t8y2/dbx#6205). Only quote when the name actually needs it (mixed case,
+/// special characters, or a reserved word) — matching the same heuristic
+/// already used for GaussDB JDBC identifier quoting in
+/// `quote_gaussdb_jdbc_identifier` above. Deliberately scoped to just
+/// GaussDB/OpenGauss rather than the wider Postgres family (see
+/// `transfer::is_postgres_family_target`): those other engines have no
+/// reported case-folding quirk, and always-quoting them is harmless.
+fn needs_conditional_quoting_for_gaussdb_family(database_type: &DatabaseType) -> bool {
+    matches!(database_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss)
+}
+
 /// Qualified table name for transfer SQL.
 ///
 /// * Without catalog: produces `schema.table` (or just `table` for MySQL family).
@@ -401,12 +510,25 @@ pub(crate) fn qualified_transfer_table(
     database_type: &DatabaseType,
     catalog: Option<&str>,
 ) -> String {
-    let table = quote_transfer_identifier(table_name, database_type);
+    qualified_transfer_table_with_gaussdb_keywords(table_name, schema, database_type, catalog, None)
+}
+
+/// Like [`qualified_transfer_table`], but threads a live GaussDB/OpenGauss
+/// reserved-keyword catalog through to [`quote_transfer_identifier_with_gaussdb_keywords`]
+/// for each part of the qualified name (t8y2/dbx#6283 follow-up).
+pub(crate) fn qualified_transfer_table_with_gaussdb_keywords(
+    table_name: &str,
+    schema: &str,
+    database_type: &DatabaseType,
+    catalog: Option<&str>,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
+    let table = quote_transfer_identifier_with_gaussdb_keywords(table_name, database_type, gaussdb_keywords);
     if let Some(catalog) = catalog {
         format!(
             "{}.{}.{}",
-            quote_transfer_identifier(catalog, database_type),
-            quote_transfer_identifier(schema, database_type),
+            quote_transfer_identifier_with_gaussdb_keywords(catalog, database_type, gaussdb_keywords),
+            quote_transfer_identifier_with_gaussdb_keywords(schema, database_type, gaussdb_keywords),
             table
         )
     } else if schema.is_empty()
@@ -414,6 +536,6 @@ pub(crate) fn qualified_transfer_table(
     {
         table
     } else {
-        format!("{}.{}", quote_transfer_identifier(schema, database_type), table)
+        format!("{}.{}", quote_transfer_identifier_with_gaussdb_keywords(schema, database_type, gaussdb_keywords), table)
     }
 }
diff --git a/crates/dbx-core/src/sql_dialect/table_select.rs b/crates/dbx-core/src/sql_dialect/table_select.rs
index 5070f5141f..019612bf0a 100644
--- a/crates/dbx-core/src/sql_dialect/table_select.rs
+++ b/crates/dbx-core/src/sql_dialect/table_select.rs
@@ -364,7 +364,7 @@ pub(crate) fn quote_table_data_identifier(
         return quote_table_identifier(database_type, name);
     };
     if matches!(database_type, Some(DatabaseType::Gaussdb | DatabaseType::OpenGauss | DatabaseType::Postgres)) {
-        return quote_gaussdb_jdbc_identifier(name, quote);
+        return quote_gaussdb_jdbc_identifier(name, quote, database_type);
     }
     if quote.is_empty() {
         return name.to_string();
diff --git a/crates/dbx-core/src/sql_dialect/tests.rs b/crates/dbx-core/src/sql_dialect/tests.rs
index 738987d2b7..c18f9e4bba 100644
--- a/crates/dbx-core/src/sql_dialect/tests.rs
+++ b/crates/dbx-core/src/sql_dialect/tests.rs
@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use super::*;
 use crate::models::connection::DatabaseType;
 
@@ -61,6 +63,130 @@ fn quotes_gaussdb_jdbc_identifiers_selectively() {
     }
 }
 
+#[test]
+fn quotes_gaussdb_only_reserved_words_not_shared_with_postgres() {
+    // t8y2/dbx#6283: `compact` and friends are reserved in GaussDB/openGauss
+    // but are not PostgreSQL keywords, so a check that only consults the
+    // Postgres reserved-word set misses them and would emit invalid unquoted
+    // DDL on the real target. This word list is cross-checked against a
+    // writable openGauss 5.0.0 instance's own `pg_get_keywords()` (not just
+    // Huawei's GaussDB(DWS) doc) — see the comment on
+    // `is_gaussdb_only_reserved_identifier` for how the DWS-derived list was
+    // trimmed down to these instance-confirmed words. `csn` through `verify`
+    // below were caught by diffing against the instance's full 653-row
+    // keyword catalog rather than spot-checking individual words.
+    for name in [
+        "authid",
+        "buckets",
+        "compact",
+        "csn",
+        "deltamerge",
+        "excluded",
+        "groupparent",
+        "hdfsdirectory",
+        "less",
+        "maxvalue",
+        "minus",
+        "modify",
+        "nocycle",
+        "performance",
+        "procedure",
+        "recyclebin",
+        "reject",
+        "rownum",
+        "shrink",
+        "sysdate",
+        "timecapsule",
+        "verify",
+    ] {
+        assert_eq!(
+            quote_table_data_identifier(Some(DatabaseType::Gaussdb), name, Some("\"")),
+            format!("\"{name}\""),
+            "GaussDB must quote target-only reserved word `{name}`"
+        );
+        assert_eq!(
+            quote_table_data_identifier(Some(DatabaseType::OpenGauss), name, Some("\"")),
+            format!("\"{name}\""),
+            "OpenGauss must quote target-only reserved word `{name}`"
+        );
+    }
+
+    // The same words are not reserved in real PostgreSQL, so the dialect-aware
+    // check must not over-quote a genuine Postgres target.
+    for name in ["compact", "buckets", "sysdate", "minus", "modify", "excluded", "rownum", "verify"] {
+        assert_eq!(
+            quote_table_data_identifier(Some(DatabaseType::Postgres), name, Some("\"")),
+            name,
+            "plain Postgres must not quote `{name}` — it is not a Postgres keyword"
+        );
+    }
+
+    // Words that were in the original DWS-derived list but turned out not to
+    // be reserved on a real GaussDB/openGauss instance must not be quoted —
+    // over-quoting them would reintroduce the case-locking bug this PR fixes.
+    for name in ["fenced", "hot", "internal", "nlssort", "plan", "tsfield", "tstag", "tstime", "warmup"] {
+        assert_eq!(
+            quote_table_data_identifier(Some(DatabaseType::Gaussdb), name, Some("\"")),
+            name,
+            "GaussDB must not quote `{name}` — confirmed not reserved on a real instance"
+        );
+    }
+}
+
+#[test]
+fn transfer_identifier_gaussdb_keywords_live_catalog_overrides_static_list() {
+    // t8y2/dbx#6283 follow-up: `maxvalue` is RESERVED_KEYWORD on openGauss
+    // 5.0 (per its kwlist.h) but UNRESERVED_KEYWORD on current/master
+    // openGauss — so a live pg_get_keywords() catalog from each version must
+    // produce different quoting for the exact same word, unlike the static
+    // list this replaces (which permanently quotes `maxvalue` everywhere).
+    let opengauss_5_0: HashSet<String> = ["maxvalue".to_string()].into_iter().collect();
+    assert_eq!(
+        quote_transfer_identifier_with_gaussdb_keywords("maxvalue", &DatabaseType::OpenGauss, Some(&opengauss_5_0)),
+        "\"maxvalue\"",
+        "openGauss 5.0's live catalog reserves maxvalue"
+    );
+
+    // A live catalog from current/master openGauss would not list `maxvalue`
+    // as reserved (some unrelated word stands in for "the rest of the real
+    // catalog" here).
+    let opengauss_current: HashSet<String> = ["compact".to_string()].into_iter().collect();
+    assert_eq!(
+        quote_transfer_identifier_with_gaussdb_keywords(
+            "maxvalue",
+            &DatabaseType::OpenGauss,
+            Some(&opengauss_current)
+        ),
+        "maxvalue",
+        "current openGauss's live catalog does not reserve maxvalue — must stay unquoted"
+    );
+}
+
+#[test]
+fn transfer_identifier_gaussdb_keywords_none_falls_back_to_static_list() {
+    // No live catalog available (probe failed, connection error, or an
+    // engine without pg_get_keywords) — must preserve the existing
+    // static-list behavior unchanged, still quoting `maxvalue`.
+    assert_eq!(
+        quote_transfer_identifier_with_gaussdb_keywords("maxvalue", &DatabaseType::Gaussdb, None),
+        "\"maxvalue\""
+    );
+    assert_eq!(quote_transfer_identifier("maxvalue", &DatabaseType::Gaussdb), "\"maxvalue\"");
+    assert_eq!(quote_transfer_identifier("maxvalue", &DatabaseType::OpenGauss), "\"maxvalue\"");
+}
+
+#[test]
+fn transfer_identifier_gaussdb_keywords_core_postgres_reserved_words_unaffected_by_live_catalog() {
+    // Core Postgres reserved words must stay quoted regardless of what the
+    // live GaussDB-only catalog contains — the live catalog only replaces
+    // `is_gaussdb_only_reserved_identifier`, never `is_postgres_reserved_identifier`.
+    let live_catalog_without_select: HashSet<String> = HashSet::new();
+    assert_eq!(
+        quote_transfer_identifier_with_gaussdb_keywords("select", &DatabaseType::Gaussdb, Some(&live_catalog_without_select)),
+        "\"select\""
+    );
+}
+
 #[test]
 fn qualifies_schema_only_for_schema_aware_databases() {
     assert_eq!(qualified_table_name(Some(DatabaseType::Postgres), Some("public"), "users"), "\"public\".\"users\"");
diff --git a/crates/dbx-core/src/transfer.rs b/crates/dbx-core/src/transfer.rs
index 4b2193553f..15bf03ff1c 100644
--- a/crates/dbx-core/src/transfer.rs
+++ b/crates/dbx-core/src/transfer.rs
@@ -13,7 +13,10 @@ use crate::query::{
     agent_execute_query_params, pool_error_action, PoolErrorAction, QueryExecutionOptions, AGENT_PROTOCOL_MAX_ROWS,
 };
 use crate::sql::{split_sql_statements, split_sql_statements_for_database};
-use crate::sql_dialect::{qualified_transfer_table, quote_transfer_identifier};
+use crate::sql_dialect::{
+    qualified_transfer_table, qualified_transfer_table_with_gaussdb_keywords, quote_transfer_identifier,
+    quote_transfer_identifier_with_gaussdb_keywords,
+};
 
 static CANCELLED: std::sync::LazyLock<RwLock<HashSet<String>>> =
     std::sync::LazyLock::new(|| RwLock::new(HashSet::new()));
@@ -235,6 +238,14 @@ pub fn quote_identifier(name: &str, db_type: &DatabaseType) -> String {
     quote_transfer_identifier(name, db_type)
 }
 
+pub(crate) fn quote_identifier_with_gaussdb_keywords(
+    name: &str,
+    db_type: &DatabaseType,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
+    quote_transfer_identifier_with_gaussdb_keywords(name, db_type, gaussdb_keywords)
+}
+
 fn quote_identifier_with_identifier_quote(
     name: &str,
     db_type: &DatabaseType,
@@ -317,6 +328,44 @@ pub fn qualified_table(table: &str, schema: &str, db_type: &DatabaseType, catalo
     qualified_transfer_table(table, schema, db_type, effective_catalog)
 }
 
+pub(crate) fn qualified_table_with_gaussdb_keywords(
+    table: &str,
+    schema: &str,
+    db_type: &DatabaseType,
+    catalog: Option<&str>,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
+    let effective_catalog = resolve_external_transfer_catalog(catalog, db_type);
+    qualified_transfer_table_with_gaussdb_keywords(table, schema, db_type, effective_catalog, gaussdb_keywords)
+}
+
+/// SQL to clear the target table before an `Overwrite`-mode transfer,
+/// shared by the SQL-source (`transfer_table`) and MongoDB-source
+/// (`transfer_mongodb_table`) write paths. Must use the `_with_gaussdb_keywords`
+/// variant of `qualified_table` — using the plain static-list version here
+/// while CREATE TABLE uses the live catalog would let the two statements
+/// disagree on whether to quote the same table name, reintroducing the
+/// case-locking bug this PR fixes (t8y2/dbx#6283 follow-up).
+fn transfer_overwrite_clear_sql(
+    target_table: &str,
+    target_schema: &str,
+    target_db_type: &DatabaseType,
+    target_catalog: Option<&str>,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
+    let full_table = qualified_table_with_gaussdb_keywords(
+        target_table,
+        target_schema,
+        target_db_type,
+        target_catalog,
+        gaussdb_keywords,
+    );
+    match target_db_type {
+        DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb => format!("DELETE FROM {full_table}"),
+        _ => format!("TRUNCATE TABLE {full_table}"),
+    }
+}
+
 fn qualified_table_with_identifier_quote(
     table: &str,
     schema: &str,
@@ -2829,6 +2878,7 @@ fn parse_mysql_row_error(error: &str) -> Option<u64> {
     at_row.trim().parse::<u64>().ok()
 }
 
+#[allow(clippy::too_many_arguments)]
 pub fn generate_create_table_ddl(
     columns: &[db::ColumnInfo],
     table: &str,
@@ -2839,7 +2889,32 @@ pub fn generate_create_table_ddl(
     table_comment: Option<&str>,
     catalog: Option<&str>,
 ) -> String {
-    let full_table = qualified_table(table, schema, target_db, catalog);
+    generate_create_table_ddl_with_gaussdb_keywords(
+        columns,
+        table,
+        source_schema,
+        schema,
+        target_db,
+        source_db,
+        table_comment,
+        catalog,
+        None,
+    )
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn generate_create_table_ddl_with_gaussdb_keywords(
+    columns: &[db::ColumnInfo],
+    table: &str,
+    source_schema: &str,
+    schema: &str,
+    target_db: &DatabaseType,
+    source_db: &DatabaseType,
+    table_comment: Option<&str>,
+    catalog: Option<&str>,
+    gaussdb_keywords: Option<&HashSet<String>>,
+) -> String {
+    let full_table = qualified_table_with_gaussdb_keywords(table, schema, target_db, catalog, gaussdb_keywords);
 
     let is_mysql_family = matches!(
         target_db,
@@ -2854,7 +2929,8 @@ pub fn generate_create_table_ddl(
     for c in columns {
         col_lines.push({
             let mapped_type = postgres_column_type_sql(c, source_schema, schema, source_db, target_db);
-            let mut line = format!("  {} {}", quote_identifier(&c.name, target_db), mapped_type);
+            let mut line =
+                format!("  {} {}", quote_identifier_with_gaussdb_keywords(&c.name, target_db, gaussdb_keywords), mapped_type);
             if let Some(default_clause) = column_default_clause(c, source_schema, schema, source_db, target_db) {
                 line.push(' ');
                 line.push_str(&default_clause);
@@ -2886,7 +2962,7 @@ pub fn generate_create_table_ddl(
     if !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
         for c in columns {
             if c.is_primary_key {
-                let qname = quote_identifier(&c.name, target_db);
+                let qname = quote_identifier_with_gaussdb_keywords(&c.name, target_db, gaussdb_keywords);
                 if is_mysql_family {
                     let mapped = map_column_type(&c.data_type, source_db, target_db);
                     if mysql_type_needs_key_prefix(&mapped) {
@@ -3039,8 +3115,25 @@ impl InsertSqlTemplate {
         catalog: Option<&str>,
         overrides_postgres_system_values: bool,
     ) -> Self {
-        let full_table = qualified_table(table, schema, db_type, catalog);
-        let col_list = columns.iter().map(|column| quote_identifier(column, db_type)).collect::<Vec<_>>().join(", ");
+        Self::new_with_gaussdb_keywords(columns, table, schema, db_type, catalog, overrides_postgres_system_values, None)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn new_with_gaussdb_keywords(
+        columns: &[String],
+        table: &str,
+        schema: &str,
+        db_type: &DatabaseType,
+        catalog: Option<&str>,
+        overrides_postgres_system_values: bool,
+        gaussdb_keywords: Option<&HashSet<String>>,
+    ) -> Self {
+        let full_table = qualified_table_with_gaussdb_keywords(table, schema, db_type, catalog, gaussdb_keywords);
+        let col_list = columns
+            .iter()
+            .map(|column| quote_identifier_with_gaussdb_keywords(column, db_type, gaussdb_keywords))
+            .collect::<Vec<_>>()
+            .join(", ");
         let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
             " OVERRIDING SYSTEM VALUE"
         } else {
@@ -3169,13 +3262,46 @@ fn generate_upsert_typed_for_transfer(
     catalog: Option<&str>,
     overrides_postgres_system_values: bool,
     mysql_spatial_markers: bool,
+) -> String {
+    generate_upsert_typed_for_transfer_with_gaussdb_keywords(
+        columns,
+        column_types,
+        rows,
+        table,
+        schema,
+        db_type,
+        pk_columns,
+        catalog,
+        overrides_postgres_system_values,
+        mysql_spatial_markers,
+        None,
+    )
+}
+
+#[allow(clippy::too_many_arguments)]
+fn generate_upsert_typed_for_transfer_with_gaussdb_keywords(
+    columns: &[String],
+    column_types: &[Option<String>],
+    rows: &[Vec<serde_json::Value>],
+    table: &str,
+    schema: &str,
+    db_type: &DatabaseType,
+    pk_columns: &[String],
+    catalog: Option<&str>,
+    overrides_postgres_system_values: bool,
+    mysql_spatial_markers: bool,
+    gaussdb_keywords: Option<&HashSet<String>>,
 ) -> String {
     if rows.is_empty() || pk_columns.is_empty() {
         return String::new();
     }
 
-    let full_table = qualified_table(table, schema, db_type, catalog);
-    let col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
+    let full_table = qualified_table_with_gaussdb_keywords(table, schema, db_type, catalog, gaussdb_keywords);
+    let col_list = columns
+        .iter()
+        .map(|c| quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords))
+        .collect::<Vec<_>>()
+        .join(", ");
 
     let value_rows = value_rows_sql(rows, column_types, db_type, mysql_spatial_markers);
 
@@ -3191,7 +3317,11 @@ fn generate_upsert_typed_for_transfer(
             if is_postgres_transfer_dialect(db_type)
                 || matches!(db_type, DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb) =>
         {
-            let pk_list = pk_columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
+            let pk_list = pk_columns
+                .iter()
+                .map(|c| quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords))
+                .collect::<Vec<_>>()
+                .join(", ");
             let overriding = if overrides_postgres_system_values && matches!(db_type, DatabaseType::Postgres) {
                 " OVERRIDING SYSTEM VALUE"
             } else {
@@ -3205,7 +3335,7 @@ fn generate_upsert_typed_for_transfer(
                 let update_set = non_pk_columns
                     .iter()
                     .map(|c| {
-                        let qc = quote_identifier(c, db_type);
+                        let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                         format!("{qc} = EXCLUDED.{qc}")
                     })
                     .collect::<Vec<_>>()
@@ -3218,13 +3348,13 @@ fn generate_upsert_typed_for_transfer(
             let mut sql = format!("INSERT INTO {full_table} ({col_list}) VALUES\n{}", value_rows.join(",\n"));
             if non_pk_columns.is_empty() {
                 sql.push_str("\nON DUPLICATE KEY UPDATE ");
-                let first_pk = quote_identifier(&pk_columns[0], db_type);
+                let first_pk = quote_identifier_with_gaussdb_keywords(&pk_columns[0], db_type, gaussdb_keywords);
                 sql.push_str(&format!("{first_pk} = {first_pk}"));
             } else {
                 let update_set = non_pk_columns
                     .iter()
                     .map(|c| {
-                        let qc = quote_identifier(c, db_type);
+                        let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                         format!("{qc} = VALUES({qc})")
                     })
                     .collect::<Vec<_>>()
@@ -3234,11 +3364,15 @@ fn generate_upsert_typed_for_transfer(
             sql
         }
         DatabaseType::SqlServer => {
-            let src_col_list = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
+            let src_col_list = columns
+                .iter()
+                .map(|c| quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords))
+                .collect::<Vec<_>>()
+                .join(", ");
             let on_clause = pk_columns
                 .iter()
                 .map(|c| {
-                    let qc = quote_identifier(c, db_type);
+                    let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                     format!("target.{qc} = src.{qc}")
                 })
                 .collect::<Vec<_>>()
@@ -3253,7 +3387,7 @@ fn generate_upsert_typed_for_transfer(
                 let update_set = non_pk_columns
                     .iter()
                     .map(|c| {
-                        let qc = quote_identifier(c, db_type);
+                        let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                         format!("target.{qc} = src.{qc}")
                     })
                     .collect::<Vec<_>>()
@@ -3261,9 +3395,16 @@ fn generate_upsert_typed_for_transfer(
                 sql.push_str(&format!("\nWHEN MATCHED THEN UPDATE SET {update_set}"));
             }
 
-            let insert_cols = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
-            let insert_vals =
-                columns.iter().map(|c| format!("src.{}", quote_identifier(c, db_type))).collect::<Vec<_>>().join(", ");
+            let insert_cols = columns
+                .iter()
+                .map(|c| quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords))
+                .collect::<Vec<_>>()
+                .join(", ");
+            let insert_vals = columns
+                .iter()
+                .map(|c| format!("src.{}", quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords)))
+                .collect::<Vec<_>>()
+                .join(", ");
             sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals});"));
             sql
         }
@@ -3275,7 +3416,7 @@ fn generate_upsert_typed_for_transfer(
                     vals.push(format!(
                         "{} AS {}",
                         escape_value_typed(v, db_type, column_types.get(index).and_then(|value| value.as_deref())),
-                        quote_identifier(c, db_type)
+                        quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords)
                     ));
                 }
                 using_rows.push(format!("SELECT {} FROM dual", vals.join(", ")));
@@ -3284,7 +3425,7 @@ fn generate_upsert_typed_for_transfer(
             let on_clause = pk_columns
                 .iter()
                 .map(|c| {
-                    let qc = quote_identifier(c, db_type);
+                    let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                     format!("t.{qc} = s.{qc}")
                 })
                 .collect::<Vec<_>>()
@@ -3297,7 +3438,7 @@ fn generate_upsert_typed_for_transfer(
                 let update_set = non_pk_columns
                     .iter()
                     .map(|c| {
-                        let qc = quote_identifier(c, db_type);
+                        let qc = quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords);
                         format!("t.{qc} = s.{qc}")
                     })
                     .collect::<Vec<_>>()
@@ -3305,14 +3446,22 @@ fn generate_upsert_typed_for_transfer(
                 sql.push_str(&format!("\nWHEN MATCHED THEN UPDATE SET {update_set}"));
             }
 
-            let insert_cols = columns.iter().map(|c| quote_identifier(c, db_type)).collect::<Vec<_>>().join(", ");
-            let insert_vals =
-                columns.iter().map(|c| format!("s.{}", quote_identifier(c, db_type))).collect::<Vec<_>>().join(", ");
+            let insert_cols = columns
+                .iter()
+                .map(|c| quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords))
+                .collect::<Vec<_>>()
+                .join(", ");
+            let insert_vals = columns
+                .iter()
+                .map(|c| format!("s.{}", quote_identifier_with_gaussdb_keywords(c, db_type, gaussdb_keywords)))
+                .collect::<Vec<_>>()
+                .join(", ");
             sql.push_str(&format!("\nWHEN NOT MATCHED THEN INSERT ({insert_cols}) VALUES ({insert_vals})"));
             sql
         }
         _ => {
-            let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, false);
+            let template =
+                InsertSqlTemplate::new_with_gaussdb_keywords(columns, table, schema, db_type, catalog, false, gaussdb_keywords);
             template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers))
         }
     }
@@ -3442,9 +3591,10 @@ fn generate_transfer_write_sql(
     catalog: Option<&str>,
     overrides_postgres_system_values: bool,
     mysql_spatial_markers: bool,
+    gaussdb_keywords: Option<&HashSet<String>>,
 ) -> String {
     match mode {
-        TransferMode::Upsert => generate_upsert_typed_for_transfer(
+        TransferMode::Upsert => generate_upsert_typed_for_transfer_with_gaussdb_keywords(
             columns,
             column_types,
             rows,
@@ -3455,13 +3605,21 @@ fn generate_transfer_write_sql(
             catalog,
             overrides_postgres_system_values,
             mysql_spatial_markers,
+            gaussdb_keywords,
         ),
         _ => {
             if rows.is_empty() {
                 return String::new();
             }
-            let template =
-                InsertSqlTemplate::new(columns, table, schema, db_type, catalog, overrides_postgres_system_values);
+            let template = InsertSqlTemplate::new_with_gaussdb_keywords(
+                columns,
+                table,
+                schema,
+                db_type,
+                catalog,
+                overrides_postgres_system_values,
+                gaussdb_keywords,
+            );
             template.build(&value_rows_sql(rows, column_types, db_type, mysql_spatial_markers))
         }
     }
@@ -3504,6 +3662,35 @@ fn generate_insert_typed_sql_batches_for_transfer(
     limits: SqlBatchLimits,
     overrides_postgres_system_values: bool,
     mysql_spatial_markers: bool,
+) -> Result<Vec<(String, usize)>, String> {
+    generate_insert_typed_sql_batches_for_transfer_with_gaussdb_keywords(
+        columns,
+        column_types,
+        rows,
+        table,
+        schema,
+        db_type,
+        catalog,
+        limits,
+        overrides_postgres_system_values,
+        mysql_spatial_markers,
+        None,
+    )
+}
+
+#[allow(clippy::too_many_arguments)]
+fn generate_insert_typed_sql_batches_for_transfer_with_gaussdb_keywords(
+    columns: &[String],
+    column_types: &[Option<String>],
+    rows: &[Vec<serde_json::Value>],
+    table: &str,
+    schema: &str,
+    db_type: &DatabaseType,
+    catalog: Option<&str>,
+    limits: SqlBatchLimits,
+    overrides_postgres_system_values: bool,
+    mysql_spatial_markers: bool,
+    gaussdb_keywords: Option<&HashSet<String>>,
 ) -> Result<Vec<(String, usize)>, String> {
     if rows.is_empty() {
         return Ok(Vec::new());
@@ -3516,7 +3703,15 @@ fn generate_insert_typed_sql_batches_for_transfer(
     });
     let target_sql_bytes = limits.target_sql_bytes.max(1);
     let batch_sql_bytes = limits.hard_sql_bytes.map_or(target_sql_bytes, |hard| target_sql_bytes.min(hard));
-    let template = InsertSqlTemplate::new(columns, table, schema, db_type, catalog, overrides_postgres_system_values);
+    let template = InsertSqlTemplate::new_with_gaussdb_keywords(
+        columns,
+        table,
+        schema,
+        db_type,
+        catalog,
+        overrides_postgres_system_values,
+        gaussdb_keywords,
+    );
     let value_rows = value_rows_sql(rows, column_types, db_type, mysql_spatial_markers);
     let value_row_bytes = value_rows.iter().map(|row| sql_text_bytes(row, db_type)).collect::<Vec<_>>();
     let mut statements = Vec::new();
@@ -3567,13 +3762,14 @@ fn generate_transfer_write_sql_batches(
     catalog: Option<&str>,
     overrides_postgres_system_values: bool,
     mysql_spatial_markers: bool,
+    gaussdb_keywords: Option<&HashSet<String>>,
 ) -> Result<Vec<String>, String> {
     if rows.is_empty() {
         return Ok(Vec::new());
     }
 
     if matches!(mode, TransferMode::Append | TransferMode::Overwrite) {
-        return Ok(generate_insert_typed_sql_batches_for_transfer(
+        return Ok(generate_insert_typed_sql_batches_for_transfer_with_gaussdb_keywords(
             columns,
             column_types,
             rows,
@@ -3584,6 +3780,7 @@ fn generate_transfer_write_sql_batches(
             SqlBatchLimits::for_database(db_type, max_transfer_write_rows(db_type, mode)),
             overrides_postgres_system_values,
             mysql_spatial_markers,
+            gaussdb_keywords,
         )?
         .into_iter()
         .map(|(sql, _)| sql)
@@ -3612,6 +3809,7 @@ fn generate_transfer_write_sql_batches(
             catalog,
             overrides_postgres_system_values,
             mysql_spatial_markers,
+            gaussdb_keywords,
         );
 
         while end < rows.len() && end - start < max_rows {
@@ -3627,6 +3825,7 @@ fn generate_transfer_write_sql_batches(
                 catalog,
                 overrides_postgres_system_values,
                 mysql_spatial_markers,
+                gaussdb_keywords,
             );
             if candidate.len() > max_sql_bytes && !accepted.is_empty() {
                 break;
@@ -6088,6 +6287,11 @@ where
             request.target_catalog.as_deref(),
         )
         .await;
+    let gaussdb_keywords = if matches!(target_db_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss) {
+        state.gaussdb_reserved_keywords(target_pool_key).await
+    } else {
+        None
+    };
     let batch_size = if request.batch_size == 0 { 1000 } else { request.batch_size };
     let mut offset: u64 = 0;
     let mut total_transferred: u64 = 0;
@@ -6219,7 +6423,7 @@ where
 
                 if request.create_table {
                     if !target_table_preexisting {
-                        let ddl = generate_create_table_ddl(
+                        let ddl = generate_create_table_ddl_with_gaussdb_keywords(
                             &sql_target_columns,
                             &target_table,
                             &request.source_schema,
@@ -6228,6 +6432,7 @@ where
                             source_db_type,
                             None,
                             request.target_catalog.as_deref(),
+                            gaussdb_keywords.as_deref(),
                         );
                         let target_table_created = transfer_create_table_created(
                             execute_on_pool(state, target_pool_key, &ddl).await.map(|_| ()),
@@ -6259,18 +6464,13 @@ where
                 }
 
                 if request.mode == TransferMode::Overwrite {
-                    let full_table = qualified_table(
+                    let truncate_sql = transfer_overwrite_clear_sql(
                         &target_table,
                         &request.target_schema,
                         target_db_type,
                         request.target_catalog.as_deref(),
+                        gaussdb_keywords.as_deref(),
                     );
-                    let truncate_sql = match target_db_type {
-                        DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb => {
-                            format!("DELETE FROM {full_table}")
-                        }
-                        _ => format!("TRUNCATE TABLE {full_table}"),
-                    };
                     execute_on_pool(state, target_pool_key, &truncate_sql)
                         .await
                         .map_err(|e| format!("Failed to truncate MongoDB transfer target table: {e}"))?;
@@ -6296,6 +6496,7 @@ where
                 request.target_catalog.as_deref(),
                 false,
                 false,
+                gaussdb_keywords.as_deref(),
             )?;
             for (statement_index, batch_sql) in write_statements.iter().enumerate() {
                 execute_on_pool(state, target_pool_key, batch_sql).await.map_err(|e| {
@@ -6463,6 +6664,11 @@ where
         )
         .await;
     let preserves_target_table_name = target_table == table;
+    let gaussdb_keywords = if matches!(target_db_type, DatabaseType::Gaussdb | DatabaseType::OpenGauss) {
+        state.gaussdb_reserved_keywords(target_pool_key).await
+    } else {
+        None
+    };
 
     // Get source columns (deduplicate by name)
     let columns = {
@@ -6640,7 +6846,7 @@ where
                         Err(err) => {
                             log::warn!("[transfer] catalog DDL read failed for {table} in catalog '{catalog}': {err}; falling back to generated DDL");
                             (
-                                generate_create_table_ddl(
+                                generate_create_table_ddl_with_gaussdb_keywords(
                                     &columns,
                                     &target_table,
                                     &request.source_schema,
@@ -6649,6 +6855,7 @@ where
                                     source_db_type,
                                     table_comment.as_deref(),
                                     request.target_catalog.as_deref(),
+                                    gaussdb_keywords.as_deref(),
                                 ),
                                 false,
                             )
@@ -6667,7 +6874,7 @@ where
                     {
                         Ok(ddl) => (ddl, true),
                         Err(_) => (
-                            generate_create_table_ddl(
+                            generate_create_table_ddl_with_gaussdb_keywords(
                                 &columns,
                                 &target_table,
                                 &request.source_schema,
@@ -6676,6 +6883,7 @@ where
                                 source_db_type,
                                 table_comment.as_deref(),
                                 request.target_catalog.as_deref(),
+                                gaussdb_keywords.as_deref(),
                             ),
                             false,
                         ),
@@ -6684,7 +6892,7 @@ where
                 if contains_oceanbase_mysql_table_options(&source_ddl)
                     && !db::oceanbase_mysql::is_profile(target_db_type, target_driver_profile.as_deref())
                 {
-                    generate_create_table_ddl(
+                    generate_create_table_ddl_with_gaussdb_keywords(
                         &columns,
                         &target_table,
                         &request.source_schema,
@@ -6693,6 +6901,7 @@ where
                         source_db_type,
                         table_comment.as_deref(),
                         request.target_catalog.as_deref(),
+                        gaussdb_keywords.as_deref(),
                     )
                 } else {
                     reused_source_ddl = source_ddl_was_read;
@@ -6705,7 +6914,7 @@ where
                     )
                 }
             } else {
-                generate_create_table_ddl(
+                generate_create_table_ddl_with_gaussdb_keywords(
                     &columns,
                     &target_table,
                     &request.source_schema,
@@ -6714,6 +6923,7 @@ where
                     source_db_type,
                     table_comment.as_deref(),
                     request.target_catalog.as_deref(),
+                    gaussdb_keywords.as_deref(),
                 )
             };
             log::info!("[transfer] creating target table: {}", ddl.chars().take(200).collect::<String>());
@@ -6765,14 +6975,13 @@ where
 
     // Truncate target if overwrite mode
     if request.mode == TransferMode::Overwrite {
-        let full_table =
-            qualified_table(&target_table, &request.target_schema, target_db_type, request.target_catalog.as_deref());
-        let truncate_sql = match target_db_type {
-            DatabaseType::Sqlite | DatabaseType::CloudflareD1 | DatabaseType::DuckDb => {
-                format!("DELETE FROM {full_table}")
-            }
-            _ => format!("TRUNCATE TABLE {full_table}"),
-        };
+        let truncate_sql = transfer_overwrite_clear_sql(
+            &target_table,
+            &request.target_schema,
+            target_db_type,
+            request.target_catalog.as_deref(),
+            gaussdb_keywords.as_deref(),
+        );
         execute_on_pool(state, target_pool_key, &truncate_sql).await.map_err(|e| format!("Failed to truncate: {e}"))?;
     }
 
@@ -6897,6 +7106,7 @@ where
                 request.target_catalog.as_deref(),
                 overrides_postgres_system_values,
                 mysql_spatial_markers,
+                gaussdb_keywords.as_deref(),
             )?;
             for (statement_index, batch_sql) in write_statements.iter().enumerate() {
                 execute_transfer_write_statement(
@@ -8827,6 +9037,330 @@ mod tests {
         );
     }
 
+    #[test]
+    fn gaussdb_create_table_does_not_quote_simple_lowercase_identifiers() {
+        // Regression test for t8y2/dbx#6205: migrating a MySQL table with plain
+        // lowercase/snake_case identifiers to GaussDB used to quote every column
+        // and the table/schema name, which locks in exact case. GaussDB can fold
+        // *unquoted* references to a different case (e.g. its Oracle-compatible
+        // mode), so a later unquoted query against the migrated table fails with
+        // "column does not exist" even though the column is right there.
+        let cols = vec![
+            db::ColumnInfo { is_primary_key: true, ..test_column("id", "bigint") },
+            test_column("created_at", "datetime"),
+            test_column("updated_at", "datetime"),
+            test_column("deleted_at", "datetime"),
+            test_column("categories_id", "bigint"),
+            test_column("background_image_id", "text"),
+            test_column("level", "int"),
+            test_column("parent_id", "bigint"),
+        ];
+
+        let ddl = generate_create_table_ddl(
+            &cols,
+            "background_categories_img",
+            "",
+            "public",
+            &DatabaseType::Gaussdb,
+            &DatabaseType::Mysql,
+            None,
+            None,
+        );
+
+        assert!(
+            ddl.starts_with("CREATE TABLE IF NOT EXISTS public.background_categories_img ("),
+            "table/schema must be unquoted, ddl: {ddl}"
+        );
+        for column in [
+            "id",
+            "created_at",
+            "updated_at",
+            "deleted_at",
+            "categories_id",
+            "background_image_id",
+            "level",
+            "parent_id",
+        ] {
+            assert!(ddl.contains(&format!("  {column} ")), "column `{column}` must be unquoted, ddl: {ddl}");
+        }
+        assert!(ddl.contains("PRIMARY KEY (id)"), "primary key must be unquoted, ddl: {ddl}");
+        assert!(!ddl.contains('"'), "no identifier should need quoting, ddl: {ddl}");
+    }
+
+    #[test]
+    fn gaussdb_create_table_still_quotes_identifiers_that_need_it() {
+        let cols = vec![
+            test_column("Order", "int"),   // mixed case
+            test_column("select", "text"), // reserved word
+        ];
+
+        let ddl = generate_create_table_ddl(
+            &cols,
+            "t",
+            "",
+            "public",
+            &DatabaseType::Gaussdb,
+            &DatabaseType::Mysql,
+            None,
+            None,
+        );
+
+        assert!(ddl.contains("\"Order\""), "mixed-case column must stay quoted, ddl: {ddl}");
+        assert!(ddl.contains("\"select\""), "reserved-word column must stay quoted, ddl: {ddl}");
+    }
+
+    #[test]
+    fn gaussdb_create_table_quotes_target_only_reserved_words() {
+        // t8y2/dbx#6283: `compact` is a plain lowercase identifier that isn't a
+        // Postgres reserved word, so relying solely on the Postgres reserved-word
+        // set would leave it unquoted — but GaussDB reserves it (Huawei's
+        // GaussDB(DWS) keyword reference lists it as "Reserved (functions and
+        // types allowed)"), and unquoted `compact` is invalid in GaussDB DDL.
+        // `rownum`/`verify` (used below as the table/schema names) were caught
+        // by a follow-up full-catalog diff — see `is_gaussdb_only_reserved_identifier`.
+        let cols = vec![test_column("compact", "int"), test_column("sysdate", "text"), test_column("excluded", "text")];
+
+        let ddl = generate_create_table_ddl(
+            &cols,
+            "rownum",
+            "",
+            "verify",
+            &DatabaseType::Gaussdb,
+            &DatabaseType::Mysql,
+            None,
+            None,
+        );
+
+        assert!(ddl.contains("\"verify\".\"rownum\""), "target-only reserved schema/table must be quoted, ddl: {ddl}");
+        assert!(ddl.contains("\"compact\""), "target-only reserved word must be quoted, ddl: {ddl}");
+        assert!(ddl.contains("\"sysdate\""), "target-only reserved word must be quoted, ddl: {ddl}");
+        assert!(ddl.contains("\"excluded\""), "target-only reserved word must be quoted, ddl: {ddl}");
+
+        let opengauss_ddl = generate_create_table_ddl(
+            &cols,
+            "rownum",
+            "",
+            "verify",
+            &DatabaseType::OpenGauss,
+            &DatabaseType::Mysql,
+            None,
+            None,
+        );
+        assert!(
+            opengauss_ddl.contains("\"compact\""),
+            "OpenGauss must also quote target-only reserved word, ddl: {opengauss_ddl}"
+        );
+        assert!(
+            opengauss_ddl.contains("\"verify\".\"rownum\""),
+            "OpenGauss must also quote target-only reserved schema/table, ddl: {opengauss_ddl}"
+        );
+    }
+
+    #[test]
+    fn gaussdb_insert_quotes_target_only_reserved_words() {
+        // t8y2/dbx#6283 follow-up: the INSERT path shares `quote_transfer_identifier`
+        // with CREATE TABLE, but had no dedicated regression test — cover table,
+        // schema, and column names here with words missed by the original
+        // Postgres-only reserved-word check (`csn`, `groupparent`, `nocycle`,
+        // `shrink` weren't covered by the CREATE TABLE test above either).
+        let columns = vec!["id".to_string(), "csn".to_string(), "groupparent".to_string(), "nocycle".to_string()];
+        let rows = vec![vec![
+            serde_json::Value::from(1),
+            serde_json::Value::from(2),
+            serde_json::Value::from(3),
+            serde_json::Value::from(4),
+        ]];
+
+        let sql = generate_insert_typed(
+            &columns,
+            &vec![None; columns.len()],
+            &rows,
+            "shrink",
+            "verify",
+            &DatabaseType::Gaussdb,
+            None,
+        );
+
+        assert!(
+            sql.starts_with("INSERT INTO \"verify\".\"shrink\""),
+            "reserved schema/table must be quoted, sql: {sql}"
+        );
+        assert!(
+            sql.contains("(id, \"csn\", \"groupparent\", \"nocycle\")"),
+            "plain `id` must stay unquoted and the reserved words must be quoted, sql: {sql}"
+        );
+    }
+
+    #[test]
+    fn gaussdb_create_table_maxvalue_quoting_is_version_dependent() {
+        // t8y2/dbx#6283 follow-up: `maxvalue` is RESERVED_KEYWORD on openGauss
+        // 5.0.0 (the instance the static list was hand-diffed against) but
+        // UNRESERVED_KEYWORD on current/master openGauss (see the kwlist.h
+        // diff linked in the PR review). A static per-word list can only be
+        // right for one of these — the live pg_get_keywords() catalog fixes
+        // that by being authoritative for whatever server is actually
+        // connected, per-connection, instead of baking in one snapshot.
+        let cols = vec![test_column("maxvalue", "int")];
+
+        let opengauss_5_0: HashSet<String> = ["maxvalue".to_string()].into_iter().collect();
+        let ddl_5_0 = generate_create_table_ddl_with_gaussdb_keywords(
+            &cols,
+            "t",
+            "",
+            "public",
+            &DatabaseType::OpenGauss,
+            &DatabaseType::Mysql,
+            None,
+            None,
+            Some(&opengauss_5_0),
+        );
+        assert!(ddl_5_0.contains("\"maxvalue\""), "openGauss 5.0's live catalog reserves maxvalue, ddl: {ddl_5_0}");
+
+        let opengauss_current: HashSet<String> = HashSet::new();
+        let ddl_current = generate_create_table_ddl_with_gaussdb_keywords(
+            &cols,
+            "t",
+            "",
+            "public",
+            &DatabaseType::OpenGauss,
+            &DatabaseType::Mysql,
+            None,
+            None,
+            Some(&opengauss_current),
+        );
+        assert!(
+            !ddl_current.contains('"'),
+            "current openGauss's live catalog does not reserve maxvalue, must stay unquoted, ddl: {ddl_current}"
+        );
+
+        // No live catalog available (probe failed/unavailable): falls back to
+        // the static list unchanged, which still classifies maxvalue as
+        // GaussDB-only reserved.
+        let ddl_fallback =
+            generate_create_table_ddl(&cols, "t", "", "public", &DatabaseType::OpenGauss, &DatabaseType::Mysql, None, None);
+        assert!(
+            ddl_fallback.contains("\"maxvalue\""),
+            "fallback path (no live catalog) must preserve existing static-list behavior, ddl: {ddl_fallback}"
+        );
+    }
+
+    #[test]
+    fn gaussdb_insert_maxvalue_quoting_is_version_dependent() {
+        // Same version-dependence as the CREATE TABLE test above, but for the
+        // actual production write path used by TransferMode::Append/Overwrite
+        // (generate_transfer_write_sql_batches -> generate_insert_typed_sql_batches_for_transfer),
+        // not just the generic dispatcher.
+        let columns = vec!["maxvalue".to_string()];
+        let rows = vec![vec![serde_json::Value::from(1)]];
+
+        let opengauss_5_0: HashSet<String> = ["maxvalue".to_string()].into_iter().collect();
+        let statements_5_0 = generate_transfer_write_sql_batches(
+            &TransferMode::Append,
+            &columns,
+            &vec![None; columns.len()],
+            &rows,
+            "t",
+            "public",
+            &DatabaseType::OpenGauss,
+            &[],
+            None,
+            false,
+            false,
+            Some(&opengauss_5_0),
+        )
+        .unwrap();
+        assert!(
+            statements_5_0[0].contains("\"maxvalue\""),
+            "openGauss 5.0's live catalog reserves maxvalue in INSERT, sql: {}",
+            statements_5_0[0]
+        );
+
+        let opengauss_current: HashSet<String> = HashSet::new();
+        let statements_current = generate_transfer_write_sql_batches(
+            &TransferMode::Append,
+            &columns,
+            &vec![None; columns.len()],
+            &rows,
+            "t",
+            "public",
+            &DatabaseType::OpenGauss,
+            &[],
+            None,
+            false,
+            false,
+            Some(&opengauss_current),
+        )
+        .unwrap();
+        assert!(
+            !statements_current[0].contains('"'),
+            "current openGauss's live catalog does not reserve maxvalue in INSERT, sql: {}",
+            statements_current[0]
+        );
+    }
+
+    #[test]
+    fn gaussdb_overwrite_truncate_maxvalue_quoting_matches_create_table() {
+        // Regression test for a code-review finding on this PR: the
+        // Overwrite-mode TRUNCATE/DELETE statement (transfer_overwrite_clear_sql)
+        // must make the SAME quoting decision as generate_create_table_ddl_with_gaussdb_keywords
+        // for the same table name — otherwise CREATE TABLE and TRUNCATE TABLE
+        // disagree on quoting a reserved-vs-not word like `maxvalue`, which
+        // reintroduces the exact case-locking bug this PR fixes, just inside
+        // the Overwrite-mode clear step instead of column DDL.
+        let opengauss_5_0: HashSet<String> = ["maxvalue".to_string()].into_iter().collect();
+        let sql_5_0 =
+            transfer_overwrite_clear_sql("maxvalue", "public", &DatabaseType::OpenGauss, None, Some(&opengauss_5_0));
+        assert_eq!(sql_5_0, "TRUNCATE TABLE public.\"maxvalue\"", "openGauss 5.0 must quote reserved maxvalue");
+
+        let opengauss_current: HashSet<String> = HashSet::new();
+        let sql_current = transfer_overwrite_clear_sql(
+            "maxvalue",
+            "public",
+            &DatabaseType::OpenGauss,
+            None,
+            Some(&opengauss_current),
+        );
+        assert_eq!(
+            sql_current, "TRUNCATE TABLE public.maxvalue",
+            "current openGauss must not quote unreserved maxvalue"
+        );
+
+        // No live catalog available: falls back to the static list, which
+        // still classifies maxvalue as GaussDB-only reserved (unchanged
+        // existing behavior, same as the CREATE TABLE fallback).
+        let sql_fallback = transfer_overwrite_clear_sql("maxvalue", "public", &DatabaseType::OpenGauss, None, None);
+        assert_eq!(sql_fallback, "TRUNCATE TABLE public.\"maxvalue\"");
+
+        // Sanity-check DuckDb/Sqlite/CloudflareD1 still use DELETE, unaffected
+        // by the gaussdb_keywords plumbing (Sqlite isn't Gaussdb/OpenGauss, so
+        // it keeps unconditionally quoting, same as before this PR).
+        assert_eq!(
+            transfer_overwrite_clear_sql("t", "", &DatabaseType::Sqlite, None, None),
+            "DELETE FROM \"t\""
+        );
+    }
+
+    #[test]
+    fn postgres_create_table_still_quotes_simple_lowercase_identifiers() {
+        // Unlike GaussDB, plain Postgres has no reported case-folding quirk, so
+        // the fix is deliberately scoped to Gaussdb/OpenGauss only — Postgres
+        // itself keeps quoting everything, unchanged.
+        let cols = vec![test_column("id", "int")];
+
+        let ddl = generate_create_table_ddl(
+            &cols,
+            "t",
+            "",
+            "public",
+            &DatabaseType::Postgres,
+            &DatabaseType::Mysql,
+            None,
+            None,
+        );
+
+        assert!(ddl.contains("\"id\""), "ddl: {ddl}");
+    }
+
     #[test]
     fn postgres_create_table_preserves_defaults_identity_and_exact_types() {
         let cols = vec![
@@ -9418,6 +9952,7 @@ mod tests {
             None,
             false,
             false,
+            None,
         )
         .unwrap();
         assert_eq!(statements, vec!["INSERT INTO `warehouse`.`events` (`id`, `binary_payload`) VALUES\n(1, '0x00ff')"]);
@@ -10649,6 +11184,7 @@ SELECT 1 FROM dual"#
             None,
             false,
             false,
+            None,
         )
         .unwrap();
 
@@ -10673,6 +11209,7 @@ SELECT 1 FROM dual"#
             None,
             false,
             false,
+            None,
         )
         .unwrap();
 
@@ -10780,6 +11317,7 @@ SELECT 1 FROM dual"#
             None,
             false,
             false,
+            None,
         )
         .unwrap();
 
@@ -10806,6 +11344,7 @@ SELECT 1 FROM dual"#
                 None,
                 false,
                 true,
+                None,
             )
             .unwrap();
 
@@ -10833,6 +11372,7 @@ SELECT 1 FROM dual"#
             None,
             false,
             true,
+            None,
         )
         .unwrap();
         let public = generate_insert_typed(
@@ -10906,6 +11446,7 @@ SELECT 1 FROM dual"#
                 None,
                 true,
                 false,
+                None,
             )
             .unwrap();
 
@@ -10931,6 +11472,7 @@ SELECT 1 FROM dual"#
             None,
             true,
             false,
+            None,
         )
         .unwrap();
 
@@ -10955,6 +11497,7 @@ SELECT 1 FROM dual"#
             None,
             false,
             false,
+            None,
         )
         .unwrap();
 
@@ -10975,6 +11518,7 @@ SELECT 1 FROM dual"#
             None,
             true,
             false,
+            None,
         )
         .unwrap();