[python] Add list partitions support for Python CLI

Author: JingsongLi

docs/content/pypaimon/cli.md

@@ -309,6 +309,38 @@ 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 List Partitions
+
+List partitions of a Paimon table. Supports optional pattern filtering to match specific partitions.
+
+```shell
+paimon table list-partitions mydb.orders
+```
+
+**Options:**
+
+- `--pattern, -p`: Partition name pattern to filter partitions
+
+**Examples:**
+
+```shell
+# List all partitions
+paimon table list-partitions mydb.orders
+
+# List partitions matching a pattern
+paimon table list-partitions mydb.orders --pattern "dt=2024*"
+```
+
+Output:
+```
+              Partition  RecordCount  FileSizeInBytes  FileCount  LastFileCreationTime       UpdatedAt  UpdatedBy
+dt=2024-01-01,region=us          500          1048576         10         1704067200000  1704153600000      admin
+dt=2024-01-02,region=eu          300           524288          5         1704153600000  1704240000000      user1
+dt=2024-01-03,region=us          200           262144          3         1704240000000  1704326400000      admin
+```
+
+**Note:** Both filesystem and REST catalogs support listing partitions.
+
 ### Table Rename
 
 Rename a table in the catalog. Both source and target must be specified in `database.table` format.

paimon-python/pypaimon/api/api_response.py

@@ -23,6 +23,7 @@
 from pypaimon.common.json_util import T, json_field
 from pypaimon.common.options import Options
 from pypaimon.schema.schema import Schema
+from pypaimon.snapshot.snapshot_commit import PartitionStatistics
 from pypaimon.snapshot.table_snapshot import TableSnapshot
 
 
@@ -132,6 +133,37 @@ def get_next_page_token(self) -> str:
         return self.next_page_token
 
 
+@dataclass
+class Partition(PartitionStatistics):
+    """Partition data model matching Java org.apache.paimon.partition.Partition."""
+
+    FIELD_DONE = "done"
+    FIELD_OPTIONS = "options"
+
+    done: bool = json_field(FIELD_DONE, default=False)
+    created_at: Optional[int] = json_field("createdAt", default=None)
+    created_by: Optional[str] = json_field("createdBy", default=None)
+    updated_at: Optional[int] = json_field("updatedAt", default=None)
+    updated_by: Optional[str] = json_field("updatedBy", default=None)
+    options: Optional[Dict[str, str]] = json_field(FIELD_OPTIONS, default=None)
+
+
+@dataclass
+class ListPartitionsResponse(PagedResponse['Partition']):
+    """Response for listing partitions."""
+    FIELD_PARTITIONS = "partitions"
+
+    partitions: Optional[List[Partition]] = json_field(FIELD_PARTITIONS)
+    next_page_token: Optional[str] = json_field(
+        PagedResponse.FIELD_NEXT_PAGE_TOKEN)
+
+    def data(self) -> Optional[List[Partition]]:
+        return self.partitions
+
+    def get_next_page_token(self) -> Optional[str]:
+        return self.next_page_token
+
+
 @dataclass
 class ListTablesResponse(PagedResponse[str]):
     FIELD_TABLES = "tables"

paimon-python/pypaimon/api/resource_paths.py

@@ -27,6 +27,7 @@ class ResourcePaths:
     DATABASES = "databases"
     TABLES = "tables"
     TABLE_DETAILS = "table-details"
+    PARTITIONS = "partitions"
 
     def __init__(self, prefix: str):
         self.base_path = "/{}/{}".format(self.V1, prefix).rstrip("/")
@@ -78,3 +79,7 @@ def rollback_table(self, database_name: str, table_name: str) -> str:
     def table_snapshot(self, database_name: str, table_name: str) -> str:
         return ("{}/{}/{}/{}/{}/snapshot".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
                                                  self.TABLES, RESTUtil.encode_string(table_name)))
+
+    def partitions(self, database_name: str, table_name: str) -> str:
+        return ("{}/{}/{}/{}/{}/{}".format(self.base_path, self.DATABASES, RESTUtil.encode_string(database_name),
+                                           self.TABLES, RESTUtil.encode_string(table_name), self.PARTITIONS))

paimon-python/pypaimon/api/rest_api.py

@@ -26,8 +26,10 @@
                                        GetDatabaseResponse, GetTableResponse,
                                        GetTableTokenResponse,
                                        ListDatabasesResponse,
+                                       ListPartitionsResponse,
                                        ListTablesResponse, PagedList,
-                                       PagedResponse, GetTableSnapshotResponse)
+                                       PagedResponse, GetTableSnapshotResponse,
+                                       Partition)
 from pypaimon.api.auth import AuthProviderFactory, RESTAuthFunction
 from pypaimon.api.client import HttpClient
 from pypaimon.api.resource_paths import ResourcePaths
@@ -48,6 +50,7 @@ class RESTApi:
     DATABASE_NAME_PATTERN = "databaseNamePattern"
     TABLE_NAME_PATTERN = "tableNamePattern"
     TABLE_TYPE = "tableType"
