From 4b8ca1b6ffc1c72529bbd4b779485d0b214842ae Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Wed, 7 Oct 2026 22:55:34 -0700 Subject: [PATCH 1/5] [python] Avoid recursive database cleanup after dropping tables --- .../pypaimon/catalog/filesystem_catalog.py | 7 ++- .../pypaimon/tests/filesystem_catalog_test.py | 47 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py b/paimon-python/pypaimon/catalog/filesystem_catalog.py index f2aad5b442c8..63b4963ca6cd 100644 --- a/paimon-python/pypaimon/catalog/filesystem_catalog.py +++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py @@ -113,13 +113,16 @@ 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." ) - self.file_io.delete(db_path, True) + self.file_io.delete(db_path, False) def list_tables(self, database_name: str) -> List[str]: try: diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 2613b824a988..6ce697ceee12 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -60,6 +60,53 @@ 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_drop_database_preserves_unlisted_contents(self): + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + for cascade in (False, True): + with self.subTest(cascade=cascade): + name = "db_" + str(cascade) + catalog.create_database(name, False) + path = catalog.get_database_path(name) + "/pending" + with open(path, "wb") as stream: + stream.write(b"in-progress table creation") + self.assertEqual([], catalog.list_tables(name)) + with self.assertRaisesRegex(OSError, "not empty"): + catalog.drop_database(name, cascade=cascade) + with open(path, "rb") as stream: + self.assertEqual(b"in-progress table creation", stream.read()) + + def test_drop_empty_database(self): + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + for cascade in (False, True): + with self.subTest(cascade=cascade): + catalog.create_database("db", False) + catalog.drop_database("db", cascade=cascade) + self.assertFalse(os.path.exists(catalog.get_database_path("db"))) + def test_table(self): fields = [ DataField.from_dict({"id": 1, "name": "f0", "type": "INT"}), From e118df0791aa9077a21f4ff7f9e0861216dfcbc7 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 8 Oct 2026 01:14:30 -0700 Subject: [PATCH 2/5] [python] Fail closed on unsafe non-recursive Arrow deletion --- .../pypaimon/filesystem/pyarrow_file_io.py | 6 ++- paimon-python/pypaimon/tests/file_io_test.py | 41 +++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index ad5d3c8a4ea9..b8c960505ab1 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -463,7 +463,11 @@ def delete(self, path: str, recursive: bool = False) -> bool: self.filesystem.delete_dir_contents(path_str) self.filesystem.delete_dir(path_str) else: - self.filesystem.delete_dir(path_str) + # Arrow delete_dir is recursive, even after an emptiness check. + if not isinstance(self.filesystem, pafs.LocalFileSystem): + raise OSError( + f"Safe non-recursive directory deletion is unsupported for {path}") + os.rmdir(path_str) else: self.filesystem.delete_file(path_str) return True diff --git a/paimon-python/pypaimon/tests/file_io_test.py b/paimon-python/pypaimon/tests/file_io_test.py index 53c8e8d29698..15bf5f049978 100644 --- a/paimon-python/pypaimon/tests/file_io_test.py +++ b/paimon-python/pypaimon/tests/file_io_test.py @@ -381,6 +381,47 @@ def test_delete_non_empty_directory_raises_error(self): finally: shutil.rmtree(temp_dir, ignore_errors=True) + def test_pyarrow_empty_directory_delete_preserves_late_child(self): + with tempfile.TemporaryDirectory() as root: + directory = os.path.join(root, "db") + os.mkdir(directory) + file_io = object.__new__(PyArrowFileIO) + file_io.filesystem = pafs.LocalFileSystem() + file_io.to_filesystem_path = lambda path: path + rmdir = os.rmdir + child = os.path.join(directory, "new", "schema", "schema-0") + + def create_before_rmdir(path): + os.makedirs(os.path.dirname(child)) + with open(child, "wb") as stream: + stream.write(b"schema") + return rmdir(path) + + with patch("pypaimon.filesystem.pyarrow_file_io.os.rmdir", + side_effect=create_before_rmdir): + with self.assertRaises(OSError): + file_io.delete(directory, False) + with open(child, "rb") as stream: + self.assertEqual(b"schema", stream.read()) + + def test_pyarrow_remote_empty_directory_delete_fails_closed(self): + file_io = object.__new__(PyArrowFileIO) + file_io.filesystem = pafs.LocalFileSystem() + file_io.to_filesystem_path = lambda path: path + for scheme in ("s3", "oss", "gs", "hdfs"): + with self.subTest(scheme=scheme): + filesystem = MagicMock() + filesystem.get_file_info.return_value = [] + file_io.filesystem = filesystem + with patch.object(file_io, "to_filesystem_path", return_value="bucket/db"), \ + patch.object(file_io, "_get_file_info", + return_value=pafs.FileInfo("bucket/db", pafs.FileType.Directory)): + with self.assertRaisesRegex(OSError, "Safe non-recursive"): + file_io.delete(scheme + "://bucket/db", False) + filesystem.delete_dir.assert_not_called() + filesystem.delete_dir_contents.assert_not_called() + filesystem.delete_file.assert_not_called() + def test_delete_returns_false_when_file_not_exists(self): temp_dir = tempfile.mkdtemp(prefix="file_io_delete_test_") try: From 33ff05bf11955127eff065e3f0b557a7c0f710f6 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 8 Oct 2026 01:35:42 -0700 Subject: [PATCH 3/5] [python] Check directory deletion support before cascading table drops --- .../pypaimon/catalog/filesystem_catalog.py | 1 + paimon-python/pypaimon/common/file_io.py | 3 +++ .../pypaimon/filesystem/caching_file_io.py | 3 +++ .../pypaimon/filesystem/pyarrow_file_io.py | 9 ++++--- .../pypaimon/tests/filesystem_catalog_test.py | 26 +++++++++++++++++++ 5 files changed, 39 insertions(+), 3 deletions(-) diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py b/paimon-python/pypaimon/catalog/filesystem_catalog.py index 63b4963ca6cd..f8d01c5d0a95 100644 --- a/paimon-python/pypaimon/catalog/filesystem_catalog.py +++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py @@ -105,6 +105,7 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: return db_path = self.get_database_path(name) + self.file_io.check_non_recursive_delete_supported(db_path) if cascade: for table_name in self.list_tables(name): diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 56c239de11e8..09c627b8ec91 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -120,6 +120,9 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: def delete(self, path: str, recursive: bool = False) -> bool: pass + def check_non_recursive_delete_supported(self, path: str) -> None: + """Reject known unsupported directory deletion before any mutation.""" + @abstractmethod def mkdirs(self, path: str) -> bool: pass diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index 1ddbb76ee02b..5e3fb6faaa00 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -417,6 +417,9 @@ def list_status(self, path: str): def exists(self, path: str) -> bool: return self._delegate.exists(path) + def check_non_recursive_delete_supported(self, path: str) -> None: + self._delegate.check_non_recursive_delete_supported(path) + def delete(self, path: str, recursive: bool = False) -> bool: return self._delegate.delete(path, recursive) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index b8c960505ab1..41a86475f00f 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -446,6 +446,11 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: for i, info in enumerate(file_infos) } + def check_non_recursive_delete_supported(self, path: str) -> None: + if not isinstance(self.filesystem, pafs.LocalFileSystem): + raise OSError( + f"Safe non-recursive directory deletion is unsupported for {path}") + def delete(self, path: str, recursive: bool = False) -> bool: path_str = self.to_filesystem_path(path) file_info = self._get_file_info(path_str) @@ -464,9 +469,7 @@ def delete(self, path: str, recursive: bool = False) -> bool: self.filesystem.delete_dir(path_str) else: # Arrow delete_dir is recursive, even after an emptiness check. - if not isinstance(self.filesystem, pafs.LocalFileSystem): - raise OSError( - f"Safe non-recursive directory deletion is unsupported for {path}") + self.check_non_recursive_delete_supported(path) os.rmdir(path_str) else: self.filesystem.delete_file(path_str) diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 6ce697ceee12..83b2df42d52c 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -22,6 +22,10 @@ from unittest.mock import MagicMock, patch import pyarrow as pa +import pyarrow.fs as pafs + +from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO +from pypaimon.filesystem.caching_file_io import CachingFileIO from pypaimon import CatalogFactory, Schema from pypaimon.catalog.catalog_exception import (DatabaseAlreadyExistException, @@ -99,6 +103,28 @@ def test_drop_database_preserves_unlisted_contents(self): with open(path, "rb") as stream: self.assertEqual(b"in-progress table creation", stream.read()) + def test_cascade_drop_rejects_unsupported_cleanup_before_deleting_tables(self): + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + catalog.create_database("db", False) + catalog.create_table("db.old", Schema(fields=[ + DataField(0, "value", AtomicType("INT"))]), False) + original_io = catalog.file_io + arrow_io = object.__new__(PyArrowFileIO) + arrow_io.filesystem = pafs.SubTreeFileSystem(self.warehouse, pafs.LocalFileSystem()) + arrow_io.to_filesystem_path = lambda path: os.path.relpath(path, self.warehouse) + wrapper = object.__new__(CachingFileIO) + wrapper._delegate = arrow_io + for file_io in (arrow_io, wrapper): + with self.subTest(cached=file_io is wrapper): + catalog.file_io = file_io + try: + with self.assertRaisesRegex(OSError, "Safe non-recursive"): + catalog.drop_database("db", cascade=True) + finally: + catalog.file_io = original_io + self.assertEqual(["old"], catalog.list_tables("db")) + self.assertIsNotNone(catalog.get_table("db.old")) + def test_drop_empty_database(self): catalog = CatalogFactory.create({"warehouse": self.warehouse}) for cascade in (False, True): From 952a34c0dcc64a062819f15baabaa2a46fae6efc Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 8 Oct 2026 01:56:08 -0700 Subject: [PATCH 4/5] [python] Forward delete capability checks through FileIO wrappers --- .../pypaimon/catalog/rest/rest_token_file_io.py | 3 +++ .../pypaimon/filesystem/resolving_file_io.py | 3 +++ .../pypaimon/tests/filesystem_catalog_test.py | 17 +++++++++++++++-- 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 293171285181..053f10b871a9 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py @@ -149,6 +149,9 @@ def list_status(self, path: str): def exists(self, path: str) -> bool: return self.file_io().exists(path) + def check_non_recursive_delete_supported(self, path: str) -> None: + self.file_io().check_non_recursive_delete_supported(path) + def delete(self, path: str, recursive: bool = False) -> bool: return self.file_io().delete(path, recursive) diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py b/paimon-python/pypaimon/filesystem/resolving_file_io.py index e4ca45394e91..76ce48b2d0b3 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -88,6 +88,9 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: result.update(fio.exists_batch(group_paths)) return result + def check_non_recursive_delete_supported(self, path: str) -> None: + self._get_fileio(path).check_non_recursive_delete_supported(path) + def delete(self, path: str, recursive: bool = False) -> bool: return self._get_fileio(path).delete(path, recursive) diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 83b2df42d52c..1a5c62e187ee 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -26,6 +26,8 @@ from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO from pypaimon.filesystem.caching_file_io import CachingFileIO +from pypaimon.filesystem.resolving_file_io import ResolvingFileIO +from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO from pypaimon import CatalogFactory, Schema from pypaimon.catalog.catalog_exception import (DatabaseAlreadyExistException, @@ -114,8 +116,19 @@ def test_cascade_drop_rejects_unsupported_cleanup_before_deleting_tables(self): arrow_io.to_filesystem_path = lambda path: os.path.relpath(path, self.warehouse) wrapper = object.__new__(CachingFileIO) wrapper._delegate = arrow_io - for file_io in (arrow_io, wrapper): - with self.subTest(cached=file_io is wrapper): + resolving_catalog = CatalogFactory.create({ + "warehouse": self.warehouse, "resolving-file-io.enabled": "true"}) + resolving = resolving_catalog.file_io + self.assertIsInstance(resolving, ResolvingFileIO) + resolving._fileio_cache[resolving._cache_key(self.warehouse)] = arrow_io + cached_resolving = object.__new__(CachingFileIO) + cached_resolving._delegate = resolving + rest_token = object.__new__(RESTTokenFileIO) + rest_token.file_io = lambda: arrow_io + for name, file_io in (("direct", arrow_io), ("cached", wrapper), + ("resolving", resolving), ("cached_resolving", cached_resolving), + ("rest_token", rest_token)): + with self.subTest(wrapper=name): catalog.file_io = file_io try: with self.assertRaisesRegex(OSError, "Safe non-recursive"): From 5d5bba3736a8a53507eb1a184052cf550b7f8a97 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Thu, 8 Oct 2026 02:23:32 -0700 Subject: [PATCH 5/5] [python] Scope database drop fix to tables observed during recheck --- .../pypaimon/catalog/filesystem_catalog.py | 3 +- .../catalog/rest/rest_token_file_io.py | 3 - paimon-python/pypaimon/common/file_io.py | 3 - .../pypaimon/filesystem/caching_file_io.py | 3 - .../pypaimon/filesystem/pyarrow_file_io.py | 9 +-- .../pypaimon/filesystem/resolving_file_io.py | 3 - paimon-python/pypaimon/tests/file_io_test.py | 41 ------------ .../pypaimon/tests/filesystem_catalog_test.py | 62 ------------------- 8 files changed, 2 insertions(+), 125 deletions(-) diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py b/paimon-python/pypaimon/catalog/filesystem_catalog.py index f8d01c5d0a95..d1264e2f9b43 100644 --- a/paimon-python/pypaimon/catalog/filesystem_catalog.py +++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py @@ -105,7 +105,6 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: return db_path = self.get_database_path(name) - self.file_io.check_non_recursive_delete_supported(db_path) if cascade: for table_name in self.list_tables(name): @@ -123,7 +122,7 @@ def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: f"Use cascade=True to drop all tables first." ) - self.file_io.delete(db_path, False) + self.file_io.delete(db_path, True) def list_tables(self, database_name: str) -> List[str]: try: diff --git a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py index 053f10b871a9..293171285181 100644 --- a/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py +++ b/paimon-python/pypaimon/catalog/rest/rest_token_file_io.py @@ -149,9 +149,6 @@ def list_status(self, path: str): def exists(self, path: str) -> bool: return self.file_io().exists(path) - def check_non_recursive_delete_supported(self, path: str) -> None: - self.file_io().check_non_recursive_delete_supported(path) - def delete(self, path: str, recursive: bool = False) -> bool: return self.file_io().delete(path, recursive) diff --git a/paimon-python/pypaimon/common/file_io.py b/paimon-python/pypaimon/common/file_io.py index 09c627b8ec91..56c239de11e8 100644 --- a/paimon-python/pypaimon/common/file_io.py +++ b/paimon-python/pypaimon/common/file_io.py @@ -120,9 +120,6 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: def delete(self, path: str, recursive: bool = False) -> bool: pass - def check_non_recursive_delete_supported(self, path: str) -> None: - """Reject known unsupported directory deletion before any mutation.""" - @abstractmethod def mkdirs(self, path: str) -> bool: pass diff --git a/paimon-python/pypaimon/filesystem/caching_file_io.py b/paimon-python/pypaimon/filesystem/caching_file_io.py index 5e3fb6faaa00..1ddbb76ee02b 100644 --- a/paimon-python/pypaimon/filesystem/caching_file_io.py +++ b/paimon-python/pypaimon/filesystem/caching_file_io.py @@ -417,9 +417,6 @@ def list_status(self, path: str): def exists(self, path: str) -> bool: return self._delegate.exists(path) - def check_non_recursive_delete_supported(self, path: str) -> None: - self._delegate.check_non_recursive_delete_supported(path) - def delete(self, path: str, recursive: bool = False) -> bool: return self._delegate.delete(path, recursive) diff --git a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py index 41a86475f00f..ad5d3c8a4ea9 100644 --- a/paimon-python/pypaimon/filesystem/pyarrow_file_io.py +++ b/paimon-python/pypaimon/filesystem/pyarrow_file_io.py @@ -446,11 +446,6 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: for i, info in enumerate(file_infos) } - def check_non_recursive_delete_supported(self, path: str) -> None: - if not isinstance(self.filesystem, pafs.LocalFileSystem): - raise OSError( - f"Safe non-recursive directory deletion is unsupported for {path}") - def delete(self, path: str, recursive: bool = False) -> bool: path_str = self.to_filesystem_path(path) file_info = self._get_file_info(path_str) @@ -468,9 +463,7 @@ def delete(self, path: str, recursive: bool = False) -> bool: self.filesystem.delete_dir_contents(path_str) self.filesystem.delete_dir(path_str) else: - # Arrow delete_dir is recursive, even after an emptiness check. - self.check_non_recursive_delete_supported(path) - os.rmdir(path_str) + self.filesystem.delete_dir(path_str) else: self.filesystem.delete_file(path_str) return True diff --git a/paimon-python/pypaimon/filesystem/resolving_file_io.py b/paimon-python/pypaimon/filesystem/resolving_file_io.py index 76ce48b2d0b3..e4ca45394e91 100644 --- a/paimon-python/pypaimon/filesystem/resolving_file_io.py +++ b/paimon-python/pypaimon/filesystem/resolving_file_io.py @@ -88,9 +88,6 @@ def exists_batch(self, paths: List[str]) -> Dict[str, bool]: result.update(fio.exists_batch(group_paths)) return result - def check_non_recursive_delete_supported(self, path: str) -> None: - self._get_fileio(path).check_non_recursive_delete_supported(path) - def delete(self, path: str, recursive: bool = False) -> bool: return self._get_fileio(path).delete(path, recursive) diff --git a/paimon-python/pypaimon/tests/file_io_test.py b/paimon-python/pypaimon/tests/file_io_test.py index 15bf5f049978..53c8e8d29698 100644 --- a/paimon-python/pypaimon/tests/file_io_test.py +++ b/paimon-python/pypaimon/tests/file_io_test.py @@ -381,47 +381,6 @@ def test_delete_non_empty_directory_raises_error(self): finally: shutil.rmtree(temp_dir, ignore_errors=True) - def test_pyarrow_empty_directory_delete_preserves_late_child(self): - with tempfile.TemporaryDirectory() as root: - directory = os.path.join(root, "db") - os.mkdir(directory) - file_io = object.__new__(PyArrowFileIO) - file_io.filesystem = pafs.LocalFileSystem() - file_io.to_filesystem_path = lambda path: path - rmdir = os.rmdir - child = os.path.join(directory, "new", "schema", "schema-0") - - def create_before_rmdir(path): - os.makedirs(os.path.dirname(child)) - with open(child, "wb") as stream: - stream.write(b"schema") - return rmdir(path) - - with patch("pypaimon.filesystem.pyarrow_file_io.os.rmdir", - side_effect=create_before_rmdir): - with self.assertRaises(OSError): - file_io.delete(directory, False) - with open(child, "rb") as stream: - self.assertEqual(b"schema", stream.read()) - - def test_pyarrow_remote_empty_directory_delete_fails_closed(self): - file_io = object.__new__(PyArrowFileIO) - file_io.filesystem = pafs.LocalFileSystem() - file_io.to_filesystem_path = lambda path: path - for scheme in ("s3", "oss", "gs", "hdfs"): - with self.subTest(scheme=scheme): - filesystem = MagicMock() - filesystem.get_file_info.return_value = [] - file_io.filesystem = filesystem - with patch.object(file_io, "to_filesystem_path", return_value="bucket/db"), \ - patch.object(file_io, "_get_file_info", - return_value=pafs.FileInfo("bucket/db", pafs.FileType.Directory)): - with self.assertRaisesRegex(OSError, "Safe non-recursive"): - file_io.delete(scheme + "://bucket/db", False) - filesystem.delete_dir.assert_not_called() - filesystem.delete_dir_contents.assert_not_called() - filesystem.delete_file.assert_not_called() - def test_delete_returns_false_when_file_not_exists(self): temp_dir = tempfile.mkdtemp(prefix="file_io_delete_test_") try: diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 1a5c62e187ee..45ced30f6ed0 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -22,12 +22,6 @@ from unittest.mock import MagicMock, patch import pyarrow as pa -import pyarrow.fs as pafs - -from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO -from pypaimon.filesystem.caching_file_io import CachingFileIO -from pypaimon.filesystem.resolving_file_io import ResolvingFileIO -from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO from pypaimon import CatalogFactory, Schema from pypaimon.catalog.catalog_exception import (DatabaseAlreadyExistException, @@ -90,62 +84,6 @@ def delete_and_create(path, recursive=False): with self.assertRaises(DatabaseNotExistException): catalog.get_database("db") - def test_drop_database_preserves_unlisted_contents(self): - catalog = CatalogFactory.create({"warehouse": self.warehouse}) - for cascade in (False, True): - with self.subTest(cascade=cascade): - name = "db_" + str(cascade) - catalog.create_database(name, False) - path = catalog.get_database_path(name) + "/pending" - with open(path, "wb") as stream: - stream.write(b"in-progress table creation") - self.assertEqual([], catalog.list_tables(name)) - with self.assertRaisesRegex(OSError, "not empty"): - catalog.drop_database(name, cascade=cascade) - with open(path, "rb") as stream: - self.assertEqual(b"in-progress table creation", stream.read()) - - def test_cascade_drop_rejects_unsupported_cleanup_before_deleting_tables(self): - catalog = CatalogFactory.create({"warehouse": self.warehouse}) - catalog.create_database("db", False) - catalog.create_table("db.old", Schema(fields=[ - DataField(0, "value", AtomicType("INT"))]), False) - original_io = catalog.file_io - arrow_io = object.__new__(PyArrowFileIO) - arrow_io.filesystem = pafs.SubTreeFileSystem(self.warehouse, pafs.LocalFileSystem()) - arrow_io.to_filesystem_path = lambda path: os.path.relpath(path, self.warehouse) - wrapper = object.__new__(CachingFileIO) - wrapper._delegate = arrow_io - resolving_catalog = CatalogFactory.create({ - "warehouse": self.warehouse, "resolving-file-io.enabled": "true"}) - resolving = resolving_catalog.file_io - self.assertIsInstance(resolving, ResolvingFileIO) - resolving._fileio_cache[resolving._cache_key(self.warehouse)] = arrow_io - cached_resolving = object.__new__(CachingFileIO) - cached_resolving._delegate = resolving - rest_token = object.__new__(RESTTokenFileIO) - rest_token.file_io = lambda: arrow_io - for name, file_io in (("direct", arrow_io), ("cached", wrapper), - ("resolving", resolving), ("cached_resolving", cached_resolving), - ("rest_token", rest_token)): - with self.subTest(wrapper=name): - catalog.file_io = file_io - try: - with self.assertRaisesRegex(OSError, "Safe non-recursive"): - catalog.drop_database("db", cascade=True) - finally: - catalog.file_io = original_io - self.assertEqual(["old"], catalog.list_tables("db")) - self.assertIsNotNone(catalog.get_table("db.old")) - - def test_drop_empty_database(self): - catalog = CatalogFactory.create({"warehouse": self.warehouse}) - for cascade in (False, True): - with self.subTest(cascade=cascade): - catalog.create_database("db", False) - catalog.drop_database("db", cascade=cascade) - self.assertFalse(os.path.exists(catalog.get_database_path("db"))) - def test_table(self): fields = [ DataField.from_dict({"id": 1, "name": "f0", "type": "INT"}),