diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py b/paimon-python/pypaimon/catalog/filesystem_catalog.py index f2aad5b442c8..d1264e2f9b43 100644 --- a/paimon-python/pypaimon/catalog/filesystem_catalog.py +++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py @@ -113,7 +113,10 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: # Check if database still has tables remaining_tables = self.list_tables(name) - if remaining_tables and not cascade: + if remaining_tables: + if cascade: + raise OSError(f"Database {name} changed during drop; " + f"remaining tables: {remaining_tables}") raise ValueError( f"Database {name} is not empty. " f"Use cascade=True to drop all tables first." diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 2613b824a988..45ced30f6ed0 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -60,6 +60,30 @@ def test_database(self): database = catalog.get_database("test_db") self.assertEqual(database.name, "test_db") + def test_cascade_drop_preserves_table_created_during_drop(self): + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + catalog.create_database("db", False) + schema = Schema(fields=[DataField(0, "value", AtomicType("INT"))]) + catalog.create_table("db.old", schema, False) + delete = catalog.file_io.delete + old_path = catalog.get_database_path("db") + "/old" + + def delete_and_create(path, recursive=False): + result = delete(path, recursive) + if path == old_path: + catalog.create_table("db.new", schema, False) + return result + + with patch.object(catalog.file_io, "delete", side_effect=delete_and_create): + with self.assertRaisesRegex(OSError, "changed during drop"): + catalog.drop_database("db", cascade=True) + + self.assertEqual(["new"], catalog.list_tables("db")) + self.assertIsNotNone(catalog.get_table("db.new")) + catalog.drop_database("db", cascade=True) + with self.assertRaises(DatabaseNotExistException): + catalog.get_database("db") + def test_table(self): fields = [ DataField.from_dict({"id": 1, "name": "f0", "type": "INT"}),