[python] Use canonical partition values when listing partitions

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,22 +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 = {field.name: str(v) for field, v in
-                zip(entry.partition.fields, 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
@@ -115,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'],
@@ -124,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.
@@ -137,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:

paimon-python/pypaimon/tests/filesystem_catalog_test.py

@@ -607,6 +607,112 @@ 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_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))
+                    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
+                    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."""
         catalog = CatalogFactory.create({"warehouse": self.warehouse})