From a2650ceb1099d9e83a2590642301800abca49688 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 15:58:12 +0800 Subject: [PATCH 1/4] [python] Read filesystem partition metadata with Native --- .../table/system/native_system_table.py | 4 +- .../pypaimon/table/system/partitions_table.py | 55 +++--- .../pypaimon/table/system/system_table.py | 7 + .../tests/native_partition_metadata_test.py | 173 ++++++++++++++++++ .../tests/native_system_table_test.py | 5 +- .../tests/system/native_system_table_test.py | 4 +- .../tests/system/partitions_table_test.py | 25 ++- 7 files changed, 241 insertions(+), 32 deletions(-) create mode 100644 paimon-python/pypaimon/tests/native_partition_metadata_test.py diff --git a/paimon-python/pypaimon/table/system/native_system_table.py b/paimon-python/pypaimon/table/system/native_system_table.py index 0123576b69ec..834dff23fe9d 100644 --- a/paimon-python/pypaimon/table/system/native_system_table.py +++ b/paimon-python/pypaimon/table/system/native_system_table.py @@ -23,11 +23,9 @@ from pypaimon.read.native_plan import _native_table, native_method_available from pypaimon.write.native_commit import _rest_catalog_supported -# Partitions need the catalog's list-partitions operation, which is not carried -# by a resolved Native Table. Keep that table on the existing catalog path. _NATIVE_TABLES = frozenset(( 'snapshots', 'schemas', 'options', 'manifests', 'files', 'buckets', - 'tags', 'branches', 'file_key_ranges', 'table_indexes', 'consumers', + 'tags', 'branches', 'file_key_ranges', 'table_indexes', 'consumers', 'partitions', )) diff --git a/paimon-python/pypaimon/table/system/partitions_table.py b/paimon-python/pypaimon/table/system/partitions_table.py index 70685fee0bad..70f7db7831dd 100644 --- a/paimon-python/pypaimon/table/system/partitions_table.py +++ b/paimon-python/pypaimon/table/system/partitions_table.py @@ -21,12 +21,17 @@ import pyarrow +from pypaimon.common.options.core_options import CoreOptions from pypaimon.index.index_file_handler import IndexFileHandler from pypaimon.manifest.index_manifest_file import IndexManifestFile from pypaimon.manifest.manifest_file_manager import ManifestFileManager from pypaimon.manifest.manifest_list_manager import ManifestListManager from pypaimon.schema.data_types import AtomicType, DataField, RowType -from pypaimon.table.system.system_table import SystemTable +from pypaimon.table.system.files_table import _render_value +from pypaimon.table.system.system_table import ( + SystemTable, metadata_snapshot, metadata_local_millis, metadata_epoch_millis) +from pypaimon.table.row.generic_row import GenericRowSerializer +from pypaimon.utils.file_store_path_factory import _is_null_or_whitespace_only TABLE_TYPE = RowType(False, [ @@ -67,7 +72,7 @@ def primary_keys(self) -> List[str]: return ["partition"] def _build_arrow_table(self) -> pyarrow.Table: - snapshot = self.base_table.snapshot_manager().get_latest_snapshot() + snapshot = metadata_snapshot(self.base_table) if snapshot is None: return self._empty_table() @@ -80,23 +85,21 @@ def _build_arrow_table(self) -> pyarrow.Table: partition_map: dict = {} for entry in entries: - partition_key = tuple(entry.partition.values) - spec_items = tuple( - (field.name, str(value)) - for field, value in zip( - entry.partition.fields, entry.partition.values)) - spec_key = spec_items + partition_key = bytes(GenericRowSerializer.to_bytes(entry.partition)) + spec_key = partition_key + partition_string = _render_partition( + entry.partition, self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME)) stats = partition_map.get(spec_key) if stats is None: stats = { - "spec_items": spec_items, + "partition_string": partition_string, "partition_key": partition_key, "record_count": 0, "file_size_in_bytes": 0, "file_count": 0, "last_update_time": None, - "buckets": set(), + "total_buckets": entry.total_buckets, } partition_map[spec_key] = stats @@ -104,11 +107,11 @@ def _build_arrow_table(self) -> pyarrow.Table: stats["file_size_in_bytes"] += int(entry.file.file_size) stats["file_count"] += 1 if entry.file.creation_time is not None: - ct_ms = entry.file.creation_time.get_millisecond() + ct_ms = metadata_epoch_millis(entry.file.creation_time.get_millisecond()) if (stats["last_update_time"] is None or ct_ms > stats["last_update_time"]): stats["last_update_time"] = ct_ms - stats["buckets"].add(int(entry.bucket)) + stats["total_buckets"] = entry.total_buckets partition_strings: List[Optional[str]] = [] record_counts: List[int] = [] @@ -118,13 +121,14 @@ def _build_arrow_table(self) -> pyarrow.Table: total_buckets: List[int] = [] deleted_counts: List[Optional[int]] = [] - for stats in partition_map.values(): - partition_strings.append(_render_partition(stats["spec_items"])) + for stats in sorted(partition_map.values(), key=lambda stats: stats["partition_string"]): + partition_strings.append(stats["partition_string"]) record_counts.append(stats["record_count"]) file_sizes.append(stats["file_size_in_bytes"]) file_counts.append(stats["file_count"]) - last_update_times.append(stats["last_update_time"]) - total_buckets.append(len(stats["buckets"])) + last_update_times.append(None if stats["last_update_time"] is None + else metadata_local_millis(stats["last_update_time"])) + total_buckets.append(stats["total_buckets"]) deleted_counts.append( deleted_record_counts.get(stats["partition_key"], 0)) @@ -155,12 +159,12 @@ def _deleted_record_counts(self, snapshot) -> dict: entries = IndexFileHandler(self.base_table).scan( snapshot, - lambda entry: entry.index_file.index_type + lambda entry: entry.kind == 0 and entry.index_file.index_type == IndexManifestFile.DELETION_VECTORS_INDEX, ) result = {} for entry in entries: - partition_key = tuple(entry.partition.values) + partition_key = bytes(GenericRowSerializer.to_bytes(entry.partition)) if partition_key in result and result[partition_key] is None: continue @@ -197,8 +201,13 @@ def _empty_table() -> pyarrow.Table: }) -def _render_partition(spec_items) -> Optional[str]: - """Render a partition spec as ``pt=v/pt2=v2`` or None when empty.""" - if not spec_items: - return None - return "/".join("{}={}".format(name, value) for name, value in spec_items) +def _render_partition(partition, default_name) -> str: + """Java PartitionsTable uses typed SQL casts and an unescaped key=value spec.""" + parts = [] + for index, field in enumerate(partition.fields): + value = partition.get_field(index) + text = default_name if value is None else _render_value(value, field.type) + if _is_null_or_whitespace_only(text): + text = default_name + parts.append("{}={}".format(field.name, text)) + return "/".join(parts) diff --git a/paimon-python/pypaimon/table/system/system_table.py b/paimon-python/pypaimon/table/system/system_table.py index a4f2c11c8216..989435c04a45 100644 --- a/paimon-python/pypaimon/table/system/system_table.py +++ b/paimon-python/pypaimon/table/system/system_table.py @@ -47,6 +47,13 @@ def metadata_local_millis(millis): return _signed(int(millis) + _local_offset(int(millis) // 1000) * 1000, 64) +def metadata_epoch_millis(millis): + """Java DataFileMeta.creationTimeEpochMillis for a stored local TIMESTAMP.""" + from pypaimon.casting.temporal_utils import _local_to_utc_seconds, _signed + seconds, fraction = divmod(int(millis), 1000) + return _signed(_local_to_utc_seconds(seconds) * 1000 + fraction, 64) + + def metadata_snapshot(table): """Use the table's captured view, including an explicitly empty snapshot.""" if '_read_snapshot' in vars(table): diff --git a/paimon-python/pypaimon/tests/native_partition_metadata_test.py b/paimon-python/pypaimon/tests/native_partition_metadata_test.py new file mode 100644 index 000000000000..3cb050b02843 --- /dev/null +++ b/paimon-python/pypaimon/tests/native_partition_metadata_test.py @@ -0,0 +1,173 @@ +# 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. + +"""Native partition metadata uses the same manifest view and Java casts as Python.""" + +from datetime import date, datetime, time +from decimal import Decimal +import json +from unittest.mock import patch + +import pyarrow as pa +import pytest + +from pypaimon import Schema +from pypaimon.read.native_plan import native_method_available +from pypaimon.index.deletion_vector_meta import DeletionVectorMeta +from pypaimon.index.index_file_meta import IndexFileMeta +from pypaimon.manifest.index_manifest_entry import IndexManifestEntry +from pypaimon.manifest.index_manifest_file import IndexManifestFile +from pypaimon.table.row.generic_row import GenericRow +from pypaimon.tests import native_plan_rest_test +from pypaimon.tests.native_system_table_test import _canonical, _read, _system + + +rest_catalog = native_plan_rest_test.rest_catalog + +pytestmark = [pytest.mark.native_plan, pytest.mark.skipif( + not native_method_available('Table', 'new_system_read'), reason='Native system-table reader required')] + + +def _create(catalog, schema, partition_keys, options=None): + settings = {'write.native.enabled': 'false', 'commit.native.enabled': 'false'} + settings.update(options or {}) + catalog.create_table('default.part_metadata', Schema.from_pyarrow_schema( + schema, partition_keys=partition_keys, options=settings), False) + return catalog.get_table('default.part_metadata') + + +def _append(table, rows, schema): + builder = table.new_batch_write_builder() + writer, commit = builder.new_write(), builder.new_commit() + try: + writer.write_arrow(pa.Table.from_pylist(rows, schema=schema)) + commit.commit(writer.prepare_commit()) + finally: + writer.close() + commit.close() + + +def _assert_parity(table): + expected = _read(_system(table, 'partitions', False)) + native = _system(table, 'partitions', True) + with patch.object(type(native), '_build_arrow_table', + side_effect=AssertionError('Python partition aggregation ran')): + actual = _read(native) + assert actual.schema == expected.schema + assert _canonical(actual, 'partitions') == _canonical(expected, 'partitions') + return actual.to_pylist() + + +@pytest.mark.parametrize('typ,value,expected', [ + (pa.bool_(), True, 'true'), + (pa.decimal128(10, 2), Decimal('1.20'), '1.20'), + (pa.date32(), date(2024, 2, 29), '2024-02-29'), + (pa.time32('ms'), time(12, 34, 56, 123000), '12:34:56'), + (pa.timestamp('us'), datetime(2024, 2, 29, 12, 34, 56, 123456), '2024-02-29 12:34:56.123456'), + (pa.string(), 'a/b=c', 'a/b=c'), + (pa.float32(), 1.401298464324817e-45, '1.4E-45'), + (pa.float64(), 5e-324, '4.9E-324'), + (pa.binary(), b'hello', 'hello'), + (pa.binary(), b'\xed\xa0\x80', '\ufffd'), + (pa.string(), '\u00a0', '\u00a0'), + (pa.string(), '\u0085', '\u0085'), + (pa.string(), '\u001c', 'MISSING'), +]) +def test_partition_values_are_java_casts(rest_catalog, typ, value, expected): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', typ)]) + table = _create(catalog, schema, ['p'], {'partition.default-name': 'MISSING'}) + _append(table, [{'id': 1, 'p': value}, {'id': 2, 'p': None}], schema) + rows = _assert_parity(table) + assert sorted(row['partition'] for row in rows) == sorted(['p=' + expected, 'p=MISSING']) + assert [row['total_buckets'] for row in rows] == [-1, -1] + assert [row['deleted_record_count'] for row in rows] == [0, 0] + + +def test_unpartitioned_metadata_uses_empty_string(rest_catalog): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32())]) + table = _create(catalog, schema, []) + _append(table, [{'id': 1}, {'id': 2}], schema) + rows = _assert_parity(table) + assert len(rows) == 1 + assert rows[0]['partition'] == '' + assert rows[0]['record_count'] == 2 + + +def test_declared_bucket_count_does_not_count_only_occupied_buckets(rest_catalog): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.string())]) + table = _create(catalog, schema, ['p'], {'bucket': '8', 'bucket-key': 'id'}) + _append(table, [{'id': 1, 'p': 'one'}], schema) + rows = _assert_parity(table) + assert rows[0]['file_count'] == 1 + assert rows[0]['total_buckets'] == 8 + + +def test_rendered_partition_collisions_are_not_aggregated(rest_catalog): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.string()), ('p2', pa.string())]) + table = _create(catalog, schema, ['p', 'p2']) + _append(table, [{'id': 1, 'p': 'a/p2=b', 'p2': 'c'}, + {'id': 2, 'p': 'a', 'p2': 'b/p2=c'}], schema) + rows = _assert_parity(table) + assert [row['partition'] for row in rows] == ['p=a/p2=b/p2=c'] * 2 + assert [row['record_count'] for row in rows] == [1, 1] + + +def test_deletion_vector_counts_read_avro_metadata_and_keep_unknown(rest_catalog): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.string())]) + table = _create(catalog, schema, ['p'], {'deletion-vectors.enabled': 'true'}) + _append(table, [{'id': 1, 'p': 'known'}, {'id': 2, 'p': 'unknown'}, + {'id': 3, 'p': 'zero'}], schema) + entries = [] + for partition, bucket, counts in [ + ('known', 0, [1, 2]), ('known', 1, [3]), + ('unknown', 0, [None]), ('unknown', 1, [5])]: + ranges = {str(i): DeletionVectorMeta(str(i), 0, 1, count) for i, count in enumerate(counts)} + entries.append(IndexManifestEntry( + 0, GenericRow([partition], table.partition_keys_fields), bucket, + IndexFileMeta('DELETION_VECTORS', partition + str(bucket), 1, 1, dv_ranges=ranges))) + # Only manifest metadata is needed; no deletion-vector payload is opened. + name = IndexManifestFile(table).write(entries) + snapshot = table.snapshot_manager().get_latest_snapshot() + path = table.snapshot_manager().snapshot_dir + '/snapshot-' + str(snapshot.id) + with table.file_io.new_input_stream(path) as stream: + data = json.loads(stream.read()) + data['indexManifest'] = name + table.file_io.overwrite_file_utf8(path, json.dumps(data)) + table = table._copy_with_snapshot(table.snapshot_manager().get_latest_snapshot()) + rows = _assert_parity(table) + assert {row['partition']: row['deleted_record_count'] for row in rows} == { + 'p=known': 6, 'p=unknown': None, 'p=zero': 0} + + +@pytest.mark.parametrize('native', [False, True]) +def test_partition_split_is_captured_before_later_commits(rest_catalog, native): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.string())]) + table = _create(catalog, schema, ['p']) + _append(table, [{'id': 1, 'p': 'old'}], schema) + system = _system(table, 'partitions', native) + builder = system.new_read_builder() + splits = builder.new_scan().plan().splits() + _append(table, [{'id': 2, 'p': 'new'}], schema) + rows = builder.new_read().to_arrow(splits).to_pylist() + assert [row['partition'] for row in rows] == ['p=old'] + assert len(_assert_parity(table)) == 2 diff --git a/paimon-python/pypaimon/tests/native_system_table_test.py b/paimon-python/pypaimon/tests/native_system_table_test.py index 7af10c298db9..7d501ceae94e 100644 --- a/paimon-python/pypaimon/tests/native_system_table_test.py +++ b/paimon-python/pypaimon/tests/native_system_table_test.py @@ -125,7 +125,7 @@ def test_projection_filter_limit_and_pickled_split(source): assert builder.new_read().to_arrow(splits).to_pylist() == [{'snapshot_id': 1}, {'snapshot_id': 2}] -@pytest.mark.parametrize('name', ['files', 'file_key_ranges', 'buckets', 'manifests', 'table_indexes']) +@pytest.mark.parametrize('name', ['files', 'file_key_ranges', 'buckets', 'manifests', 'table_indexes', 'partitions']) @pytest.mark.parametrize('view', ['first', 'empty', 'tag']) def test_captured_metadata_view(source, name, view): catalog, table, _ = source @@ -191,8 +191,7 @@ def test_native_reader_preserves_schema_on_empty_and_rejects_catalog_tables(sour assert result.schema.names[16] == 'deleteRowCount' assert result.num_rows == 0 assert [field.name() for field in read.read_type()] == result.schema.names - with pytest.raises(Exception, match='requires a catalog|unknown'): - native.new_system_read('partitions') + assert native.new_system_read('partitions').read().num_rows == 0 with pytest.raises(Exception, match='requires a catalog|unknown'): native.new_system_read('unknown') diff --git a/paimon-python/pypaimon/tests/system/native_system_table_test.py b/paimon-python/pypaimon/tests/system/native_system_table_test.py index 4383b9ae761f..60dc35eedf3b 100644 --- a/paimon-python/pypaimon/tests/system/native_system_table_test.py +++ b/paimon-python/pypaimon/tests/system/native_system_table_test.py @@ -34,9 +34,9 @@ def _table(name='files', enabled=True): row_type=lambda: RowType(False, [DataField(0, 'id', AtomicType('INT', nullable=False))])) -@pytest.mark.parametrize('reason', ['disabled', 'old-binding', 'custom-io', 'catalog-required']) +@pytest.mark.parametrize('reason', ['disabled', 'old-binding', 'custom-io', 'unknown-table']) def test_unsupported_metadata_reads_do_not_create_native_tables(reason): - table = _table('partitions' if reason == 'catalog-required' else 'files', enabled=reason != 'disabled') + table = _table('unknown' if reason == 'unknown-table' else 'files', enabled=reason != 'disabled') with patch('pypaimon.table.system.native_system_table.native_method_available', return_value=reason != 'old-binding'): with patch('pypaimon.table.system.native_system_table._rest_catalog_supported', diff --git a/paimon-python/pypaimon/tests/system/partitions_table_test.py b/paimon-python/pypaimon/tests/system/partitions_table_test.py index 108cc6d35879..ac3177fe2290 100644 --- a/paimon-python/pypaimon/tests/system/partitions_table_test.py +++ b/paimon-python/pypaimon/tests/system/partitions_table_test.py @@ -124,7 +124,7 @@ def test_aggregates_partitions_after_commit(self): for count in arrow_table.column("file_count").to_pylist(): self.assertGreaterEqual(count, 1) for total_buckets in arrow_table.column("total_buckets").to_pylist(): - self.assertGreaterEqual(total_buckets, 1) + self.assertEqual(total_buckets, -1) # FileSystem catalog does not maintain a "done" flag. for done in arrow_table.column("done").to_pylist(): self.assertFalse(done) @@ -192,6 +192,29 @@ def test_aggregates_deleted_record_count(self): self.assertEqual(3, deleted_counts["dt=2024-01-01"]) self.assertIsNone(deleted_counts["dt=2024-01-02"]) + def test_creation_times_use_java_gap_and_overlap_resolution(self): + import os + import time + from pypaimon.table.system.system_table import metadata_epoch_millis, metadata_local_millis + original = os.environ.get('TZ') + try: + os.environ['TZ'] = 'America/New_York' + time.tzset() + # 02:55 lies in the spring gap: Java shifts to 03:55 EDT, 07:55 UTC. + gap_wall = 1710039300123 + self.assertEqual(1710057300123, metadata_epoch_millis(gap_wall)) + self.assertEqual(1710042900123, metadata_local_millis(metadata_epoch_millis(gap_wall))) + # 01:30 occurs twice in the fall overlap. Java chooses 05:30 UTC (EDT). + overlap_wall = 1730597400456 + self.assertEqual(1730611800456, metadata_epoch_millis(overlap_wall)) + self.assertEqual(overlap_wall, metadata_local_millis(metadata_epoch_millis(overlap_wall))) + finally: + if original is None: + os.environ.pop('TZ', None) + else: + os.environ['TZ'] = original + time.tzset() + if __name__ == "__main__": unittest.main() From 4ed565248aafa2f9ac741bc7c64e64a940b98586 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 16:09:16 +0800 Subject: [PATCH 2/4] [python] Align partition deletion metadata lookups with Java --- .../pypaimon/table/system/partitions_table.py | 25 ++++++++++++- .../tests/native_partition_metadata_test.py | 37 +++++++++++++++++-- 2 files changed, 57 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/table/system/partitions_table.py b/paimon-python/pypaimon/table/system/partitions_table.py index 70f7db7831dd..578b79e03b4c 100644 --- a/paimon-python/pypaimon/table/system/partitions_table.py +++ b/paimon-python/pypaimon/table/system/partitions_table.py @@ -94,7 +94,8 @@ def _build_arrow_table(self) -> pyarrow.Table: if stats is None: stats = { "partition_string": partition_string, - "partition_key": partition_key, + "deletion_key": _deletion_partition_key( + entry.partition, self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME)), "record_count": 0, "file_size_in_bytes": 0, "file_count": 0, @@ -130,7 +131,7 @@ def _build_arrow_table(self) -> pyarrow.Table: else metadata_local_millis(stats["last_update_time"])) total_buckets.append(stats["total_buckets"]) deleted_counts.append( - deleted_record_counts.get(stats["partition_key"], 0)) + deleted_record_counts.get(stats["deletion_key"], 0)) n = len(partition_map) return pyarrow.table({ @@ -211,3 +212,23 @@ def _render_partition(partition, default_name) -> str: text = default_name parts.append("{}={}".format(field.name, text)) return "/".join(parts) + + +def _deletion_partition_key(partition, default_name): + """Java looks up deletion metadata after converting the spec back to typed values.""" + from pypaimon.table.row.generic_row import GenericRow, _parse_type_precision_scale + values = [] + for index, field in enumerate(partition.fields): + value = partition.get_field(index) + if value is not None: + text = _render_value(value, field.type) + if text == default_name or _is_null_or_whitespace_only(text): + value = None + elif isinstance(value, (bytes, bytearray, memoryview)): + value = text.encode('utf-8') + elif str(field.type).upper().startswith('TIME') and not str(field.type).upper().startswith('TIMESTAMP'): + precision, _ = _parse_type_precision_scale(field.type) + quantum = 10 ** (6 - min(precision, 3)) + value = value.replace(microsecond=value.microsecond // quantum * quantum) + values.append(value) + return bytes(GenericRowSerializer.to_bytes(GenericRow(values, partition.fields))) diff --git a/paimon-python/pypaimon/tests/native_partition_metadata_test.py b/paimon-python/pypaimon/tests/native_partition_metadata_test.py index 3cb050b02843..ec1ab17b2fae 100644 --- a/paimon-python/pypaimon/tests/native_partition_metadata_test.py +++ b/paimon-python/pypaimon/tests/native_partition_metadata_test.py @@ -145,6 +145,13 @@ def test_deletion_vector_counts_read_avro_metadata_and_keep_unknown(rest_catalog 0, GenericRow([partition], table.partition_keys_fields), bucket, IndexFileMeta('DELETION_VECTORS', partition + str(bucket), 1, 1, dv_ranges=ranges))) # Only manifest metadata is needed; no deletion-vector payload is opened. + table = _with_index_entries(table, entries) + rows = _assert_parity(table) + assert {row['partition']: row['deleted_record_count'] for row in rows} == { + 'p=known': 6, 'p=unknown': None, 'p=zero': 0} + + +def _with_index_entries(table, entries): name = IndexManifestFile(table).write(entries) snapshot = table.snapshot_manager().get_latest_snapshot() path = table.snapshot_manager().snapshot_dir + '/snapshot-' + str(snapshot.id) @@ -152,10 +159,34 @@ def test_deletion_vector_counts_read_avro_metadata_and_keep_unknown(rest_catalog data = json.loads(stream.read()) data['indexManifest'] = name table.file_io.overwrite_file_utf8(path, json.dumps(data)) - table = table._copy_with_snapshot(table.snapshot_manager().get_latest_snapshot()) + return table._copy_with_snapshot(table.snapshot_manager().get_latest_snapshot()) + + +@pytest.mark.parametrize('typ,value,canonical', [ + (pa.string(), 'MISSING', None), + (pa.string(), '\x1c', None), + (pa.binary(), b'\xed\xa0\x80', '\ufffd'.encode('utf-8')), + (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56)), +]) +def test_deletion_lookup_uses_java_partition_spec_round_trip(rest_catalog, typ, value, canonical): + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', typ)]) + table = _create(catalog, schema, ['p'], { + 'deletion-vectors.enabled': 'true', 'partition.default-name': 'MISSING'}) + _append(table, [{'id': i, 'p': value} for i in range(3)] + + [{'id': i + 3, 'p': canonical} for i in range(3)], schema) + entries = [] + for partition, count in [(value, 1), (canonical, 3)]: + ranges = {'file': DeletionVectorMeta('file', 0, 1, count)} + entries.append(IndexManifestEntry( + 0, GenericRow([partition], table.partition_keys_fields), 0, + IndexFileMeta('DELETION_VECTORS', str(count), 1, 1, dv_ranges=ranges))) + table = _with_index_entries(table, entries) rows = _assert_parity(table) - assert {row['partition']: row['deleted_record_count'] for row in rows} == { - 'p=known': 6, 'p=unknown': None, 'p=zero': 0} + assert len(rows) == 2 + # Java reconstructs NULL/default and UTF-8 replacement bytes from the spec, + # so both displayed rows look up the canonical partition's count. + assert [row['deleted_record_count'] for row in rows] == [3, 3] @pytest.mark.parametrize('native', [False, True]) From a99230ce302f571d99e327f49432debe1de044cb Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 18:23:33 +0800 Subject: [PATCH 3/4] [ci] Retry Maven downloads with an Apache archive fallback --- .github/workflows/ci-python.yml | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index c34c60d7457e..75a072bd3358 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -62,11 +62,6 @@ jobs: restore-keys: | ci-maven-python-${{ runner.os }}-${{ runner.arch }}- - - name: Set up Maven - uses: stCarolas/setup-maven@v5 - with: - maven-version: 3.8.8 - - name: Install system dependencies shell: bash run: | @@ -86,6 +81,22 @@ jobs: && apt-get clean \ && rm -rf /var/lib/apt/lists/* + - name: Set up Maven + shell: bash + env: + MAVEN_VERSION: '3.8.8' + run: | + maven_archive="${RUNNER_TEMP}/apache-maven-${MAVEN_VERSION}-bin.tar.gz" + # Central can return transient 404s even for an existing release. + curl --fail --location --retry 3 --retry-all-errors --retry-delay 2 \ + --output "$maven_archive" \ + "https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/${MAVEN_VERSION}/apache-maven-${MAVEN_VERSION}-bin.tar.gz" \ + || curl --fail --location --retry 3 --retry-all-errors --retry-delay 2 \ + --output "$maven_archive" \ + "https://archive.apache.org/dist/maven/maven-3/${MAVEN_VERSION}/binaries/apache-maven-${MAVEN_VERSION}-bin.tar.gz" + tar -xzf "$maven_archive" -C "$RUNNER_TEMP" + echo "${RUNNER_TEMP}/apache-maven-${MAVEN_VERSION}/bin" >> "$GITHUB_PATH" + - name: Verify Java and Maven installation run: | java -version From 76be78797939dc76a07b5e4ac4552919c9bdeef7 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Fri, 16 Oct 2026 19:48:15 +0800 Subject: [PATCH 4/4] [python] Align partition metadata normalization with Java --- .github/workflows/ci-python.yml | 1 + .../pypaimon/table/system/partitions_table.py | 40 +++++++---- .../tests/native_partition_metadata_test.py | 66 +++++++++++++++---- 3 files changed, 83 insertions(+), 24 deletions(-) diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml index 75a072bd3358..b6a0fb9f81dc 100644 --- a/.github/workflows/ci-python.yml +++ b/.github/workflows/ci-python.yml @@ -259,6 +259,7 @@ jobs: paimon-python/pypaimon/tests/native_cast_completion_test.py \ paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py \ paimon-python/pypaimon/tests/native_schema_cast_test.py \ + paimon-python/pypaimon/tests/native_partition_metadata_test.py \ -k 'TIMESTAMP or temporal' done diff --git a/paimon-python/pypaimon/table/system/partitions_table.py b/paimon-python/pypaimon/table/system/partitions_table.py index 578b79e03b4c..af0b805b07cf 100644 --- a/paimon-python/pypaimon/table/system/partitions_table.py +++ b/paimon-python/pypaimon/table/system/partitions_table.py @@ -27,7 +27,7 @@ from pypaimon.manifest.manifest_file_manager import ManifestFileManager from pypaimon.manifest.manifest_list_manager import ManifestListManager from pypaimon.schema.data_types import AtomicType, DataField, RowType -from pypaimon.table.system.files_table import _render_value +from pypaimon.table.system.files_table import _render_object, _render_value from pypaimon.table.system.system_table import ( SystemTable, metadata_snapshot, metadata_local_millis, metadata_epoch_millis) from pypaimon.table.row.generic_row import GenericRowSerializer @@ -84,18 +84,17 @@ def _build_arrow_table(self) -> pyarrow.Table: deleted_record_counts = self._deleted_record_counts(snapshot) partition_map: dict = {} + default_name = self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME) + legacy = self.base_table.options.options.get(CoreOptions.PARTITION_GENERATE_LEGACY_NAME) for entry in entries: partition_key = bytes(GenericRowSerializer.to_bytes(entry.partition)) spec_key = partition_key - partition_string = _render_partition( - entry.partition, self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME)) - stats = partition_map.get(spec_key) if stats is None: + partition = _normalize_partition(entry.partition, default_name, legacy) stats = { - "partition_string": partition_string, - "deletion_key": _deletion_partition_key( - entry.partition, self.base_table.options.options.get(CoreOptions.PARTITION_DEFAULT_NAME)), + "partition_string": _render_partition(partition, default_name), + "deletion_key": bytes(GenericRowSerializer.to_bytes(partition)), "record_count": 0, "file_size_in_bytes": 0, "file_count": 0, @@ -214,21 +213,36 @@ def _render_partition(partition, default_name) -> str: return "/".join(parts) -def _deletion_partition_key(partition, default_name): - """Java looks up deletion metadata after converting the spec back to typed values.""" - from pypaimon.table.row.generic_row import GenericRow, _parse_type_precision_scale +def _normalize_partition(partition, default_name, legacy): + """Java reconstructs one typed spec row for both display and deletion lookup.""" + from pypaimon.table.row.generic_row import ( + GenericRow, _parse_type_precision_scale, _is_ltz_type, _normalize_ltz, + _datetime_to_millis_and_nanos, _millis_nanos_to_datetime) values = [] for index, field in enumerate(partition.fields): value = partition.get_field(index) if value is not None: - text = _render_value(value, field.type) + text = _render_object(value, field.type) if legacy else _render_value(value, field.type) if text == default_name or _is_null_or_whitespace_only(text): value = None elif isinstance(value, (bytes, bytearray, memoryview)): value = text.encode('utf-8') - elif str(field.type).upper().startswith('TIME') and not str(field.type).upper().startswith('TIMESTAMP'): + elif str(field.type).upper().startswith('TIMESTAMP'): + precision, _ = _parse_type_precision_scale(field.type) + millis, nanos = _datetime_to_millis_and_nanos(_normalize_ltz(value)) + if _is_ltz_type(str(field.type).upper()): + wall_millis = millis if legacy else metadata_local_millis(millis) + millis = metadata_epoch_millis(wall_millis) + if precision <= 3: + quantum = 10 ** (3 - precision) + millis, nanos = millis // quantum * quantum, 0 + else: + quantum = 10 ** (9 - precision) + nanos = nanos // quantum * quantum + value = _millis_nanos_to_datetime(millis, nanos) + elif not legacy and str(field.type).upper().startswith('TIME'): precision, _ = _parse_type_precision_scale(field.type) quantum = 10 ** (6 - min(precision, 3)) value = value.replace(microsecond=value.microsecond // quantum * quantum) values.append(value) - return bytes(GenericRowSerializer.to_bytes(GenericRow(values, partition.fields))) + return GenericRow(values, partition.fields) diff --git a/paimon-python/pypaimon/tests/native_partition_metadata_test.py b/paimon-python/pypaimon/tests/native_partition_metadata_test.py index ec1ab17b2fae..2ca0a8d1bd71 100644 --- a/paimon-python/pypaimon/tests/native_partition_metadata_test.py +++ b/paimon-python/pypaimon/tests/native_partition_metadata_test.py @@ -20,6 +20,7 @@ from datetime import date, datetime, time from decimal import Decimal import json +import os from unittest.mock import patch import pyarrow as pa @@ -90,7 +91,8 @@ def _assert_parity(table): def test_partition_values_are_java_casts(rest_catalog, typ, value, expected): catalog, _ = rest_catalog schema = pa.schema([('id', pa.int32()), ('p', typ)]) - table = _create(catalog, schema, ['p'], {'partition.default-name': 'MISSING'}) + table = _create(catalog, schema, ['p'], { + 'partition.default-name': 'MISSING', 'partition.legacy-name': 'false'}) _append(table, [{'id': 1, 'p': value}, {'id': 2, 'p': None}], schema) rows = _assert_parity(table) assert sorted(row['partition'] for row in rows) == sorted(['p=' + expected, 'p=MISSING']) @@ -162,17 +164,22 @@ def _with_index_entries(table, entries): return table._copy_with_snapshot(table.snapshot_manager().get_latest_snapshot()) -@pytest.mark.parametrize('typ,value,canonical', [ - (pa.string(), 'MISSING', None), - (pa.string(), '\x1c', None), - (pa.binary(), b'\xed\xa0\x80', '\ufffd'.encode('utf-8')), - (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56)), +@pytest.mark.parametrize('typ,value,canonical,legacy', [ + (pa.string(), 'MISSING', None, True), + (pa.string(), 'MISSING', None, False), + (pa.string(), '\x1c', None, True), + (pa.string(), '\x1c', None, False), + # Java byte[].toString has no stable legacy spec; exercise its SQL cast. + (pa.binary(), b'\xed\xa0\x80', '\ufffd'.encode('utf-8'), False), + (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56), True), + (pa.time32('ms'), time(12, 34, 56, 123000), time(12, 34, 56), False), ]) -def test_deletion_lookup_uses_java_partition_spec_round_trip(rest_catalog, typ, value, canonical): +def test_deletion_lookup_uses_java_partition_spec_round_trip(rest_catalog, typ, value, canonical, legacy): catalog, _ = rest_catalog schema = pa.schema([('id', pa.int32()), ('p', typ)]) table = _create(catalog, schema, ['p'], { - 'deletion-vectors.enabled': 'true', 'partition.default-name': 'MISSING'}) + 'deletion-vectors.enabled': 'true', 'partition.default-name': 'MISSING', + 'partition.legacy-name': str(legacy).lower()}) _append(table, [{'id': i, 'p': value} for i in range(3)] + [{'id': i + 3, 'p': canonical} for i in range(3)], schema) entries = [] @@ -184,9 +191,46 @@ def test_deletion_lookup_uses_java_partition_spec_round_trip(rest_catalog, typ, table = _with_index_entries(table, entries) rows = _assert_parity(table) assert len(rows) == 2 - # Java reconstructs NULL/default and UTF-8 replacement bytes from the spec, - # so both displayed rows look up the canonical partition's count. - assert [row['deleted_record_count'] for row in rows] == [3, 3] + # A legacy TIME spec is an integer millisecond count, so Java keeps its + # sub-second digits even when the declared TIME precision is zero. + expected = [1, 3] if legacy and pa.types.is_time(typ) else [3, 3] + assert sorted(row['deleted_record_count'] for row in rows) == expected + + +def test_default_legacy_time_partition_keeps_deletion_counts(rest_catalog): + from pypaimon.tests.deletion_vector_path_test import _delete + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.time32('ms'))]) + table = _create(catalog, schema, ['p'], { + 'deletion-vectors.enabled': 'true', 'row-tracking.enabled': 'true', + 'data-evolution.enabled': 'true'}) + _append(table, [{'id': i, 'p': time(12, 34, 56, 120000)} for i in range(2)], schema) + _delete(table, [0]) + rows = _assert_parity(table) + assert [(row['partition'], row['deleted_record_count']) for row in rows] == [('p=12:34:56', 1)] + + +@pytest.mark.parametrize('legacy,instant,expected,deleted', [ + (True, '2024-01-01T00:00:00+00:00', '2024-01-01 00:00:00.000', 0), + (False, '2024-01-01T00:00:00+00:00', '2023-12-31 19:00:00.000', 1), + (True, '2024-11-03T06:30:00+00:00', '2024-11-03 06:30:00.000', 0), + (False, '2024-11-03T06:30:00+00:00', '2024-11-03 01:30:00.000', 0), +]) +def test_temporal_ltz_partition_specs_match_java_in_new_york(rest_catalog, legacy, instant, expected, deleted): + from pypaimon.tests.deletion_vector_path_test import _delete + # Native caches the process system zone. CI runs this golden case in a + # separate process initialized with New York, as for temporal cast tests. + if os.environ.get('TZ') != 'America/New_York': + pytest.skip('Java golden values require TZ=America/New_York') + catalog, _ = rest_catalog + schema = pa.schema([('id', pa.int32()), ('p', pa.timestamp('ms', tz='UTC'))]) + table = _create(catalog, schema, ['p'], { + 'deletion-vectors.enabled': 'true', 'row-tracking.enabled': 'true', + 'data-evolution.enabled': 'true', 'partition.legacy-name': str(legacy).lower()}) + _append(table, [{'id': i, 'p': datetime.fromisoformat(instant)} for i in range(2)], schema) + _delete(table, [0]) + rows = _assert_parity(table) + assert [(row['partition'], row['deleted_record_count']) for row in rows] == [('p=' + expected, deleted)] @pytest.mark.parametrize('native', [False, True])