From 441f1e82535a7e29c2d7ded441c1739dc5112058 Mon Sep 17 00:00:00 2001
From: Li Jiajia <lijiajia@outlook.com>
Date: Tue, 27 Oct 2026 09:03:10 -0700
Subject: [PATCH] [python] Serialize high-precision decimals with Java's
 minimal bytes

Decimals with precision above 18 store their unscaled value as two's
complement bytes. PyPaimon computed the length as
(bit_length() + 8) // 8, which adds a byte for negative values that are
exactly -2^(8k-1), so -1.28 in DECIMAL(20,2) was written as ff80 while
Java writes 80 (BigInteger.toByteArray). Such keys hashed to a different
bucket than in Java (bucket 2 instead of 7 with 10 buckets), and such
partition values had different bytes than Java's.

BinaryRow serialization and the row format writer now use the minimal
encoding through one shared helper, which the global index key
serializer also uses instead of its own copy.
---
 paimon-python/pypaimon/data/decimal.py        |  9 +++--
 .../python/globalindex/key_serializer.py      | 19 +---------
 .../python/write/writer/format_row_writer.py  |  4 +--
 .../pypaimon/table/row/generic_row.py         |  8 ++---
 .../tests/decimal_key_serializer_test.py      | 21 +++++++++++
 paimon-python/pypaimon/tests/decimal_test.py  | 35 +++++++++++++++++++
 paimon-python/pypaimon/tests/test_decimal.py  |  6 ++++
 .../tests/test_format_row_reader_writer.py    | 25 +++++++++++++
 8 files changed, 98 insertions(+), 29 deletions(-)

diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py
index 30b5101bb4d9..b5c74c0f05fe 100644
--- a/paimon-python/pypaimon/data/decimal.py
+++ b/paimon-python/pypaimon/data/decimal.py
@@ -137,9 +137,7 @@ def to_unscaled_bytes(self) -> bytes:
             if value != value.to_integral_exact():
                 raise ArithmeticError("Decimal does not exactly fit in long.")
 
