dmux · Oct 29, 2026 · Oct 25, 2026 · Oct 25, 2026 · Oct 26, 2026 · Oct 26, 2026
diff --git a/src/minilake/duckdb_pool.py b/src/minilake/duckdb_pool.py
@@ -3,7 +3,7 @@
 import asyncio
 import logging
 from pathlib import Path
-from typing import Optional
+from typing import Callable, Optional
 
 import duckdb
 
@@ -141,17 +141,41 @@ async def attach_catalog(self, catalog_name: str) -> None:
         async with self._uc_lock:
             conn.execute(f"ATTACH IF NOT EXISTS '{path}' AS \"{catalog_name}\"")
 
-    async def detach_catalog(self, catalog_name: str) -> None:
-        """DETACH a catalog and delete its underlying DuckDB file for real."""
+    async def detach_catalog(
+        self,
+        catalog_name: str,
+        *,
+        can_detach: Optional[Callable[[object], bool]] = None,
+    ) -> bool:
+        """Detach a catalog and delete its underlying DuckDB file.
+
+        ``can_detach`` runs while holding the UC connection lock, immediately before
+        DETACH. Callers can use it to make an emptiness check atomic with the detach.
+        It returns False without changing the catalog when the check fails.
+        """
         conn = await self.get_uc_connection()
         async with self._uc_lock:
-            try:
+            attachment = conn.execute(
+                "SELECT path FROM duckdb_databases() WHERE database_name = ?",
+                [catalog_name],
+            ).fetchone()
+            expected_path = self._catalog_db_path(catalog_name).resolve()
+            if can_detach is not None and attachment is None:
+                raise RuntimeError(f"Catalog '{catalog_name}' is not attached for emptiness inspection")
+            if attachment is not None:
+                attached_path = attachment[0]
+                if not attached_path or Path(attached_path).resolve() != expected_path:
+                    raise RuntimeError(
+                        f"Catalog '{catalog_name}' is attached to '{attached_path}', not its backing file '{expected_path}'"
+                    )
+            if can_detach is not None and not can_detach(conn):
+                return False
+            if attachment:
                 conn.execute(f'DETACH "{catalog_name}"')
-            except Exception as e:
-                logger.warning(f"Failed to detach catalog {catalog_name} (non-critical): {e}")
         path = self._catalog_db_path(catalog_name)
         path.unlink(missing_ok=True)
         Path(str(path) + ".wal").unlink(missing_ok=True)
+        return True
 
     async def get_warehouse_connection(self, warehouse_id: str) -> duckdb.DuckDBPyConnection:
         """Get or create a connection for a specific warehouse."""

diff --git a/src/minilake/services/unity_catalog.py b/src/minilake/services/unity_catalog.py
@@ -159,6 +159,106 @@ def _split_three_part(full_name: str, kind: str) -> tuple[str, str, str]:
     return parts[0], parts[1], parts[2]
 
 
