[python] Fix cascade drop ignoring tables found during recheck

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."

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"}),