+    PARTITION_NAME_PATTERN = "partitionNamePattern"
     TOKEN_EXPIRATION_SAFE_TIME_MILLIS = 3_600_000
 
     def __init__(self, options: Union[Options, Dict[str, str]], config_required: bool = True):
@@ -399,6 +402,29 @@ def load_snapshot(self, identifier: Identifier) -> Optional['TableSnapshot']:
             return None
         return response.get_snapshot()
 
+    def list_partitions_paged(
+            self,
+            identifier: Identifier,
+            max_results: Optional[int] = None,
+            page_token: Optional[str] = None,
+            partition_name_pattern: Optional[str] = None,
+    ) -> PagedList[Partition]:
+        database_name, table_name = self.__validate_identifier(identifier)
+
+        response = self.client.get_with_params(
+            self.resource_paths.partitions(database_name, table_name),
+            self.__build_paged_query_params(
+                max_results,
+                page_token,
+                {self.PARTITION_NAME_PATTERN: partition_name_pattern},
+            ),
+            ListPartitionsResponse,
+            self.rest_auth_function,
+        )
+
+        partitions = response.data() or []
+        return PagedList(partitions, response.get_next_page_token())
+
     @staticmethod
     def __validate_identifier(identifier: Identifier):
         if not identifier:

paimon-python/pypaimon/catalog/catalog.py

@@ -200,3 +200,28 @@ def drop_partitions(
         raise NotImplementedError(
             "drop_partitions is not supported by this catalog. Use REST catalog for partition drop."
         )
+
+    def list_partitions_paged(
+            self,
+            identifier: Union[str, Identifier],
+            max_results: Optional[int] = None,
+            page_token: Optional[str] = None,
+            partition_name_pattern: Optional[str] = None,
+    ):
+        """List partitions of a table with pagination.
+
+        Args:
+            identifier: Path of the table.
+            max_results: Maximum number of results to return per page.
+            page_token: Token for pagination.
+            partition_name_pattern: Optional pattern to filter partition names.
+
+        Returns:
+            PagedList of Partition objects.
+
+        Raises:
+            NotImplementedError: If the catalog does not support listing partitions.
+        """
+        raise NotImplementedError(
+            "list_partitions_paged is not supported by this catalog."
+        )

paimon-python/pypaimon/catalog/filesystem_catalog.py

@@ -254,3 +254,77 @@ def load_snapshot(self, identifier: Identifier):
             NotImplementedError: FileSystemCatalog does not support version management
         """
         raise NotImplementedError("Filesystem catalog does not support load_snapshot")
+
+    def list_partitions_paged(
+            self,
+            identifier: Union[str, Identifier],
+            max_results: Optional[int] = None,
+            page_token: Optional[str] = None,
+            partition_name_pattern: Optional[str] = None,
+    ):
+        from pypaimon.api.api_response import Partition, PagedList
+        from pypaimon.manifest.manifest_list_manager import ManifestListManager
+        from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+
+        if not isinstance(identifier, Identifier):
+            identifier = Identifier.from_string(identifier)
+
+        table = self.get_table(identifier)
+        snapshot = table.snapshot_manager().get_latest_snapshot()
+        if snapshot is None:
+            return PagedList(elements=[])
+
+        # Read all manifest entries (ADD - DELETE merged)
+        manifest_list_manager = ManifestListManager(table)
+        manifest_file_manager = ManifestFileManager(table)
+        manifest_files = manifest_list_manager.read_all(snapshot)
+        entries = manifest_file_manager.read_entries_parallel(manifest_files, drop_stats=True)
+
+        # Group entries by partition spec
+        partition_map = {}  # spec_key -> aggregated stats
+        for entry in entries:
+            spec = {field.name: str(v) for field, v in
+                    zip(entry.partition.fields, entry.partition.values)}
+            spec_key = tuple(sorted(spec.items()))
+
+            if spec_key not in partition_map:
+                partition_map[spec_key] = {
+                    'spec': spec,
+                    'record_count': 0,
+                    'file_size_in_bytes': 0,
+                    'file_count': 0,
+                    'last_file_creation_time': 0,
+                    'buckets': set(),
+                }
+            stats = partition_map[spec_key]
+            stats['record_count'] += entry.file.row_count
+            stats['file_size_in_bytes'] += entry.file.file_size
+            stats['file_count'] += 1
+            if entry.file.creation_time is not None:
+                ct = entry.file.creation_time.get_millisecond()
+                if ct > stats['last_file_creation_time']:
+                    stats['last_file_creation_time'] = ct
+            stats['buckets'].add(entry.bucket)
+
+        # Convert to Partition objects
+        partitions = []
+        for stats in partition_map.values():
+            partitions.append(Partition(
+                spec=stats['spec'],
+                record_count=stats['record_count'],
+                file_size_in_bytes=stats['file_size_in_bytes'],
+                file_count=stats['file_count'],
+                last_file_creation_time=stats['last_file_creation_time'],
+                total_buckets=len(stats['buckets']),
+            ))
+
+        # Apply pattern filter
+        if partition_name_pattern:
+            import re
+            regex = re.compile(partition_name_pattern.replace('*', '.*'))
+            partitions = [
+                p for p in partitions
+                if regex.fullmatch(','.join(f'{k}={v}' for k, v in p.spec.items()))
+            ]
+
+        return PagedList(elements=partitions)

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

@@ -267,6 +267,27 @@ def drop_partitions(
         finally:
             commit.close()
 
+    def list_partitions_paged(
+            self,
+            identifier: Union[str, Identifier],
+            max_results: Optional[int] = None,
+            page_token: Optional[str] = None,
+            partition_name_pattern: Optional[str] = None,
+    ):
+        if not isinstance(identifier, Identifier):
+            identifier = Identifier.from_string(identifier)
+        try:
+            return self.rest_api.list_partitions_paged(
+                identifier,
+                max_results,
+                page_token,
+                partition_name_pattern
+            )
+        except NoSuchResourceException as e:
+            raise TableNotExistException(identifier) from e
+        except ForbiddenException as e:
+            raise TableNoPermissionException(identifier) from e
+
     def alter_table(
         self,
         identifier: Union[str, Identifier],

paimon-python/pypaimon/cli/cli_table.py

@@ -573,6 +573,70 @@ def cmd_table_alter(args):
         sys.exit(1)
 
 
+def cmd_table_list_partitions(args):
+    """
+    Execute the 'table list-partitions' command.
+
+    Lists partitions of a Paimon table with optional pattern filtering.
+
+    Args:
+        args: Parsed command line arguments.
+    """
+    from pypaimon.cli.cli import load_catalog_config, create_catalog
+
+    # Load catalog configuration
+    config_path = args.config
+    config = load_catalog_config(config_path)
+
+    # Create catalog
+    catalog = create_catalog(config)
+
+    # Parse table identifier
+    table_identifier = args.table
+    parts = table_identifier.split('.')
+    if len(parts) != 2:
+        print(f"Error: Invalid table identifier '{table_identifier}'. "
+              f"Expected format: 'database.table'", file=sys.stderr)
+        sys.exit(1)
+
+    # List partitions with pagination
+    pattern = getattr(args, 'pattern', None)
+    try:
+        paged_list = catalog.list_partitions_paged(
+            table_identifier,
+            partition_name_pattern=pattern,
+        )
+        import pandas as pd
+
+        partitions = paged_list.elements
+        if not partitions:
+            print("No partitions found.")
+            return
+
+        data = []
+        for p in partitions:
+            spec_str = ",".join(f"{k}={v}" for k, v in p.spec.items())
+            data.append({
+                'Partition': spec_str,
+                'RecordCount': p.record_count,
+                'FileSizeInBytes': p.file_size_in_bytes,
+                'FileCount': p.file_count,
+                'LastFileCreationTime': p.last_file_creation_time,
+                'UpdatedAt': p.updated_at,
+                'UpdatedBy': p.updated_by or '',
+            })
+
+        df = pd.DataFrame(data)
+        print(df.to_string(index=False))
+
+    except NotImplementedError as e:
+        print(f"Error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:
+        print(f"Error: Failed to list partitions: {e}", file=sys.stderr)
+        sys.exit(1)
+
+
 def add_table_subcommands(table_parser):
     """
     Add table subcommands to the parser.
@@ -669,6 +733,20 @@ def add_table_subcommands(table_parser):
     )
     import_parser.set_defaults(func=cmd_table_import)
 
