This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 93089092f8 [python] Render min/max_partition_stats in the $manifests
system table (#8842)
93089092f8 is described below
commit 93089092f8138628c7ad1a4c138c45b210f8d8f6
Author: Vova Kolmakov <[email protected]>
AuthorDate: Tue Jul 28 20:16:20 2026 +0700
[python] Render min/max_partition_stats in the $manifests system table
(#8842)
---
paimon-python/pypaimon/casting/row_to_string.py | 145 +++++++++++++++++++
.../pypaimon/table/system/manifests_table.py | 11 +-
paimon-python/pypaimon/tests/row_to_string_test.py | 156 +++++++++++++++++++++
.../pypaimon/tests/system/manifests_table_test.py | 101 ++++++++++++-
4 files changed, 401 insertions(+), 12 deletions(-)
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 0000000000..e775ca457c
--- /dev/null
+++ b/paimon-python/pypaimon/casting/row_to_string.py
@@ -0,0 +1,145 @@
+# 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.
+
+Only the types whose output provably matches Java are rendered. A row holding
+one of the types left out is not rendered at all, so the caller keeps the NULL
+it would have emitted anyway:
+
+- FLOAT and DOUBLE go through ``Float/Double.toString``, which up to JDK 18 is
+ the legacy ``FloatingDecimal`` and does not produce the shortest round
+ tripping decimal, while JDK 19+ does: ``2.68873286E11`` against
+ ``2.6887329E11`` for the same float.
+- TIMESTAMP WITH LOCAL TIME ZONE is formatted in ``TimeZone.getDefault()``, so
+ one manifest reads ``2024-01-02 03:04:05`` on a UTC JVM and
+ ``2024-01-02 06:04:05`` on a Europe/Moscow one.
+- BINARY, VARBINARY and BYTES are decoded as UTF-8, and malformed input leaves
+ the JDK decoder and Python disagreeing on how many replacement characters to
+ emit: ``ED A0 80`` is one in Java and three here.
+"""
+
+from typing import Any, Optional
+
+from pypaimon.table.row.generic_row import _parse_type_precision_scale
+
+_UNSUPPORTED_TYPES = ("FLOAT", "REAL", "DOUBLE", "TIMESTAMP_LTZ",
+ "BINARY", "VARBINARY", "BYTES")
+
+
+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.
+ Returns None when a field has a type this module does not render.
+ """
+ if row is None:
+ return None
+ fields = getattr(row, "fields", None) or []
+ # on the declared type, not on the value, so every manifest of a table
+ # answers the same way no matter which stats happen to be null
+ if any(_is_unsupported(field.type) for field in fields):
+ return None
+ 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 _is_unsupported(data_type):
+ raise ValueError(
+ "{} has no portable string form, see the module docstring".format(
+ type_name))
+ if type_name in ("BOOLEAN", "BOOL"):
+ return "true" if value else "false"
+ if type_name.startswith("DECIMAL") or type_name.startswith("NUMERIC"):
+ # Java Decimal.toString is BigDecimal.toPlainString, which never uses
+ # scientific notation, while str(Decimal) does below 1e-6
+ return "{:f}".format(value)
+ 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 _is_unsupported(data_type) -> bool:
+ type_name = _type_name(data_type)
+ # the other spelling of TIMESTAMP_LTZ, see data_types.py
+ if "WITH LOCAL TIME ZONE" in type_name:
+ return True
+ # first token only, so "FLOAT NOT NULL" is caught as well
+ head = type_name.split("(", 1)[0].split()
+ return bool(head) and head[0] in _UNSUPPORTED_TYPES
+
+
+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]``.
+
+ Digit by digit off the millisecond part, exactly as
+ ``DateTimeUtils.formatTimestampMillis`` does: it stops only once nothing
+ but zeros is left, so a truncated non zero tail keeps the zero before it
+ (``.10`` for 101 ms at precision 2, not ``.1``).
+ """
+ text = "{:02d}:{:02d}:{:02d}".format(value.hour, value.minute,
value.second)
+ if precision <= 0:
+ return text
+ millis = value.microsecond // 1000
+ digits = []
+ while precision > 0:
+ digits.append(str(millis // 100))
+ millis = millis % 100 * 10
+ if millis == 0:
+ break
+ precision -= 1
+ return text + "." + "".join(digits)
diff --git a/paimon-python/pypaimon/table/system/manifests_table.py
b/paimon-python/pypaimon/table/system/manifests_table.py
index 1975a4208a..a2bcb90877 100644
--- a/paimon-python/pypaimon/table/system/manifests_table.py
+++ b/paimon-python/pypaimon/table/system/manifests_table.py
@@ -21,6 +21,7 @@ from typing import List, Optional
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 @@ class ManifestsTable(SystemTable):
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 0000000000..654aaae9bd
--- /dev/null
+++ b/paimon-python/pypaimon/tests/row_to_string_test.py
@@ -0,0 +1,156 @@
+# 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 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_decimal_never_uses_scientific_notation(self):
+ # Java Decimal.toString is BigDecimal.toPlainString; str(Decimal)
+ # would give "0E-9" and "-1E-9" here
+ self.assertEqual("0.000000000",
+ _cast("DECIMAL(20, 9)", Decimal("0E-9")))
+ self.assertEqual("-0.000000001",
+ _cast("DECIMAL(20, 9)", Decimal("-1E-9")))
+ self.assertEqual("0.000000000000000000",
+ _cast("DECIMAL(38, 18)", Decimal("0E-18")))
+
+ def test_row_with_a_float_field_is_not_rendered(self):
+ self.assertIsNone(cast_row_to_string(_row(("INT", 1),
+ ("FLOAT", 1.5))))
+ self.assertIsNone(cast_row_to_string(_row(("DOUBLE", 1.5))))
+ self.assertIsNone(cast_row_to_string(_row(("REAL", 1.5))))
+ self.assertIsNone(cast_row_to_string(_row(("FLOAT NOT NULL", 1.5))))
+
+ def test_float_field_stops_the_row_even_when_its_value_is_null(self):
+ # the check is on the declared type, so all rows of a table agree
+ self.assertIsNone(cast_row_to_string(_row(("INT", 1),
+ ("DOUBLE", None))))
+
+ def test_row_with_a_local_zoned_timestamp_is_not_rendered(self):
+ # Java formats these in TimeZone.getDefault(), so the same manifest
+ # reads differently depending on where the query runs
+ value = datetime(2024, 1, 2, 3, 4, 5)
+ self.assertIsNone(cast_row_to_string(_row(("TIMESTAMP_LTZ(3)",
value))))
+ self.assertIsNone(cast_row_to_string(
+ _row(("TIMESTAMP(3) WITH LOCAL TIME ZONE", value))))
+
+ def test_unrenderable_values_are_rejected(self):
+ self.assertRaises(ValueError, _cast, "FLOAT", 1.5)
+ self.assertRaises(ValueError, _cast, "DOUBLE", 1.5)
+ self.assertRaises(ValueError, _cast, "TIMESTAMP_LTZ(3)",
+ datetime(2024, 1, 2, 3, 4, 5))
+ self.assertRaises(ValueError, _cast, "BYTES", b"ab")
+
+ def test_row_with_a_binary_field_is_not_rendered(self):
+ # malformed UTF-8 yields a different replacement char count than the
+ # JDK decoder gives, so binary is not rendered at all
+ self.assertIsNone(cast_row_to_string(_row(("BYTES", b"ab"))))
+ self.assertIsNone(cast_row_to_string(_row(("BINARY(2)", b"ab"))))
+ self.assertIsNone(cast_row_to_string(_row(("VARBINARY(10)", 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_time_keeps_a_zero_that_a_truncated_tail_sits_behind(self):
+ # 101 ms at precision 2 is ".10" in Java, not ".1": it stops only
+ # once the remaining fraction is exactly zero
+ self.assertEqual("03:04:05.10", _cast("TIME(2)", time(3, 4, 5,
101000)))
+ self.assertEqual("03:04:05.00", _cast("TIME(2)", time(3, 4, 5, 1000)))
+ self.assertEqual("03:04:05.0", _cast("TIME(2)", time(3, 4, 5)))
+ self.assertEqual("03:04:05.01", _cast("TIME(2)", time(3, 4, 5, 10000)))
+
+ 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 825ae23934..d88f4fda4c 100644
--- a/paimon-python/pypaimon/tests/system/manifests_table_test.py
+++ b/paimon-python/pypaimon/tests/system/manifests_table_test.py
@@ -63,6 +63,46 @@ class ManifestsTableTest(unittest.TestCase):
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 _write_partition(self, name, partition_type, partition_column):
+ fields = [
+ DataField.from_dict({"id": 0, "name": "id", "type": "INT"}),
+ DataField.from_dict({"id": 1, "name": "pt", "type":
partition_type}),
+ ]
+ self.catalog.create_table(
+ "db." + name, Schema(fields=fields, partition_keys=["pt"]), False)
+ table = self.catalog.get_table("db." + name)
+ 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": partition_column,
+ }))
+ 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 +140,61 @@ class ManifestsTableTest(unittest.TestCase):
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())
+
+ def test_partition_stats_stay_null_for_a_float_partition(self):
+ # Java prints these with Double.toString, whose output depends on the
+ # JVM version, so the columns keep the NULL they had before
+ self._write_partition("d", "DOUBLE",
+ pa.array([1.5, 2.5], type=pa.float64()))
+ arrow_table = _read(self.catalog.get_table("db.d$manifests"))
+ self.assertEqual(1, arrow_table.num_rows)
+
+ self.assertEqual([None],
+ arrow_table.column("min_partition_stats").to_pylist())
+ self.assertEqual([None],
+ arrow_table.column("max_partition_stats").to_pylist())
+
+ def test_partition_stats_stay_null_for_a_binary_partition(self):
+ self._write_partition("b", "BYTES",
+ pa.array([b"a", b"b"], type=pa.binary()))
+ arrow_table = _read(self.catalog.get_table("db.b$manifests"))
+ self.assertEqual(1, arrow_table.num_rows)
+
+ self.assertEqual([None],
+ arrow_table.column("min_partition_stats").to_pylist())
+ self.assertEqual([None],
+ arrow_table.column("max_partition_stats").to_pylist())
+
+ def test_decimal_partition_stats_carry_the_full_scale(self):
+ # str(Decimal) would render the zero as "0E-9" here, Java never does
+ self._write_partition("n", "DECIMAL(20, 9)",
+ pa.array([0, 1], type=pa.decimal128(20, 9)))
+ arrow_table = _read(self.catalog.get_table("db.n$manifests"))
+ self.assertEqual(1, arrow_table.num_rows)
+
+ self.assertEqual(["{0.000000000}"],
+ arrow_table.column("min_partition_stats").to_pylist())
+ self.assertEqual(["{1.000000000}"],
+ arrow_table.column("max_partition_stats").to_pylist())
if __name__ == "__main__":
