From fbaf395541b0e3c5417cca26a075f40227a1274b Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 02:22:51 -0700 Subject: [PATCH 1/3] [python] Use canonical partition values when listing partitions --- .../pypaimon/catalog/catalog_utils.py | 5 ++- .../pypaimon/tests/filesystem_catalog_test.py | 44 +++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py b/paimon-python/pypaimon/catalog/catalog_utils.py index 8deb937c1cd3..a01e88914cd0 100644 --- a/paimon-python/pypaimon/catalog/catalog_utils.py +++ b/paimon-python/pypaimon/catalog/catalog_utils.py @@ -91,9 +91,10 @@ def list_partitions_from_file_system( entries = manifest_file_manager.read_entries_parallel(manifest_files, drop_stats=True) partition_map = {} # spec_key -> aggregated stats + path_factory = table.path_factory() for entry in entries: - spec = {field.name: str(v) for field, v in - zip(entry.partition.fields, entry.partition.values)} + spec = dict(zip( + table.partition_keys, path_factory._canonical_partition(tuple(entry.partition.values)))) spec_key = tuple(sorted(spec.items())) if spec_key not in partition_map: diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index f30e44a6b003..afff5ca8f6e6 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -607,6 +607,50 @@ def _create_partitioned_table_with_data(self, catalog, identifier, partitions_da table_write.close() table_commit.close() + def test_list_partitions_null_values(self): + from datetime import date + + from pypaimon.common.options.core_options import CoreOptions + + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + catalog.create_database("test_db", False) + cases = [ + (pa.date32(), date(2026, 8, 26), "2026-08-26"), + (pa.string(), "None", "None"), + ] + for default_name in [None, "custom_null"]: + for data_type, value, text in cases: + with self.subTest(default_name=default_name, data_type=data_type): + identifier = "test_db.null_parts_{}_{}".format( + default_name or "default", "date" if pa.types.is_date32(data_type) else "string") + pa_schema = pa.schema([('dt', data_type), ('value', pa.int32())]) + options = {"partition.legacy-name": "false"} + if default_name is not None: + options["partition.default-name"] = default_name + schema = Schema.from_pyarrow_schema( + pa_schema, partition_keys=['dt'], options=options) + catalog.create_table(identifier, schema, False) + table = catalog.get_table(identifier) + builder = table.new_batch_write_builder() + writer = builder.new_write() + committer = builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pydict( + {'dt': [None, value], 'value': [1, 2]}, schema=pa_schema)) + committer.commit(writer.prepare_commit()) + finally: + writer.close() + committer.close() + null_name = default_name or CoreOptions.PARTITION_DEFAULT_NAME.default_value() + partitions = catalog.list_partitions_paged(identifier).elements + self.assertEqual( + {p.spec['dt']: p.record_count for p in partitions}, + {null_name: 1, text: 1}) + filtered = catalog.list_partitions_paged( + identifier, partition_name_pattern='dt=' + null_name) + self.assertEqual([p.spec for p in filtered.elements], [{'dt': null_name}]) + catalog.drop_table(identifier) + def test_list_partitions_paged_pagination(self): """Test list_partitions_paged pagination with max_results and page_token.""" catalog = CatalogFactory.create({"warehouse": self.warehouse}) From 8c4a8fc22524e1dad145e946abdadc1e5fc28425 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 06:52:23 -0700 Subject: [PATCH 2/3] Group listed partitions by serialized rows --- .../pypaimon/catalog/catalog_utils.py | 17 +++--- .../pypaimon/tests/filesystem_catalog_test.py | 53 +++++++++++++++++++ 2 files changed, 62 insertions(+), 8 deletions(-) diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py b/paimon-python/pypaimon/catalog/catalog_utils.py index a01e88914cd0..64c46b80dfca 100644 --- a/paimon-python/pypaimon/catalog/catalog_utils.py +++ b/paimon-python/pypaimon/catalog/catalog_utils.py @@ -79,6 +79,7 @@ def list_partitions_from_file_system( """ from pypaimon.manifest.manifest_file_manager import ManifestFileManager from pypaimon.manifest.manifest_list_manager import ManifestListManager + from pypaimon.table.row.generic_row import GenericRowSerializer snapshot = table.snapshot_manager().get_latest_snapshot() if snapshot is None: @@ -90,23 +91,23 @@ def list_partitions_from_file_system( 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 + partition_map = {} # serialized partition row -> aggregated stats path_factory = table.path_factory() for entry in entries: - spec = dict(zip( - table.partition_keys, path_factory._canonical_partition(tuple(entry.partition.values)))) - spec_key = tuple(sorted(spec.items())) + partition_key = GenericRowSerializer.to_bytes(entry.partition) - if spec_key not in partition_map: - partition_map[spec_key] = { - 'spec': spec, + if partition_key not in partition_map: + partition_map[partition_key] = { + 'spec': dict(zip( + table.partition_keys, + path_factory._canonical_partition(tuple(entry.partition.values)))), 'record_count': 0, 'file_size_in_bytes': 0, 'file_count': 0, 'last_file_creation_time': 0, 'buckets': set(), } - stats = partition_map[spec_key] + stats = partition_map[partition_key] stats['record_count'] += entry.file.row_count stats['file_size_in_bytes'] += entry.file.file_size stats['file_count'] += 1 diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index afff5ca8f6e6..919e81fe0edd 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -651,6 +651,59 @@ def test_list_partitions_null_values(self): self.assertEqual([p.spec for p in filtered.elements], [{'dt': null_name}]) catalog.drop_table(identifier) + def test_list_partitions_preserves_colliding_specs(self): + from pypaimon.api.rest_exception import NotImplementedException + from pypaimon.catalog.rest.rest_catalog import RESTCatalog + from pypaimon.common.options.core_options import CoreOptions + + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + catalog.create_database("test_db", False) + for default_name in [CoreOptions.PARTITION_DEFAULT_NAME.default_value(), "custom_null"]: + with self.subTest(default_name=default_name): + identifier = "test_db.collisions_" + default_name + pa_schema = pa.schema([('dt', pa.string()), ('value', pa.int32())]) + schema = Schema.from_pyarrow_schema( + pa_schema, partition_keys=['dt'], options={ + "partition.default-name": default_name, + "partition.legacy-name": "false"}) + catalog.create_table(identifier, schema, False) + table = catalog.get_table(identifier) + for _ in range(2): + builder = table.new_batch_write_builder() + writer = builder.new_write() + committer = builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pydict({ + 'dt': [None, " ", " ", default_name, default_name, default_name], + 'value': list(range(6))}, schema=pa_schema)) + committer.commit(writer.prepare_commit()) + finally: + writer.close() + committer.close() + + rest_catalog = object.__new__(RESTCatalog) + rest_catalog.rest_api = MagicMock() + rest_catalog.rest_api.list_partitions_paged.side_effect = NotImplementedException( + "partition listing not implemented") + rest_catalog.get_table = MagicMock(return_value=table) + for reader in [catalog, rest_catalog]: + partitions = reader.list_partitions_paged(identifier).elements + self.assertEqual(len(partitions), 3) + self.assertEqual(sorted(p.record_count for p in partitions), [2, 4, 6]) + self.assertTrue(all(p.spec == {'dt': default_name} for p in partitions)) + self.assertTrue(all(p.file_count == 2 for p in partitions)) + counts = [] + token = None + while True: + page = reader.list_partitions_paged( + identifier, max_results=1, page_token=token, + partition_name_pattern='dt=' + default_name) + counts.extend(p.record_count for p in page.elements) + token = page.next_page_token + if token is None: + break + self.assertEqual(sorted(counts), [2, 4, 6]) + def test_list_partitions_paged_pagination(self): """Test list_partitions_paged pagination with max_results and page_token.""" catalog = CatalogFactory.create({"warehouse": self.warehouse}) From 895438beffeef69a2629f099e4c934682170c17b Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 30 Oct 2026 07:14:35 -0700 Subject: [PATCH 3/3] Stabilize pagination for colliding partition names --- .../pypaimon/catalog/catalog_utils.py | 8 +++--- .../pypaimon/tests/filesystem_catalog_test.py | 27 ++++++++++++------- 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py b/paimon-python/pypaimon/catalog/catalog_utils.py index 64c46b80dfca..e670563b1789 100644 --- a/paimon-python/pypaimon/catalog/catalog_utils.py +++ b/paimon-python/pypaimon/catalog/catalog_utils.py @@ -117,6 +117,9 @@ def list_partitions_from_file_system( stats['last_file_creation_time'] = ct stats['buckets'].add(entry.bucket) + # Break display-name ties with the raw partition key for stable pagination. + sorted_partitions = sorted(partition_map.items(), key=lambda item: ( + ','.join(f'{k}={v}' for k, v in sorted(item[1]['spec'].items())), item[0])) partitions = [ Partition( spec=stats['spec'], @@ -126,7 +129,7 @@ def list_partitions_from_file_system( last_file_creation_time=stats['last_file_creation_time'], total_buckets=len(stats['buckets']), ) - for stats in partition_map.values() + for _, stats in sorted_partitions ] # Apply pattern filter with proper regex escaping. @@ -139,9 +142,6 @@ def list_partitions_from_file_system( 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: diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 919e81fe0edd..e72a5f4ef879 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -692,17 +692,26 @@ def test_list_partitions_preserves_colliding_specs(self): self.assertEqual(sorted(p.record_count for p in partitions), [2, 4, 6]) self.assertTrue(all(p.spec == {'dt': default_name} for p in partitions)) self.assertTrue(all(p.file_count == 2 for p in partitions)) + from pypaimon.manifest.manifest_file_manager import ManifestFileManager + from pypaimon.manifest.manifest_list_manager import ManifestListManager + + manifests = ManifestListManager(table).read_all( + table.snapshot_manager().get_latest_snapshot()) + entries = ManifestFileManager(table).read_entries_parallel(manifests, drop_stats=True) + reordered = list(reversed(entries)) counts = [] token = None - while True: - page = reader.list_partitions_paged( - identifier, max_results=1, page_token=token, - partition_name_pattern='dt=' + default_name) - counts.extend(p.record_count for p in page.elements) - token = page.next_page_token - if token is None: - break - self.assertEqual(sorted(counts), [2, 4, 6]) + with patch.object(ManifestFileManager, 'read_entries_parallel', + side_effect=[entries, reordered, reordered]): + while True: + page = reader.list_partitions_paged( + identifier, max_results=1, page_token=token, + partition_name_pattern='dt=' + default_name) + counts.extend(p.record_count for p in page.elements) + token = page.next_page_token + if token is None: + break + self.assertEqual(counts, [p.record_count for p in partitions]) def test_list_partitions_paged_pagination(self): """Test list_partitions_paged pagination with max_results and page_token."""