+    # table list-partitions command
+    list_partitions_parser = table_subparsers.add_parser('list-partitions', help='List partitions of a table')
+    list_partitions_parser.add_argument(
+        'table',
+        help='Table identifier in format: database.table'
+    )
+    list_partitions_parser.add_argument(
+        '--pattern', '-p',
+        type=str,
+        default=None,
+        help='Partition name pattern to filter partitions (e.g., "dt=2024*")'
+    )
+    list_partitions_parser.set_defaults(func=cmd_table_list_partitions)
+
     # table rename command
     rename_parser = table_subparsers.add_parser('rename', help='Rename a table')
     rename_parser.add_argument(

paimon-python/pypaimon/common/json_util.py

@@ -101,10 +101,12 @@ def __from_dict(data: Dict[str, Any], target_class: Type[T]) -> T:
             field_type = field_info.type
             if origin_type is Union and len(args) == 2:
                 field_type = args[0]
+                origin_type = getattr(field_type, '__origin__', None)
+                args = getattr(field_type, '__args__', None)
             if is_dataclass(field_type):
                 type_mapping[json_name] = field_type
-            elif origin_type in (list, List) and is_dataclass(args[0]):
-                type_mapping[json_name] = field_info.type
+            elif origin_type in (list, List) and args and is_dataclass(args[0]):
+                type_mapping[json_name] = field_type
 
         # Map JSON data to field names
         kwargs = {}