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 414080b718 [python] Preserve high-precision DECIMAL values in the row
file format (#10124)
414080b718 is described below
commit 414080b71820570fe2a33063a5cb47049c2721c5
Author: jackylee <[email protected]>
AuthorDate: Thu Sep 24 10:46:26 2026 +0800
[python] Preserve high-precision DECIMAL values in the row file format
(#10124)
---
.../pypaimon/read/reader/format_row_reader.py | 10 ++--
.../tests/test_format_row_reader_writer.py | 62 ++++++++++++++++++++++
.../pypaimon/write/writer/format_row_writer.py | 17 +++---
3 files changed, 77 insertions(+), 12 deletions(-)
diff --git a/paimon-python/pypaimon/read/reader/format_row_reader.py
b/paimon-python/pypaimon/read/reader/format_row_reader.py
index 56ddd15fc0..21b7e75c3f 100644
--- a/paimon-python/pypaimon/read/reader/format_row_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_row_reader.py
@@ -16,7 +16,7 @@
# under the License.
import struct
-from decimal import Decimal
+from decimal import Decimal, localcontext
from typing import Any, List, Optional, Tuple
import pyarrow as pa
@@ -483,11 +483,15 @@ def _read_field(decoder: _RowDecoder, data_type) -> Any:
precision, scale = _parse_decimal_params(type_name)
if precision <= 18:
unscaled = decoder.read_long()
- return Decimal(unscaled) / Decimal(10 ** scale)
else:
raw = decoder.read_bytes()
unscaled = int.from_bytes(raw, byteorder='big', signed=True)
- return Decimal(unscaled) / Decimal(10 ** scale)
+ # Rescale under a context wide enough for the column: the default
28-digit
+ # precision would round a DECIMAL(p) value with more than 28
significant
+ # digits, silently corrupting it. Mirrors pypaimon/data/decimal.py.
+ with localcontext() as ctx:
+ ctx.prec = max(precision + abs(scale), 38)
+ return Decimal(unscaled).scaleb(-scale)
elif type_name.startswith('TIMESTAMP'):
precision = _parse_timestamp_precision(type_name)
millis = decoder.read_long()
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 2023dd3fdd..d8b5d9b7d1 100644
--- a/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
+++ b/paimon-python/pypaimon/tests/test_format_row_reader_writer.py
@@ -88,6 +88,68 @@ class TestFormatRowReaderWriter:
finally:
os.unlink(path)
+ def test_high_precision_decimal(self):
+ # DECIMAL(p) with more than 28 significant digits overflows Python's
default
+ # decimal context, which silently rounded the value on both write and
read.
+ # Cover positive and negative (signed-byte path) high-precision values
and a
+ # null, plus the 19..28 band and the compact p<=18 path.
+ fields = [
+ DataField(0, "d_int", AtomicType("DECIMAL(38, 0)")),
+ DataField(1, "d_frac", AtomicType("DECIMAL(38, 10)")),
+ DataField(2, "d_band", AtomicType("DECIMAL(28, 4)")),
+ DataField(3, "d_small", AtomicType("DECIMAL(18, 2)")),
+ ]
+ d_int = [Decimal("12345678901234567890123456789012345678"),
+ Decimal("-12345678901234567890123456789012345678"), None]
+ d_frac = [Decimal("1234567890123456789012345678.9012345678"),
+ Decimal("-1234567890123456789012345678.9012345678"), None]
+ d_band = [Decimal("123456789012345678901234.5678"),
+ Decimal("-123456789012345678901234.5678"), None]
+ d_small = [Decimal("1234.56"), Decimal("-1234.56"), None]
+ data = pa.table({
+ "d_int": pa.array(d_int, type=pa.decimal128(38, 0)),
+ "d_frac": pa.array(d_frac, type=pa.decimal128(38, 10)),
+ "d_band": pa.array(d_band, type=pa.decimal128(28, 4)),
+ "d_small": pa.array(d_small, type=pa.decimal128(18, 2)),
+ })
+
+ with tempfile.NamedTemporaryFile(suffix=".row", delete=False) as tmp:
+ path = tmp.name
+
+ try:
+ _write_row_file(path, fields, data)
+ result = _read_row_file(path, fields)
+ assert result.column("d_int").to_pylist() == d_int
+ assert result.column("d_frac").to_pylist() == d_frac
+ assert result.column("d_band").to_pylist() == d_band
+ assert result.column("d_small").to_pylist() == d_small
+ finally:
+ os.unlink(path)
+
+ def test_high_precision_decimal_decoded_from_wire(self):
+ # Independent of the writer: decode a hand-built signed unscaled byte
+ # sequence (the row-file wire form shared with the Java
implementation) and
+ # assert the exact Decimal, so a symmetric writer+reader scaling
mistake
+ # cannot round-trip undetected.
+ from pypaimon.read.reader.format_row_reader import _read_field,
_RowDecoder
+
+ def _varint(x):
+ out = bytearray()
+ while True:
+ b = x & 0x7F
+ x >>= 7
+ if x:
+ out.append(b | 0x80)
+ else:
+ out.append(b)
+ return bytes(out)
+
+ unscaled = 12345678901234567890123456789012345678 # 38 significant
digits
+ raw = unscaled.to_bytes((unscaled.bit_length() + 8) // 8, 'big',
signed=True)
+ buf = _varint(len(raw)) + raw
+ got = _read_field(_RowDecoder(buf, 0), AtomicType("DECIMAL(38, 10)"))
+ assert got == Decimal("1234567890123456789012345678.9012345678")
+
def test_all_primitive_types(self):
fields = [
DataField(0, "bool_col", AtomicType("BOOLEAN")),
diff --git a/paimon-python/pypaimon/write/writer/format_row_writer.py
b/paimon-python/pypaimon/write/writer/format_row_writer.py
index f31c1eb58c..249189d111 100644
--- a/paimon-python/pypaimon/write/writer/format_row_writer.py
+++ b/paimon-python/pypaimon/write/writer/format_row_writer.py
@@ -18,7 +18,7 @@
import datetime
import re
import struct
-from decimal import Decimal
+from decimal import Decimal, localcontext
from typing import Any, List
import pyarrow as pa
@@ -268,17 +268,16 @@ def _write_field(buf: _BlockBuffer, value: Any,
data_type) -> None:
buf.write_bytes_with_length(value)
elif type_name.startswith('DECIMAL'):
precision, scale = _parse_decimal_params(type_name)
+ dec = value if isinstance(value, Decimal) else Decimal(str(value))
+ # Compute the unscaled value under a context wide enough for the
column:
+ # the default 28-digit precision would round a DECIMAL(p) value
with more
+ # than 28 significant digits, silently corrupting it. Mirrors the
read path.
+ with localcontext() as ctx:
+ ctx.prec = max(precision + abs(scale), 38)
+ unscaled = int(dec.scaleb(scale))
if precision <= 18:
- if isinstance(value, Decimal):
- unscaled = int(value * (10 ** scale))
- else:
- unscaled = int(Decimal(str(value)) * (10 ** scale))
buf.write_long_le(unscaled)
else:
- if isinstance(value, Decimal):
- unscaled = int(value * (10 ** scale))
- else:
- unscaled = int(Decimal(str(value)) * (10 ** scale))
raw = unscaled.to_bytes(
(unscaled.bit_length() + 8) // 8, byteorder='big',
signed=True)
buf.write_bytes_with_length(raw)
