[python] Render min/max_partition_stats in the $manifests system table

paimon-python/pypaimon/casting/row_to_string.py

@@ -0,0 +1,165 @@
+# 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 math
+import struct
+from decimal import Decimal
+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_java_float(value, True)
+    if type_name == "DOUBLE":
+        return _format_java_float(value, False)
+    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_java_float(value, is_float32: bool) -> str:
+    """Format like Java ``Float.toString`` / ``Double.toString``.
+
+    Finds the shortest decimal that round trips, then lays it out with Java's
+    rules: plain decimal when ``1 <= D <= 7`` or ``-2 <= D <= 0`` (writing
+    ``value = 0.<digits> x 10^D``), otherwise ``d.dddEexp`` with an upper case
+    ``E``, a sign only when negative, and no zero padded exponent. The shortest
+    form differs from a pre JDK 19 JVM only at the denormal extremes
+    (``1.0E-45`` vs ``1.4E-45``), which partition stats never hold.
+    """
+    if value != value:
+        return "NaN"
+    if value == math.inf:
+        return "Infinity"
+    if value == -math.inf:
+        return "-Infinity"
+    sign = "-" if math.copysign(1.0, value) < 0 else ""
+    if value == 0.0:
+        return sign + "0.0"
+    digits, decimal_exp = _shortest_digits(abs(value), is_float32)
+    n = len(digits)
+    if 0 < decimal_exp <= 7:
+        if decimal_exp >= n:
+            body = digits + "0" * (decimal_exp - n) + ".0"
+        else:
+            body = digits[:decimal_exp] + "." + digits[decimal_exp:]
+    elif -3 < decimal_exp <= 0:
+        body = "0." + "0" * (-decimal_exp) + digits
+    else:
+        body = digits[0] + "." + (digits[1:] or "0") + "E" + str(decimal_exp - 1)
+    return sign + body
+
+
+def _shortest_digits(magnitude, is_float32: bool):
+    """Return ``(digits, D)`` with ``magnitude = 0.<digits> x 10^D``.
+
+    ``digits`` carries no leading or trailing zeros. ``repr`` already yields the
+    shortest float64 form; for float32 the shortest ``%g`` that round trips
+    through four bytes is used, so a widened ``0.1f`` renders as ``0.1``.
+    """
+    if is_float32:
+        packed = struct.pack("<f", magnitude)
+        text = repr(magnitude)
+        for precision in range(1, 10):
+            candidate = "%.{}g".format(precision) % magnitude
+            if struct.pack("<f", float(candidate)) == packed:
+                text = candidate
+                break
+    else:
+        text = repr(magnitude)
+    _, digit_tuple, exp = Decimal(text).as_tuple()
+    digits = list(digit_tuple)
+    while len(digits) > 1 and digits[-1] == 0:
+        digits.pop()
+        exp += 1
+    digit_str = "".join(map(str, digits))
+    return digit_str, exp + len(digit_str)

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(

paimon-python/pypaimon/tests/row_to_string_test.py

@@ -0,0 +1,163 @@
+# 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))
+
+
+def _f32(value):
+    # a FLOAT is widened to float64 on the way in, as the reader does
+    widened = struct.unpack("<f", struct.pack("<f", value))[0]
+    return _cast("FLOAT", widened)
+
+
+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_decimal_range(self):
+        self.assertEqual("1.5", _cast("DOUBLE", 1.5))
+        self.assertEqual("1.0", _cast("DOUBLE", 1.0))
+        self.assertEqual("0.001", _cast("DOUBLE", 0.001))
+        self.assertEqual("123456.0", _cast("DOUBLE", 123456.0))
+        self.assertEqual("9999999.0", _cast("DOUBLE", 9999999.0))
+        self.assertEqual("-1.5", _cast("DOUBLE", -1.5))
+
+    def test_double_scientific_matches_java(self):
+        # Java Double.toString: upper case E, no + sign, no zero padding
+        self.assertEqual("1.0E7", _cast("DOUBLE", 1e7))
+        self.assertEqual("1.0E-5", _cast("DOUBLE", 1e-5))
+        self.assertEqual("1.0E-4", _cast("DOUBLE", 1e-4))
+        self.assertEqual("1.0E20", _cast("DOUBLE", 1e20))
+        self.assertEqual("1.0E-20", _cast("DOUBLE", 1e-20))
+        self.assertEqual("1.23456789E300", _cast("DOUBLE", 1.23456789e300))
+        self.assertEqual("1.23456789012345E14",
+                         _cast("DOUBLE", 123456789012345.0))
+
+    def test_double_signed_zero(self):
+        self.assertEqual("0.0", _cast("DOUBLE", 0.0))
+        self.assertEqual("-0.0", _cast("DOUBLE", -0.0))
+
+    def test_double_non_finite_matches_java(self):
+        self.assertEqual("NaN", _cast("DOUBLE", float("nan")))
+        self.assertEqual("Infinity", _cast("DOUBLE", float("inf")))
+        self.assertEqual("-Infinity", _cast("DOUBLE", float("-inf")))
+
+    def test_float_prints_the_shortest_float32_form(self):
+        # a float32 0.1 widened to float64 is 0.10000000149011612
+        self.assertEqual("0.1", _f32(0.1))
+        self.assertEqual("1.0", _f32(1.0))
+        self.assertEqual("12345.678", _f32(12345.678))
+
+    def test_float_scientific_matches_java(self):
+        self.assertEqual("1.0E7", _f32(1e7))
+        self.assertEqual("1.0E-5", _f32(1e-5))
+        self.assertEqual("1.0E-4", _f32(1e-4))
+        self.assertEqual("3.4E38", _f32(3.4e38))
+        self.assertEqual("0.001", _f32(1e-3))
+        self.assertEqual("9999999.0", _f32(9999999.0))
+        self.assertEqual("-1.5", _f32(-1.5))
+
+    def test_float_signed_zero_and_non_finite(self):
+        self.assertEqual("0.0", _f32(0.0))
+        self.assertEqual("-0.0", _f32(-0.0))
+        self.assertEqual("NaN", _f32(float("nan")))
+        self.assertEqual("Infinity", _f32(float("inf")))
+
+    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()