diff --git a/paimon-python/pypaimon/catalog/catalog_utils.py b/paimon-python/pypaimon/catalog/catalog_utils.py index 8deb937c1cd3..e670563b1789 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,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: diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index f30e44a6b003..154d154ed26c 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -607,6 +607,150 @@ 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_and_drop_nanosecond_partitions(self): + import pandas as pd + + catalog = CatalogFactory.create({"warehouse": self.warehouse}) + catalog.create_database("test_db", False) + for legacy in [True, False]: + with self.subTest(legacy=legacy): + identifier = "test_db.timestamp_{}".format(legacy) + arrow_schema = pa.schema([('dt', pa.timestamp('ns')), ('value', pa.int32())]) + catalog.create_table(identifier, Schema.from_pyarrow_schema( + arrow_schema, partition_keys=['dt'], + options={"partition.legacy-name": str(legacy).lower()}), False) + table = catalog.get_table(identifier) + base = pd.Timestamp('2024-01-02 03:04:05') + values = [base + pd.Timedelta(n, unit='ns') for n in range(3)] + builder = table.new_batch_write_builder() + writer = builder.new_write() + committer = builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pydict({ + 'dt': [values[0], values[1], values[1], values[2], values[2], values[2]], + 'value': list(range(6)), + }, schema=arrow_schema)) + committer.commit(writer.prepare_commit()) + finally: + writer.close() + committer.close() + partitions = catalog.list_partitions_paged(identifier).elements + self.assertEqual(len(partitions), 3) + self.assertEqual(len({p.spec['dt'] for p in partitions}), 3) + target = next(p for p in partitions if p.record_count == 2) + self.assertTrue(target.spec['dt'].endswith('.000000001')) + catalog.drop_partitions(identifier, [target.spec]) + remaining = catalog.list_partitions_paged(identifier).elements + self.assertEqual(sorted(p.record_count for p in remaining), [1, 3]) + self.assertEqual({p.spec['dt'] for p in remaining}, + {p.spec['dt'] for p in partitions if p.record_count != 2}) + + 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}) diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py b/paimon-python/pypaimon/utils/file_store_path_factory.py index b66b54f40643..2641f129a3be 100644 --- a/paimon-python/pypaimon/utils/file_store_path_factory.py +++ b/paimon-python/pypaimon/utils/file_store_path_factory.py @@ -181,7 +181,11 @@ def _canonical_partition(self, partition: Tuple) -> Tuple[str, ...]: text = str((value - date(1970, 1, 1)).days) elif self.legacy_partition_name and type_name.startswith('TIMESTAMP'): text = value.isoformat(timespec='minutes') - if value.second or value.microsecond: + nanos = getattr(value, 'nanosecond', 0) + if nanos: + text = '{}.{:09d}'.format( + value.isoformat(timespec='seconds'), value.microsecond * 1000 + nanos) + elif value.second or value.microsecond: text = value.isoformat(timespec='microseconds' if value.microsecond else 'seconds') if value.microsecond and value.microsecond % 1000 == 0: text = text[:-3] diff --git a/paimon-python/pypaimon/utils/temporal_string.py b/paimon-python/pypaimon/utils/temporal_string.py index c43ccaac8c20..58cba9470515 100644 --- a/paimon-python/pypaimon/utils/temporal_string.py +++ b/paimon-python/pypaimon/utils/temporal_string.py @@ -20,8 +20,10 @@ def format_timestamp(value, precision, nanos=None): """Keep significant fraction digits, with declared precision as a minimum.""" + if nanos is None: + nanos = value.microsecond * 1000 + getattr(value, 'nanosecond', 0) return format_timestamp_parts(value.year, value.month, value.day, value.hour, value.minute, - value.second, value.microsecond * 1000 if nanos is None else nanos, precision) + value.second, nanos, precision) def format_timestamp_parts(year, month, day, hour, minute, second, nanos, precision):