[python] Reject dropping the main branch (#10855)

Author: plusplusjiajia

paimon-python/pypaimon/branch/filesystem_branch_manager.py

@@ -180,8 +180,10 @@ def drop_branch(self, branch_name: str) -> None:
             branch_name: Name of the branch to drop
 
         Raises:
-            ValueError: If branch doesn't exist
+            ValueError: If branch is main or doesn't exist
         """
+        if BranchManager.is_main_branch(branch_name):
+            raise ValueError("The main branch cannot be deleted.")
         if not self.branch_exists(branch_name):
             raise ValueError(f"Branch name '{branch_name}' doesn't exist.")
 

paimon-python/pypaimon/tests/branch_manager_test.py

@@ -184,6 +184,25 @@ def test_branch_exists(self):
         self.assertTrue(manager.branch_exists("feature"))
         self.assertFalse(manager.branch_exists("develop"))
 
+    def test_main_branch_validation_before_file_access(self):
+        from unittest.mock import Mock
+
+        file_io = Mock()
+        manager = FileSystemBranchManager(file_io, self.table_path, None, None, None)
+        for operation, args, message in (
+            (manager.drop_branch, ("main",), "The main branch cannot be deleted."),
+            (manager.create_branch, ("main",),
+             "Branch name 'main' is the default branch and cannot be used."),
+            (manager.create_branch, ("main", "absent"),
+             "Branch name 'main' is the default branch and cannot be used."),
+            (manager.fast_forward, ("main",), "Branch name 'main' do not use in fast-forward."),
+        ):
+            with self.subTest(operation=operation.__name__, args=args):
+                with self.assertRaises(ValueError) as cm:
+                    operation(*args)
+                self.assertEqual(str(cm.exception), message)
+        self.assertEqual(file_io.mock_calls, [])
+
 
 class SnapshotManagerBranchAwarenessTest(unittest.TestCase):
     """Pin down SnapshotManager's branch-aware path computation.

paimon-python/pypaimon/tests/filesystem_catalog_branch_test.py

@@ -238,6 +238,16 @@ def test_drop_branch_missing_raises(self):
             self.catalog.drop_branch(self.identifier, "absent")
         self.assertEqual(cm.exception.branch, "absent")
 
+    def test_drop_main_branch_preserves_table(self):
+        table = self.catalog.get_table(self.identifier)
+        with self.assertRaisesRegex(ValueError, r"^The main branch cannot be deleted\.$"):
+            self.catalog.drop_branch(self.identifier, "main")
+        self.assertTrue(os.path.isdir(table.table_path))
+        self.assertEqual(
+            self._read(self.catalog.get_table(self.identifier)).to_pydict(),
+            {"id": [1, 2, 3], "value": ["a", "b", "c"]},
+        )
+
     # -- fast_forward ---------------------------------------------------------
 
     def test_fast_forward_missing_raises(self):

paimon-python/pypaimon/tests/jdbc_catalog_test.py

@@ -22,6 +22,8 @@
 import tempfile
 import unittest
 
+import pyarrow as pa
+
 from pypaimon import CatalogFactory, Schema
 from pypaimon.catalog.catalog_exception import (
     DatabaseAlreadyExistException,
@@ -190,6 +192,30 @@ def fail_insert_table_properties(identifier, properties):
         self.assertEqual(table_count, 0)
         self.assertFalse(os.path.exists(os.path.join(self.warehouse, "test_db.db", "test_table")))
 
+    def test_drop_main_branch_preserves_table(self):
+        with CatalogFactory.create(self.options) as catalog:
+            catalog.create_database("test_db", False)
+            data = pa.table({"id": [1]})
+            catalog.create_table("test_db.test_table", Schema.from_pyarrow_schema(data.schema), False)
+            table = catalog.get_table("test_db.test_table")
+            builder = table.new_batch_write_builder()
+            writer = builder.new_write()
+            try:
+                writer.write_arrow(data)
+                builder.new_commit().commit(writer.prepare_commit())
+            finally:
+                writer.close()
+
+            with self.assertRaisesRegex(ValueError, r"^The main branch cannot be deleted\.$"):
+                table.branch_manager().drop_branch("main")
+            self.assertTrue(os.path.isdir(table.table_path))
+            table = catalog.get_table("test_db.test_table")
+            reader = table.new_read_builder()
+            self.assertEqual(
+                reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pydict(),
+                {"id": [1]},
+            )
+
     def test_rename_table_keeps_metadata_when_file_move_fails(self):
         fields = [DataField.from_dict({"id": 1, "name": "f0", "type": "INT"})]
         catalog = CatalogFactory.create(self.options)