-        unscaled = int(value)
-        length = self._get_compact_byte_length(unscaled)
-        return unscaled.to_bytes(length, byteorder="big", signed=True)
+        return _signed_big_endian_bytes(int(value))
 
     def is_compact(self) -> bool:
         """
@@ -505,3 +503,8 @@ def _get_compact_byte_length(val: int) -> int:
             bits = (~val).bit_length()
 
         return bits // 8 + 1
+
+
+def _signed_big_endian_bytes(value: int) -> bytes:
+    length = Decimal._get_compact_byte_length(value)
+    return value.to_bytes(length, byteorder="big", signed=True)
diff --git a/paimon-python/pypaimon/execution/python/globalindex/key_serializer.py b/paimon-python/pypaimon/execution/python/globalindex/key_serializer.py
index e8c43c493799..d7347c4172f8 100644
--- a/paimon-python/pypaimon/execution/python/globalindex/key_serializer.py
+++ b/paimon-python/pypaimon/execution/python/globalindex/key_serializer.py
@@ -25,6 +25,7 @@
 from typing import Callable, Tuple
 import struct
 
+from pypaimon.data.decimal import _signed_big_endian_bytes
 from pypaimon.schema.data_types import AtomicType
 from pypaimon.schema.data_types import DataType
 
@@ -327,24 +328,6 @@ def _decimal_unscaled(value: object, scale: int) -> int:
     return int(rounded.scaleb(scale, context=context))
 
 
-def _signed_big_endian_bytes(value: int) -> bytes:
-    length = max(1, (value.bit_length() + 8) // 8)
-    while True:
-        try:
-            encoded = value.to_bytes(length, byteorder='big', signed=True)
-            break
-        except OverflowError:
-            length += 1
-    while len(encoded) > 1:
-        if encoded[0] == 0x00 and encoded[1] < 0x80:
-            encoded = encoded[1:]
-        elif encoded[0] == 0xFF and encoded[1] >= 0x80:
-            encoded = encoded[1:]
-        else:
-            break
-    return encoded
-
-
 def _date_to_epoch_days(value: object) -> int:
     if isinstance(value, datetime.datetime):
         value = value.date()
diff --git a/paimon-python/pypaimon/execution/python/write/writer/format_row_writer.py b/paimon-python/pypaimon/execution/python/write/writer/format_row_writer.py
index 249189d11128..3fb5f1c00e4f 100644
--- a/paimon-python/pypaimon/execution/python/write/writer/format_row_writer.py
+++ b/paimon-python/pypaimon/execution/python/write/writer/format_row_writer.py
@@ -24,6 +24,7 @@
 import pyarrow as pa
 
 from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
+from pypaimon.data.decimal import _signed_big_endian_bytes
 from pypaimon.schema.data_types import (
     ArrayType, AtomicType, DataField, MapType, MultisetType, RowType, VectorType
 )
@@ -278,8 +279,7 @@ def _write_field(buf: _BlockBuffer, value: Any, data_type) -> None:
             if precision <= 18:
                 buf.write_long_le(unscaled)
             else:
-                raw = unscaled.to_bytes(
-                    (unscaled.bit_length() + 8) // 8, byteorder='big', signed=True)
+                raw = _signed_big_endian_bytes(unscaled)
                 buf.write_bytes_with_length(raw)
         elif type_name.startswith('TIMESTAMP'):
             precision = _parse_timestamp_precision(type_name)
diff --git a/paimon-python/pypaimon/table/row/generic_row.py b/paimon-python/pypaimon/table/row/generic_row.py
index 31a7701edb9b..3012d65c42ea 100644
--- a/paimon-python/pypaimon/table/row/generic_row.py
+++ b/paimon-python/pypaimon/table/row/generic_row.py
@@ -23,6 +23,7 @@
 from decimal import Decimal
 from typing import Any, List, Union
 
+from pypaimon.data.decimal import _signed_big_endian_bytes
 from pypaimon.data.generic_variant import GenericVariant
 from pypaimon.schema.data_types import AtomicType, DataField, DataType
 from pypaimon.table.row.binary_row import BinaryRow
@@ -528,12 +529,7 @@ def to_bytes(cls, row: Union[GenericRow, BinaryRow]) -> bytes:
                     (absolute_offset << 32) | length)
             elif is_var_len_type or is_high_precision_decimal:
                 if is_high_precision_decimal:
-                    # Big-endian signed bytes
-                    if unscaled_value == 0:
-                        value_bytes = b'\x00'
-                    else:
-                        byte_length = (unscaled_value.bit_length() + 8) // 8  # +8 for sign bit
-                        value_bytes = unscaled_value.to_bytes(byte_length, byteorder='big', signed=True)
+                    value_bytes = _signed_big_endian_bytes(unscaled_value)
                 elif any(type_name.startswith(p) for p in ['CHAR', 'VARCHAR', 'STRING']):
                     value_bytes = str(value).encode('utf-8')
                 elif type_name == 'BLOB':
diff --git a/paimon-python/pypaimon/tests/decimal_key_serializer_test.py b/paimon-python/pypaimon/tests/decimal_key_serializer_test.py
index 531efc9d75a2..af25ece22a50 100644
--- a/paimon-python/pypaimon/tests/decimal_key_serializer_test.py
+++ b/paimon-python/pypaimon/tests/decimal_key_serializer_test.py
@@ -66,6 +66,27 @@ def test_exact_bytes_and_decode_with_low_precision(self, precision, scale, value
             self.assertEqual(3, context.prec)
             self.assertFalse(any(context.flags.values()))
 
+    @parameterized.expand([
+        (0, '00'),
+        (-1, 'ff'),
+        (127, '7f'),
+        (128, '0080'),
+        (-128, '80'),
+        (-129, 'ff7f'),
+        (-32768, '8000'),
+        (-8388608, '800000'),
+    ])
+    def test_java_decimal_bytes(self, unscaled, expected):
+        value = Decimal(unscaled).scaleb(-2)
+        raw = bytes.fromhex(expected)
+        for precision in (18, 20):
+            serializer = create_serializer(AtomicType('DECIMAL(%s,2)' % precision))
+            encoded = struct.pack('<q', unscaled) if precision <= 18 else raw
+            self.assertEqual(serializer.serialize(value), encoded)
+            self.assertEqual(serializer.deserialize(encoded), value)
+        padded = (b'\xff' if unscaled < 0 else b'\x00') + raw
+        self.assertEqual(serializer.deserialize(padded), value)
+
     def test_decode_does_not_round_at_default_precision(self):
         serializer = create_serializer(AtomicType('DECIMAL(38,2)'))
         unscaled = 12345678901234567890123456789012345678
diff --git a/paimon-python/pypaimon/tests/decimal_test.py b/paimon-python/pypaimon/tests/decimal_test.py
index fa90b3d8a2d4..217f624fb330 100644
--- a/paimon-python/pypaimon/tests/decimal_test.py
+++ b/paimon-python/pypaimon/tests/decimal_test.py
@@ -19,7 +19,11 @@
 import unittest
 from decimal import Decimal
 
+from parameterized import parameterized
+
+from pypaimon.execution.python.write.input_routing import FixedBucketInputRouter
 from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.schema.table_schema import TableSchema
 from pypaimon.table.row.generic_row import (GenericRow, GenericRowDeserializer,
                                             GenericRowSerializer,
                                             _decimal_to_unscaled_with_check)
@@ -158,6 +162,37 @@ def test_decimal_not_compact_binary_format(self):
         # Decimal("5.55000") with scale=5 => unscaled = 555000
         self.assertEqual(unscaled_value, 555000)
 
+    @parameterized.expand([
+        (0, '00'),
+        (-1, 'ff'),
+        (127, '7f'),
+        (128, '0080'),
+        (-128, '80'),
+        (-129, 'ff7f'),
+        (-32768, '8000'),
+        (-8388608, '800000'),
+    ])
+    def test_decimal_not_compact_java_bytes(self, unscaled, expected):
+        fields = [DataField(0, 'd', AtomicType('DECIMAL(20, 2)'))]
+        value = Decimal(unscaled).scaleb(-2)
+        raw = bytes.fromhex(expected)
+        prefix = struct.pack('>i', 1) + b'\x00' * 8
+        expected_row = prefix + struct.pack('<q', (16 << 32) | len(raw)) + raw.ljust(16, b'\x00')
+        self.assertEqual(GenericRowSerializer.to_bytes(GenericRow([value], fields)), expected_row)
+
+        for encoded in (raw, (b'\xff' if unscaled < 0 else b'\x00') + raw):
+            row = prefix + struct.pack('<q', (16 << 32) | len(encoded)) + encoded.ljust(16, b'\x00')
+            self.assertEqual(GenericRowDeserializer.from_bytes(row, fields).values, [value])
+
+    def test_decimal_fixed_bucket_matches_java(self):
+        fields = [DataField(0, 'd', AtomicType('DECIMAL(20, 2)'))]
+        schema = TableSchema(fields=fields, primary_keys=['d'], options={'bucket': '10'})
+        router = FixedBucketInputRouter(schema)
+        value = Decimal('-1.28')
+        # Same hash and bucket as Java.
+        self.assertEqual(router._binary_row_hash_code((value,), fields), 1196242097)
+        self.assertEqual(router.extract_partition_bucket_row({'d': value}), ((), 7))
+
     def test_decimal_boundary_precision(self):
         """Test boundary: DECIMAL(18, ...) is compact, DECIMAL(19, ...) is non-compact."""
         # precision=18: last compact
diff --git a/paimon-python/pypaimon/tests/test_decimal.py b/paimon-python/pypaimon/tests/test_decimal.py
index 7853aecc0e34..675bb0b37e18 100644
--- a/paimon-python/pypaimon/tests/test_decimal.py
+++ b/paimon-python/pypaimon/tests/test_decimal.py
@@ -163,6 +163,7 @@ def test_high_precision_scaled_round_trip(self):
 
     def test_to_unscaled_bytes_java_compatible(self):
         cases = [
+            (-8388608, "800000"),
             (-32769, "ff7fff"),
             (-32768, "8000"),
             (-129, "ff7f"),
@@ -195,7 +196,12 @@ def test_to_unscaled_bytes_java_compatible(self):
 
     def test_from_unscaled_bytes_java_compatible(self):
         cases = [
+            ("800000", -8388608),
+            ("ff800000", -8388608),
+            ("8000", -32768),
+            ("ff8000", -32768),
             ("80", -128),
+            ("ff80", -128),
             ("ff7f", -129),
             ("ff", -1),
             ("00", 0),
diff --git a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
index 34def649da18..de747e8eb3c7 100644
--- a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
+++ b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
@@ -296,6 +296,31 @@ def test_decimal(self):
         finally:
             os.unlink(path)
 
+    @pytest.mark.parametrize('unscaled,expected', [
+        (0, '00'),
+        (-1, 'ff'),
+        (127, '7f'),
+        (128, '0080'),
+        (-128, '80'),
+        (-129, 'ff7f'),
+        (-32768, '8000'),
+        (-8388608, '800000'),
+    ])
+    def test_decimal_minimal_bytes(self, unscaled, expected):
+        from pypaimon.execution.python.read.reader.format_row_reader import _read_field, _RowDecoder
+        from pypaimon.execution.python.write.writer.format_row_writer import _BlockBuffer, _write_field
+
+        data_type = AtomicType('DECIMAL(20, 2)')
+        value = Decimal(unscaled).scaleb(-2)
+        raw = bytes.fromhex(expected)
+        buf = _BlockBuffer()
+        _write_field(buf, value, data_type)
+        assert bytes(buf.buffer[:buf.position]) == bytes([len(raw)]) + raw
+
+        for encoded in (raw, (b'\xff' if unscaled < 0 else b'\x00') + raw):
+            decoder = _RowDecoder(bytes([len(encoded)]) + encoded, 0)
+            assert _read_field(decoder, data_type) == value
+
     def test_timestamp(self):
         fields = [
             DataField(0, "ts_millis", AtomicType("TIMESTAMP(3)")),