[python] Introduce database related commands to Python CLI

docs/content/pypaimon/cli.md

@@ -70,7 +70,7 @@ paimon [OPTIONS] COMMAND [ARGS]...
 - `-c, --config PATH`: Path to catalog configuration file (default: `paimon.yaml`)
 - `--help`: Show help message and exit
 
-## Commands
+## Table Commands
 
 ### Table Read
 
@@ -230,6 +230,21 @@ Successfully imported 3 rows into 'mydb.users'.
 - Data types should be compatible with the table schema
 - The import operation appends data to the existing table
 
+### Table Rename
+
+Rename a table in the catalog. Both source and target must be specified in `database.table` format.
+
+```shell
+paimon table rename mydb.old_name mydb.new_name
+```
+
+Output:
+```
+Table 'mydb.old_name' renamed to 'mydb.new_name' successfully.
+```
+
+**Note:** Both filesystem and REST catalogs support table rename. For filesystem catalogs, the rename is performed by renaming the underlying table directory.
+
 ### Table Drop
 
 Drop a table from the catalog. This will permanently delete the table and all its data.
@@ -352,3 +367,81 @@ paimon table alter mydb.users alter-column -n age -t BIGINT -c 'User age in year
 ```shell
 paimon table alter mydb.users update-comment -c "Updated user information table"
 ```
+
+## Database Commands
+
+### DB Get
+
+Get and display database information in JSON format.
+
+```shell
+paimon db get mydb
+```
+
+Output:
+```json
+{
+  "name": "mydb",
+  "options": {}
+}
+```
+
+### DB Create
+
+Create a new database.
+
+```shell
+# Create a simple database
+paimon db create mydb
+
+# Create with properties
+paimon db create mydb -p '{"key1": "value1", "key2": "value2"}'
+
+# Create and ignore if already exists
+paimon db create mydb -i
+```
+
+### DB Drop
+
+Drop an existing database.
+
+```shell
+# Drop a database
+paimon db drop mydb
+
+# Drop and ignore if not exists
+paimon db drop mydb --ignore-if-not-exists
+
+# Drop with all tables (cascade)
+paimon db drop mydb --cascade
+```
+
+### DB Alter
+
+Alter database properties by setting or removing properties.
+
+```shell
+# Set properties
+paimon db alter mydb --set '{"key1": "value1", "key2": "value2"}'
+
+# Remove properties
+paimon db alter mydb --remove key1 key2
+
+# Set and remove properties in one command
+paimon db alter mydb --set '{"key1": "new_value"}' --remove key2
+```
+
+### DB List Tables
+
+List all tables in a database.
+
+```shell
+paimon db list-tables mydb
+```
+
+Output:
+```
+orders
+products
+users
+```

paimon-python/pypaimon/catalog/catalog.py

@@ -47,6 +47,41 @@ def get_database(self, name: str) -> 'Database':
     def create_database(self, name: str, ignore_if_exists: bool, properties: Optional[dict] = None):
         """Create a database with properties."""
 
+    @abstractmethod
+    def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: bool = False):
+        """Drop a database.
+
+        Args:
+            name: Name of the database to drop.
+            ignore_if_not_exists: If True, do not raise error if database does not exist.
+            cascade: If True, drop all tables in the database before dropping it.
+        """
+
+    @abstractmethod
+    def list_tables(self, database_name: str) -> List[str]:
+        """List all table names in the given database.
+
+        Args:
+            database_name: Name of the database.
+
+        Returns:
+            List of table names.
+        """
+
+    def alter_database(self, name: str, changes: list):
+        """Alter database properties.
+
+        Args:
+            name: Name of the database.
+            changes: List of PropertyChange objects.
+
+        Raises:
+            NotImplementedError: If the catalog does not support alter database.
+        """
+        raise NotImplementedError(
+            "alter_database is not supported by this catalog."
+        )
+
     @abstractmethod
     def get_table(self, identifier: Union[str, Identifier]) -> 'Table':
         """Get paimon table identified by the given Identifier."""