+def _directory_has_children(path: Path) -> bool:
+    """Whether a volume directory contains a volume or other on-disk content."""
+    try:
+        return next(path.iterdir(), None) is not None
+    except FileNotFoundError:
+        return False
+
+
+def _schema_has_registered_children(catalog_name: str, schema_name: str) -> bool:
+    full_name = f"{catalog_name}.{schema_name}"
+    prefix = f"{full_name}."
+    for registry in ("tables", "external_tables", "volumes", "functions"):
+        if any(name.startswith(prefix) for name in _state[registry]):
+            return True
+
+    # Volumes are directories, so they can outlive the in-memory registry (for example,
+    # after restarting without snapshot persistence). An empty volume directory still
+    # represents a child and must be explicitly deleted before a non-forced schema drop.
+    return _directory_has_children(_get_volumes_dir() / catalog_name / schema_name)
+
+
+def _catalog_has_registered_children(catalog_name: str) -> bool:
+    prefix = f"{catalog_name}."
+    # Catalog creation always adds this synthetic default schema. Its existence alone
+    # must not make an otherwise empty catalog undeletable.
+    if any(name.startswith(prefix) and name != f"{catalog_name}.default" for name in _state["schemas"]):
+        return True
+    for registry in ("tables", "external_tables", "volumes", "functions"):
+        if any(name.startswith(prefix) for name in _state[registry]):
+            return True
+
+    # Count volume directories, including volumes whose registry entry was lost. Empty
+    # schema directories left after volume deletion do not prevent catalog deletion.
+    catalog_volumes = _get_volumes_dir() / catalog_name
+    try:
+        for schema_path in catalog_volumes.iterdir():
+            if not schema_path.is_dir() or _directory_has_children(schema_path):
+                return True
+    except FileNotFoundError:
+        pass
+    return False
+
+
+def _duckdb_schema_has_children(conn, catalog_name: str, schema_name: str) -> bool:
+    """Check DuckDB objects that can be removed by DROP SCHEMA, including raw SQL DDL."""
+    row = conn.execute(
+        """
+        SELECT 1 FROM information_schema.tables
+        WHERE table_catalog = ? AND table_schema = ?
+        UNION ALL
+        SELECT 1 FROM duckdb_sequences()
+        WHERE database_name = ? AND schema_name = ?
+        UNION ALL
+        SELECT 1 FROM duckdb_functions()
+        WHERE database_name = ? AND schema_name = ?
+        UNION ALL
+        SELECT 1 FROM duckdb_types()
+        WHERE database_name = ? AND schema_name = ? AND NOT internal
+        LIMIT 1
+        """,
+        [catalog_name, schema_name] * 4,
+    ).fetchone()
+    return row is not None
+
+
+def _duckdb_catalog_has_children(conn, catalog_name: str) -> bool:
+    """Check attached database objects, including objects created outside the UC API."""
+    if conn.execute(
+        """
+        SELECT 1 FROM information_schema.schemata
+        WHERE catalog_name = ?
+          AND schema_name NOT IN ('main', 'default', 'information_schema', 'pg_catalog')
+        LIMIT 1
+        """,
+        [catalog_name],
+    ).fetchone():
+        return True
+
+    for schema_name in ("main", "default"):
+        if _duckdb_schema_has_children(conn, catalog_name, schema_name):
+            return True
+    return False
+
+
+def _not_empty_error(kind: str, name: str) -> DatabricksError:
+    return DatabricksError(
+        error_code="INVALID_STATE",
+        message=f"{kind.capitalize()} '{name}' is not empty; set force=true to delete it",
+        status_code=409,
+    )
+
+
+def _inspection_error(kind: str, name: str, error: Exception) -> DatabricksError:
+    return DatabricksError(
+        error_code="INVALID_STATE",
+        message=f"Could not verify whether {kind} '{name}' is empty; no deletion was performed: {error}",
+        status_code=500,
+    )
+
+
 async def _managed_table_exists(catalog_name: str, schema_name: str, table_name: str) -> bool:
     """Whether a native DuckDB table exists, regardless of what UC knows about it."""
     pool = get_duckdb_pool()
@@ -380,26 +480,43 @@ async def update_catalog(name: str, req: UpdateCatalogRequest) -> CatalogInfo:
 
 
 @router.delete("/catalogs/{name}")
-async def delete_catalog(name: str) -> dict[str, str]:
-    """Delete a catalog."""
+async def delete_catalog(name: str, force: bool = Query(False)) -> dict[str, str]:
+    """Delete a catalog, requiring force when it contains child objects."""
     if name not in _state["catalogs"]:
         raise DatabricksError(
             error_code="NOT_FOUND",
             message=f"Catalog '{name}' not found",
             status_code=404,
         )
-    # Also drop this catalog's schemas, tables and volumes from in-memory state.
-    for registry in ("schemas", "tables", "external_tables", "volumes"):
-        for full_name in [k for k in _state[registry] if k.startswith(f"{name}.")]:
-            del _state[registry][full_name]
 
     pool = get_duckdb_pool()
     if pool:
         try:
-            await pool.detach_catalog(name)
-        except Exception as e:
-            logger.warning(f"Failed to detach catalog {name} in DuckDB: {e}")
+            # Attach first so a persisted catalog that is not currently attached is
+            # inspected before its file can be removed.
+            await pool.attach_catalog(name)
+            can_detach = None
+            if not force:
+
+                def catalog_is_empty(conn):
+                    return not _catalog_has_registered_children(name) and not _duckdb_catalog_has_children(conn, name)
 
