[python] Serialize high-precision decimals with Java's minimal bytes

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)

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()

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)

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':

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

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

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),

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)")),