This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 90e0a2a5bd [python] Fall back to table metadata when REST partition
listing is unimplemented (#10176)
90e0a2a5bd is described below
commit 90e0a2a5bd87b99f9c3a1ceef185a1147d9dc8c1
Author: jackylee <[email protected]>
AuthorDate: Sat Oct 3 17:12:34 2026 +0800
[python] Fall back to table metadata when REST partition listing is
unimplemented (#10176)
---
paimon-python/pypaimon/catalog/catalog_utils.py | 121 +++++++++++++++++++++
.../pypaimon/catalog/filesystem_catalog.py | 87 +--------------
.../pypaimon/catalog/rest/rest_catalog.py | 21 ++++
.../rest/rest_list_partitions_fallback_test.py | 104 ++++++++++++++++++
4 files changed, 249 insertions(+), 84 deletions(-)
diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py
b/paimon-python/pypaimon/catalog/catalog_utils.py
new file mode 100644
index 0000000000..a3f70ab5b7
--- /dev/null
+++ b/paimon-python/pypaimon/catalog/catalog_utils.py
@@ -0,0 +1,121 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Helpers shared across catalog implementations."""
+
+import re
+from typing import Optional
+
+from pypaimon.api.api_response import PagedList, Partition
+from pypaimon.table.table import Table
+
+
+def list_partitions_from_file_system(
+ table: Table,
+ max_results: Optional[int] = None,
+ page_token: Optional[str] = None,
+ partition_name_pattern: Optional[str] = None,
+) -> PagedList[Partition]:
+ """Compute partitions from a table's own metadata (manifest entries).
+
+ Mirrors Java ``CatalogUtils.listPartitionsFromFileSystem``: the aggregation
+ is shared by the filesystem catalog and by the REST catalog, which falls
+ back to it when the server does not implement the partition-listing
+ endpoint.
+ """
+ from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+ from pypaimon.manifest.manifest_list_manager import ManifestListManager
+
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if snapshot is None:
+ return PagedList(elements=[])
+
+ # Read all manifest entries (ADD - DELETE merged) and group by partition.
+ 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)
+
+ 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)
+
+ partitions = [
+ 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']),
+ )
+ for stats in partition_map.values()
+ ]
+
+ # Apply pattern filter with proper regex escaping.
+ if partition_name_pattern:
+ # 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/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index e71f9cc18a..f2aad5b442 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -307,95 +307,14 @@ class FileSystemCatalog(Catalog):
page_token: Optional[str] = None,
partition_name_pattern: Optional[str] = None,
) -> PagedList[Partition]:
- from pypaimon.manifest.manifest_file_manager import ManifestFileManager
- from pypaimon.manifest.manifest_list_manager import ManifestListManager
+ from pypaimon.catalog.catalog_utils import
list_partitions_from_file_system
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)
+ return list_partitions_from_file_system(
+ table, max_results, page_token, partition_name_pattern)
def drop_partitions(
self,
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index c8ebbfa10e..12fc512c29 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -368,6 +368,27 @@ class RESTCatalog(Catalog):
raise TableNotExistException(identifier) from e
except ForbiddenException as e:
raise TableNoPermissionException(identifier) from e
+ except NotImplementedException:
+ # The server does not implement the partition-listing endpoint.
+ # Mirror Java RESTCatalog#listPartitionsPaged and fall back to
+ # computing partitions from the table's own metadata.
+ #
+ # Scope: this Python fallback reads Paimon manifests, so it only
+ # supports data tables (FileStoreTable). Java re-raises solely for
+ # catalog-managed-partition tables (a FormatTable whose
+ # partitionManager is set) and otherwise scans the table's file
+ # system, including an unmanaged format table's directory layout
+ # (CatalogUtils.listPartitionsFromFileSystem). PyPaimon does not
+ # scan format/object tables here yet, so those still surface the
+ # server's NotImplemented error rather than being listed.
+ from pypaimon.catalog.catalog_utils import (
+ list_partitions_from_file_system)
+
+ table = self.get_table(identifier)
+ if not isinstance(table, FileStoreTable):
+ raise
+ return list_partitions_from_file_system(
+ table, max_results, page_token, partition_name_pattern)
def alter_table(
self,
diff --git
a/paimon-python/pypaimon/tests/rest/rest_list_partitions_fallback_test.py
b/paimon-python/pypaimon/tests/rest/rest_list_partitions_fallback_test.py
new file mode 100644
index 0000000000..c0e2de6089
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_list_partitions_fallback_test.py
@@ -0,0 +1,104 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from pypaimon import Schema
+from pypaimon.api.rest_exception import NotImplementedException
+from pypaimon.common.identifier import Identifier
+from pypaimon.tests.rest.rest_base_test import RESTBaseTest
+
+
+class RESTListPartitionsFallbackTest(RESTBaseTest):
+ """When the REST server does not implement partition listing, the catalog
+ must fall back to computing partitions from the table's own metadata,
+ mirroring the Java RESTCatalog."""
+
+ def _create_partitioned_table(self, name):
+ schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
+ self.rest_catalog.create_table(name, schema, False)
+ table = self.rest_catalog.get_table(name)
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(self.expected)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+ return Identifier.from_string(name)
+
+ def _make_endpoint_not_implemented(self):
+ def _raise(*args, **kwargs):
+ raise NotImplementedException("partition listing not implemented")
+ self.rest_catalog.rest_api.list_partitions_paged = _raise
+
+ def _create_format_table(self, name):
+ schema = Schema.from_pyarrow_schema(
+ self.pa_schema,
+ partition_keys=['dt'],
+ options={"type": "format-table", "file.format": "parquet"})
+ self.rest_catalog.create_table(name, schema, False)
+ table = self.rest_catalog.get_table(name)
+ write_builder = table.new_batch_write_builder()
+ table_write = write_builder.new_write()
+ table_commit = write_builder.new_commit()
+ table_write.write_arrow(self.expected)
+ table_commit.commit(table_write.prepare_commit())
+ table_write.close()
+ table_commit.close()
+ return Identifier.from_string(name)
+
+ def test_falls_back_to_metadata(self):
+ identifier = self._create_partitioned_table('default.parts_fallback')
+ self._make_endpoint_not_implemented()
+
+ result = self.rest_catalog.list_partitions_paged(identifier)
+
+ specs = sorted(tuple(sorted(p.spec.items())) for p in result.elements)
+ self.assertEqual(specs, [(('dt', 'p1'),), (('dt', 'p2'),)])
+ # Stats come from the manifests, not zero placeholders.
+ by_dt = {p.spec['dt']: p for p in result.elements}
+ self.assertEqual(by_dt['p1'].record_count, 4)
+ self.assertEqual(by_dt['p2'].record_count, 4)
+ for p in result.elements:
+ self.assertGreater(p.file_count, 0)
+
+ def test_format_table_still_raises(self):
+ # The manifest-based fallback only supports data tables. Unlike Java
+ # (which scans an unmanaged format table's directory), PyPaimon does
+ # not list format-table partitions here, so the server's
+ # NotImplemented error must surface rather than be silently swallowed.
+ identifier = self._create_format_table('default.parts_fallback_format')
+ self._make_endpoint_not_implemented()
+
+ with self.assertRaises(NotImplementedException):
+ self.rest_catalog.list_partitions_paged(identifier)
+
+ def test_fallback_honors_pattern_and_pagination(self):
+ identifier =
self._create_partitioned_table('default.parts_fallback_paged')
+ self._make_endpoint_not_implemented()
+
+ only_p1 = self.rest_catalog.list_partitions_paged(
+ identifier, partition_name_pattern="dt=p1")
+ self.assertEqual([p.spec for p in only_p1.elements], [{"dt": "p1"}])
+
+ first = self.rest_catalog.list_partitions_paged(identifier,
max_results=1)
+ self.assertEqual(len(first.elements), 1)
+ self.assertIsNotNone(first.next_page_token)
+ second = self.rest_catalog.list_partitions_paged(
+ identifier, max_results=1, page_token=first.next_page_token)
+ self.assertEqual(len(second.elements), 1)
+ self.assertIsNone(second.next_page_token)
+ self.assertNotEqual(first.elements[0].spec, second.elements[0].spec)