+                can_detach = catalog_is_empty
+            detached = await pool.detach_catalog(name, can_detach=can_detach)
+        except Exception as e:
+            if not force:
+                raise _inspection_error("catalog", name, e) from e
+            raise DatabricksError(
+                error_code="INVALID_STATE",
+                message=f"Could not delete catalog '{name}': {e}",
+                status_code=500,
+            ) from e
+        if not detached:
+            raise _not_empty_error("catalog", name)
+    elif not force and _catalog_has_registered_children(name):
+        raise _not_empty_error("catalog", name)
+
+    _delete_catalog_children(name)
     del _state["catalogs"][name]
     return {"message": f"Catalog '{name}' deleted"}
 
@@ -497,36 +614,69 @@ async def update_schema(full_name: str, req: UpdateSchemaRequest) -> SchemaInfo:
 
 
 @router.delete("/schemas/{full_name}")
-async def delete_schema(full_name: str) -> dict[str, str]:
-    """Delete a schema."""
+async def delete_schema(full_name: str, force: bool = Query(False)) -> dict[str, str]:
+    """Delete a schema, requiring force when it contains child objects."""
     if full_name not in _state["schemas"]:
         raise DatabricksError(
             error_code="NOT_FOUND",
             message=f"Schema '{full_name}' not found",
             status_code=404,
         )
 
-    # Delete from DuckDB
+    cat, schema = full_name.split(".", 1)
     pool = get_duckdb_pool()
     if pool:
-        cat, schema = full_name.split(".", 1)
         try:
+            await pool.attach_catalog(cat)
             conn = await pool.get_uc_connection()
             async with await pool.get_uc_lock():
-                conn.execute(f'DROP SCHEMA IF EXISTS "{cat}"."{schema}" CASCADE')
+                if not force and (
+                    _schema_has_registered_children(cat, schema) or _duckdb_schema_has_children(conn, cat, schema)
+                ):
+                    raise _not_empty_error("schema", full_name)
+                drop_mode = "CASCADE" if force else "RESTRICT"
+                conn.execute(f'DROP SCHEMA IF EXISTS "{cat}"."{schema}" {drop_mode}')
+        except DatabricksError:
+            raise
         except Exception as e:
-            logger.warning(f"Failed to drop schema in DuckDB: {e}")
+            if not force:
+                raise _inspection_error("schema", full_name, e) from e
+            raise DatabricksError(
+                error_code="INVALID_STATE",
+                message=f"Could not delete schema '{full_name}': {e}",
+                status_code=500,
+            ) from e
+    elif not force and _schema_has_registered_children(cat, schema):
+        raise _not_empty_error("schema", full_name)
+
+    _delete_schema_children(full_name)
+    del _state["schemas"][full_name]
+    return {"message": f"Schema '{full_name}' deleted"}
+
+
+def _delete_catalog_children(name: str) -> None:
+    """Remove registered children after deleting their backing catalog."""
+    volume_path = _get_volumes_dir() / name
+    if volume_path.exists():
+        shutil.rmtree(volume_path)
+
+    for registry in ("schemas", "tables", "external_tables", "volumes", "functions"):
+        for full_name in [k for k in _state[registry] if k.startswith(f"{name}.")]:
+            del _state[registry][full_name]
+
+
+def _delete_schema_children(full_name: str) -> None:
+    """Remove registered children after deleting their backing schema."""
+    cat, schema = full_name.split(".", 1)
+    volume_path = _get_volumes_dir() / cat / schema
+    if volume_path.exists():
+        shutil.rmtree(volume_path)
 
-    # Drop the schema's tables from both registries. Leaving external_tables behind kept
-    # the table queryable through delta_scan() after its schema was gone.
     prefix = f"{full_name}."
-    for registry in ("tables", "external_tables"):
+    for registry in ("tables", "external_tables", "volumes", "functions"):
         for table_name in [k for k in _state[registry] if k.startswith(prefix)]:
             del _state[registry][table_name]
 
