use super::{CatalogIndexRow, ColumnType, DropKind, DropStmt, Engine, SQLError, SQLResult};
use crate::engine_capabilities::RelationResolution;
mod index_dependencies;
pub(in crate::sql) fn run_drop(engine: &Engine, stmt: DropStmt) -> Result<SQLResult, SQLError> {
if stmt.kind == DropKind::Table {
for name in &stmt.names {
if let Some(canonical) = crate::sql::resolve_age_label_relation_name(engine, name)? {
let relation =
crate::RelationIdentity::from_legacy_name(&canonical).map_err(|error| {
SQLError::Internal(format!(
"resolve AGE label relation `{canonical}` for DROP TABLE: {error}"
))
})?;
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"table \"{}\" is for label \"{}\"",
relation.name, relation.name
),
});
}
}
}
if stmt.kind == DropKind::Index {
return run_drop_index(engine, stmt);
}
if stmt.cascade
&& stmt.kind == DropKind::Schema
&& !only_graph_namespaces(engine, &stmt.names, stmt.if_exists)?
{
return Err(SQLError::Unsupported(
"DROP SCHEMA CASCADE is not supported; no objects were changed".into(),
));
}
let mut lock_targets = std::collections::BTreeSet::new();
match stmt.kind {
DropKind::Table | DropKind::ForeignTable | DropKind::View | DropKind::MaterializedView => {
let mut table_targets = Vec::new();
for name in &stmt.names {
if let Some((canonical, kind)) = engine.try_resolve_visible_relation_kind(name)? {
if stmt.kind == DropKind::Table && kind == "table" {
table_targets.push(canonical.clone());
}
lock_targets.insert(canonical);
}
}
if stmt.kind == DropKind::Table {
let (hierarchy_targets, _) =
engine.hierarchy_drop_targets(&table_targets, stmt.cascade);
lock_targets.extend(hierarchy_targets);
}
}
DropKind::Index => unreachable!("DROP INDEX has a bound execution path"),
DropKind::Schema => {
for name in &stmt.names {
for table in engine
.tables_in_schema(name)
.map_err(|err| ddl_storage_error("DROP SCHEMA relation lock", err))?
{
lock_targets.insert(format!("{name}.{table}"));
}
}
}
DropKind::Sequence => {}
}
for table in lock_targets {
engine.lock_relation(&table, crate::row_locks::RelationLockMode::AccessExclusive)?;
}
engine.with_implicit_transaction(move |engine| run_drop_inner(engine, stmt))
}
fn only_graph_namespaces(
engine: &Engine,
names: &[String],
if_exists: bool,
) -> Result<bool, SQLError> {
for name in names {
let is_graph = engine
.has_graph(name)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?;
let is_schema = engine
.has_schema(name)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?;
if if_exists && !is_graph && !is_schema {
continue;
}
if !is_graph || is_schema {
return Ok(false);
}
}
Ok(!names.is_empty())
}
#[expect(
clippy::too_many_lines,
reason = "preserves DDL dependency and action order"
)]
fn run_drop_inner(engine: &Engine, stmt: DropStmt) -> Result<SQLResult, SQLError> {
match stmt.kind {
DropKind::Table => {
let mut tables = Vec::new();
for name in &stmt.names {
match engine.try_resolve_visible_relation_kind(name)? {
Some((canonical, "table")) => tables.push(canonical),
Some((canonical, kind)) => {
return Err(SQLError::Unsupported(format!(
"DROP TABLE: relation `{canonical}` is a {kind}, not a table"
)));
}
None if stmt.if_exists => {}
None => {
return Err(SQLError::Unsupported(format!(
"DROP TABLE: relation `{name}` does not exist"
)));
}
}
}
for table in &tables {
engine.ensure_table_drop_authority(table)?;
}
let (tables, dependents) = engine.hierarchy_drop_targets(&tables, stmt.cascade);
if !dependents.is_empty() {
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"cannot drop table {} because other objects depend on it",
tables.join(", ")
),
});
}
if !stmt.cascade {
let restrict_dependents = engine
.try_drop_table_restrict_dependents(&tables)
.map_err(|err| ddl_storage_error("DROP TABLE dependency preflight", err))?;
if !restrict_dependents.is_empty() {
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"cannot drop table {} because other objects depend on it: {}",
tables.join(", "),
restrict_dependents.join(", ")
),
});
}
}
for table in &tables {
engine.ensure_no_pending_trigger_events(table, "DROP TABLE")?;
}
engine
.try_drop_tables(&tables, stmt.cascade)
.map_err(|err| ddl_storage_error("DROP TABLE", err))?;
}
DropKind::ForeignTable => {
let mut foreign_tables = Vec::new();
let mut seen = std::collections::BTreeSet::new();
for name in &stmt.names {
match engine.resolve_visible_relation_kind(name)? {
RelationResolution::Found(canonical, "foreign table") => {
if seen.insert(canonical.clone()) {
foreign_tables.push(canonical);
}
}
RelationResolution::Found(_, _) => {
return Err(SQLError::Routine {
sqlstate: "42809".into(),
message: format!("\"{name}\" is not a foreign table"),
});
}
RelationResolution::MissingSchema(schema) if stmt.if_exists => {
engine.push_sql_notice(
"NOTICE",
&format!("schema \"{schema}\" does not exist, skipping"),
);
}
RelationResolution::MissingRelation if stmt.if_exists => {
engine.push_sql_notice(
"NOTICE",
&format!("foreign table \"{name}\" does not exist, skipping"),
);
}
RelationResolution::MissingSchema(schema) => {
return Err(SQLError::Routine {
sqlstate: "3F000".into(),
message: format!("schema \"{schema}\" does not exist"),
});
}
RelationResolution::MissingRelation => {
return Err(SQLError::Routine {
sqlstate: "42P01".into(),
message: format!("foreign table \"{name}\" does not exist"),
});
}
}
}
for table in &foreign_tables {
engine.ensure_foreign_table_drop_authority(table)?;
}
let target_names = foreign_tables.iter().cloned().collect();
let owned_sequences = engine
.foreign_table_owned_sequence_names(&foreign_tables)
.map_err(|error| {
ddl_storage_error("DROP FOREIGN TABLE sequence ownership", error)
})?;
let mut dependents = std::collections::BTreeSet::new();
for table in &foreign_tables {
dependents.extend(
engine
.views_depending_on_relation(table)
.map_err(|error| {
ddl_storage_error("DROP FOREIGN TABLE dependency preflight", error)
})?
.into_iter()
.map(|view| format!("view {view}")),
);
}
dependents.extend(
engine
.rules_depending_on_relations(&foreign_tables)
.map_err(|error| {
ddl_storage_error("DROP FOREIGN TABLE dependency preflight", error)
})?
.into_iter()
.map(|(table, rule)| {
format!("rule {rule} on table {}", table.qualified_name())
}),
);
for sequence in &owned_sequences {
dependents.extend(
engine
.sequence_external_dependents_for_owner_drop(sequence, &target_names)
.map_err(|error| {
ddl_storage_error(
"DROP FOREIGN TABLE owned-sequence dependency preflight",
error,
)
})?,
);
}
if !stmt.cascade && !dependents.is_empty() {
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"cannot drop foreign table {} because other objects depend on it: {}",
foreign_tables.join(", "),
dependents.into_iter().collect::<Vec<_>>().join(", ")
),
});
}
if stmt.cascade {
engine
.drop_rules_depending_on_relations_inner(&foreign_tables)
.map_err(|error| ddl_storage_error("DROP FOREIGN TABLE CASCADE", error))?;
engine
.drop_views_depending_on_relations(&foreign_tables)
.map_err(|error| ddl_storage_error("DROP FOREIGN TABLE CASCADE", error))?;
}
for table in foreign_tables {
let removed = engine.drop_foreign_table_inner(&table).map_err(|error| {
SQLError::Internal(format!(
"DROP FOREIGN TABLE failed in storage backend: {error}"
))
})?;
if !removed {
return Err(SQLError::Internal(format!(
"foreign table `{table}` disappeared after DROP preflight"
)));
}
}
for sequence in owned_sequences {
engine
.drop_owned_sequence(&sequence, stmt.cascade)
.map_err(|error| {
ddl_storage_error("DROP FOREIGN TABLE owned sequence", error)
})?;
}
}
DropKind::Index => unreachable!("DROP INDEX has a bound execution path"),
DropKind::View | DropKind::MaterializedView => {
let expected_kind = if stmt.kind == DropKind::View {
"view"
} else {
"materialized view"
};
let command = if stmt.kind == DropKind::View {
"DROP VIEW"
} else {
"DROP MATERIALIZED VIEW"
};
let mut views = Vec::new();
for name in &stmt.names {
match engine.try_resolve_visible_relation_kind(name)? {
Some((canonical, kind)) if kind == expected_kind => views.push(canonical),
Some((canonical, kind)) => {
return Err(SQLError::Routine {
sqlstate: "42809".into(),
message: format!(
"{command}: relation `{canonical}` is a {kind}, not a {expected_kind}"
),
});
}
None if stmt.if_exists => {}
None => {
return Err(SQLError::Routine {
sqlstate: "42P01".into(),
message: format!("{command}: relation `{name}` does not exist"),
});
}
}
}
engine.drop_views(&views, stmt.cascade)?;
}
DropKind::Sequence => {
let mut sequences = Vec::new();
let mut seen = std::collections::BTreeSet::new();
for name in &stmt.names {
match engine.resolve_visible_relation_kind(name)? {
RelationResolution::Found(canonical, "sequence") => {
if seen.insert(canonical.clone()) {
sequences.push(canonical);
}
}
RelationResolution::Found(_canonical, _kind) => {
return Err(SQLError::Routine {
sqlstate: "42809".into(),
message: format!("\"{name}\" is not a sequence"),
});
}
RelationResolution::MissingRelation | RelationResolution::MissingSchema(_)
if stmt.if_exists =>
{
engine.push_sql_notice(
"NOTICE",
&format!("sequence \"{name}\" does not exist, skipping"),
);
}
RelationResolution::MissingSchema(schema) => {
return Err(SQLError::Routine {
sqlstate: "3F000".into(),
message: format!("schema \"{schema}\" does not exist"),
});
}
RelationResolution::MissingRelation => {
return Err(SQLError::Routine {
sqlstate: "42P01".into(),
message: format!("sequence \"{name}\" does not exist"),
});
}
}
}
engine.drop_sequences_sql_inner(&sequences, stmt.cascade)?;
}
DropKind::Schema => {
let mut schemas = Vec::new();
let mut graphs = Vec::new();
for name in &stmt.names {
let exists = engine
.preflight_drop_schema(name)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?;
if exists {
schemas.push(name.clone());
continue;
}
if engine
.has_graph(name)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?
{
if !stmt.cascade {
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"cannot drop schema {name} because other objects depend on it"
),
});
}
graphs.push(name.clone());
} else if !stmt.if_exists {
return Err(SQLError::Unsupported(format!(
"DROP SCHEMA: schema `{name}` does not exist"
)));
}
}
for schema in schemas {
engine
.drop_schema(&schema)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?;
}
for graph in graphs {
engine
.drop_graph(&graph)
.map_err(|err| ddl_storage_error("DROP SCHEMA", err))?;
}
}
}
Ok(SQLResult::empty())
}
fn run_drop_index(engine: &Engine, stmt: DropStmt) -> Result<SQLResult, SQLError> {
let mut indexes = Vec::new();
let mut seen = std::collections::BTreeSet::new();
for requested in &stmt.names {
match engine.resolve_visible_relation_kind(requested)? {
RelationResolution::Found(canonical, "index") => {
let relation = crate::RelationIdentity::from_legacy_name(&canonical)
.map_err(SQLError::Internal)?;
if !seen.insert(relation.clone()) {
continue;
}
let row = engine
.bound_catalog_index(&canonical)
.map_err(|error| ddl_storage_error("DROP INDEX", error))?
.ok_or_else(|| {
SQLError::Internal(format!(
"resolved index `{canonical}` has no bound catalog row"
))
})?;
engine.require_index_drop_authority(&row)?;
if engine
.catalog_read_view()
.has_constraint_index(&row.relation)
{
return Err(SQLError::Routine {
sqlstate: "2BP01".into(),
message: format!(
"cannot drop index {} because constraint {} on table {} requires it",
row.relation.name, row.relation.name, row.table_name
),
});
}
indexes.push(row);
}
RelationResolution::Found(_, _) => {
return Err(SQLError::Routine {
sqlstate: "42809".into(),
message: format!("\"{requested}\" is not an index"),
});
}
RelationResolution::MissingSchema(schema) if stmt.if_exists => {
engine.push_sql_notice(
"NOTICE",
&format!("schema \"{schema}\" does not exist, skipping"),
);
}
RelationResolution::MissingSchema(schema) => {
return Err(SQLError::Routine {
sqlstate: "3F000".into(),
message: format!("schema \"{schema}\" does not exist"),
});
}
RelationResolution::MissingRelation if stmt.if_exists => {
let local = crate::RelationIdentity::parse_reference(requested)
.map_err(SQLError::Internal)?
.1;
engine.push_sql_notice(
"NOTICE",
&format!("index \"{local}\" does not exist, skipping"),
);
}
RelationResolution::MissingRelation => {
let local = crate::RelationIdentity::parse_reference(requested)
.map_err(SQLError::Internal)?
.1;
return Err(SQLError::Routine {
sqlstate: "42704".into(),
message: format!("index \"{local}\" does not exist"),
});
}
}
}
let dependents = index_dependencies::dependents(engine, &indexes, stmt.cascade)?;
for row in &indexes {
engine.lock_relation(
&row.table_name,
crate::row_locks::RelationLockMode::AccessExclusive,
)?;
}
engine.with_implicit_transaction(move |engine| {
for (table, name) in dependents {
super::alter_table::drop_constraint_dependency(engine, &table, &name)?;
}
for row in indexes {
drop_index_side_effects(engine, &row)?;
engine
.try_drop_catalog_index_relation(&row.relation)
.map_err(|error| ddl_storage_error("DROP INDEX", error))?;
}
Ok(SQLResult::empty())
})
}
pub(super) fn ddl_storage_error(action: &str, err: impl std::error::Error + 'static) -> SQLError {
let mut source: Option<&(dyn std::error::Error + 'static)> = Some(&err);
while let Some(error) = source {
if let Some(error) = error.downcast_ref::<SQLError>() {
return SQLError::Routine {
sqlstate: error.sqlstate().unwrap_or("XX000").into(),
message: error.to_string(),
};
}
source = error.source();
}
SQLError::Internal(format!("{action} failed in storage backend: {err}"))
}
fn drop_index_side_effects(engine: &Engine, row: &CatalogIndexRow) -> Result<(), SQLError> {
if row.index_type.eq_ignore_ascii_case("gin") {
drop_gin_index_side_effects(engine, row)?;
} else if row.index_type.eq_ignore_ascii_case("ivf")
|| row.index_type.eq_ignore_ascii_case("hnsw")
{
drop_vector_index_side_effects(engine, row)?;
}
Ok(())
}
fn catalog_index_columns(row: &CatalogIndexRow, action: &str) -> Result<Vec<String>, SQLError> {
serde_json::from_str(&row.columns_json).map_err(|e| {
SQLError::Internal(format!(
"{action} `{}`: invalid index column metadata: {e}",
row.relation.qualified_name()
))
})
}
fn drop_gin_index_side_effects(engine: &Engine, row: &CatalogIndexRow) -> Result<(), SQLError> {
let fields: std::collections::BTreeSet<String> = catalog_index_columns(row, "DROP INDEX")?
.into_iter()
.collect();
let indexes = engine
.list_catalog_indexes()
.map_err(|err| ddl_storage_error("DROP INDEX", err))?;
for field in fields {
let mut still_referenced = false;
for candidate in &indexes {
if candidate.relation == row.relation
|| candidate.table_name != row.table_name
|| !candidate.index_type.eq_ignore_ascii_case("gin")
{
continue;
}
if catalog_index_columns(candidate, "DROP INDEX")?
.iter()
.any(|candidate_field| candidate_field == &field)
{
still_referenced = true;
break;
}
}
if !still_referenced {
engine
.drop_fts_field(&row.table_name, &field)
.map_err(|err| {
SQLError::Internal(format!(
"DROP INDEX `{}`: failed to remove FTS field `{}`.`{field}`: {err}",
row.relation.qualified_name(),
row.table_name
))
})?;
}
}
Ok(())
}
fn drop_vector_index_side_effects(engine: &Engine, row: &CatalogIndexRow) -> Result<(), SQLError> {
let columns = catalog_index_columns(row, "DROP INDEX")?;
for col in columns {
match engine
.column_type(&row.table_name, &col)
.map_err(|err| ddl_storage_error("DROP INDEX", err))?
{
Some(ColumnType::Vector(dim) | ColumnType::Tensor(dim)) => {
if !engine
.drop_vector_field_index(&row.table_name, col.clone(), dim)
.map_err(|err| ddl_storage_error("DROP INDEX vector field", err))?
{
return Err(SQLError::Unsupported(format!(
"DROP INDEX `{}`: relation `{}` does not exist",
row.relation.qualified_name(),
row.table_name
)));
}
engine
.drop_vector_index_metadata(&row.table_name, &col)
.map_err(|e| {
SQLError::Internal(format!(
"DROP INDEX `{}`: failed to drop vector-index metadata for `{}`.`{col}`: {e}",
row.relation.qualified_name(), row.table_name
))
})?;
}
Some(other) => {
return Err(SQLError::Unsupported(format!(
"DROP INDEX `{}`: vector-index column `{}`.`{col}` is no longer VECTOR or TENSOR, got {other:?}",
row.relation.qualified_name(), row.table_name
)));
}
None => {
return Err(SQLError::Unsupported(format!(
"DROP INDEX `{}`: column `{}`.`{col}` does not exist",
row.relation.qualified_name(),
row.table_name
)));
}
}
}
Ok(())
}
pub(crate) fn drop_index_dependency(
engine: &Engine,
relation: &crate::RelationIdentity,
) -> Result<(), SQLError> {
let row = engine
.bound_catalog_index(&relation.qualified_name())
.map_err(|error| ddl_storage_error("DROP INDEX dependency", error))?
.ok_or_else(|| SQLError::Internal("dependent index disappeared".into()))?;
drop_index_side_effects(engine, &row)?;
engine
.try_drop_catalog_index_relation(relation)
.map_err(|error| ddl_storage_error("DROP INDEX dependency", error))?;
Ok(())
}