From 4a1fe4eab1ff21c538f574d41640b655e0dab01b Mon Sep 17 00:00:00 2001
From: Vova Kolmakov <wombatukun@apache.org>
Date: Fri, 24 Jul 2026 15:05:10 +0700
Subject: [PATCH] [python] Render min/max_partition_stats in the $manifests
 system table

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 .../pypaimon/casting/row_to_string.py         | 124 ++++++++++++++++++
 .../pypaimon/table/system/manifests_table.py  |  11 +-
 .../pypaimon/tests/row_to_string_test.py      | 118 +++++++++++++++++
 .../tests/system/manifests_table_test.py      |  46 ++++++-
 4 files changed, 287 insertions(+), 12 deletions(-)
 create mode 100644 paimon-python/pypaimon/casting/row_to_string.py
 create mode 100644 paimon-python/pypaimon/tests/row_to_string_test.py

diff --git a/paimon-python/pypaimon/casting/row_to_string.py b/paimon-python/pypaimon/casting/row_to_string.py
new file mode 100644
index 000000000000..2d2aa7332557
--- /dev/null
+++ b/paimon-python/pypaimon/casting/row_to_string.py
@@ -0,0 +1,124 @@
+# 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.
+
+"""Render a row as the string Java produces when casting it to STRING.
+
+Mirrors ``RowToStringCastRule`` and the per type ``*ToStringCastRule`` rules of
+``paimon-common``, so system tables show the same text in both languages.
+"""
+
+import struct
+from typing import Any, Optional
+
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+
+def cast_row_to_string(row) -> Optional[str]:
+    """Render ``row`` as ``{v1, v2}``, with ``null`` for null fields.
+
+    An empty row renders as ``{}``, which is what an unpartitioned table has.
+    """
+    if row is None:
+        return None
+    fields = getattr(row, "fields", None) or []
+    parts = []
+    for i in range(len(fields)):
+        value = row.get_field(i)
+        parts.append("null" if value is None
+                     else cast_value_to_string(value, fields[i].type))
+    return "{" + ", ".join(parts) + "}"
+
+
+def cast_value_to_string(value: Any, data_type) -> str:
+    """Cast a single field value to string the way the Java cast rules do."""
+    type_name = _type_name(data_type)
+
+    if type_name in ("BOOLEAN", "BOOL"):
+        return "true" if value else "false"
+    if type_name in ("FLOAT", "REAL"):
+        return _format_float(value)
+    if (type_name.startswith("BINARY") or type_name.startswith("VARBINARY")
+            or type_name == "BYTES"):
+        return bytes(value).decode("utf-8", "replace")
+    if type_name == "DATE":
+        return value.isoformat()
+    # TIMESTAMP has to be tested before TIME, it starts with it
+    if type_name.startswith("TIMESTAMP"):
+        precision, _ = _parse_type_precision_scale(data_type)
+        return _format_timestamp(value, precision)
+    if type_name.startswith("TIME"):
+        precision, _ = _parse_type_precision_scale(data_type)
+        return _format_time(value, precision)
+    return str(value)
+
+
+def _type_name(data_type) -> str:
+    name = getattr(data_type, "type", None)
+    return (name if isinstance(name, str) else str(data_type)).upper().strip()
+
+
+def _format_timestamp(value, precision: int) -> str:
+    """Format as ``yyyy-MM-dd HH:mm:ss[.fraction]``.
+
+    The fraction is padded to nine digits and then stripped of trailing zeros
+    down to ``precision`` digits, as ``DateTimeUtils.formatTimestamp`` does.
+    Python datetimes are microsecond resolution, so the last three digits of a
+    TIMESTAMP(7..9) value are always zero.
+    """
+    text = "{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
+        value.year, value.month, value.day,
+        value.hour, value.minute, value.second)
+    fraction = "{:09d}".format(value.microsecond * 1000)
+    while len(fraction) > precision and fraction.endswith("0"):
+        fraction = fraction[:-1]
+    return (text + "." + fraction) if fraction else text
+
+
+def _format_time(value, precision: int) -> str:
+    """Format as ``HH:mm:ss[.fraction]``.
+
+    ``DateTimeUtils.formatTimestampMillis`` emits at most ``precision`` digits
+    of the millisecond part and stops early once nothing but zeros is left, but
+    always keeps one digit when precision allows any.
+    """
+    text = "{:02d}:{:02d}:{:02d}".format(value.hour, value.minute, value.second)
+    if precision <= 0:
+        return text
+    digits = "{:03d}".format(value.microsecond // 1000)[:min(precision, 3)]
+    return text + "." + (digits.rstrip("0") or digits[:1])
+
+
+def _format_float(value) -> str:
+    """Format like ``Float.toString``: shortest text that round trips as float32.
+
+    ``repr`` would print the float64 the reader widened the value to, turning
+    ``0.1f`` into ``0.10000000149011612``.
+    """
+    try:
+        packed = struct.pack("<f", value)
+    except (OverflowError, struct.error):
+        return repr(value)
+    text = repr(value)
+    for digits in range(1, 10):
+        candidate = "%.{}g".format(digits) % value
+        if struct.pack("<f", float(candidate)) == packed:
+            text = candidate
+            break
+    # Java always prints a decimal point for a finite float
+    if text.isdigit() or (text.startswith("-") and text[1:].isdigit()):
+        return text + ".0"
+    return text
diff --git a/paimon-python/pypaimon/table/system/manifests_table.py b/paimon-python/pypaimon/table/system/manifests_table.py
index 1975a4208af5..a2bcb90877ce 100644
--- a/paimon-python/pypaimon/table/system/manifests_table.py
+++ b/paimon-python/pypaimon/table/system/manifests_table.py
@@ -21,6 +21,7 @@
 
 import pyarrow
 
+from pypaimon.casting.row_to_string import cast_row_to_string
 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
@@ -75,12 +76,10 @@ def _build_arrow_table(self) -> pyarrow.Table:
             num_added.append(int(meta.num_added_files))
             num_deleted.append(int(meta.num_deleted_files))
             schema_ids.append(int(meta.schema_id))
-            # TODO: render min/max_partition_stats by casting partition
-            # rows to their string form. pypaimon
-            # has SimpleStats but no shared partition-row-to-string
-            # helper yet; emit NULL to preserve the column shape.
-            min_partition_stats.append(None)
-            max_partition_stats.append(None)
+            min_partition_stats.append(
+                cast_row_to_string(meta.partition_stats.min_values))
+            max_partition_stats.append(
+                cast_row_to_string(meta.partition_stats.max_values))
             min_row_ids.append(
                 None if meta.min_row_id is None else int(meta.min_row_id))
             max_row_ids.append(
diff --git a/paimon-python/pypaimon/tests/row_to_string_test.py b/paimon-python/pypaimon/tests/row_to_string_test.py
new file mode 100644
index 000000000000..6bbd561e8a49
--- /dev/null
+++ b/paimon-python/pypaimon/tests/row_to_string_test.py
@@ -0,0 +1,118 @@
+# 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.
+
+"""Tests that cast_row_to_string reproduces the Java cast-to-string rules."""
+
+import struct
+import unittest
+from datetime import date, datetime, time
+from decimal import Decimal
+
+from pypaimon.casting.row_to_string import (cast_row_to_string,
+                                            cast_value_to_string)
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.generic_row import GenericRow
+
+
+def _row(*pairs):
+    fields = [DataField(i, "f" + str(i), AtomicType(t))
+              for i, (t, _) in enumerate(pairs)]
+    return GenericRow([v for _, v in pairs], fields)
+
+
+def _cast(type_name, value):
+    return cast_value_to_string(value, AtomicType(type_name))
+
+
+class RowToStringTest(unittest.TestCase):
+
+    def test_none_row(self):
+        self.assertIsNone(cast_row_to_string(None))
+
+    def test_empty_row_of_unpartitioned_table(self):
+        self.assertEqual("{}", cast_row_to_string(_row()))
+
+    def test_fields_are_comma_separated(self):
+        self.assertEqual("{1, a}", cast_row_to_string(_row(("INT", 1),
+                                                           ("STRING", "a"))))
+
+    def test_null_field_is_a_literal(self):
+        self.assertEqual("{null, a}", cast_row_to_string(_row(("INT", None),
+                                                              ("STRING", "a"))))
+
+    def test_boolean_is_lower_case(self):
+        self.assertEqual("true", _cast("BOOLEAN", True))
+        self.assertEqual("false", _cast("BOOLEAN", False))
+
+    def test_integers(self):
+        self.assertEqual("-7", _cast("TINYINT", -7))
+        self.assertEqual("42", _cast("INT", 42))
+        self.assertEqual("9223372036854775807", _cast("BIGINT",
+                                                      9223372036854775807))
+
+    def test_decimal_keeps_its_scale(self):
+        self.assertEqual("1.50", _cast("DECIMAL(10, 2)", Decimal("1.50")))
+
+    def test_double(self):
+        self.assertEqual("1.5", _cast("DOUBLE", 1.5))
+        self.assertEqual("1.0", _cast("DOUBLE", 1.0))
+
+    def test_float_prints_the_shortest_float32_form(self):
+        # a float32 0.1 widened to float64 is 0.10000000149011612
+        widened = struct.unpack("<f", struct.pack("<f", 0.1))[0]
+        self.assertEqual("0.1", _cast("FLOAT", widened))
+        self.assertEqual("1.0", _cast("FLOAT", 1.0))
+
+    def test_binary_is_decoded_as_utf8(self):
+        self.assertEqual("ab", _cast("BYTES", b"ab"))
+
+    def test_date(self):
+        self.assertEqual("2024-01-02", _cast("DATE", date(2024, 1, 2)))
+
+    def test_timestamp_separator_is_a_space(self):
+        value = datetime(2024, 1, 2, 3, 4, 5, 123456)
+        self.assertEqual("2024-01-02 03:04:05.123456",
+                         _cast("TIMESTAMP(6)", value))
+
+    def test_timestamp_fraction_is_kept_up_to_precision(self):
+        value = datetime(2024, 1, 2, 3, 4, 5, 0)
+        self.assertEqual("2024-01-02 03:04:05", _cast("TIMESTAMP(0)", value))
+        self.assertEqual("2024-01-02 03:04:05.000", _cast("TIMESTAMP(3)", value))
+        self.assertEqual("2024-01-02 03:04:05.000000",
+                         _cast("TIMESTAMP(6)", value))
+
+    def test_timestamp_trailing_zeros_are_stripped_down_to_precision(self):
+        value = datetime(2024, 1, 2, 3, 4, 5, 120000)
+        self.assertEqual("2024-01-02 03:04:05.12", _cast("TIMESTAMP(0)", value))
+        self.assertEqual("2024-01-02 03:04:05.120", _cast("TIMESTAMP(3)", value))
+
+    def test_time(self):
+        self.assertEqual("03:04:05", _cast("TIME", time(3, 4, 5)))
+        self.assertEqual("03:04:05", _cast("TIME(0)", time(3, 4, 5)))
+        self.assertEqual("03:04:05.123",
+                         _cast("TIME(3)", time(3, 4, 5, 123000)))
+
+    def test_time_keeps_one_fraction_digit_at_least(self):
+        self.assertEqual("03:04:05.0", _cast("TIME(3)", time(3, 4, 5)))
+        self.assertEqual("03:04:05.5", _cast("TIME(3)", time(3, 4, 5, 500000)))
+
+    def test_unknown_type_falls_back_to_str(self):
+        self.assertEqual("x", _cast("SOMETHING_NEW", "x"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/paimon-python/pypaimon/tests/system/manifests_table_test.py b/paimon-python/pypaimon/tests/system/manifests_table_test.py
index 825ae239345e..ae0016c404fa 100644
--- a/paimon-python/pypaimon/tests/system/manifests_table_test.py
+++ b/paimon-python/pypaimon/tests/system/manifests_table_test.py
@@ -63,6 +63,27 @@ def _write_one_commit(self):
         writer.close()
         commit.close()
 
+    def _write_two_partitions(self):
+        fields = [
+            DataField.from_dict({"id": 0, "name": "id", "type": "INT"}),
+            DataField.from_dict({"id": 1, "name": "pt", "type": "INT"}),
+            DataField.from_dict({"id": 2, "name": "dt", "type": "STRING"}),
+        ]
+        self.catalog.create_table(
+            "db.p", Schema(fields=fields, partition_keys=["pt", "dt"]), False)
+        table = self.catalog.get_table("db.p")
+        write_builder = table.new_batch_write_builder()
+        writer = write_builder.new_write()
+        commit = write_builder.new_commit()
+        writer.write_arrow(pa.table({
+            "id": pa.array([1, 2], type=pa.int32()),
+            "pt": pa.array([1, 2], type=pa.int32()),
+            "dt": ["2024-01-01", "2024-01-02"],
+        }))
+        commit.commit(writer.prepare_commit())
+        writer.close()
+        commit.close()
+
     def test_manifests_table_loaded_via_catalog(self):
         table = self.catalog.get_table("db.t$manifests")
         self.assertIsInstance(table, ManifestsTable)
@@ -100,12 +121,25 @@ def test_lists_manifests_of_latest_snapshot(self):
         for n_added in arrow_table.column("num_added_files").to_pylist():
             self.assertGreaterEqual(n_added, 0)
 
-        # min/max_partition_stats are placeholders until the partition
-        # cast-to-string helper lands; pin the placeholder contract.
-        for value in arrow_table.column("min_partition_stats").to_pylist():
-            self.assertIsNone(value)
-        for value in arrow_table.column("max_partition_stats").to_pylist():
-            self.assertIsNone(value)
+    def test_partition_stats_of_unpartitioned_table(self):
+        self._write_one_commit()
+        arrow_table = _read(self.catalog.get_table("db.t$manifests"))
+
+        # an empty partition row renders as "{}", the same as in Java
+        self.assertEqual(["{}"] * arrow_table.num_rows,
+                         arrow_table.column("min_partition_stats").to_pylist())
+        self.assertEqual(["{}"] * arrow_table.num_rows,
+                         arrow_table.column("max_partition_stats").to_pylist())
+
+    def test_partition_stats_span_the_written_partitions(self):
+        self._write_two_partitions()
+        arrow_table = _read(self.catalog.get_table("db.p$manifests"))
+        self.assertEqual(1, arrow_table.num_rows)
+
+        self.assertEqual(["{1, 2024-01-01}"],
+                         arrow_table.column("min_partition_stats").to_pylist())
+        self.assertEqual(["{2, 2024-01-02}"],
+                         arrow_table.column("max_partition_stats").to_pylist())
 
 
 if __name__ == "__main__":