fix

Author: JingsongLi

docs/content/pypaimon/cli.md

@@ -339,8 +339,6 @@ dt=2024-01-02,region=eu          300           524288          5         1704153
 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/catalog/filesystem_catalog.py

@@ -245,14 +245,6 @@ def commit_snapshot(
         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.
-
-        Args:
-            identifier: Path of the table
-
-        Raises:
-            NotImplementedError: FileSystemCatalog does not support version management
-        """
         raise NotImplementedError("Filesystem catalog does not support load_snapshot")
 
     def list_partitions_paged(
@@ -318,13 +310,36 @@ def list_partitions_paged(
                 total_buckets=len(stats['buckets']),
             ))
 
-        # Apply pattern filter
+        # Apply pattern filter with proper regex escaping
         if partition_name_pattern:
             import re
-            regex = re.compile(partition_name_pattern.replace('*', '.*'))
+            # 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()))
             ]
 
-        return PagedList(elements=partitions)
+        # 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)

paimon-python/pypaimon/tests/filesystem_catalog_test.py

@@ -19,6 +19,8 @@
 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,162 @@ def test_get_database_propagates_exists_error(self):
 
         # Restore original method
         filesystem_catalog.file_io.exists = original_exists
+
+    def _create_partitioned_table_with_data(self, catalog, identifier, partitions_data):
+        """Helper to create a partitioned table and write data for each partition.
+
+        Args:
+            catalog: The catalog instance.
+            identifier: Table identifier string (e.g. 'test_db.tbl').
+            partitions_data: List of dicts, each with 'dt' and rows count.
+                e.g. [{'dt': '2024-01-01', 'rows': 2}, {'dt': '2024-01-02', 'rows': 3}]
+        """
+        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(self):
+        """Test list_partitions_paged with real data from manifest files."""
+        catalog = CatalogFactory.create({"warehouse": self.warehouse})
+        catalog.create_database("test_db", False)
+
+        identifier = "test_db.part_tbl"
+        self._create_partitioned_table_with_data(catalog, identifier, [
+            {'dt': '2024-01-03', 'rows': 3},
+            {'dt': '2024-01-01', 'rows': 2},
+            {'dt': '2024-01-02', 'rows': 5},
+        ])
+
+        # List all partitions
+        result = catalog.list_partitions_paged(identifier)
+        self.assertEqual(len(result.elements), 3)
+        self.assertIsNone(result.next_page_token)
+
+        # Verify partitions are sorted by spec
+        specs = [p.spec['dt'] for p in result.elements]
+        self.assertEqual(specs, sorted(specs))
+
+        # Verify aggregated statistics
+        part_map = {p.spec['dt']: p for p in result.elements}
+        self.assertEqual(part_map['2024-01-01'].record_count, 2)
+        self.assertEqual(part_map['2024-01-02'].record_count, 5)
+        self.assertEqual(part_map['2024-01-03'].record_count, 3)
+        for p in result.elements:
+            self.assertGreater(p.file_size_in_bytes, 0)
+            self.assertGreater(p.file_count, 0)
+            self.assertGreater(p.last_file_creation_time, 0)
+
+    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)
+
+    def test_list_partitions_paged_empty(self):
+        """Test list_partitions_paged on a table with no data."""
+        catalog = CatalogFactory.create({"warehouse": self.warehouse})
+        catalog.create_database("test_db", False)
+
+        pa_schema = pa.schema([('dt', pa.string()), ('val', pa.int32())])
+        schema = Schema.from_pyarrow_schema(pa_schema, partition_keys=['dt'])
+        catalog.create_table('test_db.empty_tbl', schema, False)
+
+        result = catalog.list_partitions_paged('test_db.empty_tbl')
+        self.assertEqual(len(result.elements), 0)
+        self.assertIsNone(result.next_page_token)
+
+    def test_list_partitions_paged_invalid_token(self):
+        """Test list_partitions_paged with invalid page_token falls back to start."""
+        catalog = CatalogFactory.create({"warehouse": self.warehouse})
+        catalog.create_database("test_db", False)
+
+        identifier = "test_db.token_tbl"
+        self._create_partitioned_table_with_data(catalog, identifier, [
+            {'dt': '2024-01-01', 'rows': 1},
+            {'dt': '2024-01-02', 'rows': 1},
+        ])
+
+        # Invalid page_token should fall back to start
+        result = catalog.list_partitions_paged(
+            identifier, max_results=1, page_token='invalid'
+        )
+        self.assertEqual(len(result.elements), 1)
+        self.assertEqual(result.elements[0].spec['dt'], '2024-01-01')
+        self.assertIsNotNone(result.next_page_token)

paimon-python/pypaimon/tests/rest/rest_server.py

@@ -471,8 +471,8 @@ def _handle_table_resource(self, method: str, path_parts: List[str],
                 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]:
+    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)
@@ -1111,8 +1111,8 @@ def _generate_final_list_tables_response(self, parameters: Dict[str, str],
 
         return self._mock_response(response, 200)
 
-    def _generate_final_list_partitions_response(self, parameters: Dict[str, str],
-                                                   partitions: List[Partition]) -> Tuple[str, int]:
+    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)