feat(postgres): support concurrent index creation
This commit is contained in:
1 parent
ebf412b144
commit
aa6d60533c
38 files changed
+1917
-34
No files matched your search
@@ -5473,6 +5473,30 @@ pub async fn list_indexes(pool: &Pool, schema: &str, table: &str) -> Result<Vec<
|
||||
}
|
||||
}
|
||||
|
||||
/// Names of same-table indexes whose `pg_index.indisvalid` is `false`.
|
||||
///
|
||||
/// A cancelled `CREATE INDEX CONCURRENTLY` leaves an INVALID index behind with
|
||||
/// the requested name; until it is dropped, any retry of the same build fails
|
||||
/// with `relation already exists`. The structure editor consults this before
|
||||
/// applying a concurrent build so it can surface the leftover explicitly
|
||||
/// instead of failing silently.
|
||||
const POSTGRES_INVALID_INDEXES_SQL: &str = "SELECT idx.relname \
|
||||
FROM pg_catalog.pg_index ix \
|
||||
JOIN pg_catalog.pg_class idx ON idx.oid = ix.indexrelid \
|
||||
JOIN pg_catalog.pg_class t ON t.oid = ix.indrelid \
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = t.relnamespace \
|
||||
WHERE n.nspname = $1 AND t.relname = $2 AND ix.indisvalid = false \
|
||||
ORDER BY idx.relname";
|
||||
|
||||
pub async fn list_invalid_indexes(pool: &Pool, schema: &str, table: &str) -> Result<Vec<String>, String> {
|
||||
let schema = if schema.is_empty() { "public" } else { schema };
|
||||
let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
|
||||
let rows = postgres_query_cached(&client, POSTGRES_INVALID_INDEXES_SQL, &[&schema, &table])
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(rows.iter().map(|row| pg_row_try_string(row, 0)).collect())
|
||||
}
|
||||
|
||||
fn postgres_foreign_keys_sql() -> &'static str {
|
||||
"SELECT fk.constraint_name, fk.column_name, \
|
||||
pk.table_schema AS ref_schema, pk.table_name AS ref_table, pk.column_name AS ref_column, \
|
||||
|
||||
@@ -6273,6 +6273,63 @@ pub async fn list_partitions_core(
|
||||
.await
|
||||
}
|
||||
|
||||
/// PostgreSQL partition classification of a single table, used by the table
|
||||
/// structure editor to decide whether `CREATE INDEX CONCURRENTLY` applies.
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct TablePartitionStatus {
|
||||
/// The table is a partitioned parent (`pg_class.relkind = 'p'`); PostgreSQL
|
||||
/// rejects `CREATE INDEX CONCURRENTLY` directly on it — the supported
|
||||
/// approach is building child indexes concurrently and attaching them.
|
||||
pub is_partitioned_parent: bool,
|
||||
/// The table is itself a partition of a parent (`pg_class.relispartition`).
|
||||
pub is_partition: bool,
|
||||
}
|
||||
|
||||
pub async fn table_partition_status_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
) -> Result<TablePartitionStatus, String> {
|
||||
retry_metadata_connection(state, connection_id, Some(database), || async {
|
||||
let pool_key = state.get_or_create_metadata_pool_for_session(connection_id, Some(database), None).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(&pool_key) {
|
||||
Some(PoolKind::Postgres(pool)) => {
|
||||
let info = db::postgres::get_table_partition_info(pool, schema, table).await?;
|
||||
Ok(TablePartitionStatus {
|
||||
is_partitioned_parent: info.key.is_some() && !info.is_partition,
|
||||
is_partition: info.is_partition,
|
||||
})
|
||||
}
|
||||
_ => Ok(TablePartitionStatus::default()),
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
/// Same-table index names whose `pg_index.indisvalid` is `false` (left behind
|
||||
/// by a cancelled `CREATE INDEX CONCURRENTLY`). Empty for non-PostgreSQL pools.
|
||||
pub async fn list_invalid_indexes_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
schema: &str,
|
||||
table: &str,
|
||||
) -> Result<Vec<String>, String> {
|
||||
retry_metadata_connection(state, connection_id, Some(database), || async {
|
||||
let pool_key = state.get_or_create_metadata_pool_for_session(connection_id, Some(database), None).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(&pool_key) {
|
||||
Some(PoolKind::Postgres(pool)) => db::postgres::list_invalid_indexes(pool, schema, table).await,
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn list_subpartitions_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
|
||||
@@ -25,9 +25,16 @@ use comments::build_table_comment_sql;
|
||||
use foreign_keys::build_foreign_key_sql;
|
||||
use indexes::build_index_sql;
|
||||
use triggers::build_trigger_sql;
|
||||
use validation::validate_draft;
|
||||
use validation::{validate_concurrent_index_scope, validate_draft};
|
||||
|
||||
pub fn build_table_structure_change_sql(options: TableStructureSqlOptions) -> TableStructureSqlResult {
|
||||
// Fail closed: an unsupported concurrent-index request (existing index, or
|
||||
// partitioned parent table) must never degrade into blocking index DDL
|
||||
// behind the caller's back, so the whole plan is refused up front.
|
||||
let concurrent_errors = validate_concurrent_index_scope(&options);
|
||||
if !concurrent_errors.is_empty() {
|
||||
return TableStructureSqlResult { statements: Vec::new(), warnings: concurrent_errors };
|
||||
}
|
||||
let mut warnings = validate_draft(&options);
|
||||
let mut statements = build_column_sql(&options, &mut warnings);
|
||||
statements.extend(build_index_sql(&options, &mut warnings));
|
||||
|
||||
@@ -9,12 +9,16 @@ use super::indexes::build_create_index_statements;
|
||||
use super::triggers::build_trigger_sql;
|
||||
use super::types::{TableStructureSqlOptions, TableStructureSqlResult};
|
||||
use super::util::{clean, format_default_for_sql, normalize_default, qualified_table, quote_ident, quote_string};
|
||||
use super::validation::{validate_columns, validate_dameng_identity};
|
||||
use super::validation::{validate_columns, validate_concurrent_index_scope, validate_dameng_identity};
|
||||
use crate::models::connection::DatabaseType;
|
||||
|
||||
pub fn build_create_table_sql(mut options: TableStructureSqlOptions) -> TableStructureSqlResult {
|
||||
options.table_name = clean(&options.table_name);
|
||||
let mut warnings = Vec::new();
|
||||
// Fail closed: a concurrent-index request on a partitioned parent (or on an
|
||||
// existing index in a hand-built draft) is refused up front instead of
|
||||
// degrading into blocking index DDL.
|
||||
warnings.extend(validate_concurrent_index_scope(&options));
|
||||
if options.table_name.is_empty() {
|
||||
warnings.push("Table name is required.".to_string());
|
||||
}
|
||||
@@ -183,6 +187,7 @@ pub fn build_create_table_sql(mut options: TableStructureSqlOptions) -> TableStr
|
||||
options.schema.as_deref(),
|
||||
&options.table_name,
|
||||
false,
|
||||
capabilities.index_concurrent,
|
||||
));
|
||||
}
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ pub(super) struct TableStructureCapabilities {
|
||||
pub(super) index_include: bool,
|
||||
pub(super) index_filter: bool,
|
||||
pub(super) index_comment: bool,
|
||||
pub(super) index_concurrent: bool,
|
||||
pub(super) alter_primary_key: bool,
|
||||
pub(super) foreign_key: bool,
|
||||
}
|
||||
@@ -59,6 +60,7 @@ impl Default for TableStructureCapabilities {
|
||||
index_include: false,
|
||||
index_filter: false,
|
||||
index_comment: false,
|
||||
index_concurrent: false,
|
||||
alter_primary_key: false,
|
||||
foreign_key: false,
|
||||
}
|
||||
@@ -118,24 +120,35 @@ pub(super) fn capabilities_for(database_type: Option<DatabaseType>) -> TableStru
|
||||
| DatabaseType::Vastbase
|
||||
| DatabaseType::Kingbase
|
||||
| DatabaseType::Firebird,
|
||||
) => TableStructureCapabilities {
|
||||
dialect: StructureDialect::Postgres,
|
||||
add_column: true,
|
||||
drop_column: true,
|
||||
rename_column: true,
|
||||
alter_existing_column: true,
|
||||
comment: true,
|
||||
create_index: true,
|
||||
drop_index: true,
|
||||
rebuild_index: true,
|
||||
index_type: true,
|
||||
index_include: true,
|
||||
index_filter: true,
|
||||
index_comment: true,
|
||||
alter_primary_key: true,
|
||||
foreign_key: true,
|
||||
..base
|
||||
},
|
||||
) => {
|
||||
let mut caps = TableStructureCapabilities {
|
||||
dialect: StructureDialect::Postgres,
|
||||
add_column: true,
|
||||
drop_column: true,
|
||||
rename_column: true,
|
||||
alter_existing_column: true,
|
||||
comment: true,
|
||||
create_index: true,
|
||||
drop_index: true,
|
||||
rebuild_index: true,
|
||||
index_type: true,
|
||||
index_include: true,
|
||||
index_filter: true,
|
||||
index_comment: true,
|
||||
alter_primary_key: true,
|
||||
foreign_key: true,
|
||||
..base
|
||||
};
|
||||
// CREATE INDEX CONCURRENTLY is PostgreSQL-specific. Other PostgreSQL-family
|
||||
// engines (Kingbase, GaussDB, OpenGauss, HighGo, UXDB, Vastbase, KWDB,
|
||||
// Firebird) share the Postgres dialect but have not been verified against a
|
||||
// real environment, so they keep this capability off until individually
|
||||
// validated.
|
||||
if database_type == Some(DatabaseType::Postgres) {
|
||||
caps.index_concurrent = true;
|
||||
}
|
||||
caps
|
||||
}
|
||||
Some(DatabaseType::Questdb) => TableStructureCapabilities {
|
||||
dialect: StructureDialect::Questdb,
|
||||
add_column: true,
|
||||
@@ -151,6 +164,7 @@ pub(super) fn capabilities_for(database_type: Option<DatabaseType>) -> TableStru
|
||||
index_include: false,
|
||||
index_filter: false,
|
||||
index_comment: false,
|
||||
index_concurrent: false,
|
||||
alter_primary_key: false,
|
||||
foreign_key: false,
|
||||
},
|
||||
|
||||
@@ -66,6 +66,7 @@ pub(super) fn build_index_sql(options: &TableStructureSqlOptions, warnings: &mut
|
||||
options.schema.as_deref(),
|
||||
&options.table_name,
|
||||
or_replace,
|
||||
capabilities.index_concurrent,
|
||||
));
|
||||
continue;
|
||||
}
|
||||
@@ -82,6 +83,7 @@ pub(super) fn build_index_sql(options: &TableStructureSqlOptions, warnings: &mut
|
||||
options.schema.as_deref(),
|
||||
&options.table_name,
|
||||
false,
|
||||
capabilities.index_concurrent,
|
||||
));
|
||||
}
|
||||
|
||||
@@ -217,6 +219,7 @@ pub(super) fn build_create_index_statements(
|
||||
schema: Option<&str>,
|
||||
table_name: &str,
|
||||
or_replace: bool,
|
||||
concurrently_supported: bool,
|
||||
) -> Vec<String> {
|
||||
let capabilities = capabilities_for(database_type_for_dialect(dialect));
|
||||
let name = clean(&index.name);
|
||||
@@ -226,6 +229,14 @@ pub(super) fn build_create_index_statements(
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
// `CREATE INDEX CONCURRENTLY` is PostgreSQL-only: the flag is honored only when
|
||||
// the caller's real database type advertises the capability, so a stale or
|
||||
// forged `concurrently: true` from another engine is ignored and the original
|
||||
// SQL is generated unchanged. Unsupported concurrent requests (existing index,
|
||||
// partitioned parent) are rejected up front by `validate_concurrent_index_scope`
|
||||
// before any statement is built — this function never downgrades them.
|
||||
let concurrently = index.concurrently && concurrently_supported && dialect == StructureDialect::Postgres;
|
||||
|
||||
let unique = if index.is_unique { "UNIQUE " } else { "" };
|
||||
let replace = if or_replace { "OR REPLACE " } else { "" };
|
||||
let key_is_expression = key_expression_flags(index, &columns);
|
||||
@@ -296,8 +307,9 @@ pub(super) fn build_create_index_statements(
|
||||
let create_table =
|
||||
if dialect == StructureDialect::Sqlite { quote_ident(dialect, table_name) } else { table.to_string() };
|
||||
let create_sql = if dialect == StructureDialect::Postgres {
|
||||
let concurrent_clause = if concurrently { "CONCURRENTLY " } else { "" };
|
||||
format!(
|
||||
"CREATE {replace}{unique}{type_prefix}INDEX {index_name} ON {create_table}{using_clause} ({cols}){include_clause}{where_clause};"
|
||||
"CREATE {replace}{unique}INDEX {concurrent_clause}{index_name} ON {create_table}{using_clause} ({cols}){include_clause}{where_clause};"
|
||||
)
|
||||
} else {
|
||||
format!(
|
||||
|
||||
@@ -1516,6 +1516,7 @@ mod tests {
|
||||
triggers: Vec::new(),
|
||||
table_comment: None,
|
||||
original_table_comment: None,
|
||||
partitioned: false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1691,6 +1692,7 @@ mod tests {
|
||||
index_type: String::new(),
|
||||
included_columns: Vec::new(),
|
||||
comment: String::new(),
|
||||
concurrently: false,
|
||||
original: None,
|
||||
marked_for_drop: false,
|
||||
});
|
||||
@@ -2009,6 +2011,7 @@ mod tests {
|
||||
index_type: String::new(),
|
||||
included_columns: Vec::new(),
|
||||
comment: String::new(),
|
||||
concurrently: false,
|
||||
original: Some(IndexInfo {
|
||||
name: "uq_items_value".to_string(),
|
||||
columns: vec!["value".to_string()],
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -96,6 +96,8 @@ pub struct EditableStructureIndex {
|
||||
pub included_columns: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub comment: String,
|
||||
#[serde(default)]
|
||||
pub concurrently: bool,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub original: Option<IndexInfo>,
|
||||
#[serde(default)]
|
||||
@@ -208,6 +210,13 @@ pub struct TableStructureSqlOptions {
|
||||
pub table_comment: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub original_table_comment: Option<String>,
|
||||
/// Whether the target table is a partitioned parent table (PostgreSQL
|
||||
/// `relkind = 'p'`). PostgreSQL rejects `CREATE INDEX CONCURRENTLY` on
|
||||
/// partitioned parents, so the builder refuses such a request up front
|
||||
/// (`validate_concurrent_index_scope`) instead of emitting SQL the server
|
||||
/// will reject or downgrading to a blocking `CREATE INDEX`.
|
||||
#[serde(default)]
|
||||
pub partitioned: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
|
||||
@@ -4,6 +4,40 @@ use super::indexes::has_existing_index_change;
|
||||
use super::types::{EditableStructureColumn, TableStructureSqlOptions};
|
||||
use super::util::clean;
|
||||
|
||||
/// Hard errors for `CREATE INDEX CONCURRENTLY` requests the editor cannot
|
||||
/// honor safely. Unlike the advisory warnings in [`validate_draft`], these
|
||||
/// reject the whole save plan: the caller asked for a non-blocking build and
|
||||
/// the only way to honor that is the requested concurrent DDL, so falling back
|
||||
/// to blocking index DDL must never happen silently.
|
||||
///
|
||||
/// Only real PostgreSQL advertises `index_concurrent`, so stale/forged
|
||||
/// `concurrently` flags on other engines are ignored here (they cannot express
|
||||
/// a concurrent request in the first place).
|
||||
pub(super) fn validate_concurrent_index_scope(options: &TableStructureSqlOptions) -> Vec<String> {
|
||||
let mut errors = Vec::new();
|
||||
let capabilities = capabilities_for(options.database_type);
|
||||
if !capabilities.index_concurrent || capabilities.dialect != StructureDialect::Postgres {
|
||||
return errors;
|
||||
}
|
||||
for index in options.indexes.iter().filter(|index| !index.marked_for_drop) {
|
||||
if !index.concurrently {
|
||||
continue;
|
||||
}
|
||||
if index.original.is_some() {
|
||||
errors.push(format!(
|
||||
"CREATE INDEX CONCURRENTLY is only supported for newly created indexes. Editing an existing index \"{}\" with Concurrent enabled is not supported.",
|
||||
index.name
|
||||
));
|
||||
} else if options.partitioned {
|
||||
errors.push(
|
||||
"CREATE INDEX CONCURRENTLY is not supported on PostgreSQL partitioned parent tables. Create indexes concurrently on individual partitions and attach them separately."
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
}
|
||||
errors
|
||||
}
|
||||
|
||||
pub(super) fn validate_draft(options: &TableStructureSqlOptions) -> Vec<String> {
|
||||
let mut warnings = Vec::new();
|
||||
let active_columns: Vec<_> = options.columns.iter().filter(|column| !column.marked_for_drop).collect();
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
use std::time::Duration;
|
||||
|
||||
use dbx_core::db::postgres;
|
||||
use dbx_core::models::connection::DatabaseType;
|
||||
use dbx_core::table_structure_sql::{
|
||||
build_table_structure_change_sql, EditableStructureIndex, TableStructureSqlOptions,
|
||||
};
|
||||
|
||||
fn live_postgres_url() -> String {
|
||||
let host = std::env::var("DBX_LIVE_POSTGRES_HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
|
||||
let port = std::env::var("DBX_LIVE_POSTGRES_PORT").ok().and_then(|value| value.parse().ok()).unwrap_or(5432);
|
||||
let user = std::env::var("DBX_LIVE_POSTGRES_USER").unwrap_or_else(|_| "postgres".to_string());
|
||||
let password = std::env::var("DBX_LIVE_POSTGRES_PASSWORD").unwrap_or_default();
|
||||
let database = std::env::var("DBX_LIVE_POSTGRES_DATABASE").unwrap_or_else(|_| "postgres".to_string());
|
||||
format!("postgresql://{user}:{password}@{host}:{port}/{database}")
|
||||
}
|
||||
|
||||
fn concurrent_index(columns: Vec<&str>) -> EditableStructureIndex {
|
||||
EditableStructureIndex {
|
||||
id: "idx_users_email".to_string(),
|
||||
name: "idx_users_email".to_string(),
|
||||
columns: columns.into_iter().map(str::to_string).collect(),
|
||||
is_unique: false,
|
||||
is_primary: false,
|
||||
filter: String::new(),
|
||||
index_type: String::new(),
|
||||
included_columns: Vec::new(),
|
||||
comment: String::new(),
|
||||
concurrently: true,
|
||||
original: None,
|
||||
marked_for_drop: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// `CREATE INDEX CONCURRENTLY` generated by the table structure editor must
|
||||
/// succeed on a real PostgreSQL server and leave a valid index
|
||||
/// (`pg_index.indisvalid = true`).
|
||||
#[tokio::test]
|
||||
#[ignore = "requires DBX_LIVE_POSTGRES_HOST/PORT/USER/PASSWORD/DATABASE pointing at a writable PostgreSQL database"]
|
||||
async fn live_postgres_concurrent_index_builds_valid_index() {
|
||||
let pool = postgres::connect(&live_postgres_url(), Duration::from_secs(10)).await.expect("connect PostgreSQL");
|
||||
let suffix = uuid::Uuid::new_v4().simple().to_string();
|
||||
let schema = format!("dbx_cic_{}", &suffix[..8]);
|
||||
let cleanup = vec![format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE")];
|
||||
let _ = postgres::execute_batch(&pool, &cleanup).await;
|
||||
let setup = vec![
|
||||
format!("CREATE SCHEMA \"{schema}\""),
|
||||
format!("CREATE TABLE \"{schema}\".users (id integer PRIMARY KEY, email text, name text, status text)"),
|
||||
];
|
||||
postgres::execute_batch(&pool, &setup).await.expect("create live test schema");
|
||||
|
||||
let result = build_table_structure_change_sql(TableStructureSqlOptions {
|
||||
database_type: Some(DatabaseType::Postgres),
|
||||
schema: Some(schema.clone()),
|
||||
table_name: "users".to_string(),
|
||||
columns: Vec::new(),
|
||||
indexes: vec![concurrent_index(vec!["email"])],
|
||||
foreign_keys: Vec::new(),
|
||||
triggers: Vec::new(),
|
||||
table_comment: None,
|
||||
original_table_comment: None,
|
||||
partitioned: false,
|
||||
});
|
||||
assert!(result.warnings.is_empty(), "{:?}", result.warnings);
|
||||
assert_eq!(
|
||||
result.statements,
|
||||
vec![format!("CREATE INDEX CONCURRENTLY \"idx_users_email\" ON \"{schema}\".\"users\" (\"email\");")]
|
||||
);
|
||||
|
||||
postgres::execute_batch(&pool, &result.statements).await.expect("execute concurrent index build");
|
||||
let client = pool.get().await.expect("checkout connection");
|
||||
let row = client
|
||||
.query_one(
|
||||
"SELECT indisvalid FROM pg_index WHERE indexrelid = to_regclass($1)",
|
||||
&[&format!("{schema}.idx_users_email")],
|
||||
)
|
||||
.await
|
||||
.expect("read pg_index");
|
||||
let valid: bool = row.get(0);
|
||||
assert!(valid, "concurrently-built index must be valid");
|
||||
|
||||
postgres::execute_batch(&pool, &cleanup).await.expect("cleanup live test schema");
|
||||
}
|
||||
|
||||
/// PostgreSQL forbids `CREATE INDEX CONCURRENTLY` inside an explicit
|
||||
/// transaction block; the editor relies on per-statement autocommit, so the
|
||||
/// generated statement must only ever run outside `BEGIN`.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires DBX_LIVE_POSTGRES_HOST/PORT/USER/PASSWORD/DATABASE pointing at a writable PostgreSQL database"]
|
||||
async fn live_postgres_concurrent_index_rejects_transaction_block() {
|
||||
let pool = postgres::connect(&live_postgres_url(), Duration::from_secs(10)).await.expect("connect PostgreSQL");
|
||||
let suffix = uuid::Uuid::new_v4().simple().to_string();
|
||||
let schema = format!("dbx_cic_{}", &suffix[..8]);
|
||||
let cleanup = vec![format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE")];
|
||||
let _ = postgres::execute_batch(&pool, &cleanup).await;
|
||||
let setup = vec![
|
||||
format!("CREATE SCHEMA \"{schema}\""),
|
||||
format!("CREATE TABLE \"{schema}\".users (id integer PRIMARY KEY, email text)"),
|
||||
];
|
||||
postgres::execute_batch(&pool, &setup).await.expect("create live test schema");
|
||||
|
||||
let client = pool.get().await.expect("checkout connection");
|
||||
client.batch_execute("BEGIN").await.expect("begin transaction");
|
||||
let err = client
|
||||
.batch_execute(&format!("CREATE INDEX CONCURRENTLY \"idx_users_email\" ON \"{schema}\".\"users\" (\"email\")"))
|
||||
.await
|
||||
.expect_err("CREATE INDEX CONCURRENTLY must fail inside a transaction block");
|
||||
let _ = client.batch_execute("ROLLBACK").await;
|
||||
let message = err.as_db_error().map(|db_error| db_error.message().to_string()).unwrap_or_else(|| err.to_string());
|
||||
assert!(
|
||||
message.contains("cannot run inside a transaction block"),
|
||||
"expected transaction-block error, got: {message}"
|
||||
);
|
||||
|
||||
postgres::execute_batch(&pool, &cleanup).await.expect("cleanup live test schema");
|
||||
}
|
||||
|
||||
/// A cancelled `CREATE INDEX CONCURRENTLY` leaves a same-name INVALID index
|
||||
/// behind (`pg_index.indisvalid = false`); `list_invalid_indexes` must report
|
||||
/// it so the editor can block a doomed retry and point the user at the fix.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires DBX_LIVE_POSTGRES_HOST/PORT/USER/PASSWORD/DATABASE pointing at a writable PostgreSQL database"]
|
||||
async fn live_postgres_invalid_index_leftover_is_detected() {
|
||||
let pool = postgres::connect(&live_postgres_url(), Duration::from_secs(10)).await.expect("connect PostgreSQL");
|
||||
let suffix = uuid::Uuid::new_v4().simple().to_string();
|
||||
let schema = format!("dbx_cic_{}", &suffix[..8]);
|
||||
let cleanup = vec![format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE")];
|
||||
let _ = postgres::execute_batch(&pool, &cleanup).await;
|
||||
let setup = vec![
|
||||
format!("CREATE SCHEMA \"{schema}\""),
|
||||
format!("CREATE TABLE \"{schema}\".users (id integer PRIMARY KEY, email text)"),
|
||||
// Simulate the leftover of a cancelled concurrent build. PostgreSQL
|
||||
// does not accept a schema-qualified index name in CREATE INDEX; the
|
||||
// index inherits the table's schema, so only the table is qualified.
|
||||
format!("CREATE INDEX \"idx_users_email\" ON \"{schema}\".\"users\" (\"email\")"),
|
||||
];
|
||||
postgres::execute_batch(&pool, &setup).await.expect("create live test schema");
|
||||
|
||||
let invalid = postgres::list_invalid_indexes(&pool, &schema, "users").await.expect("list invalid indexes");
|
||||
assert!(invalid.is_empty(), "fresh index must be valid, got: {invalid:?}");
|
||||
|
||||
// Flip indisvalid to false the same way an interrupted concurrent build does.
|
||||
let client = pool.get().await.expect("checkout connection");
|
||||
client
|
||||
.batch_execute(&format!(
|
||||
"UPDATE pg_index SET indisvalid = false WHERE indexrelid = to_regclass('{schema}.idx_users_email')"
|
||||
))
|
||||
.await
|
||||
.expect("mark index invalid");
|
||||
|
||||
let invalid = postgres::list_invalid_indexes(&pool, &schema, "users").await.expect("list invalid indexes");
|
||||
assert_eq!(invalid, vec!["idx_users_email".to_string()], "invalid leftover must be reported");
|
||||
|
||||
postgres::execute_batch(&pool, &cleanup).await.expect("cleanup live test schema");
|
||||
}
|
||||
|
||||
/// The structure editor must never emit `CREATE INDEX CONCURRENTLY` for a
|
||||
/// partitioned parent table — the builder rejects the request up front
|
||||
/// (fail closed) and no SQL reaches the server at all, instead of downgrading
|
||||
/// to a blocking `CREATE INDEX`.
|
||||
#[tokio::test]
|
||||
#[ignore = "requires DBX_LIVE_POSTGRES_HOST/PORT/USER/PASSWORD/DATABASE pointing at a writable PostgreSQL database"]
|
||||
async fn live_postgres_partitioned_parent_concurrent_request_rejected() {
|
||||
let pool = postgres::connect(&live_postgres_url(), Duration::from_secs(10)).await.expect("connect PostgreSQL");
|
||||
let suffix = uuid::Uuid::new_v4().simple().to_string();
|
||||
let schema = format!("dbx_cic_{}", &suffix[..8]);
|
||||
let cleanup = vec![format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE")];
|
||||
let _ = postgres::execute_batch(&pool, &cleanup).await;
|
||||
let setup = vec![
|
||||
format!("CREATE SCHEMA \"{schema}\""),
|
||||
format!(
|
||||
"CREATE TABLE \"{schema}\".events (id integer, created_at date) PARTITION BY RANGE (created_at)"
|
||||
),
|
||||
format!("CREATE TABLE \"{schema}\".events_2026 PARTITION OF \"{schema}\".events FOR VALUES FROM ('2026-01-01') TO ('2027-01-01')"),
|
||||
];
|
||||
postgres::execute_batch(&pool, &setup).await.expect("create live test schema");
|
||||
|
||||
let mut idx = concurrent_index(vec!["id"]);
|
||||
idx.name = "idx_events_id".to_string();
|
||||
idx.id = "idx_events_id".to_string();
|
||||
let result = build_table_structure_change_sql(TableStructureSqlOptions {
|
||||
database_type: Some(DatabaseType::Postgres),
|
||||
schema: Some(schema.clone()),
|
||||
table_name: "events".to_string(),
|
||||
columns: Vec::new(),
|
||||
indexes: vec![idx],
|
||||
foreign_keys: Vec::new(),
|
||||
triggers: Vec::new(),
|
||||
table_comment: None,
|
||||
original_table_comment: None,
|
||||
partitioned: true,
|
||||
});
|
||||
assert_eq!(
|
||||
result.warnings,
|
||||
vec![
|
||||
"CREATE INDEX CONCURRENTLY is not supported on PostgreSQL partitioned parent tables. Create indexes concurrently on individual partitions and attach them separately."
|
||||
]
|
||||
);
|
||||
assert!(
|
||||
result.statements.is_empty(),
|
||||
"no SQL may be generated for an unsupported concurrent request, got: {:?}",
|
||||
result.statements
|
||||
);
|
||||
// Nothing was executed: the partitioned parent still has no such index.
|
||||
let client = pool.get().await.expect("checkout connection");
|
||||
let row = client
|
||||
.query_one(
|
||||
"SELECT count(*) FROM pg_catalog.pg_class WHERE relname = $1 AND relnamespace = to_regnamespace($2)",
|
||||
&[&"idx_events_id", &schema],
|
||||
)
|
||||
.await
|
||||
.expect("count index relations");
|
||||
let count: i64 = row.get(0);
|
||||
assert_eq!(count, 0, "rejected request must not create any index on the server");
|
||||
|
||||
postgres::execute_batch(&pool, &cleanup).await.expect("cleanup live test schema");
|
||||
}
|
||||
@@ -1005,6 +1005,7 @@ async fn live_sqlserver_table_structure_default_changes_drop_existing_constraint
|
||||
triggers: Vec::new(),
|
||||
table_comment: None,
|
||||
original_table_comment: None,
|
||||
partitioned: false,
|
||||
});
|
||||
assert_eq!(result.warnings, Vec::<String>::new());
|
||||
assert_eq!(result.statements.len(), 4);
|
||||
|
||||
@@ -420,6 +420,8 @@ async fn main() {
|
||||
.route("/schema/triggers", get(routes::schema::list_triggers))
|
||||
.route("/schema/constraints", get(routes::schema::list_constraints))
|
||||
.route("/schema/partitions", get(routes::schema::list_partitions))
|
||||
.route("/schema/table-partition-status", get(routes::schema::get_table_partition_status))
|
||||
.route("/schema/invalid-indexes", get(routes::schema::list_invalid_indexes))
|
||||
.route("/schema/subpartitions", get(routes::schema::list_subpartitions))
|
||||
.route("/schema/functions", get(routes::schema::list_functions))
|
||||
.route("/schema/sequences", get(routes::schema::list_sequences))
|
||||
|
||||
@@ -623,6 +623,32 @@ pub async fn list_partitions(
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError::from(e.to_string()))?))
|
||||
}
|
||||
|
||||
pub async fn get_table_partition_status(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Query(q): Query<SchemaQuery>,
|
||||
) -> Result<Json<dbx_core::schema::TablePartitionStatus>, AppError> {
|
||||
let database = q.database.as_deref().unwrap_or("");
|
||||
let schema = q.schema.as_deref().unwrap_or("");
|
||||
let table = q.table.as_deref().unwrap_or("");
|
||||
dbx_core::schema::table_partition_status_core(&state.app, &q.connection_id, database, schema, table)
|
||||
.await
|
||||
.map(Json)
|
||||
.map_err(AppError::from)
|
||||
}
|
||||
|
||||
pub async fn list_invalid_indexes(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Query(q): Query<SchemaQuery>,
|
||||
) -> Result<Json<Vec<String>>, AppError> {
|
||||
let database = q.database.as_deref().unwrap_or("");
|
||||
let schema = q.schema.as_deref().unwrap_or("");
|
||||
let table = q.table.as_deref().unwrap_or("");
|
||||
dbx_core::schema::list_invalid_indexes_core(&state.app, &q.connection_id, database, schema, table)
|
||||
.await
|
||||
.map(Json)
|
||||
.map_err(AppError::from)
|
||||
}
|
||||
|
||||
pub async fn list_subpartitions(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Query(q): Query<SchemaQuery>,
|
||||
|
||||
Reference in new issue
Block a user