You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/03/17 07:30:48 UTC
(paimon) branch master updated: [python] Add list partitions support for Python CLI (#7442)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 81760b0dec [python] Add list partitions support for Python CLI (#7442)
81760b0dec is described below
commit 81760b0dec2fcc14500f61a79ddfd570f4e18547
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Tue Mar 17 15:30:43 2026 +0800
[python] Add list partitions support for Python CLI (#7442)
---
docs/content/pypaimon/cli.md | 30 ++++++
paimon-python/pypaimon/api/api_response.py | 32 ++++++
paimon-python/pypaimon/api/resource_paths.py | 5 +
paimon-python/pypaimon/api/rest_api.py | 28 +++++-
paimon-python/pypaimon/catalog/catalog.py | 25 +++++
.../pypaimon/catalog/filesystem_catalog.py | 103 +++++++++++++++++--
.../pypaimon/catalog/rest/rest_catalog.py | 21 ++++
paimon-python/pypaimon/cli/cli_table.py | 78 +++++++++++++++
paimon-python/pypaimon/common/json_util.py | 6 +-
paimon-python/pypaimon/tests/cli_table_test.py | 109 +++++++++++++++++++++
.../pypaimon/tests/filesystem_catalog_test.py | 90 +++++++++++++++++
.../pypaimon/tests/rest/rest_base_test.py | 72 +++++++++++++-
paimon-python/pypaimon/tests/rest/rest_server.py | 53 +++++++++-
13 files changed, 640 insertions(+), 12 deletions(-)
diff --git a/docs/content/pypaimon/cli.md b/docs/content/pypaimon/cli.md
index 3588c7c974..e0bd1c423f 100644
--- a/docs/content/pypaimon/cli.md
+++ b/docs/content/pypaimon/cli.md
@@ -309,6 +309,36 @@ 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
+```
+
### Table Rename
Rename a table in the catalog. Both source and target must be specified in `database.table` format.
diff --git a/paimon-python/pypaimon/api/api_response.py b/paimon-python/pypaimon/api/api_response.py
index 0cfa73c855..64afc50b9b 100644
--- a/paimon-python/pypaimon/api/api_response.py
+++ b/paimon-python/pypaimon/api/api_response.py
@@ -23,6 +23,7 @@ from typing import Dict, Generic, List, Optional
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 @@ class ListDatabasesResponse(PagedResponse[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"
diff --git a/paimon-python/pypaimon/api/resource_paths.py b/paimon-python/pypaimon/api/resource_paths.py
index 03a7c43fab..9b2351c7e3 100644
--- a/paimon-python/pypaimon/api/resource_paths.py
+++ b/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 @@ class ResourcePaths:
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))
diff --git a/paimon-python/pypaimon/api/rest_api.py b/paimon-python/pypaimon/api/rest_api.py
index b7c816c9d9..8db56fad0f 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -26,8 +26,10 @@ from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
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 @@ class RESTApi:
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:
diff --git a/paimon-python/pypaimon/catalog/catalog.py b/paimon-python/pypaimon/catalog/catalog.py
index d5f6cdee99..38331c64e2 100644
--- a/paimon-python/pypaimon/catalog/catalog.py
+++ b/paimon-python/pypaimon/catalog/catalog.py
@@ -200,3 +200,28 @@ class Catalog(ABC):
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."
+ )
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index 73a7f6e3cd..0e4bd7d40c 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -245,12 +245,101 @@ class FileSystemCatalog(Catalog):
raise NotImplementedError("This catalog does not support commit catalog")
def load_snapshot(self, identifier: Identifier):
- """Load the snapshot of table identified by the given Identifier.
+ raise NotImplementedError("Filesystem catalog does not support load_snapshot")
- Args:
- identifier: Path of the table
+ 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
- Raises:
- NotImplementedError: FileSystemCatalog does not support version management
- """
- raise NotImplementedError("Filesystem catalog does not support load_snapshot")
+ 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 with proper regex escaping
+ if partition_name_pattern:
+ import re
+ # Escape special regex chars except '*', then replace '*' with '.*'
+ escaped_pattern = re.escape(partition_name_pattern).replace(r'\*', '.*')
+ regex = re.compile(escaped_pattern)
+ partitions = [
+ p for p in partitions
+ if regex.fullmatch(','.join(f'{k}={v}' for k, v in p.spec.items()))
+ ]
+
+ # Sort partitions by name (partition spec string)
+ partitions.sort(key=lambda p: ','.join(f'{k}={v}' for k, v in sorted(p.spec.items())))
+
+ # Apply pagination
+ start_index = 0
+ if page_token is not None:
+ try:
+ start_index = int(page_token)
+ except ValueError:
+ # Invalid token, start from beginning
+ start_index = 0
+
+ end_index = len(partitions)
+ if max_results is not None and max_results > 0:
+ end_index = min(start_index + max_results, len(partitions))
+
+ result_partitions = partitions[start_index:end_index]
+ next_page_token = None
+ if max_results is not None and end_index < len(partitions):
+ next_page_token = str(end_index)
+
+ return PagedList(elements=result_partitions, next_page_token=next_page_token)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 399d4e1a86..a8f232a19c 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -267,6 +267,27 @@ class RESTCatalog(Catalog):
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],
diff --git a/paimon-python/pypaimon/cli/cli_table.py b/paimon-python/pypaimon/cli/cli_table.py
index 9c68ff653e..1056b7c679 100644
--- a/paimon-python/pypaimon/cli/cli_table.py
+++ b/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(
diff --git a/paimon-python/pypaimon/common/json_util.py b/paimon-python/pypaimon/common/json_util.py
index 6fdd7bbb62..fb257cfe3c 100644
--- a/paimon-python/pypaimon/common/json_util.py
+++ b/paimon-python/pypaimon/common/json_util.py
@@ -101,10 +101,12 @@ class JSON:
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 = {}
diff --git a/paimon-python/pypaimon/tests/cli_table_test.py b/paimon-python/pypaimon/tests/cli_table_test.py
index dfe6f7e431..9eab3f00b9 100644
--- a/paimon-python/pypaimon/tests/cli_table_test.py
+++ b/paimon-python/pypaimon/tests/cli_table_test.py
@@ -1168,5 +1168,114 @@ class CliTableTest(unittest.TestCase):
with self.assertRaises(Exception):
self.catalog.get_table('test_db.rename_source')
+ @classmethod
+ def _create_partitioned_table(cls):
+ """Create a partitioned test table and insert sample data."""
+ pa_schema = pa.schema([
+ ('dt', pa.string()),
+ ('region', pa.string()),
+ ('id', pa.int32()),
+ ('value', pa.string())
+ ])
+
+ schema = Schema.from_pyarrow_schema(pa_schema, partition_keys=['dt', 'region'])
+ cls.catalog.create_table('test_db.partitioned', schema, True)
+
+ table = cls.catalog.get_table('test_db.partitioned')
+
+ # Write data for partition dt=2024-01-01, region=us
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ data = pa.Table.from_pydict({
+ 'dt': ['2024-01-01', '2024-01-01'],
+ 'region': ['us', 'us'],
+ 'id': [1, 2],
+ 'value': ['a', 'b'],
+ }, schema=pa_schema)
+ table_write.write_arrow(data)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ # Write data for partition dt=2024-01-02, region=eu
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ data = pa.Table.from_pydict({
+ 'dt': ['2024-01-02', '2024-01-02', '2024-01-02'],
+ 'region': ['eu', 'eu', 'eu'],
+ 'id': [3, 4, 5],
+ 'value': ['c', 'd', 'e'],
+ }, schema=pa_schema)
+ table_write.write_arrow(data)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ def test_cli_table_list_partitions(self):
+ """Test list-partitions command with real partitioned table."""
+ self._create_partitioned_table()
+
+ with patch('sys.argv',
+ ['paimon', '-c', self.config_file,
+ 'table', 'list-partitions', 'test_db.partitioned']):
+ with patch('sys.stdout', new_callable=StringIO) as mock_stdout:
+ try:
+ main()
+ except SystemExit:
+ pass
+
+ output = mock_stdout.getvalue()
+ # Verify header columns
+ self.assertIn('Partition', output)
+ self.assertIn('RecordCount', output)
+ self.assertIn('FileSizeInBytes', output)
+ self.assertIn('FileCount', output)
+ # Verify partition specs
+ self.assertIn('dt=2024-01-01,region=us', output)
+ self.assertIn('dt=2024-01-02,region=eu', output)
+
+ def test_cli_table_list_partitions_with_pattern(self):
+ """Test list-partitions command with --pattern filter."""
+ self._create_partitioned_table()
+
+ with patch('sys.argv',
+ ['paimon', '-c', self.config_file,
+ 'table', 'list-partitions', 'test_db.partitioned',
+ '--pattern', 'dt=2024-01-01*']):
+ with patch('sys.stdout', new_callable=StringIO) as mock_stdout:
+ try:
+ main()
+ except SystemExit:
+ pass
+
+ output = mock_stdout.getvalue()
+ self.assertIn('dt=2024-01-01,region=us', output)
+ self.assertNotIn('dt=2024-01-02', output)
+
+ def test_cli_table_list_partitions_empty(self):
+ """Test list-partitions on non-partitioned table (no snapshot = empty)."""
+ # Create a table with no data
+ pa_schema = pa.schema([
+ ('id', pa.int32()),
+ ('val', pa.string())
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema)
+ self.catalog.create_table('test_db.empty_part', schema, True)
+
+ with patch('sys.argv',
+ ['paimon', '-c', self.config_file,
+ 'table', 'list-partitions', 'test_db.empty_part']):
+ with patch('sys.stdout', new_callable=StringIO) as mock_stdout:
+ try:
+ main()
+ except SystemExit:
+ pass
+
+ output = mock_stdout.getvalue()
+ self.assertIn('No partitions found', output)
+
+
if __name__ == '__main__':
unittest.main()
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
index c778c27082..dbbc18bb3e 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
@@ -19,6 +19,8 @@ import tempfile
import unittest
from unittest.mock import MagicMock
+import pyarrow as pa
+
from pypaimon import CatalogFactory, Schema
from pypaimon.catalog.catalog_exception import (DatabaseAlreadyExistException,
DatabaseNotExistException,
@@ -276,3 +278,91 @@ class FileSystemCatalogTest(unittest.TestCase):
# Restore original method
filesystem_catalog.file_io.exists = original_exists
+
+ def _create_partitioned_table_with_data(self, catalog, identifier, partitions_data):
+ pa_schema = pa.schema([
+ ('dt', pa.string()),
+ ('col1', pa.int32()),
+ ])
+ schema = Schema.from_pyarrow_schema(pa_schema, partition_keys=['dt'])
+ catalog.create_table(identifier, schema, True)
+ table = catalog.get_table(identifier)
+
+ for part in partitions_data:
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ data = pa.Table.from_pydict({
+ 'dt': [part['dt']] * part['rows'],
+ 'col1': list(range(part['rows'])),
+ }, schema=pa_schema)
+ table_write.write_arrow(data)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+
+ def test_list_partitions_paged_pagination(self):
+ """Test list_partitions_paged pagination with max_results and page_token."""
+ catalog = CatalogFactory.create({"warehouse": self.warehouse})
+ catalog.create_database("test_db", False)
+
+ identifier = "test_db.paged_tbl"
+ self._create_partitioned_table_with_data(catalog, identifier, [
+ {'dt': '2024-01-01', 'rows': 1},
+ {'dt': '2024-01-02', 'rows': 1},
+ {'dt': '2024-01-03', 'rows': 1},
+ ])
+
+ # First page: max_results=2
+ page1 = catalog.list_partitions_paged(identifier, max_results=2)
+ self.assertEqual(len(page1.elements), 2)
+ self.assertIsNotNone(page1.next_page_token)
+
+ # Second page: use next_page_token
+ page2 = catalog.list_partitions_paged(
+ identifier, max_results=2, page_token=page1.next_page_token
+ )
+ self.assertEqual(len(page2.elements), 1)
+ self.assertIsNone(page2.next_page_token)
+
+ # All specs across pages should cover all 3 partitions
+ all_specs = [p.spec['dt'] for p in page1.elements + page2.elements]
+ self.assertEqual(sorted(all_specs), ['2024-01-01', '2024-01-02', '2024-01-03'])
+
+ # max_results larger than total returns all
+ result = catalog.list_partitions_paged(identifier, max_results=100)
+ self.assertEqual(len(result.elements), 3)
+ self.assertIsNone(result.next_page_token)
+
+ def test_list_partitions_paged_pattern(self):
+ """Test list_partitions_paged with partition_name_pattern filter."""
+ catalog = CatalogFactory.create({"warehouse": self.warehouse})
+ catalog.create_database("test_db", False)
+
+ identifier = "test_db.pattern_tbl"
+ self._create_partitioned_table_with_data(catalog, identifier, [
+ {'dt': '2024-01-01', 'rows': 1},
+ {'dt': '2024-02-01', 'rows': 1},
+ {'dt': '2024-02-15', 'rows': 1},
+ ])
+
+ # Exact match
+ result = catalog.list_partitions_paged(
+ identifier, partition_name_pattern='dt=2024-01-01'
+ )
+ self.assertEqual(len(result.elements), 1)
+ self.assertEqual(result.elements[0].spec['dt'], '2024-01-01')
+
+ # Wildcard match
+ result = catalog.list_partitions_paged(
+ identifier, partition_name_pattern='dt=2024-02*'
+ )
+ self.assertEqual(len(result.elements), 2)
+ specs = sorted(p.spec['dt'] for p in result.elements)
+ self.assertEqual(specs, ['2024-02-01', '2024-02-15'])
+
+ # No match
+ result = catalog.list_partitions_paged(
+ identifier, partition_name_pattern='dt=2025*'
+ )
+ self.assertEqual(len(result.elements), 0)
diff --git a/paimon-python/pypaimon/tests/rest/rest_base_test.py b/paimon-python/pypaimon/tests/rest/rest_base_test.py
index 0c60d4af9a..eb0905e692 100644
--- a/paimon-python/pypaimon/tests/rest/rest_base_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_base_test.py
@@ -26,7 +26,7 @@ import uuid
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
-from pypaimon.api.api_response import ConfigResponse
+from pypaimon.api.api_response import ConfigResponse, Partition
from pypaimon.api.auth import BearTokenAuthProvider
from pypaimon.common.options import Options
from pypaimon.catalog.catalog_context import CatalogContext
@@ -275,3 +275,73 @@ class RESTBaseTest(unittest.TestCase):
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()
+
+ def test_list_partitions_paged(self):
+ """Test list_partitions_paged returns partitions from server store."""
+ identifier = Identifier.from_string('default.test_reader_iterator')
+
+ # Add test partitions to server store
+ p1 = Partition(
+ spec={"dt": "p1"},
+ record_count=4,
+ file_size_in_bytes=1024,
+ file_count=1,
+ last_file_creation_time=1000,
+ total_buckets=1,
+ done=False,
+ )
+ p2 = Partition(
+ spec={"dt": "p2"},
+ record_count=4,
+ file_size_in_bytes=2048,
+ file_count=1,
+ last_file_creation_time=2000,
+ total_buckets=1,
+ done=True,
+ )
+ p3 = Partition(
+ spec={"dt": "p3"},
+ record_count=2,
+ file_size_in_bytes=512,
+ file_count=1,
+ last_file_creation_time=3000,
+ total_buckets=1,
+ done=False,
+ )
+ self.server.table_partitions_store[identifier.get_full_name()] = [p1, p2, p3]
+
+ # Test: list all partitions
+ result = self.rest_catalog.list_partitions_paged(identifier)
+ self.assertEqual(len(result.elements), 3)
+
+ # Test: list with max_results
+ result = self.rest_catalog.list_partitions_paged(identifier, max_results=2)
+ self.assertEqual(len(result.elements), 2)
+ self.assertIsNotNone(result.next_page_token)
+
+ # Test: list next page using page_token
+ result2 = self.rest_catalog.list_partitions_paged(
+ identifier, max_results=2, page_token=result.next_page_token
+ )
+ self.assertEqual(len(result2.elements), 1)
+ self.assertIsNone(result2.next_page_token)
+
+ # Test: list with pattern filter
+ result = self.rest_catalog.list_partitions_paged(
+ identifier, partition_name_pattern="dt=p1"
+ )
+ self.assertEqual(len(result.elements), 1)
+ self.assertEqual(result.elements[0].spec, {"dt": "p1"})
+
+ # Test: list with pattern using wildcard
+ result = self.rest_catalog.list_partitions_paged(
+ identifier, partition_name_pattern="dt=p%"
+ )
+ self.assertEqual(len(result.elements), 3)
+
+ def test_list_partitions_paged_empty(self):
+ """Test list_partitions_paged returns empty when no partitions."""
+ identifier = Identifier.from_string('default.test_reader_iterator')
+ result = self.rest_catalog.list_partitions_paged(identifier)
+ self.assertEqual(len(result.elements), 0)
+ self.assertIsNone(result.next_page_token)
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py b/paimon-python/pypaimon/tests/rest/rest_server.py
index ee84553932..8eaebcffa4 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -33,7 +33,8 @@ from pypaimon.api.api_request import (AlterTableRequest, CreateDatabaseRequest,
CreateTableRequest, RenameTableRequest)
from pypaimon.api.api_response import (ConfigResponse, GetDatabaseResponse,
GetTableResponse, ListDatabasesResponse,
- ListTablesResponse, PagedList,
+ ListPartitionsResponse, ListTablesResponse,
+ PagedList, Partition,
RESTResponse, ErrorResponse)
from pypaimon.api.resource_paths import ResourcePaths
from pypaimon.api.rest_util import RESTUtil
@@ -381,6 +382,8 @@ class RESTCatalogServer:
if resource_type == ResourcePaths.TABLES:
return self._handle_table_resource(method, path_parts, identifier, data, parameters)
+ elif resource_type == ResourcePaths.PARTITIONS:
+ return self._table_partitions_handle(method, identifier, parameters)
return self._mock_response(ErrorResponse(None, None, "Not Found", 404), 404)
@@ -462,10 +465,24 @@ class RESTCatalogServer:
return self._table_rollback_handle(method, data, lookup_identifier)
elif operation == "snapshot":
return self._table_snapshot_handle(method, lookup_identifier)
+ elif operation == ResourcePaths.PARTITIONS:
+ return self._table_partitions_handle(method, lookup_identifier, parameters)
else:
return self._mock_response(ErrorResponse(None, None, "Not Found", 404), 404)
return self._mock_response(ErrorResponse(None, None, "Not Found", 404), 404)
+ def _table_partitions_handle(
+ self, method: str, identifier: Identifier, parameters: Dict[str, str]) -> Tuple[str, int]:
+ """Handle table partitions listing"""
+ if method != "GET":
+ return self._mock_response(ErrorResponse(None, None, "Method Not Allowed", 405), 405)
+
+ if identifier.get_full_name() not in self.table_metadata_store:
+ raise TableNotExistException(identifier)
+
+ partitions = self._list_partitions(identifier, parameters)
+ return self._generate_final_list_partitions_response(parameters, partitions)
+
def _databases_api_handler(self, method: str, data: str,
parameters: Dict[str, str]) -> Tuple[str, int]:
"""Handle databases API requests"""
@@ -914,6 +931,8 @@ class RESTCatalogServer:
"""Get paging key for entity"""
if isinstance(entity, str):
return entity
+ elif isinstance(entity, Partition):
+ return "/".join(f"{k}={v}" for k, v in sorted(entity.spec.items()))
elif hasattr(entity, 'get_name'):
return entity.get_name()
elif hasattr(entity, 'get_full_name'):
@@ -1043,6 +1062,22 @@ class RESTCatalogServer:
return tables
+ def _list_partitions(self, identifier: Identifier, parameters: Dict[str, str]) -> List[Partition]:
+ """List partitions for a table from the partitions store."""
+ partition_name_pattern = parameters.get(PARTITION_NAME_PATTERN)
+ partitions = self.table_partitions_store.get(identifier.get_full_name(), [])
+ if partition_name_pattern:
+ partitions = [
+ p for p in partitions
+ if self._match_partition_name_pattern(p, partition_name_pattern)
+ ]
+ return partitions
+
+ def _match_partition_name_pattern(self, partition: Partition, pattern: str) -> bool:
+ """Match partition spec against a name pattern."""
+ partition_name = "/".join(f"{k}={v}" for k, v in sorted(partition.spec.items()))
+ return self._match_name_pattern(partition_name, pattern)
+
# Response generation methods
def _generate_final_list_databases_response(self, parameters: Dict[str, str],
databases: List[str]) -> Tuple[str, int]:
@@ -1076,6 +1111,22 @@ class RESTCatalogServer:
return self._mock_response(response, 200)
+ def _generate_final_list_partitions_response(
+ self, parameters: Dict[str, str], partitions: List[Partition]) -> Tuple[str, int]:
+ """Generate final list partitions response"""
+ if partitions:
+ max_results = self._get_max_results(parameters)
+ page_token = parameters.get(PAGE_TOKEN)
+ paged_partitions = self._build_paged_entities(partitions, max_results, page_token)
+ response = ListPartitionsResponse(
+ partitions=paged_partitions.elements,
+ next_page_token=paged_partitions.next_page_token
+ )
+ else:
+ response = ListPartitionsResponse(partitions=[], next_page_token=None)
+
+ return self._mock_response(response, 200)
+
def add_no_permission_database(self, database: str) -> None:
"""Add no permission database"""
self.no_permission_databases.append(database)