fix(postgres): include partition children when generating table DDL
This commit is contained in:
1 parent
f997a8a39b
commit
1345e3f57b
7 files changed
+1993
-98
No files matched your search
@@ -151,17 +151,32 @@ pub fn append_table_modifiers(ddl: &str, modifiers: &TableModifiers) -> Result<S
|
||||
}
|
||||
|
||||
fn render_external_table_ddl(ddl: &str, external: &ExternalTableDefinition) -> Result<String, String> {
|
||||
// The fallback DDL may already read `CREATE FOREIGN TABLE` — a foreign
|
||||
// table's own relkind ('f') makes the generic renderer emit that header
|
||||
// directly — or plain `CREATE TABLE` from an older/other rendering path.
|
||||
// Accept either so this doesn't regress when the upstream header changes.
|
||||
let create_foreign_table = "CREATE FOREIGN TABLE ";
|
||||
let create_table = "CREATE TABLE ";
|
||||
if !ddl.starts_with(create_table) {
|
||||
let header_len = if ddl.starts_with(create_foreign_table) {
|
||||
create_foreign_table.len()
|
||||
} else if ddl.starts_with(create_table) {
|
||||
create_table.len()
|
||||
} else {
|
||||
return Err("Cloudberry external-table fallback expected CREATE TABLE DDL".to_string());
|
||||
}
|
||||
};
|
||||
let insertion = ddl
|
||||
.find(";\n")
|
||||
.or_else(|| ddl.find(';'))
|
||||
.ok_or_else(|| "Cloudberry fallback DDL has no CREATE TABLE terminator".to_string())?;
|
||||
// When the fallback DDL already read `CREATE FOREIGN TABLE`, the generic
|
||||
// renderer also already appended its own ` SERVER ... [OPTIONS (...)]`
|
||||
// clause (from the same relkind='f' catalog data). Drop that so this
|
||||
// doesn't emit the clause twice — Cloudberry's own `external` definition
|
||||
// above is authoritative here.
|
||||
let body_end = ddl[header_len..insertion].find(" SERVER \"").map(|pos| header_len + pos).unwrap_or(insertion);
|
||||
let mut output = String::with_capacity(ddl.len() + external.options.len() * 24 + 48);
|
||||
output.push_str("CREATE FOREIGN TABLE ");
|
||||
output.push_str(&ddl[create_table.len()..insertion]);
|
||||
output.push_str(&ddl[header_len..body_end]);
|
||||
output.push_str("\nSERVER ");
|
||||
output.push_str(&db::postgres::pg_quote_ident(&external.server));
|
||||
if !external.options.is_empty() {
|
||||
@@ -324,6 +339,31 @@ mod tests {
|
||||
assert!(rendered.contains("\"null\" ''"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn renders_external_table_from_already_foreign_ddl_without_duplicate_server_clause() {
|
||||
// The generic renderer now derives `is_foreign` straight from relkind
|
||||
// and may hand back a `CREATE FOREIGN TABLE ... SERVER ...` fallback
|
||||
// DDL directly, not just plain `CREATE TABLE`.
|
||||
let ddl =
|
||||
"CREATE FOREIGN TABLE \"public\".\"external_events\" (\n \"id\" integer\n) SERVER \"stale_server\";\n";
|
||||
let rendered = append_table_modifiers(
|
||||
ddl,
|
||||
&TableModifiers {
|
||||
external: Some(ExternalTableDefinition {
|
||||
server: "gp_exttable_server".to_string(),
|
||||
options: vec!["format=csv".to_string()],
|
||||
}),
|
||||
..modifiers(None)
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(rendered.starts_with("CREATE FOREIGN TABLE \"public\".\"external_events\""));
|
||||
assert_eq!(rendered.matches("SERVER").count(), 1);
|
||||
assert!(rendered.contains("SERVER \"gp_exttable_server\""));
|
||||
assert!(!rendered.contains("stale_server"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ddl_queries_avoid_privileged_external_catalog_view() {
|
||||
assert_eq!(CLOUD_BERRY_TABLE_DDL_SQL, "SELECT pg_get_tabledef($1, $2, true)");
|
||||
|
||||
+1078
-62
File diff suppressed because it is too large.
Load diff
+503
-29
@@ -7,7 +7,7 @@ use crate::db;
|
||||
use crate::models::connection::{ConnectionConfig, DatabaseType};
|
||||
use crate::query::{agent_execute_query_params, should_discard_pool_after_error, QueryExecutionOptions};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::{BTreeMap, BTreeSet, HashMap};
|
||||
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
@@ -6495,6 +6495,26 @@ pub async fn list_owners_core(
|
||||
.await
|
||||
}
|
||||
|
||||
/// Whether to widen or normalize a single-table DDL fetch for its caller.
|
||||
///
|
||||
/// Database export and table transfer render one relation at a time because
|
||||
/// they already iterate every relation themselves. Interactive display paths
|
||||
/// include PostgreSQL access statements and recurse through the partition
|
||||
/// tree. Oracle export additionally requests portable DDL normalization.
|
||||
#[derive(Clone, Copy)]
|
||||
struct TableDdlOptions {
|
||||
include_postgres_access: bool,
|
||||
include_partitions: bool,
|
||||
portable_oracle: bool,
|
||||
}
|
||||
|
||||
impl TableDdlOptions {
|
||||
const SINGLE_RELATION: Self =
|
||||
Self { include_postgres_access: false, include_partitions: false, portable_oracle: false };
|
||||
const EXPORT: Self = Self { include_postgres_access: false, include_partitions: false, portable_oracle: true };
|
||||
const DISPLAY: Self = Self { include_postgres_access: true, include_partitions: true, portable_oracle: false };
|
||||
}
|
||||
|
||||
pub async fn get_table_ddl_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
@@ -6503,7 +6523,16 @@ pub async fn get_table_ddl_core(
|
||||
table: &str,
|
||||
object_type: Option<db::ObjectSourceKind>,
|
||||
) -> Result<String, String> {
|
||||
get_table_ddl_core_with_options(state, connection_id, database, schema, table, object_type, false, false).await
|
||||
get_table_ddl_core_with_options(
|
||||
state,
|
||||
connection_id,
|
||||
database,
|
||||
schema,
|
||||
table,
|
||||
object_type,
|
||||
TableDdlOptions::SINGLE_RELATION,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn get_table_export_ddl_core(
|
||||
@@ -6514,7 +6543,8 @@ pub async fn get_table_export_ddl_core(
|
||||
table: &str,
|
||||
object_type: Option<db::ObjectSourceKind>,
|
||||
) -> Result<String, String> {
|
||||
get_table_ddl_core_with_options(state, connection_id, database, schema, table, object_type, false, true).await
|
||||
get_table_ddl_core_with_options(state, connection_id, database, schema, table, object_type, TableDdlOptions::EXPORT)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn get_table_display_ddl_core(
|
||||
@@ -6525,7 +6555,16 @@ pub async fn get_table_display_ddl_core(
|
||||
table: &str,
|
||||
object_type: Option<db::ObjectSourceKind>,
|
||||
) -> Result<String, String> {
|
||||
get_table_ddl_core_with_options(state, connection_id, database, schema, table, object_type, true, false).await
|
||||
get_table_ddl_core_with_options(
|
||||
state,
|
||||
connection_id,
|
||||
database,
|
||||
schema,
|
||||
table,
|
||||
object_type,
|
||||
TableDdlOptions::DISPLAY,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn get_table_ddl_core_with_options(
|
||||
@@ -6535,8 +6574,7 @@ async fn get_table_ddl_core_with_options(
|
||||
schema: &str,
|
||||
table: &str,
|
||||
object_type: Option<db::ObjectSourceKind>,
|
||||
include_postgres_access: bool,
|
||||
portable_oracle: bool,
|
||||
options: TableDdlOptions,
|
||||
) -> Result<String, String> {
|
||||
if crate::sql_dialect::parse_sqlserver_linked_schema_ref(schema).is_some() {
|
||||
return Err("DDL is not supported for SQL Server linked server tables".to_string());
|
||||
@@ -6582,19 +6620,33 @@ async fn get_table_ddl_core_with_options(
|
||||
}
|
||||
|
||||
retry_metadata_connection(state, connection_id, Some(database), || {
|
||||
get_table_ddl_once(state, connection_id, database, schema, table, include_postgres_access, portable_oracle)
|
||||
get_table_ddl_once(state, connection_id, database, schema, table, options)
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
/// `pg_ddl_with_partitions` when the caller wants the whole partition tree,
|
||||
/// otherwise plain single-relation `pg_ddl`.
|
||||
async fn pg_ddl_for_options(
|
||||
pool: &deadpool_postgres::Pool,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
include_partitions: bool,
|
||||
) -> Result<String, String> {
|
||||
if include_partitions {
|
||||
pg_ddl_with_partitions(pool, schema, table).await
|
||||
} else {
|
||||
pg_ddl(pool, schema, table).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn get_table_ddl_once(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
include_postgres_access: bool,
|
||||
portable_oracle: bool,
|
||||
options: TableDdlOptions,
|
||||
) -> Result<String, String> {
|
||||
let pool_key = state.get_or_create_metadata_pool_for_session(connection_id, Some(database), None).await?;
|
||||
let db_config = connection_config(state, connection_id).await;
|
||||
@@ -6642,10 +6694,11 @@ async fn get_table_ddl_once(
|
||||
if let Some(config) = db_config.as_ref().filter(|config| is_agent_postgres_metadata_fallback_config(config))
|
||||
{
|
||||
match native_postgres_metadata_pool(state, connection_id, database, config).await {
|
||||
Ok(Some(pool)) => match pg_ddl(&pool, schema, table).await {
|
||||
Ok(ddl) => return Ok(ddl),
|
||||
Err(error) => {
|
||||
log::warn!(
|
||||
Ok(Some(pool)) => {
|
||||
match pg_ddl_for_options(&pool, schema, table, options.include_partitions).await {
|
||||
Ok(ddl) => return Ok(ddl),
|
||||
Err(error) => {
|
||||
log::warn!(
|
||||
"[schema][agent:get_table_ddl:postgres-compatible-native-fallback-failed] connection_id={} database={} schema={} table={} error={}",
|
||||
connection_id,
|
||||
database,
|
||||
@@ -6653,8 +6706,9 @@ async fn get_table_ddl_once(
|
||||
table,
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(error) => {
|
||||
log::warn!(
|
||||
@@ -6674,7 +6728,7 @@ async fn get_table_ddl_once(
|
||||
database,
|
||||
schema,
|
||||
table,
|
||||
portable_oracle,
|
||||
options.portable_oracle,
|
||||
agent_metadata_timeout(db_config.as_ref()),
|
||||
)
|
||||
.await;
|
||||
@@ -6702,27 +6756,27 @@ async fn get_table_ddl_once(
|
||||
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_opengauss_family_config) => {
|
||||
match opengauss_table_ddl(p, schema, table).await {
|
||||
Ok(ddl) => Ok(ddl),
|
||||
Err(_) => pg_ddl(p, schema, table).await,
|
||||
Err(_) => pg_ddl_for_options(p, schema, table, options.include_partitions).await,
|
||||
}
|
||||
}
|
||||
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_questdb_config) => {
|
||||
match db::questdb::questdb_table_or_view_ddl(p, table).await {
|
||||
Ok(ddl) => Ok(ddl),
|
||||
Err(_) => pg_ddl(p, schema, table).await,
|
||||
Err(_) => pg_ddl_for_options(p, schema, table, options.include_partitions).await,
|
||||
}
|
||||
}
|
||||
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_cloudberry_config) => {
|
||||
cloudberry_ddl(p, schema, table).await
|
||||
cloudberry_ddl(p, schema, table, options.include_partitions).await
|
||||
}
|
||||
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(db::opentenbase::is_config) => {
|
||||
opentenbase_ddl(p, schema, table).await
|
||||
opentenbase_ddl(p, schema, table, options.include_partitions).await
|
||||
}
|
||||
PoolKind::Postgres(p)
|
||||
if include_postgres_access && db_config.as_ref().is_some_and(is_native_postgres_config) =>
|
||||
if options.include_postgres_access && db_config.as_ref().is_some_and(is_native_postgres_config) =>
|
||||
{
|
||||
pg_display_ddl(p, schema, table).await
|
||||
}
|
||||
PoolKind::Postgres(p) => pg_ddl(p, schema, table).await,
|
||||
PoolKind::Postgres(p) => pg_ddl_for_options(p, schema, table, options.include_partitions).await,
|
||||
PoolKind::Sqlite(p) => sqlite_ddl(p, schema, table).await,
|
||||
PoolKind::Rqlite(client) => db::rqlite_driver::table_ddl(client, table).await,
|
||||
PoolKind::Turso(client) => db::turso_driver::table_ddl(client, table).await,
|
||||
@@ -6844,6 +6898,14 @@ fn pg_ident(value: &str) -> String {
|
||||
format!("\"{}\"", value.replace('"', "\"\""))
|
||||
}
|
||||
|
||||
/// Whether a `pg_get_constraintdef` result ends in the ` NOT VALID` suffix
|
||||
/// Postgres appends for an unvalidated constraint. That suffix is only legal
|
||||
/// after `ALTER TABLE ADD CONSTRAINT`, never inside a `CREATE TABLE` column
|
||||
/// list.
|
||||
fn is_not_valid_constraintdef(definition: &str) -> bool {
|
||||
definition.to_ascii_uppercase().trim_end().ends_with("NOT VALID")
|
||||
}
|
||||
|
||||
fn sqlserver_ident(value: &str) -> String {
|
||||
format!("[{}]", value.replace(']', "]]"))
|
||||
}
|
||||
@@ -8644,6 +8706,7 @@ mod ddl_tests {
|
||||
&[column("created_at", "timestamp without time zone")],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
key: Some(partition_key.to_string()),
|
||||
@@ -8665,6 +8728,7 @@ mod ddl_tests {
|
||||
&[column("id", "integer")],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo::default(),
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
@@ -8695,11 +8759,13 @@ mod ddl_tests {
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("FOR VALUES FROM ('2026-01-01') TO ('2027-01-01')".to_string()),
|
||||
key: Some("HASH (payload)".to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
let partition_local_objects = db::postgres::PostgresTablePartitionLocalObjects {
|
||||
has_primary_key: true,
|
||||
foreign_keys: BTreeSet::new(),
|
||||
indexes: BTreeSet::from(["events_payload_idx".to_string()]),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
@@ -8708,6 +8774,7 @@ mod ddl_tests {
|
||||
&[id, column("payload", "text")],
|
||||
&indexes,
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&partition_info,
|
||||
&partition_local_objects,
|
||||
@@ -8742,6 +8809,7 @@ mod ddl_tests {
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("DEFAULT".to_string()),
|
||||
key: None,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
@@ -8750,6 +8818,7 @@ mod ddl_tests {
|
||||
&[id],
|
||||
&indexes,
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&partition_info,
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
@@ -8758,6 +8827,175 @@ mod ddl_tests {
|
||||
assert_eq!(ddl, "CREATE TABLE \"public\".\"events_default\" PARTITION OF \"public\".\"events\" DEFAULT;\n");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_table_ddl_renders_check_constraints_for_ordinary_tables() {
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"users",
|
||||
&[column("age", "integer")],
|
||||
&[],
|
||||
&[],
|
||||
&[("users_age_check".to_string(), "CHECK (age >= 0)".to_string())],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo::default(),
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
);
|
||||
|
||||
assert!(ddl.contains("CONSTRAINT \"users_age_check\" CHECK (age >= 0)"), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_partition_ddl_only_renders_local_check_constraints() {
|
||||
let mut partition_local_objects = db::postgres::PostgresTablePartitionLocalObjects::default();
|
||||
partition_local_objects.check_constraints.insert("child_only_check".to_string());
|
||||
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"events_2026",
|
||||
&[column("payload", "text")],
|
||||
&[],
|
||||
&[],
|
||||
&[
|
||||
("parent_check".to_string(), "CHECK (payload IS NOT NULL)".to_string()),
|
||||
("child_only_check".to_string(), "CHECK (payload <> '')".to_string()),
|
||||
],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
is_partition: true,
|
||||
parent_schema: Some("public".to_string()),
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("DEFAULT".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
&partition_local_objects,
|
||||
);
|
||||
|
||||
assert!(ddl.contains("CONSTRAINT \"child_only_check\" CHECK (payload <> '')"), "ddl: {ddl}");
|
||||
assert!(!ddl.contains("parent_check"), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_partition_ddl_overrides_local_column_default() {
|
||||
let mut status = column("status", "text");
|
||||
status.column_default = Some("'archived'::text".to_string());
|
||||
let mut partition_local_objects = db::postgres::PostgresTablePartitionLocalObjects::default();
|
||||
partition_local_objects
|
||||
.column_defaults
|
||||
.insert("status".to_string(), db::postgres::PostgresColumnDefaultState::Overridden);
|
||||
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"events_2026",
|
||||
&[status],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
is_partition: true,
|
||||
parent_schema: Some("public".to_string()),
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("DEFAULT".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
&partition_local_objects,
|
||||
);
|
||||
|
||||
assert!(ddl.contains("\"status\" WITH OPTIONS DEFAULT 'archived'::text"), "ddl: {ddl}");
|
||||
// The partition's own column list is otherwise omitted (inherited
|
||||
// from the parent), so a plain (non-override) column declaration
|
||||
// must not appear alongside the WITH OPTIONS clause.
|
||||
assert!(!ddl.contains("\"status\" text"), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_partition_ddl_emits_drop_default_for_locally_dropped_column() {
|
||||
// A partition that ran `ALTER TABLE ONLY child ALTER COLUMN status
|
||||
// DROP DEFAULT` has no default of its own to report here — the
|
||||
// column's `column_default` is `None`, distinct from the
|
||||
// "overridden" case (which has its own, different, Some value).
|
||||
let status = column("status", "text");
|
||||
let mut partition_local_objects = db::postgres::PostgresTablePartitionLocalObjects::default();
|
||||
partition_local_objects
|
||||
.column_defaults
|
||||
.insert("status".to_string(), db::postgres::PostgresColumnDefaultState::Dropped);
|
||||
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"events_2026",
|
||||
&[status],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
is_partition: true,
|
||||
parent_schema: Some("public".to_string()),
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("DEFAULT".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
&partition_local_objects,
|
||||
);
|
||||
|
||||
assert!(
|
||||
ddl.contains("ALTER TABLE ONLY \"public\".\"events_2026\" ALTER COLUMN \"status\" DROP DEFAULT;"),
|
||||
"ddl: {ddl}"
|
||||
);
|
||||
assert!(!ddl.contains("WITH OPTIONS DEFAULT"), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_table_ddl_renders_foreign_table_with_server_and_options() {
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"remote_users",
|
||||
&[column("id", "integer")],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
is_foreign: true,
|
||||
foreign_server: Some("loopback".to_string()),
|
||||
foreign_options: vec![("schema_name".to_string(), "public".to_string())],
|
||||
..Default::default()
|
||||
},
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
);
|
||||
|
||||
assert!(ddl.starts_with("CREATE FOREIGN TABLE \"public\".\"remote_users\""), "ddl: {ddl}");
|
||||
assert!(ddl.contains("SERVER \"loopback\""), "ddl: {ddl}");
|
||||
assert!(ddl.contains("OPTIONS (\"schema_name\" 'public')"), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_partition_ddl_uses_foreign_table_syntax_for_foreign_partitions() {
|
||||
let ddl = render_postgres_table_ddl_with_partition_info(
|
||||
"public",
|
||||
"events_remote",
|
||||
&[column("id", "integer")],
|
||||
&[],
|
||||
&[],
|
||||
&[],
|
||||
None,
|
||||
&db::postgres::PostgresTablePartitionInfo {
|
||||
is_partition: true,
|
||||
parent_schema: Some("public".to_string()),
|
||||
parent_table: Some("events".to_string()),
|
||||
bound: Some("FOR VALUES FROM ('2027-01-01') TO ('2028-01-01')".to_string()),
|
||||
is_foreign: true,
|
||||
foreign_server: Some("loopback".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
);
|
||||
|
||||
assert!(ddl.starts_with("CREATE FOREIGN TABLE \"public\".\"events_remote\" PARTITION OF"), "ddl: {ddl}");
|
||||
assert!(ddl.contains("SERVER \"loopback\""), "ddl: {ddl}");
|
||||
assert!(!ddl.contains("CREATE TABLE \"public\".\"events_remote\""), "ddl: {ddl}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn postgres_table_ddl_keeps_composite_foreign_key_together() {
|
||||
let columns = vec![column("a", "integer"), column("b", "integer"), column("c", "integer")];
|
||||
@@ -9290,14 +9528,20 @@ pub async fn sqlite_ddl(pool: &db::sqlite::SqliteHandle, schema: &str, table: &s
|
||||
.map_err(|e| e.to_string())?
|
||||
}
|
||||
|
||||
/// DDL for a single relation. Callers that already iterate a relation set
|
||||
/// themselves (database export, table transfer) must use this rather than
|
||||
/// `pg_ddl_with_partitions` — recursing into partition children here would
|
||||
/// duplicate every partition's `CREATE TABLE` (once from the parent's DDL,
|
||||
/// once from the caller's own loop over that same child relation).
|
||||
pub async fn pg_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
||||
let (columns, indexes, fkeys, table_comment, partition_info, trigger_definitions) = tokio::try_join!(
|
||||
let (columns, indexes, fkeys, table_comment, partition_info, trigger_definitions, check_constraints) = tokio::try_join!(
|
||||
db::postgres::get_columns(pool, schema, table),
|
||||
db::postgres::list_indexes(pool, schema, table),
|
||||
db::postgres::list_foreign_keys(pool, schema, table),
|
||||
async { db::postgres::get_table_comment(pool, schema, table).await },
|
||||
db::postgres::get_table_partition_info(pool, schema, table),
|
||||
db::postgres::list_trigger_definitions(pool, schema, table),
|
||||
db::postgres::list_check_constraints(pool, schema, table),
|
||||
)?;
|
||||
let partition_local_objects = if partition_info.is_partition {
|
||||
db::postgres::get_table_partition_local_objects(pool, schema, table).await?
|
||||
@@ -9312,6 +9556,7 @@ pub async fn pg_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -
|
||||
&columns,
|
||||
&indexes,
|
||||
&fkeys,
|
||||
&check_constraints,
|
||||
table_comment.as_deref(),
|
||||
&partition_info,
|
||||
&partition_local_objects,
|
||||
@@ -9320,8 +9565,156 @@ pub async fn pg_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -
|
||||
))
|
||||
}
|
||||
|
||||
/// Like `pg_ddl`, but for a partitioned table also emits `CREATE TABLE ...
|
||||
/// PARTITION OF` for every existing partition, at any depth. Used only by the
|
||||
/// interactive "view DDL" paths (`get_table_display_ddl_core`) — callers that
|
||||
/// iterate relations themselves must use `pg_ddl` instead (see its doc
|
||||
/// comment). Fetches the whole tree's metadata via a handful of batched,
|
||||
/// tree-wide queries (see `db::postgres::fetch_postgres_partition_tree` and
|
||||
/// its `_for_relations` siblings) instead of recursing per relation, which
|
||||
/// would rerun the full metadata query chain once per node and scale request
|
||||
/// count linearly with the number of partitions.
|
||||
pub async fn pg_ddl_with_partitions(
|
||||
pool: &deadpool_postgres::Pool,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
) -> Result<String, String> {
|
||||
let tree = db::postgres::fetch_postgres_partition_tree(pool, schema, table).await?;
|
||||
let tree_oids: HashSet<i64> = tree.iter().map(|node| node.oid).collect();
|
||||
// The requested relation is the tree root: it's the only node whose parent
|
||||
// (if it has one at all — it may itself be a partition of a table outside
|
||||
// this tree) isn't also a node we fetched.
|
||||
let Some(root) =
|
||||
tree.iter().find(|node| !node.parent_oid.is_some_and(|parent_oid| tree_oids.contains(&parent_oid)))
|
||||
else {
|
||||
return Err(format!(
|
||||
"relation \"{schema}\".\"{table}\" was not found or is not a table/partition/foreign table"
|
||||
));
|
||||
};
|
||||
|
||||
let oids: Vec<i64> = tree.iter().map(|node| node.oid).collect();
|
||||
let relations: Vec<(i64, String, String)> =
|
||||
tree.iter().map(|node| (node.oid, node.schema.clone(), node.table.clone())).collect();
|
||||
let relation_pairs: Vec<(String, String)> =
|
||||
tree.iter().map(|node| (node.schema.clone(), node.table.clone())).collect();
|
||||
|
||||
let (
|
||||
columns_by_oid,
|
||||
indexes_by_oid,
|
||||
fkeys_by_relation,
|
||||
comments_by_oid,
|
||||
triggers_by_oid,
|
||||
checks_by_oid,
|
||||
local_objects_by_oid,
|
||||
) = tokio::try_join!(
|
||||
db::postgres::get_columns_for_relations(pool, &relations),
|
||||
db::postgres::list_indexes_for_relations(pool, &relations),
|
||||
db::postgres::list_foreign_keys_for_relations(pool, &relation_pairs),
|
||||
db::postgres::get_table_comments_for_relations(pool, &oids),
|
||||
db::postgres::list_trigger_definitions_for_relations(pool, &oids),
|
||||
db::postgres::list_check_constraints_for_relations(pool, &oids),
|
||||
db::postgres::get_table_partition_local_objects_for_relations(pool, &oids),
|
||||
)?;
|
||||
|
||||
// Group children by parent oid, each group ordered by relname to match
|
||||
// `list_table_partitions`' `ORDER BY c.relname` (today's traversal order).
|
||||
let mut children_by_parent: HashMap<i64, Vec<&db::postgres::PostgresPartitionTreeNode>> = HashMap::new();
|
||||
for node in &tree {
|
||||
if let Some(parent_oid) = node.parent_oid {
|
||||
children_by_parent.entry(parent_oid).or_default().push(node);
|
||||
}
|
||||
}
|
||||
for children in children_by_parent.values_mut() {
|
||||
children.sort_by(|a, b| a.table.cmp(&b.table));
|
||||
}
|
||||
|
||||
let mut ddl = String::new();
|
||||
render_postgres_partition_tree_node(
|
||||
root,
|
||||
&children_by_parent,
|
||||
&columns_by_oid,
|
||||
&indexes_by_oid,
|
||||
&fkeys_by_relation,
|
||||
&comments_by_oid,
|
||||
&triggers_by_oid,
|
||||
&checks_by_oid,
|
||||
&local_objects_by_oid,
|
||||
&mut ddl,
|
||||
);
|
||||
Ok(ddl)
|
||||
}
|
||||
|
||||
/// Renders `root` and every descendant reachable through `children_by_parent`.
|
||||
/// Iterative (an explicit stack, not function-call recursion) so that a
|
||||
/// pathologically deep partition hierarchy can't overflow the stack; `visited`
|
||||
/// additionally guards against a corrupted catalog (or non-PostgreSQL fork)
|
||||
/// whose `pg_inherits` data forms a cycle, which would otherwise loop forever.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn render_postgres_partition_tree_node(
|
||||
root: &db::postgres::PostgresPartitionTreeNode,
|
||||
children_by_parent: &HashMap<i64, Vec<&db::postgres::PostgresPartitionTreeNode>>,
|
||||
columns_by_oid: &HashMap<i64, Vec<db::ColumnInfo>>,
|
||||
indexes_by_oid: &HashMap<i64, Vec<db::IndexInfo>>,
|
||||
fkeys_by_relation: &HashMap<(String, String), Vec<db::ForeignKeyInfo>>,
|
||||
comments_by_oid: &HashMap<i64, Option<String>>,
|
||||
triggers_by_oid: &HashMap<i64, Vec<String>>,
|
||||
checks_by_oid: &HashMap<i64, Vec<(String, String)>>,
|
||||
local_objects_by_oid: &HashMap<i64, db::postgres::PostgresTablePartitionLocalObjects>,
|
||||
ddl: &mut String,
|
||||
) {
|
||||
let empty_columns = Vec::new();
|
||||
let empty_indexes = Vec::new();
|
||||
let empty_fkeys = Vec::new();
|
||||
let empty_triggers = Vec::new();
|
||||
let empty_checks = Vec::new();
|
||||
let empty_local_objects = db::postgres::PostgresTablePartitionLocalObjects::default();
|
||||
|
||||
let mut visited: HashSet<i64> = HashSet::new();
|
||||
let mut stack: Vec<&db::postgres::PostgresPartitionTreeNode> = vec![root];
|
||||
while let Some(node) = stack.pop() {
|
||||
if !visited.insert(node.oid) {
|
||||
continue;
|
||||
}
|
||||
|
||||
let columns = columns_by_oid.get(&node.oid).unwrap_or(&empty_columns);
|
||||
let indexes = indexes_by_oid.get(&node.oid).unwrap_or(&empty_indexes);
|
||||
let fkeys = fkeys_by_relation.get(&(node.schema.clone(), node.table.clone())).unwrap_or(&empty_fkeys);
|
||||
let comment = comments_by_oid.get(&node.oid).cloned().flatten();
|
||||
let triggers = triggers_by_oid.get(&node.oid).unwrap_or(&empty_triggers);
|
||||
let checks = checks_by_oid.get(&node.oid).unwrap_or(&empty_checks);
|
||||
let local_objects = local_objects_by_oid.get(&node.oid).unwrap_or(&empty_local_objects);
|
||||
|
||||
if !ddl.is_empty() {
|
||||
ddl.push('\n');
|
||||
}
|
||||
ddl.push_str(&append_postgres_trigger_definitions(
|
||||
render_postgres_table_ddl_with_partition_info(
|
||||
&node.schema,
|
||||
&node.table,
|
||||
columns,
|
||||
indexes,
|
||||
fkeys,
|
||||
checks,
|
||||
comment.as_deref(),
|
||||
&node.partition_info,
|
||||
local_objects,
|
||||
),
|
||||
triggers,
|
||||
));
|
||||
|
||||
if let Some(children) = children_by_parent.get(&node.oid) {
|
||||
// Push in reverse so children are popped (and rendered) in their
|
||||
// original relname-sorted order.
|
||||
for child in children.iter().rev() {
|
||||
stack.push(child);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn pg_display_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
||||
let (ddl, access) = tokio::join!(pg_ddl(pool, schema, table), db::postgres::get_table_access(pool, schema, table));
|
||||
let (ddl, access) =
|
||||
tokio::join!(pg_ddl_with_partitions(pool, schema, table), db::postgres::get_table_access(pool, schema, table));
|
||||
let ddl = ddl?;
|
||||
match access {
|
||||
Ok(access) => Ok(append_postgres_access_ddl(ddl, schema, table, &access)),
|
||||
@@ -9577,11 +9970,16 @@ fn append_opengauss_trigger_definitions(mut ddl: String, trigger_definitions: &[
|
||||
ddl
|
||||
}
|
||||
|
||||
pub async fn cloudberry_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
||||
pub async fn cloudberry_ddl(
|
||||
pool: &deadpool_postgres::Pool,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
include_partitions: bool,
|
||||
) -> Result<String, String> {
|
||||
match db::cloudberry::table_ddl(pool, schema, table).await {
|
||||
Ok(ddl) => Ok(ddl),
|
||||
Err(native_error) => {
|
||||
let base_ddl = pg_ddl(pool, schema, table).await.map_err(|fallback_error| {
|
||||
let base_ddl = pg_ddl_for_options(pool, schema, table, include_partitions).await.map_err(|fallback_error| {
|
||||
format!(
|
||||
"Cloudberry pg_get_tabledef failed: {native_error}; PostgreSQL DDL fallback failed: {fallback_error}"
|
||||
)
|
||||
@@ -9598,8 +9996,13 @@ pub async fn cloudberry_ddl(pool: &deadpool_postgres::Pool, schema: &str, table:
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn opentenbase_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
||||
let ddl = pg_ddl(pool, schema, table).await?;
|
||||
pub async fn opentenbase_ddl(
|
||||
pool: &deadpool_postgres::Pool,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
include_partitions: bool,
|
||||
) -> Result<String, String> {
|
||||
let ddl = pg_ddl_for_options(pool, schema, table, include_partitions).await?;
|
||||
match db::opentenbase::table_distribution(pool, schema, table).await {
|
||||
Ok(Some(distribution)) => match db::opentenbase::append_distribution_clause(&ddl, &distribution) {
|
||||
Ok(ddl) => Ok(ddl),
|
||||
@@ -9640,6 +10043,7 @@ pub fn render_postgres_table_ddl(
|
||||
columns,
|
||||
indexes,
|
||||
fkeys,
|
||||
&[],
|
||||
table_comment,
|
||||
&db::postgres::PostgresTablePartitionInfo::default(),
|
||||
&db::postgres::PostgresTablePartitionLocalObjects::default(),
|
||||
@@ -9652,6 +10056,7 @@ fn render_postgres_table_ddl_with_partition_info(
|
||||
columns: &[db::ColumnInfo],
|
||||
indexes: &[db::IndexInfo],
|
||||
fkeys: &[db::ForeignKeyInfo],
|
||||
check_constraints: &[(String, String)],
|
||||
table_comment: Option<&str>,
|
||||
partition_info: &db::postgres::PostgresTablePartitionInfo,
|
||||
partition_local_objects: &db::postgres::PostgresTablePartitionLocalObjects,
|
||||
@@ -9728,7 +10133,42 @@ fn render_postgres_table_ddl_with_partition_info(
|
||||
ref_columns
|
||||
));
|
||||
}
|
||||
// `pg_get_constraintdef` appends ` NOT VALID` for an unvalidated CHECK
|
||||
// constraint, but that suffix is only legal after `ALTER TABLE ADD
|
||||
// CONSTRAINT` — it's a syntax error inside a `CREATE TABLE` column list.
|
||||
// Emit those as a separate statement below instead, so the constraint's
|
||||
// unvalidated state round-trips instead of producing invalid DDL.
|
||||
let mut not_valid_check_constraints: Vec<(&str, &str)> = Vec::new();
|
||||
for (name, definition) in check_constraints {
|
||||
if is_partition && !partition_local_objects.check_constraints.contains(name) {
|
||||
continue;
|
||||
}
|
||||
let definition = definition.trim();
|
||||
if is_not_valid_constraintdef(definition) {
|
||||
not_valid_check_constraints.push((name.as_str(), definition));
|
||||
continue;
|
||||
}
|
||||
definition_lines.push(format!(" CONSTRAINT {} {}", pg_ident(name), definition));
|
||||
}
|
||||
if is_partition {
|
||||
// A partition can override a column's default independently of the
|
||||
// parent; PostgreSQL only accepts that override through `column_name
|
||||
// WITH OPTIONS DEFAULT ...` since the partition's own column list is
|
||||
// otherwise inherited (and thus omitted) from the parent's.
|
||||
for column in columns {
|
||||
if partition_local_objects.column_defaults.get(&column.name)
|
||||
!= Some(&db::postgres::PostgresColumnDefaultState::Overridden)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let Some(default) = column.column_default.as_deref() else {
|
||||
continue;
|
||||
};
|
||||
definition_lines.push(format!(" {} WITH OPTIONS DEFAULT {default}", pg_ident(&column.name)));
|
||||
}
|
||||
}
|
||||
|
||||
let create = if partition_info.is_foreign { "CREATE FOREIGN TABLE" } else { "CREATE TABLE" };
|
||||
let mut ddl = if let Some((parent_schema, parent_table, bound)) = partition_parent {
|
||||
let parent_name = format!("{}.{}", pg_ident(parent_schema), pg_ident(parent_table));
|
||||
let definitions = if definition_lines.is_empty() {
|
||||
@@ -9736,15 +10176,49 @@ fn render_postgres_table_ddl_with_partition_info(
|
||||
} else {
|
||||
format!(" (\n{}\n)", definition_lines.join(",\n"))
|
||||
};
|
||||
format!("CREATE TABLE {table_name} PARTITION OF {parent_name}{definitions} {bound}")
|
||||
format!("{create} {table_name} PARTITION OF {parent_name}{definitions} {bound}")
|
||||
} else {
|
||||
format!("CREATE TABLE {table_name} (\n{}\n)", definition_lines.join(",\n"))
|
||||
format!("{create} {table_name} (\n{}\n)", definition_lines.join(",\n"))
|
||||
};
|
||||
if let Some(server) = partition_info.foreign_server.as_deref().filter(|server| !server.trim().is_empty()) {
|
||||
ddl.push_str(&format!(" SERVER {}", pg_ident(server)));
|
||||
if !partition_info.foreign_options.is_empty() {
|
||||
let options = partition_info
|
||||
.foreign_options
|
||||
.iter()
|
||||
.map(|(key, value)| format!("{} {}", pg_ident(key), sql_string(value)))
|
||||
.collect::<Vec<_>>()
|
||||
.join(", ");
|
||||
ddl.push_str(&format!(" OPTIONS ({options})"));
|
||||
}
|
||||
}
|
||||
if let Some(partition_key) = partition_info.key.as_deref().filter(|key| !key.trim().is_empty()) {
|
||||
ddl.push_str(&format!(" PARTITION BY {partition_key}"));
|
||||
}
|
||||
ddl.push_str(";\n");
|
||||
|
||||
for (name, definition) in ¬_valid_check_constraints {
|
||||
ddl.push_str(&format!("\nALTER TABLE {table_name} ADD CONSTRAINT {} {};", pg_ident(name), definition));
|
||||
}
|
||||
|
||||
if is_partition {
|
||||
// A dropped default has no counterpart in the PARTITION OF column
|
||||
// list syntax used above for overrides — it must be replayed as a
|
||||
// standalone statement, or restore would silently reintroduce the
|
||||
// parent's default (PostgreSQL auto-copies it onto every partition
|
||||
// at creation time unless explicitly dropped).
|
||||
for column in columns {
|
||||
if partition_local_objects.column_defaults.get(&column.name)
|
||||
== Some(&db::postgres::PostgresColumnDefaultState::Dropped)
|
||||
{
|
||||
ddl.push_str(&format!(
|
||||
"\nALTER TABLE ONLY {table_name} ALTER COLUMN {} DROP DEFAULT;",
|
||||
pg_ident(&column.name)
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(comment) = table_comment.filter(|comment| !comment.trim().is_empty()) {
|
||||
ddl.push_str(&format!("\nCOMMENT ON TABLE {table_name} IS {};", sql_string(comment)));
|
||||
}
|
||||
|
||||
@@ -258,10 +258,10 @@ pub(in crate::schema) async fn table_ddl(
|
||||
}
|
||||
}
|
||||
PoolKind::Postgres(p) if config.is_some_and(is_cloudberry_config) => {
|
||||
super::super::cloudberry_ddl(p, schema, table).await
|
||||
super::super::cloudberry_ddl(p, schema, table, false).await
|
||||
}
|
||||
PoolKind::Postgres(p) if config.is_some_and(db::opentenbase::is_config) => {
|
||||
super::super::opentenbase_ddl(p, schema, table).await
|
||||
super::super::opentenbase_ddl(p, schema, table, false).await
|
||||
}
|
||||
PoolKind::Postgres(p) => super::super::pg_ddl(p, schema, table).await,
|
||||
PoolKind::Sqlite(p) => super::super::sqlite_ddl(p, schema, table).await,
|
||||
|
||||
@@ -0,0 +1,365 @@
|
||||
//! Regression coverage for PR review points on partition DDL:
|
||||
//! (1) database export must not duplicate a partition's `CREATE TABLE`
|
||||
//! (once via the parent's embedded tree, once via the export loop's own
|
||||
//! iteration over that same child relation) and the exported SQL must
|
||||
//! replay cleanly into a fresh schema; (2) a partition-local dropped
|
||||
//! column default must be modeled and replayed as a standalone
|
||||
//! `ALTER TABLE ONLY ... DROP DEFAULT`, not silently reintroduced;
|
||||
//! (3) fetching a whole partition tree's DDL must issue a small,
|
||||
//! roughly-constant number of queries rather than one scaling with the
|
||||
//! number of partitions.
|
||||
//! Needs local Docker; skips cleanly when unavailable.
|
||||
|
||||
use std::process::Command;
|
||||
|
||||
use dbx_core::connection::AppState;
|
||||
use dbx_core::database_export::{export_database_sql_core, DatabaseExportRequest};
|
||||
use dbx_core::models::connection::{ConnectionConfig, DatabaseType};
|
||||
use dbx_core::storage::Storage;
|
||||
|
||||
struct DockerPostgres {
|
||||
name: String,
|
||||
port: u16,
|
||||
}
|
||||
|
||||
impl Drop for DockerPostgres {
|
||||
fn drop(&mut self) {
|
||||
let _ = Command::new("docker").args(["rm", "-f", &self.name]).status();
|
||||
}
|
||||
}
|
||||
|
||||
fn docker_ready() -> bool {
|
||||
Command::new("docker")
|
||||
.args(["version", "--format", "{{.Server.Version}}"])
|
||||
.output()
|
||||
.map(|output| output.status.success())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
fn start_docker_postgres(name_prefix: &str) -> Option<DockerPostgres> {
|
||||
if !docker_ready() {
|
||||
eprintln!("skipping docker-backed partition ddl test because Docker is unavailable");
|
||||
return None;
|
||||
}
|
||||
|
||||
let port = portpicker::pick_unused_port().expect("pick unused postgres port");
|
||||
let container = DockerPostgres { name: format!("{name_prefix}-{}", uuid::Uuid::new_v4()), port };
|
||||
|
||||
let status = Command::new("docker")
|
||||
.args([
|
||||
"run",
|
||||
"-d",
|
||||
"--rm",
|
||||
"--name",
|
||||
&container.name,
|
||||
"-e",
|
||||
"POSTGRES_PASSWORD=postgres",
|
||||
"-e",
|
||||
"POSTGRES_USER=postgres",
|
||||
"-e",
|
||||
"POSTGRES_DB=postgres",
|
||||
"-p",
|
||||
&format!("{port}:5432"),
|
||||
"postgres:16-alpine",
|
||||
])
|
||||
.status()
|
||||
.expect("start docker postgres");
|
||||
assert!(status.success(), "docker run postgres container should succeed");
|
||||
|
||||
let mut consecutive_ok = 0;
|
||||
for _ in 0..120 {
|
||||
let ready = Command::new("docker")
|
||||
.args(["exec", &container.name, "psql", "-U", "postgres", "-d", "postgres", "-c", "SELECT 1"])
|
||||
.output()
|
||||
.map(|output| output.status.success())
|
||||
.unwrap_or(false);
|
||||
consecutive_ok = if ready { consecutive_ok + 1 } else { 0 };
|
||||
if consecutive_ok >= 2 {
|
||||
return Some(container);
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(500));
|
||||
}
|
||||
panic!("postgres container did not become ready");
|
||||
}
|
||||
|
||||
fn psql(container: &DockerPostgres, sql: &str) {
|
||||
let output = Command::new("docker")
|
||||
.args(["exec", &container.name, "psql", "-U", "postgres", "-d", "postgres", "-v", "ON_ERROR_STOP=1", "-c", sql])
|
||||
.output()
|
||||
.expect("run psql");
|
||||
assert!(output.status.success(), "psql failed: {}", String::from_utf8_lossy(&output.stderr));
|
||||
}
|
||||
|
||||
fn psql_allow_failure(container: &DockerPostgres, sql: &str) -> bool {
|
||||
Command::new("docker")
|
||||
.args(["exec", &container.name, "psql", "-U", "postgres", "-d", "postgres", "-v", "ON_ERROR_STOP=1", "-c", sql])
|
||||
.output()
|
||||
.map(|output| output.status.success())
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
fn postgres_test_config(id: &str, port: u16) -> ConnectionConfig {
|
||||
ConnectionConfig {
|
||||
docs_notes_path: None,
|
||||
id: id.to_string(),
|
||||
name: id.to_string(),
|
||||
note: String::new(),
|
||||
db_type: DatabaseType::Postgres,
|
||||
driver_profile: None,
|
||||
driver_label: None,
|
||||
url_params: None,
|
||||
agent_java_options: Vec::new(),
|
||||
host: "127.0.0.1".to_string(),
|
||||
port,
|
||||
username: "postgres".to_string(),
|
||||
password: "postgres".to_string(),
|
||||
database: Some("postgres".to_string()),
|
||||
default_schema: None,
|
||||
visible_databases: None,
|
||||
visible_schemas: None,
|
||||
attached_databases: Vec::new(),
|
||||
init_script: None,
|
||||
color: None,
|
||||
transport_layers: Vec::new(),
|
||||
connect_timeout_secs: 5,
|
||||
query_timeout_secs: 30,
|
||||
idle_timeout_secs: 60,
|
||||
keepalive_interval_secs: 0,
|
||||
ssl: false,
|
||||
ca_cert_path: String::new(),
|
||||
client_cert_path: String::new(),
|
||||
client_key_path: String::new(),
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: None,
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
redis_cluster_nodes: String::new(),
|
||||
redis_key_separator: dbx_core::models::connection::default_redis_key_separator(),
|
||||
redis_scan_page_size: None,
|
||||
redis_database_aliases: Default::default(),
|
||||
etcd_endpoints: String::new(),
|
||||
gbase_server: String::new(),
|
||||
informix_server: String::new(),
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
one_time: false,
|
||||
save_password: true,
|
||||
read_only: false,
|
||||
is_production: false,
|
||||
production_databases: vec![],
|
||||
show_system_schemas: false,
|
||||
database_info: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Full database export of a partitioned schema must not duplicate any
|
||||
/// relation's `CREATE TABLE`/`CREATE FOREIGN TABLE`, must faithfully replay
|
||||
/// a partition-local dropped default, and the exported SQL must replay
|
||||
/// cleanly into a fresh schema (the actual "restore" check the review
|
||||
/// comment asked for).
|
||||
#[tokio::test]
|
||||
async fn database_export_of_partition_tree_has_no_duplicates_and_replays() {
|
||||
let Some(container) = start_docker_postgres("dbx-export-partition-ddl") else {
|
||||
return;
|
||||
};
|
||||
|
||||
psql(
|
||||
&container,
|
||||
"CREATE EXTENSION IF NOT EXISTS postgres_fdw;\
|
||||
CREATE SERVER dbx_export_loopback FOREIGN DATA WRAPPER postgres_fdw \
|
||||
OPTIONS (host 'localhost', dbname 'postgres', port '5432');\
|
||||
CREATE USER MAPPING FOR CURRENT_USER SERVER dbx_export_loopback \
|
||||
OPTIONS (user 'postgres', password 'postgres');\
|
||||
CREATE TABLE foreign_child_data (id integer, status text);\
|
||||
CREATE TABLE events (id bigint NOT NULL, status text DEFAULT 'active', created_at timestamp NOT NULL) \
|
||||
PARTITION BY RANGE (created_at);\
|
||||
CREATE TABLE events_2026 PARTITION OF events FOR VALUES FROM ('2026-01-01') TO ('2027-01-01');\
|
||||
CREATE TABLE events_2027 PARTITION OF events FOR VALUES FROM ('2027-01-01') TO ('2028-01-01') PARTITION BY RANGE (created_at);\
|
||||
CREATE TABLE events_2027_h1 PARTITION OF events_2027 FOR VALUES FROM ('2027-01-01') TO ('2027-07-01');\
|
||||
CREATE TABLE events_2027_h2 PARTITION OF events_2027 FOR VALUES FROM ('2027-07-01') TO ('2028-01-01');\
|
||||
CREATE FOREIGN TABLE events_foreign PARTITION OF events FOR VALUES FROM ('2028-01-01') TO ('2029-01-01') \
|
||||
SERVER dbx_export_loopback OPTIONS (schema_name 'public', table_name 'foreign_child_data');\
|
||||
ALTER TABLE ONLY events_2027_h1 ALTER COLUMN status DROP DEFAULT;\
|
||||
ALTER TABLE ONLY events_2027_h2 ALTER COLUMN status SET DEFAULT 'archived';",
|
||||
);
|
||||
|
||||
let dir = std::env::temp_dir().join(format!("dbx-export-partition-ddl-{}", uuid::Uuid::new_v4()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
|
||||
let state = AppState::new(storage);
|
||||
|
||||
let connection_id = "export-partition-ddl-conn";
|
||||
state.configs.write().await.insert(connection_id.to_string(), postgres_test_config(connection_id, container.port));
|
||||
|
||||
let file_path = dir.join("export.sql");
|
||||
let request = DatabaseExportRequest {
|
||||
export_id: format!("export-partition-ddl-{}", uuid::Uuid::new_v4()),
|
||||
connection_id: connection_id.to_string(),
|
||||
database: "postgres".to_string(),
|
||||
schema: "public".to_string(),
|
||||
file_path: file_path.to_string_lossy().to_string(),
|
||||
selected_tables: Vec::new(),
|
||||
excluded_tables: Vec::new(),
|
||||
include_structure: true,
|
||||
include_data: false,
|
||||
include_objects: false,
|
||||
include_create_database: false,
|
||||
drop_table_if_exists: false,
|
||||
omit_auto_increment: false,
|
||||
fail_on_error: true,
|
||||
snapshot_session_id: None,
|
||||
batch_size: 1000,
|
||||
};
|
||||
|
||||
export_database_sql_core(&state, &request, |_progress| {}).await.expect("export should succeed");
|
||||
|
||||
let exported = std::fs::read_to_string(&file_path).expect("read exported file");
|
||||
assert!(!exported.contains("-- ERROR"), "exported SQL should not contain errors:\n{exported}");
|
||||
|
||||
for table in ["events", "events_2026", "events_2027", "events_2027_h1", "events_2027_h2", "events_foreign"] {
|
||||
let needle = format!("TABLE \"public\".\"{table}\"");
|
||||
let count = exported.matches(&needle).count();
|
||||
assert_eq!(count, 1, "expected exactly one CREATE (FOREIGN) TABLE for {table}, found {count}:\n{exported}");
|
||||
}
|
||||
assert!(exported.contains("CREATE FOREIGN TABLE \"public\".\"events_foreign\""), "exported: {exported}");
|
||||
|
||||
assert!(
|
||||
exported.contains("ALTER TABLE ONLY \"public\".\"events_2027_h1\" ALTER COLUMN \"status\" DROP DEFAULT;"),
|
||||
"dropped default should be replayed as a standalone statement:\n{exported}"
|
||||
);
|
||||
assert!(
|
||||
exported.contains("\"status\" WITH OPTIONS DEFAULT 'archived'::text"),
|
||||
"overridden default should still render inline:\n{exported}"
|
||||
);
|
||||
// The untouched sibling (events_2026) must not gain either a drop or an
|
||||
// inline override — it purely inherits the parent's default.
|
||||
let events_2026_ddl_start = exported.find("PARTITION OF \"public\".\"events\"").expect("events_2026 ddl present");
|
||||
let events_2026_ddl = &exported[events_2026_ddl_start..(events_2026_ddl_start + 400).min(exported.len())];
|
||||
assert!(!events_2026_ddl.contains("DROP DEFAULT"), "events_2026 ddl: {events_2026_ddl}");
|
||||
assert!(!events_2026_ddl.contains("WITH OPTIONS DEFAULT"), "events_2026 ddl: {events_2026_ddl}");
|
||||
|
||||
// The actual "restore" check: replay the exported SQL into a fresh
|
||||
// schema and confirm no `relation already exists` (or any other) error.
|
||||
psql(&container, "CREATE SCHEMA replay");
|
||||
let replayable = exported.replace("\"public\".", "\"replay\".").replace("SCHEMA \"public\"", "SCHEMA \"replay\"");
|
||||
let replay_file = dir.join("replay.sql");
|
||||
std::fs::write(&replay_file, &replayable).unwrap();
|
||||
let output = Command::new("docker")
|
||||
.args(["cp", replay_file.to_str().unwrap(), &format!("{}:/tmp/replay.sql", container.name)])
|
||||
.output()
|
||||
.expect("copy replay file into container");
|
||||
assert!(output.status.success(), "docker cp failed: {}", String::from_utf8_lossy(&output.stderr));
|
||||
let replay_output = Command::new("docker")
|
||||
.args([
|
||||
"exec",
|
||||
&container.name,
|
||||
"psql",
|
||||
"-U",
|
||||
"postgres",
|
||||
"-d",
|
||||
"postgres",
|
||||
"-v",
|
||||
"ON_ERROR_STOP=1",
|
||||
"-f",
|
||||
"/tmp/replay.sql",
|
||||
])
|
||||
.output()
|
||||
.expect("replay exported sql");
|
||||
assert!(
|
||||
replay_output.status.success(),
|
||||
"replaying the exported SQL should succeed (no duplicate/`already exists` errors): {}",
|
||||
String::from_utf8_lossy(&replay_output.stderr)
|
||||
);
|
||||
}
|
||||
|
||||
/// Fetching a whole partition tree's DDL (the "view DDL" interactive path)
|
||||
/// must issue a small, roughly-constant number of queries — not one that
|
||||
/// scales linearly with the number of leaf partitions.
|
||||
#[tokio::test]
|
||||
async fn partition_tree_ddl_query_count_does_not_scale_with_partition_count() {
|
||||
let Some(container) = start_docker_postgres("dbx-partition-ddl-query-count") else {
|
||||
return;
|
||||
};
|
||||
|
||||
if !psql_allow_failure(&container, "ALTER SYSTEM SET log_statement = 'all'") {
|
||||
eprintln!("skipping query-count test: could not enable log_statement");
|
||||
return;
|
||||
}
|
||||
psql(&container, "SELECT pg_reload_conf()");
|
||||
|
||||
let build_tree = |leaves: u32| -> String {
|
||||
let mut sql = String::from(
|
||||
"CREATE TABLE t (id bigint NOT NULL, bucket integer NOT NULL, status text DEFAULT 'active') PARTITION BY RANGE (bucket);",
|
||||
);
|
||||
for i in 0..leaves {
|
||||
sql.push_str(&format!(
|
||||
"CREATE TABLE t_{i} PARTITION OF t FOR VALUES FROM ({start}) TO ({end});",
|
||||
i = i,
|
||||
start = i * 10,
|
||||
end = (i + 1) * 10
|
||||
));
|
||||
}
|
||||
sql
|
||||
};
|
||||
|
||||
let dir = std::env::temp_dir().join(format!("dbx-partition-query-count-{}", uuid::Uuid::new_v4()));
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
|
||||
let state = AppState::new(storage);
|
||||
let connection_id = "partition-query-count-conn";
|
||||
state.configs.write().await.insert(connection_id.to_string(), postgres_test_config(connection_id, container.port));
|
||||
|
||||
async fn fetch_ddl(state: &AppState, connection_id: &str) -> String {
|
||||
dbx_core::schema::get_table_display_ddl_core(state, connection_id, "postgres", "public", "t", None)
|
||||
.await
|
||||
.expect("fetch partition tree ddl")
|
||||
}
|
||||
|
||||
let logs_between = |container: &DockerPostgres, start_marker: &str, end_marker: &str| -> usize {
|
||||
let output = Command::new("docker").args(["logs", &container.name]).output().expect("docker logs");
|
||||
let combined =
|
||||
format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr));
|
||||
let start = combined.rfind(start_marker).map(|i| i + start_marker.len());
|
||||
let end = combined.rfind(end_marker);
|
||||
match (start, end) {
|
||||
(Some(s), Some(e)) if e > s => {
|
||||
combined[s..e].lines().filter(|line| line.contains("statement:") || line.contains("execute")).count()
|
||||
}
|
||||
_ => usize::MAX,
|
||||
}
|
||||
};
|
||||
|
||||
psql(&container, &format!("DROP TABLE IF EXISTS t CASCADE; {}", build_tree(3)));
|
||||
psql(&container, "SELECT 'MARKER_SMALL_START'");
|
||||
let small_ddl = fetch_ddl(&state, connection_id).await;
|
||||
psql(&container, "SELECT 'MARKER_SMALL_END'");
|
||||
assert_eq!(small_ddl.matches("PARTITION OF").count(), 3, "small tree ddl: {small_ddl}");
|
||||
|
||||
psql(&container, &format!("DROP TABLE IF EXISTS t CASCADE; {}", build_tree(40)));
|
||||
psql(&container, "SELECT 'MARKER_LARGE_START'");
|
||||
let large_ddl = fetch_ddl(&state, connection_id).await;
|
||||
psql(&container, "SELECT 'MARKER_LARGE_END'");
|
||||
assert_eq!(large_ddl.matches("PARTITION OF").count(), 40, "large tree ddl: {large_ddl}");
|
||||
|
||||
// Give the container's log a moment to flush before reading it back.
|
||||
std::thread::sleep(std::time::Duration::from_millis(500));
|
||||
let small_count = logs_between(&container, "MARKER_SMALL_START", "MARKER_SMALL_END");
|
||||
let large_count = logs_between(&container, "MARKER_LARGE_START", "MARKER_LARGE_END");
|
||||
|
||||
assert_ne!(small_count, usize::MAX, "could not locate small-tree marker window in container logs");
|
||||
assert_ne!(large_count, usize::MAX, "could not locate large-tree marker window in container logs");
|
||||
// A naive per-relation recursion issues ~9 queries per relation, so a
|
||||
// 40-leaf tree (41 relations) would log roughly 13x the 3-leaf tree's
|
||||
// (4 relations) query count. The batched implementation should stay
|
||||
// close to flat instead.
|
||||
assert!(
|
||||
large_count <= small_count + 10,
|
||||
"query count should not scale with partition count: small={small_count} large={large_count}"
|
||||
);
|
||||
}
|
||||
@@ -63,7 +63,7 @@ async fn cloudberry_metadata_and_ddl_round_trip() {
|
||||
("external_events", "CREATE FOREIGN TABLE"),
|
||||
];
|
||||
for (table, expected) in cases {
|
||||
let ddl = schema::cloudberry_ddl(&pool, &source_schema, table).await?;
|
||||
let ddl = schema::cloudberry_ddl(&pool, &source_schema, table, false).await?;
|
||||
assert!(ddl.contains(expected), "{table} DDL did not contain {expected}: {ddl}");
|
||||
if table == "external_events" {
|
||||
assert!(ddl.contains("SERVER \"gp_exttable_server\""), "external DDL: {ddl}");
|
||||
|
||||
@@ -46,7 +46,7 @@ async fn opentenbase_metadata_and_ddl_round_trip() {
|
||||
("replicated_dimensions", "DISTRIBUTE BY REPLICATION"),
|
||||
];
|
||||
for (table, expected) in cases {
|
||||
let ddl = schema::opentenbase_ddl(&pool, &source_schema, table).await?;
|
||||
let ddl = schema::opentenbase_ddl(&pool, &source_schema, table, false).await?;
|
||||
if !ddl.contains(expected) {
|
||||
return Err(format!("{table} DDL did not contain {expected}: {ddl}"));
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user