@@ -67,6 +102,20 @@ def drop_table(self, identifier: Union[str, Identifier], ignore_if_not_exists: b
             TableNotExistException: If table does not exist and ignore_if_not_exists is False
         """
 
+    def rename_table(self, source_identifier: Union[str, Identifier], target_identifier: Union[str, Identifier]):
+        """Rename a table.
+
+        Args:
+            source_identifier: Current table identifier.
+            target_identifier: New table identifier.
+
+        Raises:
+            NotImplementedError: If the catalog does not support rename table.
+        """
+        raise NotImplementedError(
+            "rename_table is not supported by this catalog."
+        )
+
     @abstractmethod
     def alter_table(
         self,

paimon-python/pypaimon/catalog/filesystem_catalog.py

@@ -65,6 +65,48 @@ def create_database(self, name: str, ignore_if_exists: bool, properties: Optiona
             path = self.get_database_path(name)
             self.file_io.mkdirs(path)
 
+    def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: bool = False):
+        try:
+            self.get_database(name)
+        except DatabaseNotExistException:
+            if not ignore_if_not_exists:
+                raise
+            return
+
+        db_path = self.get_database_path(name)
+
+        if cascade:
+            for table_name in self.list_tables(name):
+                table_path = f"{db_path}/{table_name}"
+                self.file_io.delete(table_path, True)
+
+        # Check if database still has tables
+        remaining_tables = self.list_tables(name)
+        if remaining_tables and not cascade:
+            raise ValueError(
+                f"Database {name} is not empty. "
+                f"Use cascade=True to drop all tables first."
+            )
+
+        self.file_io.delete(db_path, True)
+
+    def list_tables(self, database_name: str) -> list:
+        try:
+            self.get_database(database_name)
+        except DatabaseNotExistException:
+            raise
+
+        db_path = self.get_database_path(database_name)
+        statuses = self.file_io.list_status(db_path)
+        table_names = []
+        for status in statuses:
+            import pyarrow.fs as pafs
+            is_directory = hasattr(status, 'type') and status.type == pafs.FileType.Directory
+            name = status.base_name if hasattr(status, 'base_name') else ""
+            if is_directory and name and not name.startswith("."):
+                table_names.append(name)
+        return sorted(table_names)
+
     def get_table(self, identifier: Union[str, Identifier]) -> Table:
         if not isinstance(identifier, Identifier):
             identifier = Identifier.from_string(identifier)
@@ -140,6 +182,32 @@ def alter_table(
         except Exception as e:
             raise RuntimeError(f"Failed to alter table {identifier.get_full_name()}: {e}") from e
 
+    def rename_table(self, source_identifier: Union[str, Identifier], target_identifier: Union[str, Identifier]):
+        if not isinstance(source_identifier, Identifier):
+            source_identifier = Identifier.from_string(source_identifier)
+        if not isinstance(target_identifier, Identifier):
+            target_identifier = Identifier.from_string(target_identifier)
+
+        # Verify source table exists
+        try:
+            self.get_table(source_identifier)
+        except TableNotExistException:
+            raise
+
+        # Verify target database exists
+        self.get_database(target_identifier.get_database_name())
+
+        # Verify target table does not exist
+        try:
+            self.get_table(target_identifier)
+            raise TableAlreadyExistException(target_identifier)
+        except TableNotExistException:
+            pass
+
+        source_path = self.get_table_path(source_identifier)
+        target_path = self.get_table_path(target_identifier)
+        self.file_io.rename(source_path, target_path)
+
     def drop_table(self, identifier: Union[str, Identifier], ignore_if_not_exists: bool = False):
         if not isinstance(identifier, Identifier):
             identifier = Identifier.from_string(identifier)

paimon-python/pypaimon/catalog/rest/rest_catalog.py

@@ -139,12 +139,24 @@ def get_database(self, name: str) -> Database:
         except ForbiddenException as e:
             raise DatabaseNoPermissionException(name) from e
 
-    def drop_database(self, name: str, ignore_if_not_exists: bool = False):
+    def drop_database(self, name: str, ignore_if_not_exists: bool = False, cascade: bool = False):
+        if not cascade:
+            try:
+                tables = self.list_tables(name)
+                if tables:
+                    raise ValueError(
+                        f"Database {name} is not empty. "
+                        f"Use cascade=True to drop all tables first."
+                    )
+            except DatabaseNotExistException:
+                if not ignore_if_not_exists:
+                    raise
+                return
+
         try:
             self.rest_api.drop_database(name)
         except NoSuchResourceException as e:
             if not ignore_if_not_exists:
-                # Convert REST API exception to catalog exception
                 raise DatabaseNotExistException(name) from e
         except ForbiddenException as e:
             raise DatabaseNoPermissionException(name) from e
@@ -206,6 +218,20 @@ def create_table(self, identifier: Union[str, Identifier], schema: Schema, ignor
             if not ignore_if_exists:
                 raise TableAlreadyExistException(identifier) from e
 
+    def rename_table(self, source_identifier: Union[str, Identifier], target_identifier: Union[str, Identifier]):
+        if not isinstance(source_identifier, Identifier):
+            source_identifier = Identifier.from_string(source_identifier)
+        if not isinstance(target_identifier, Identifier):
+            target_identifier = Identifier.from_string(target_identifier)
+        try:
+            self.rest_api.rename_table(source_identifier, target_identifier)
+        except NoSuchResourceException as e:
+            raise TableNotExistException(source_identifier) from e
+        except AlreadyExistsException as e:
+            raise TableAlreadyExistException(target_identifier) from e
+        except ForbiddenException as e:
+            raise TableNoPermissionException(source_identifier) from e
+
     def drop_table(self, identifier: Union[str, Identifier], ignore_if_not_exists: bool = False):
         if not isinstance(identifier, Identifier):
             identifier = Identifier.from_string(identifier)

paimon-python/pypaimon/cli/cli.py

@@ -107,6 +107,13 @@ def main():
     from pypaimon.cli.cli_table import add_table_subcommands
     add_table_subcommands(table_parser)
 
+    # Database commands
+    db_parser = subparsers.add_parser('db', help='Database operations')
+
+    # Import and add database subcommands
+    from pypaimon.cli.cli_db import add_db_subcommands
+    add_db_subcommands(db_parser)
+
     args = parser.parse_args()
 
     if args.command is None: