[python] Introduce database related commands to Python CLI

Author: JingsongLi

docs/content/pypaimon/cli.md

@@ -76,7 +76,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
 
@@ -203,4 +203,82 @@ The schema JSON file follows the same format as output by `table get`:
 2. Create a new table with the same schema:
    ```shell
    paimon table create mydb.users_copy --schema users_schema.json
-   ```
+   ```
+
+## 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."""

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)

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: