[python] Read partition metadata with Native SystemRead

.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
@@ -248,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
 

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',
 ))
 
 

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_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
+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()
 
@@ -79,36 +84,34 @@ 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 = 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
             stats = partition_map.get(spec_key)
             if stats is None:
+                partition = _normalize_partition(entry.partition, default_name, legacy)
                 stats = {
-                    "spec_items": spec_items,
-                    "partition_key": partition_key,
+                    "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,
                     "last_update_time": None,
-                    "buckets": set(),
+                    "total_buckets": entry.total_buckets,
                 }
                 partition_map[spec_key] = stats
 
             stats["record_count"] += int(entry.file.row_count)
             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,15 +121,16 @@ 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))
+                deleted_record_counts.get(stats["deletion_key"], 0))
 
         n = len(partition_map)
         return pyarrow.table({
@@ -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,48 @@ 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)
+
+
+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_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('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 GenericRow(values, partition.fields)

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):