-    del _state["schemas"][full_name]
-    return {"message": f"Schema '{full_name}' deleted"}
-
 
 # ============================================================================
 # Tables (backed by real DuckDB)

diff --git a/tests/unity_catalog/test_catalogs.py b/tests/unity_catalog/test_catalogs.py
@@ -4,6 +4,7 @@
 
 import pytest
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors import DatabricksError
 
 
 @pytest.mark.crud
@@ -71,6 +72,84 @@ def test_catalog_delete_removes_from_list(workspace_client: WorkspaceClient):
     print(f"✓ Catalog deleted: {cat_name}")
 
 
+@pytest.mark.serial
+@pytest.mark.workflow
+def test_catalog_delete_without_force_preserves_sql_created_table(catalog, workspace_client: WorkspaceClient):
+    """Omitting force must reject a populated catalog without deleting its database."""
+    table_name = f"preserve_{uuid4().hex[:8]}"
+    warehouse = workspace_client.warehouses.create(name=f"delete_wh_{uuid4().hex[:6]}", cluster_size="Small")
+
+    try:
+        workspace_client.statement_execution.execute_statement(
+            warehouse_id=warehouse.id,
+            statement=f'CREATE TABLE "{catalog.name}"."default"."{table_name}" AS SELECT 42 AS id',
+        )
+
+        # This exercises the SDK's omitted-force request, which must behave as false.
+        with pytest.raises(DatabricksError, match="not empty"):
+            workspace_client.catalogs.delete(name=catalog.name)
+
+        result = workspace_client.statement_execution.execute_statement(
+            warehouse_id=warehouse.id,
+            statement=f'SELECT id FROM "{catalog.name}"."default"."{table_name}"',
+        )
+        assert result.result.data_array == [[42]]
+
+        workspace_client.catalogs.delete(name=catalog.name, force=True)
+        with pytest.raises(DatabricksError):
+            workspace_client.statement_execution.execute_statement(
+                warehouse_id=warehouse.id,
+                statement=f'SELECT id FROM "{catalog.name}"."default"."{table_name}"',
+            )
+    finally:
+        try:
+            workspace_client.catalogs.delete(name=catalog.name, force=True)
+        except Exception:
+            pass
+        workspace_client.warehouses.delete(id=warehouse.id)
+
+
+@pytest.mark.serial
+@pytest.mark.workflow
+def test_catalog_delete_rejects_replaced_attachment(catalog, workspace_client: WorkspaceClient):
+    """A same-name attachment to another database must not let deletion unlink this catalog."""
+    table_name = f"preserve_{uuid4().hex[:8]}"
+    warehouse = workspace_client.warehouses.create(name=f"delete_wh_{uuid4().hex[:6]}", cluster_size="Small")
+
+    def execute(statement):
+        return workspace_client.statement_execution.execute_statement(
+            warehouse_id=warehouse.id,
+            statement=statement,
+        )
+
+    try:
+        execute(f'CREATE TABLE "{catalog.name}"."default"."{table_name}" AS SELECT 42 AS id')
+        execute(f'DETACH "{catalog.name}"')
+        execute(f"ATTACH ':memory:' AS \"{catalog.name}\"")
+
+        with pytest.raises(DatabricksError):
+            workspace_client.catalogs.delete(name=catalog.name)
+
+        # Once the conflicting alias is removed, deletion must attach and inspect the
+        # original file, proving that the failed request left its table intact.
+        execute(f'DETACH "{catalog.name}"')
+        with pytest.raises(DatabricksError, match="not empty"):
+            workspace_client.catalogs.delete(name=catalog.name)
+
+        result = execute(f'SELECT id FROM "{catalog.name}"."default"."{table_name}"')
+        assert result.result.data_array == [[42]]
+    finally:
+        try:
+            execute(f'DETACH "{catalog.name}"')
+        except Exception:
+            pass
+        try:
+            workspace_client.catalogs.delete(name=catalog.name, force=True)
+        except Exception:
+            pass
+        workspace_client.warehouses.delete(id=warehouse.id)
+
+
 @pytest.mark.error
 def test_catalog_create_duplicate_fails(workspace_client: WorkspaceClient):
     """Test: Creating duplicate catalog raises error."""