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..e775ca457c3b --- /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 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..654aaae9bd3b --- /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 825ae239345e..d88f4fda4cb6 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 @@ 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 _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 @@ 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()) + + 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__":