From f7fe935fa402d2bd8e3ad33734f4c1040239bdce Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 17:22:34 +0800 Subject: [PATCH 01/12] [python] Implement Decimal data structure for pypaimon. --- paimon-python/pypaimon/data/__init__.py | 3 +- paimon-python/pypaimon/data/decimal.py | 329 +++++++++++++++++++ paimon-python/pypaimon/tests/test_decimal.py | 125 +++++++ 3 files changed, 456 insertions(+), 1 deletion(-) create mode 100644 paimon-python/pypaimon/data/decimal.py create mode 100644 paimon-python/pypaimon/tests/test_decimal.py diff --git a/paimon-python/pypaimon/data/__init__.py b/paimon-python/pypaimon/data/__init__.py index 6308bf139b66..97f36f3d5278 100644 --- a/paimon-python/pypaimon/data/__init__.py +++ b/paimon-python/pypaimon/data/__init__.py @@ -16,5 +16,6 @@ # under the License. from pypaimon.data.timestamp import Timestamp +from pypaimon.data.decimal import Decimal -__all__ = ['Timestamp'] +__all__ = ['Timestamp', 'Decimal'] diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py new file mode 100644 index 000000000000..4ae72302a992 --- /dev/null +++ b/paimon-python/pypaimon/data/decimal.py @@ -0,0 +1,329 @@ +# 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. + +from __future__ import annotations + +import re +from decimal import Decimal as BigDecimal +from decimal import ROUND_HALF_UP +from typing import Optional, Tuple + +_DECIMAL_PATTERN = re.compile( + r"^(DECIMAL|NUMERIC|DEC)(?:\((\d+)(?:,\s*(\d+))?\))?$", + re.IGNORECASE, +) + + +class Decimal: + """ + An internal data structure representing data of DecimalType. + + This data structure is immutable and might store decimal values in a + compact representation (as an integer value) if values are small enough. + + Python implementation note: + ``decimal_val`` is an instance of Python's standard library + ``decimal.Decimal`` (imported as ``BigDecimal``), corresponding to + Java's ``java.math.BigDecimal``. + """ + + # Maximum number of decimal digits a long integer can represent. + # (1e18 < Long.MAX_VALUE < 1e19) + MAX_LONG_DIGITS = 18 + + MAX_COMPACT_PRECISION = 18 + + # Powers of 10 used for compact decimal conversion. + POW10 = tuple(10 ** i for i in range(MAX_COMPACT_PRECISION + 1)) + + # The semantics of the fields are as follows: + # + # - precision and scale represent the SQL decimal type. + # - If decimal_val is set, it stores the complete decimal value. + # - Otherwise, the decimal value is represented by + # long_val / (10 ** scale). + # + # Note that (precision, scale) must always be correct. + # + # If precision > MAX_COMPACT_PRECISION: + # decimal_val stores the value and long_val is undefined. + # Otherwise: + # (long_val, scale) represents the value and decimal_val may be + # lazily initialized and cached. + def __init__( + self, + precision: int, + scale: int, + long_val: int, + decimal_val: Optional[BigDecimal], + ): + self.precision = precision + self.scale = scale + self.long_val = long_val + self.decimal_val = decimal_val + + # ---------------------------------------------------------------------- + # Public Interfaces + # ---------------------------------------------------------------------- + + def to_big_decimal(self) -> BigDecimal: + """ + Converts this Decimal into a decimal.Decimal instance. + """ + if self.decimal_val is None: + self.decimal_val = ( + BigDecimal(self.long_val) + / BigDecimal(self.POW10[self.scale]) + ) + return self.decimal_val + + def to_unscaled_long(self) -> int: + """ + Returns the unscaled integer value of this Decimal. + + Raises: + ArithmeticError: + If this Decimal does not exactly fit into a long integer. + """ + if self.is_compact(): + return self.long_val + + value = self.to_big_decimal() * self.POW10[self.scale] + + if value != value.to_integral_exact(): + raise ArithmeticError("Decimal does not exactly fit in long.") + + return int(value) + + def to_unscaled_bytes(self) -> bytes: + """ + Returns the unscaled value encoded as a signed byte array. + """ + value = self.to_unscaled_long() + length = max(1, (value.bit_length() + 8) // 8) + return value.to_bytes(length, byteorder="big", signed=True) + + def is_compact(self) -> bool: + """ + Returns whether the decimal value is small enough to be stored in a long integer. + """ + return self.precision <= self.MAX_COMPACT_PRECISION + + def copy(self) -> "Decimal": + """ + Returns a copy of this Decimal. + """ + return Decimal( + self.precision, + self.scale, + self.long_val, + self.decimal_val, + ) + + @staticmethod + def extract_decimal_precision_scale(type_str: str) -> Tuple[int, int]: + """ + Extracts the precision and scale from a DECIMAL/NUMERIC/DEC type string. + + Examples: + DECIMAL -> (10, 0) + DECIMAL(10) -> (10, 0) + DECIMAL(10,2) -> (10, 2) + NUMERIC(20,5) -> (20, 5) + DEC(18,6) -> (18, 6) + """ + match = _DECIMAL_PATTERN.fullmatch(type_str.strip()) + if match is None: + raise ValueError(f"Invalid decimal type: {type_str}") + + precision = match.group(2) + scale = match.group(3) + + if precision is None: + return 10, 0 + + if scale is None: + return int(precision), 0 + + return int(precision), int(scale) + + # ---------------------------------------------------------------------- + # Comparison + # ---------------------------------------------------------------------- + + def __eq__(self, other): + if not isinstance(other, Decimal): + return False + return self.compare_to(other) == 0 + + def __lt__(self, other): + return self.compare_to(other) < 0 + + def compare_to(self, other: "Decimal") -> int: + """ + Compares this Decimal with another Decimal. + """ + if ( + self.is_compact() + and other.is_compact() + and self.scale == other.scale + ): + if self.long_val < other.long_val: + return -1 + if self.long_val > other.long_val: + return 1 + return 0 + + a = self.to_big_decimal() + b = other.to_big_decimal() + + if a < b: + return -1 + if a > b: + return 1 + return 0 + + def __hash__(self): + return hash(self.to_big_decimal()) + + def __str__(self): + return format(self.to_big_decimal(), "f") + + # ---------------------------------------------------------------------- + # Constructor Utilities + # ---------------------------------------------------------------------- + + @classmethod + def from_big_decimal( + cls, + value: BigDecimal, + precision: int, + scale: int, + ) -> Optional["Decimal"]: + """ + Creates a ``Decimal`` from a Python's built-in ``decimal.Decimal`` with the given precision and scale. + + The value is rounded to the requested scale using ROUND_HALF_UP. + If the resulting precision exceeds the specified precision, None is returned. + """ + + quant = BigDecimal(1).scaleb(-scale) + + value = value.quantize( + quant, + rounding=ROUND_HALF_UP, + ) + + digits = len(value.as_tuple().digits) + + if digits > precision: + return None + + long_val = -1 + + if precision <= cls.MAX_COMPACT_PRECISION: + unscaled = value * cls.POW10[scale] + + if unscaled != unscaled.to_integral_exact(): + raise ArithmeticError( + "Decimal does not exactly fit in long." + ) + + long_val = int(unscaled) + + return cls( + precision, + scale, + long_val, + value, + ) + + @classmethod + def from_unscaled_long( + cls, + unscaled_long: int, + precision: int, + scale: int, + ) -> "Decimal": + """ + Creates a Decimal from an unscaled integer value with the given precision and scale. + """ + if precision <= 0 or precision > cls.MAX_LONG_DIGITS: + raise ValueError( + "precision must be between 1 and {}".format( + cls.MAX_LONG_DIGITS + ) + ) + + return cls( + precision, + scale, + unscaled_long, + None, + ) + + @classmethod + def from_unscaled_bytes( + cls, + unscaled_bytes: bytes, + precision: int, + scale: int, + ) -> Optional["Decimal"]: + """ + Creates a Decimal from an unscaled signed byte array. + """ + value = int.from_bytes( + unscaled_bytes, + byteorder="big", + signed=True, + ) + + bd = BigDecimal(value) / BigDecimal(cls.POW10[scale]) + + return cls.from_big_decimal( + bd, + precision, + scale, + ) + + @classmethod + def zero(cls, precision: int, scale: int) -> Optional["Decimal"]: + """ + Creates a Decimal representing zero with the given precision and scale. + + If the precision exceeds the supported range, None is returned. + """ + if precision <= cls.MAX_COMPACT_PRECISION: + return cls( + precision, + scale, + 0, + None, + ) + + return cls.from_big_decimal( + BigDecimal(0), + precision, + scale, + ) + + @staticmethod + def is_compact_precision(precision: int) -> bool: + """ + Returns whether the specified precision can be stored in compact form. + """ + return precision <= Decimal.MAX_COMPACT_PRECISION diff --git a/paimon-python/pypaimon/tests/test_decimal.py b/paimon-python/pypaimon/tests/test_decimal.py new file mode 100644 index 000000000000..12abd6f8512c --- /dev/null +++ b/paimon-python/pypaimon/tests/test_decimal.py @@ -0,0 +1,125 @@ +# 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. + +import unittest +from decimal import Decimal as BigDecimal + +from pypaimon.data.decimal import Decimal + + +class DecimalTest(unittest.TestCase): + + def test_from_big_decimal(self): + d = Decimal.from_big_decimal(BigDecimal("12.34"), precision=10, scale=2) + + self.assertIsNotNone(d) + self.assertEqual(d.precision, 10) + self.assertEqual(d.scale, 2) + self.assertEqual(d.to_big_decimal(), BigDecimal("12.34")) + self.assertEqual(d.to_unscaled_long(), 1234) + + def test_from_big_decimal_overflow(self): + d = Decimal.from_big_decimal(BigDecimal("12345678901.23"), precision=10, scale=2) + self.assertIsNone(d) + + def test_from_unscaled_long(self): + d = Decimal.from_unscaled_long(1234, precision=10, scale=2) + + self.assertEqual(d.precision, 10) + self.assertEqual(d.scale, 2) + self.assertEqual(d.to_big_decimal(), BigDecimal("12.34")) + + def test_from_unscaled_bytes(self): + d = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + + d2 = Decimal.from_unscaled_bytes( + d.to_unscaled_bytes(), + precision=10, + scale=2, + ) + + self.assertEqual(d2, d) + + def test_zero(self): + d = Decimal.zero(precision=10, scale=2) + self.assertEqual(d.to_big_decimal(), BigDecimal("0.00")) + + def test_copy(self): + d = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + copy = d.copy() + + self.assertEqual(copy, d) + self.assertIsNot(copy, d) + + def test_compare(self): + d1 = Decimal.from_big_decimal(BigDecimal("1.23"), 10, 2) + d2 = Decimal.from_big_decimal(BigDecimal("2.34"), 10, 2) + + self.assertLess(d1.compare_to(d2), 0) + self.assertGreater(d2.compare_to(d1), 0) + self.assertEqual(d1.compare_to(d1.copy()), 0) + + def test_hash(self): + d1 = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + d2 = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + + self.assertEqual(hash(d1), hash(d2)) + + def test_to_big_decimal(self): + d = Decimal.from_unscaled_long(1234, 10, 2) + self.assertEqual(d.to_big_decimal(), BigDecimal("12.34")) + + def test_to_unscaled_long(self): + d = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + self.assertEqual(d.to_unscaled_long(), 1234) + + def test_to_unscaled_bytes(self): + d = Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2) + + self.assertEqual( + int.from_bytes(d.to_unscaled_bytes(), "big", signed=True), + 1234, + ) + + def test_to_string(self): + self.assertEqual( + str(Decimal.from_big_decimal(BigDecimal("12.34"), 10, 2)), + "12.34", + ) + + self.assertEqual( + str(Decimal.from_big_decimal(BigDecimal("0.0000000000000000001"), 39, 19)), + "0.0000000000000000001", + ) + + def test_is_compact(self): + self.assertTrue(Decimal.is_compact_precision(18)) + self.assertFalse(Decimal.is_compact_precision(19)) + + self.assertTrue(Decimal.from_big_decimal(BigDecimal("1"), 18, 0).is_compact()) + self.assertFalse(Decimal.from_big_decimal(BigDecimal("1"), 19, 0).is_compact()) + + def test_extract_decimal_precision_scale(self): + self.assertEqual(Decimal.extract_decimal_precision_scale("DECIMAL"), (10, 0)) + self.assertEqual(Decimal.extract_decimal_precision_scale("DECIMAL(20)"), (20, 0)) + self.assertEqual(Decimal.extract_decimal_precision_scale("DECIMAL(20,5)"), (20, 5)) + self.assertEqual(Decimal.extract_decimal_precision_scale("NUMERIC(18,2)"), (18, 2)) + self.assertEqual(Decimal.extract_decimal_precision_scale("DEC(8,3)"), (8, 3)) + + def test_extract_decimal_precision_scale_invalid(self): + with self.assertRaises(ValueError): + Decimal.extract_decimal_precision_scale("INT") From 3dc6dee1f366e9088f1a68fa459330e5f2e0ff2a Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 17:26:08 +0800 Subject: [PATCH 02/12] [python] Implement Decimal data structure for pypaimon. --- paimon-python/pypaimon/data/decimal.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py index 4ae72302a992..5176c2c73ebe 100644 --- a/paimon-python/pypaimon/data/decimal.py +++ b/paimon-python/pypaimon/data/decimal.py @@ -86,8 +86,7 @@ def to_big_decimal(self) -> BigDecimal: """ if self.decimal_val is None: self.decimal_val = ( - BigDecimal(self.long_val) - / BigDecimal(self.POW10[self.scale]) + BigDecimal(self.long_val) / BigDecimal(self.POW10[self.scale]) ) return self.decimal_val From 7a6992ec37285ed4e58c32e2de6f3827373b18b7 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 17:31:51 +0800 Subject: [PATCH 03/12] [python] Introduce 'product' aggregator function. --- .../pypaimon/read/merge_engine_support.py | 3 +- .../read/reader/aggregate/aggregators.py | 168 +++++++++++++++++ .../pypaimon/tests/test_field_aggregators.py | 177 +++++++++++++++++- 3 files changed, 343 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/read/merge_engine_support.py b/paimon-python/pypaimon/read/merge_engine_support.py index cefcee2e6701..0989fb81c075 100644 --- a/paimon-python/pypaimon/read/merge_engine_support.py +++ b/paimon-python/pypaimon/read/merge_engine_support.py @@ -64,6 +64,7 @@ "bool_or", "bool_and", "listagg", "nested_update", + "product", ]) _FIELDS_PREFIX = "fields." _FIELD_SEQUENCE_GROUP_SUFFIX = ".sequence-group" @@ -209,7 +210,7 @@ def check_supported(table) -> None: "built-in aggregators ({}); retract opt-ins " "(aggregation.remove-record-on-delete, " "fields..ignore-retract) " - "and other aggregators (product / listagg / collect / " + "and other aggregators (listagg / collect / " "merge_map* / nested_update* / theta_sketch / " "hll_sketch / roaring_bitmap_*) are not yet supported. " "Open an issue to track support.".format( diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py index 11b182edaf44..a2fe70977fc6 100644 --- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py +++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py @@ -37,6 +37,7 @@ from pypaimon.common.options import CoreOptions from pypaimon.common.options.core_options import NestedKeyNullStrategy +from pypaimon.data.decimal import Decimal from pypaimon.read.reader.aggregate import register_aggregator from pypaimon.read.reader.aggregate.field_aggregator import FieldAggregator from pypaimon.schema.data_types import AtomicType, DataType, ArrayType, RowType @@ -54,6 +55,7 @@ NAME_FIRST_VALUE = "first_value" NAME_FIRST_NON_NULL_VALUE = "first_non_null_value" NAME_SUM = "sum" +NAME_PRODUCT = "product" NAME_MAX = "max" NAME_MIN = "min" NAME_BOOL_OR = "bool_or" @@ -62,6 +64,16 @@ NAME_NESTED_UPDATE = "nested_update" +# Min/Max values for each numeric type +_BYTE_MIN = -128 +_BYTE_MAX = 127 +_SHORT_MIN = -32768 +_SHORT_MAX = 32767 +_INT_MIN = -(1 << 31) +_INT_MAX = (1 << 31) - 1 +_LONG_MIN = -(1 << 63) +_LONG_MAX = (1 << 63) - 1 + # Base SQL type names treated as numeric for sum/product-style # aggregators. NUMERIC / DEC are SQL synonyms accepted by the parser; # treat them the same as DECIMAL. @@ -70,6 +82,10 @@ "FLOAT", "DOUBLE", "DECIMAL", "NUMERIC", "DEC", ]) +_DECIMAL_TYPES = frozenset({"DECIMAL", "NUMERIC", "DEC"}) +_INT_TYPES = frozenset({"INT", "INTEGER"}) +_FLOAT_TYPES = frozenset({"FLOAT", "DOUBLE"}) + def _atomic_base_name(field_type: DataType): """Extract the bare SQL type name from an :class:`AtomicType`, @@ -351,6 +367,157 @@ def agg(self, accumulator: Any, input_field: Any) -> Any: return accumulator + input_field +class FieldProductAgg(FieldAggregator): + """ + Numeric product aggregator. + + Null values are ignored and the non-null operand is returned. + Otherwise, returns the product of accumulator and input value. + """ + def __init__(self, name: str, field_type: DataType): + super().__init__(name, field_type) + self._base_type = _atomic_base_name(field_type) + if self._base_type in _DECIMAL_TYPES: + self._precision, self._scale = Decimal.extract_decimal_precision_scale(field_type.type) + else: + self._precision = None + self._scale = None + + def agg(self, accumulator: Any, input_field: Any) -> Any: + if accumulator is None or input_field is None: + return accumulator if input_field is None else input_field + + if self._base_type in _DECIMAL_TYPES: + merge_field = Decimal.from_big_decimal(accumulator, self._precision, self._scale) + input_field = Decimal.from_big_decimal(input_field, self._precision, self._scale) + + assert merge_field.scale == input_field.scale, "Inconsistent scale of aggregate Decimal!" + assert merge_field.precision == input_field.precision, "Inconsistent precision of aggregate Decimal!" + + big_decimal = merge_field.to_big_decimal() + input_big_decimal = input_field.to_big_decimal() + result = big_decimal * input_big_decimal + + return Decimal.from_big_decimal( + result, + merge_field.precision, + merge_field.scale, + ).to_big_decimal() + + elif self._base_type == "TINYINT": + value = accumulator * input_field + if value < _BYTE_MIN or value > _BYTE_MAX: + raise ArithmeticError( + "byte overflow: {} * {}".format(accumulator, input_field) + ) + return value + + elif self._base_type == "SMALLINT": + value = accumulator * input_field + if value < _SHORT_MIN or value > _SHORT_MAX: + raise ArithmeticError( + "short overflow: {} * {}".format(accumulator, input_field) + ) + return value + + elif self._base_type in _INT_TYPES: + value = accumulator * input_field + if value < _INT_MIN or value > _INT_MAX: + raise ArithmeticError( + "int overflow: {} * {}".format(accumulator, input_field) + ) + return value + + elif self._base_type == "BIGINT": + value = accumulator * input_field + if value < _LONG_MIN or value > _LONG_MAX: + raise ArithmeticError( + "long overflow: {} * {}".format(accumulator, input_field) + ) + return value + + elif self._base_type in _FLOAT_TYPES: + return accumulator * input_field + + raise ValueError( + "type {} not support in {}".format(self._base_type, self.__class__.__name__) + ) + + def retract(self, accumulator: Any, input_field: Any) -> Any: + if accumulator is None or input_field is None: + return accumulator + + if self._base_type in _DECIMAL_TYPES: + merge_field = Decimal.from_big_decimal(accumulator, self._precision, self._scale) + input_field = Decimal.from_big_decimal(input_field, self._precision, self._scale) + + assert merge_field.scale == input_field.scale, "Inconsistent scale of aggregate Decimal!" + assert merge_field.precision == input_field.precision, "Inconsistent precision of aggregate Decimal!" + + big_decimal = merge_field.to_big_decimal() + input_big_decimal = input_field.to_big_decimal() + result = big_decimal / input_big_decimal + + return Decimal.from_big_decimal( + result, + merge_field.precision, + merge_field.scale, + ).to_big_decimal() + + elif self._base_type == "TINYINT": + value = int(accumulator / input_field) + if value > _BYTE_MAX or value < _BYTE_MIN: + raise ArithmeticError( + "byte overflow: {} / {} = {}".format( + accumulator, input_field, value + ) + ) + return value + + elif self._base_type == "SMALLINT": + value = int(accumulator / input_field) + if value > _SHORT_MAX or value < _SHORT_MIN: + raise ArithmeticError( + "short overflow: {} / {} = {}".format( + accumulator, input_field, value + ) + ) + return value + + elif self._base_type in _INT_TYPES: + if accumulator == _INT_MIN and input_field == -1: + raise ArithmeticError( + "int overflow: {} / {}".format( + accumulator, input_field + ) + ) + return int(accumulator / input_field) + + elif self._base_type == "BIGINT": + if accumulator == _LONG_MIN and input_field == -1: + raise ArithmeticError( + "long overflow: {} / {}".format( + accumulator, input_field + ) + ) + + # Java integer division truncates toward zero, while Python's "//" + # floors toward negative infinity. Divide absolute values first, then + # restore the sign to match Java semantics without converting through + # float (which would lose precision for BIGINT values). + if (accumulator >= 0) == (input_field >= 0): + return abs(accumulator) // abs(input_field) + else: + return -(abs(accumulator) // abs(input_field)) + + elif self._base_type in _FLOAT_TYPES: + return accumulator / input_field + + raise ValueError( + "type {} not support in {}".format(self._base_type, self.__class__.__name__) + ) + + class FieldMaxAgg(FieldAggregator): """Maximum value. ``None`` on either side returns the non-null operand. Uses Python's native ``<`` so any orderable type @@ -726,6 +893,7 @@ def _factory(field_type, field_name, options): _build_no_type_check(FieldFirstNonNullValueAgg, NAME_FIRST_NON_NULL_VALUE), ) register_aggregator(NAME_SUM, _build_numeric(FieldSumAgg, NAME_SUM)) +register_aggregator(NAME_PRODUCT, _build_numeric(FieldProductAgg, NAME_PRODUCT)) register_aggregator(NAME_MAX, _build_no_type_check(FieldMaxAgg, NAME_MAX)) register_aggregator(NAME_MIN, _build_no_type_check(FieldMinAgg, NAME_MIN)) register_aggregator( diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py b/paimon-python/pypaimon/tests/test_field_aggregators.py index 9c9d5876fe9f..ca0245a457cc 100644 --- a/paimon-python/pypaimon/tests/test_field_aggregators.py +++ b/paimon-python/pypaimon/tests/test_field_aggregators.py @@ -26,12 +26,12 @@ import datetime import unittest -from decimal import Decimal +from decimal import Decimal as BigDecimal from functools import reduce from typing import List from pypaimon.common.options import CoreOptions, Options -from pypaimon.data import Timestamp +from pypaimon.data import Timestamp, Decimal from pypaimon.read.reader.aggregate import create_field_aggregator from pypaimon.read.reader.aggregate.aggregators import ( FieldBoolAndAgg, @@ -46,6 +46,7 @@ FieldSumAgg, FieldListaggAgg, FieldNestedUpdateAgg, + FieldProductAgg, ) from pypaimon.schema.data_types import AtomicType, DataField, RowType, ArrayType from pypaimon.table.row.generic_row import GenericRow @@ -151,8 +152,8 @@ def test_float_sum(self): def test_decimal_sum(self): agg = _make("sum", "DECIMAL(10,2)") - result = agg.agg(Decimal("1.23"), Decimal("4.56")) - self.assertEqual(result, Decimal("5.79")) + result = agg.agg(BigDecimal("1.23"), BigDecimal("4.56")) + self.assertEqual(result, BigDecimal("5.79")) def test_null_inputs_return_non_null_operand(self): agg = _make("sum", "INT") @@ -166,6 +167,174 @@ def test_non_numeric_type_rejected_at_construction(self): self.assertIn("numeric", str(ctx.exception)) +class FieldProductAggTest(unittest.TestCase): + + @staticmethod + def to_decimal(value, precision: int = 10, scale: int = 0) -> "BigDecimal": + return Decimal.from_big_decimal(BigDecimal(str(value)), precision, scale).to_big_decimal() + + def test_int(self): + agg = _make("product", "INT") + self.assertIsInstance(agg, FieldProductAgg) + + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 10) + self.assertEqual(agg.retract(10, 5), 2) + self.assertIsNone(agg.retract(None, 5)) + + def test_byte(self): + agg = _make("product", "TINYINT") + + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 10) + self.assertEqual(agg.retract(10, 5), 2) + self.assertIsNone(agg.retract(None, 5)) + + def test_short(self): + agg = _make("product", "SMALLINT") + + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 10) + self.assertEqual(agg.retract(10, 5), 2) + self.assertIsNone(agg.retract(None, 5)) + + def test_long(self): + agg = _make("product", "BIGINT") + + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 10) + self.assertEqual(agg.retract(10, 5), 2) + self.assertIsNone(agg.retract(None, 5)) + + def test_float(self): + agg = _make("product", "FLOAT") + + self.assertEqual(agg.agg(None, 10.0), 10.0) + self.assertEqual(agg.agg(1.0, 10.0), 10.0) + self.assertEqual(agg.retract(10.0, 5.0), 2.0) + self.assertIsNone(agg.retract(None, 5.0)) + + def test_double(self): + agg = _make("product", "DOUBLE") + + self.assertEqual(agg.agg(None, 10.0), 10.0) + self.assertEqual(agg.agg(1.0, 10.0), 10.0) + self.assertEqual(agg.retract(10.0, 5.0), 2.0) + self.assertIsNone(agg.retract(None, 5.0)) + + def test_decimal_no_precision_scale(self): + agg = _make("product", "DECIMAL") + + self.assertEqual(agg.agg(None, self.to_decimal(10)), self.to_decimal(10)) + self.assertEqual(agg.agg(self.to_decimal(1), self.to_decimal(10)), self.to_decimal(10)) + self.assertEqual(agg.agg(self.to_decimal(1.3), self.to_decimal(10)), self.to_decimal(10)) + self.assertEqual(agg.agg(self.to_decimal(1.5), self.to_decimal(10)), self.to_decimal(20)) + self.assertEqual(agg.retract(self.to_decimal(10), self.to_decimal(5)), self.to_decimal(2)) + self.assertIsNone(agg.retract(None, self.to_decimal(5))) + + def test_decimal(self): + agg = _make("product", "DECIMAL(8,2)") + + self.assertEqual(agg.agg(BigDecimal("1.50"), BigDecimal("2.00")), BigDecimal("3.00")) + self.assertEqual(agg.agg(BigDecimal("1.50"), BigDecimal("2.01")), BigDecimal("3.02")) + self.assertEqual(agg.retract(BigDecimal("3.00"), BigDecimal("2.00")), BigDecimal("1.50")) + self.assertEqual(agg.agg(None, self.to_decimal(10.15)), self.to_decimal(10.15)) + self.assertIsNone(agg.retract(None, self.to_decimal(5.02))) + + def test_numeric(self): + agg = _make("product", "NUMERIC(12,2)") + + self.assertEqual(agg.agg(None, self.to_decimal(10, 12, 2)), self.to_decimal(10, 12, 2)) + self.assertEqual(agg.agg(self.to_decimal(1), self.to_decimal(10, 12, 2)), self.to_decimal(10, 12, 2)) + self.assertEqual(agg.retract(self.to_decimal(10), self.to_decimal(5, 12, 2)), self.to_decimal(2, 12, 2)) + self.assertIsNone(agg.retract(None, self.to_decimal(5, 12, 2))) + + def test_dec(self): + agg = _make("product", "DEC(12)") + + self.assertEqual(agg.agg(None, self.to_decimal(10, 12)), self.to_decimal(10, 12)) + self.assertEqual(agg.agg(self.to_decimal(1), self.to_decimal(10, 12)), self.to_decimal(10, 12)) + self.assertEqual(agg.retract(self.to_decimal(10), self.to_decimal(5, 12)), self.to_decimal(2, 12)) + self.assertIsNone(agg.retract(None, self.to_decimal(5, 12))) + + def test_byte_overflow(self): + agg = _make("product", "TINYINT") + + with self.assertRaises(ArithmeticError): + agg.agg(64, 2) + + with self.assertRaises(ArithmeticError): + agg.agg(-64, 4) + + def test_short_overflow(self): + agg = _make("product", "SMALLINT") + + with self.assertRaises(ArithmeticError): + agg.agg(1000, 100) + + with self.assertRaises(ArithmeticError): + agg.agg(-32768, 2) + + def test_int_overflow(self): + agg = _make("product", "INT") + + with self.assertRaises(ArithmeticError): + agg.agg(100000, 100000) + + with self.assertRaises(ArithmeticError): + agg.agg(-2147483648, -1) + + def test_long_overflow(self): + agg = _make("product", "BIGINT") + + with self.assertRaises(ArithmeticError): + agg.agg(9223372036854775807, 2) + + with self.assertRaises(ArithmeticError): + agg.agg(-9223372036854775808, -1) + + def test_byte_retract_overflow(self): + agg = _make("product", "TINYINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-128, -1) + + def test_short_retract_overflow(self): + agg = _make("product", "SMALLINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-32768, -1) + + def test_int_retract_overflow(self): + agg = _make("product", "INT") + + with self.assertRaises(ArithmeticError): + agg.retract(-2147483648, -1) + + def test_long_retract_overflow(self): + agg = _make("product", "BIGINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-9223372036854775808, -1) + + def test_null_inputs(self): + agg = _make("product", "INT") + + self.assertEqual(agg.agg(None, 5), 5) + self.assertEqual(agg.agg(5, None), 5) + self.assertIsNone(agg.agg(None, None)) + + self.assertEqual(agg.retract(5, None), 5) + self.assertIsNone(agg.retract(None, 5)) + self.assertIsNone(agg.retract(None, None)) + + def test_non_numeric_type_rejected_at_construction(self): + with self.assertRaises(ValueError) as ctx: + _make("product", "VARCHAR") + + self.assertIn("numeric", str(ctx.exception)) + + class FieldMaxAggTest(unittest.TestCase): def test_numeric_max(self): From 841ece3684067b530729f2984d57b5c868c0c5ee Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 18:13:05 +0800 Subject: [PATCH 04/12] [python] Introduce 'product' aggregator function. --- .../read/reader/aggregate/aggregators.py | 20 ++++++------------- 1 file changed, 6 insertions(+), 14 deletions(-) diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py index a2fe70977fc6..317c902ed08b 100644 --- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py +++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py @@ -408,7 +408,7 @@ def agg(self, accumulator: Any, input_field: Any) -> Any: value = accumulator * input_field if value < _BYTE_MIN or value > _BYTE_MAX: raise ArithmeticError( - "byte overflow: {} * {}".format(accumulator, input_field) + "byte overflow: {} * {} = {}".format(accumulator, input_field, value) ) return value @@ -416,7 +416,7 @@ def agg(self, accumulator: Any, input_field: Any) -> Any: value = accumulator * input_field if value < _SHORT_MIN or value > _SHORT_MAX: raise ArithmeticError( - "short overflow: {} * {}".format(accumulator, input_field) + "short overflow: {} * {} = {}".format(accumulator, input_field, value) ) return value @@ -468,9 +468,7 @@ def retract(self, accumulator: Any, input_field: Any) -> Any: value = int(accumulator / input_field) if value > _BYTE_MAX or value < _BYTE_MIN: raise ArithmeticError( - "byte overflow: {} / {} = {}".format( - accumulator, input_field, value - ) + "byte overflow: {} / {} = {}".format(accumulator, input_field, value) ) return value @@ -478,27 +476,21 @@ def retract(self, accumulator: Any, input_field: Any) -> Any: value = int(accumulator / input_field) if value > _SHORT_MAX or value < _SHORT_MIN: raise ArithmeticError( - "short overflow: {} / {} = {}".format( - accumulator, input_field, value - ) + "short overflow: {} / {} = {}".format(accumulator, input_field, value) ) return value elif self._base_type in _INT_TYPES: if accumulator == _INT_MIN and input_field == -1: raise ArithmeticError( - "int overflow: {} / {}".format( - accumulator, input_field - ) + "int overflow: {} / {}".format(accumulator, input_field) ) return int(accumulator / input_field) elif self._base_type == "BIGINT": if accumulator == _LONG_MIN and input_field == -1: raise ArithmeticError( - "long overflow: {} / {}".format( - accumulator, input_field - ) + "long overflow: {} / {}".format(accumulator, input_field) ) # Java integer division truncates toward zero, while Python's "//" From 0e4c70565b00362ce950f6ba0c6ac7db58934dbb Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 19:20:20 +0800 Subject: [PATCH 05/12] [python] Fix Python 3.6 compatibility for Decimal. --- paimon-python/pypaimon/data/decimal.py | 31 +++++++++++++------------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py index 5176c2c73ebe..0609a70263c5 100644 --- a/paimon-python/pypaimon/data/decimal.py +++ b/paimon-python/pypaimon/data/decimal.py @@ -1,21 +1,20 @@ -# 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 +################################################################################ +# 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 +# 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. - -from __future__ import annotations +# 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. +################################################################################ import re from decimal import Decimal as BigDecimal From dcb9022a6fbef0796319c0995dd96acb76434b2a Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 20:18:18 +0800 Subject: [PATCH 06/12] [python] adjust test_aggregation_e2e::test_out_of_scope_default_aggregator_rejected default aggregator to unsupported `collect`. --- paimon-python/pypaimon/tests/test_aggregation_e2e.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py b/paimon-python/pypaimon/tests/test_aggregation_e2e.py index 0d557799e9c2..d32e39cc91b3 100644 --- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py +++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py @@ -307,7 +307,7 @@ def test_out_of_scope_field_aggregator_rejected(self): def test_out_of_scope_default_aggregator_rejected(self): self._create_and_expect_unsupported( 'agg_reject_default_collect', - {'fields.default-aggregate-function': 'product'}, + {'fields.default-aggregate-function': 'collect'}, 'fields.default-aggregate-function', ) From c158c82d18a61f8de6e8cbb3a57b4da436200156 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 22:10:19 +0800 Subject: [PATCH 07/12] [python] adjust decimal type calc logic. --- .../read/reader/aggregate/aggregators.py | 50 +++++++------------ 1 file changed, 19 insertions(+), 31 deletions(-) diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py index 317c902ed08b..77eda391edbe 100644 --- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py +++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py @@ -64,7 +64,7 @@ NAME_NESTED_UPDATE = "nested_update" -# Min/Max values for each numeric type +# Integer range limits used for overflow checking. _BYTE_MIN = -128 _BYTE_MAX = 127 _SHORT_MIN = -32768 @@ -82,8 +82,14 @@ "FLOAT", "DOUBLE", "DECIMAL", "NUMERIC", "DEC", ]) +# SQL type names treated as decimal. NUMERIC / DEC are SQL +# synonyms accepted by the parser; treat them the same as DECIMAL. _DECIMAL_TYPES = frozenset({"DECIMAL", "NUMERIC", "DEC"}) + +# SQL type names treated as integer. _INT_TYPES = frozenset({"INT", "INTEGER"}) + +# SQL type names treated as floating-point. _FLOAT_TYPES = frozenset({"FLOAT", "DOUBLE"}) @@ -388,21 +394,12 @@ def agg(self, accumulator: Any, input_field: Any) -> Any: return accumulator if input_field is None else input_field if self._base_type in _DECIMAL_TYPES: - merge_field = Decimal.from_big_decimal(accumulator, self._precision, self._scale) - input_field = Decimal.from_big_decimal(input_field, self._precision, self._scale) - - assert merge_field.scale == input_field.scale, "Inconsistent scale of aggregate Decimal!" - assert merge_field.precision == input_field.precision, "Inconsistent precision of aggregate Decimal!" - - big_decimal = merge_field.to_big_decimal() - input_big_decimal = input_field.to_big_decimal() - result = big_decimal * input_big_decimal - - return Decimal.from_big_decimal( - result, - merge_field.precision, - merge_field.scale, - ).to_big_decimal() + value = Decimal.from_big_decimal( + accumulator * input_field, + self._precision, + self._scale, + ) + return None if value is None else value.to_big_decimal() elif self._base_type == "TINYINT": value = accumulator * input_field @@ -448,21 +445,12 @@ def retract(self, accumulator: Any, input_field: Any) -> Any: return accumulator if self._base_type in _DECIMAL_TYPES: - merge_field = Decimal.from_big_decimal(accumulator, self._precision, self._scale) - input_field = Decimal.from_big_decimal(input_field, self._precision, self._scale) - - assert merge_field.scale == input_field.scale, "Inconsistent scale of aggregate Decimal!" - assert merge_field.precision == input_field.precision, "Inconsistent precision of aggregate Decimal!" - - big_decimal = merge_field.to_big_decimal() - input_big_decimal = input_field.to_big_decimal() - result = big_decimal / input_big_decimal - - return Decimal.from_big_decimal( - result, - merge_field.precision, - merge_field.scale, - ).to_big_decimal() + value = Decimal.from_big_decimal( + accumulator / input_field, + self._precision, + self._scale, + ) + return None if value is None else value.to_big_decimal() elif self._base_type == "TINYINT": value = int(accumulator / input_field) From cfbb196356711661d2bb231c4cd7694c610d84c2 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Sat, 18 Jul 2026 22:27:35 +0800 Subject: [PATCH 08/12] [python] Align FieldSumAgg with Java implementation. --- .../read/reader/aggregate/aggregators.py | 156 +++++++++++++++++- .../pypaimon/tests/test_field_aggregators.py | 154 +++++++++++++++-- 2 files changed, 289 insertions(+), 21 deletions(-) diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py index 77eda391edbe..829f1f74efe8 100644 --- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py +++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py @@ -360,17 +360,161 @@ def reset(self) -> None: class FieldSumAgg(FieldAggregator): - """Numeric sum. ``None`` on either side returns the non-null - operand. Python's native ``+`` works uniformly for int / float / - Decimal — the values produced by the pyarrow read path already - arrive as the right Python primitive for the column's SQL type, so - no per-type branching is needed. """ + Numeric sum aggregator. + + Returns the non-null operand if either side is ``None``. Performs + overflow checking for integral types and preserves decimal + precision and scale for DECIMAL values. + """ + def __init__(self, name: str, field_type: DataType): + super().__init__(name, field_type) + self._base_type = _atomic_base_name(field_type) + if self._base_type in _DECIMAL_TYPES: + self._precision, self._scale = Decimal.extract_decimal_precision_scale(field_type.type) + else: + self._precision = None + self._scale = None def agg(self, accumulator: Any, input_field: Any) -> Any: if accumulator is None or input_field is None: return accumulator if input_field is None else input_field - return accumulator + input_field + + if self._base_type in _DECIMAL_TYPES: + value = Decimal.from_big_decimal( + accumulator + input_field, + self._precision, + self._scale + ) + return None if value is None else value.to_big_decimal() + + elif self._base_type == "TINYINT": + value = accumulator + input_field + if value < _BYTE_MIN or value > _BYTE_MAX: + raise ArithmeticError( + "byte overflow: {} + {} = {}".format(accumulator, input_field, value) + ) + return value + + elif self._base_type == "SMALLINT": + value = accumulator + input_field + if value < _SHORT_MIN or value > _SHORT_MAX: + raise ArithmeticError( + "short overflow: {} + {} = {}".format(accumulator, input_field, value) + ) + return value + + elif self._base_type in _INT_TYPES: + value = accumulator + input_field + if value < _INT_MIN or value > _INT_MAX: + raise ArithmeticError( + "int overflow: {} + {}".format(accumulator, input_field) + ) + return value + + elif self._base_type == "BIGINT": + value = accumulator + input_field + if value < _LONG_MIN or value > _LONG_MAX: + raise ArithmeticError( + "long overflow: {} + {}".format(accumulator, input_field) + ) + return value + + elif self._base_type in _FLOAT_TYPES: + return accumulator + input_field + + raise ValueError( + "type {} not support in {}".format(self._base_type, self.__class__.__name__) + ) + + def retract(self, accumulator: Any, retract_field: Any) -> Any: + if accumulator is None or retract_field is None: + return self._negative(retract_field) if accumulator is None else accumulator + + if self._base_type in _DECIMAL_TYPES: + value = Decimal.from_big_decimal( + accumulator - retract_field, + self._precision, + self._scale, + ) + return None if value is None else value.to_big_decimal() + + elif self._base_type == "TINYINT": + value = accumulator - retract_field + if value < _BYTE_MIN or value > _BYTE_MAX: + raise ArithmeticError( + "byte overflow: {} - {} = {}".format(accumulator, retract_field, value) + ) + return value + + elif self._base_type == "SMALLINT": + value = accumulator - retract_field + if value < _SHORT_MIN or value > _SHORT_MAX: + raise ArithmeticError( + "short overflow: {} - {} = {}".format(accumulator, retract_field, value) + ) + return value + + elif self._base_type in _INT_TYPES: + value = accumulator - retract_field + if value < _INT_MIN or value > _INT_MAX: + raise ArithmeticError( + "int overflow: {} - {}".format(accumulator, retract_field) + ) + return value + + elif self._base_type == "BIGINT": + value = accumulator - retract_field + if value < _LONG_MIN or value > _LONG_MAX: + raise ArithmeticError( + "long overflow: {} - {}".format(accumulator, retract_field) + ) + return value + + elif self._base_type in _FLOAT_TYPES: + return accumulator - retract_field + + raise ValueError( + "type {} not support in {}".format(self._base_type, self.__class__.__name__) + ) + + def _negative(self, value: Any) -> Any: + if value is None: + return None + + if self._base_type in _DECIMAL_TYPES: + return -value + + elif self._base_type == "TINYINT": + result = -value + if result < _BYTE_MIN or result > _BYTE_MAX: + raise ArithmeticError("byte overflow: -{} = {}".format(value, result)) + return result + + elif self._base_type == "SMALLINT": + result = -value + if result < _SHORT_MIN or result > _SHORT_MAX: + raise ArithmeticError("short overflow: -{} = {}".format(value, result)) + return result + + elif self._base_type in _INT_TYPES: + result = -value + if result < _INT_MIN or result > _INT_MAX: + raise ArithmeticError("int overflow: -{}".format(value)) + return result + + elif self._base_type == "BIGINT": + result = -value + if result < _LONG_MIN or result > _LONG_MAX: + raise ArithmeticError("long overflow: -{}".format(value)) + return result + + elif self._base_type in _FLOAT_TYPES: + return -value + + raise ValueError( + "type {} not support in {}".format(self._base_type, self.__class__.__name__) + ) class FieldProductAgg(FieldAggregator): diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py b/paimon-python/pypaimon/tests/test_field_aggregators.py index ca0245a457cc..464d13a875ef 100644 --- a/paimon-python/pypaimon/tests/test_field_aggregators.py +++ b/paimon-python/pypaimon/tests/test_field_aggregators.py @@ -140,21 +140,6 @@ def test_reset_re_arms_first_non_null(self): class FieldSumAggTest(unittest.TestCase): - def test_int_sum(self): - agg = _make("sum", "BIGINT") - self.assertIsInstance(agg, FieldSumAgg) - self.assertEqual(agg.agg(None, 5), 5) - self.assertEqual(agg.agg(5, 7), 12) - - def test_float_sum(self): - agg = _make("sum", "DOUBLE") - self.assertAlmostEqual(agg.agg(1.5, 2.25), 3.75) - - def test_decimal_sum(self): - agg = _make("sum", "DECIMAL(10,2)") - result = agg.agg(BigDecimal("1.23"), BigDecimal("4.56")) - self.assertEqual(result, BigDecimal("5.79")) - def test_null_inputs_return_non_null_operand(self): agg = _make("sum", "INT") self.assertEqual(agg.agg(None, 5), 5) @@ -166,6 +151,145 @@ def test_non_numeric_type_rejected_at_construction(self): _make("sum", "VARCHAR") self.assertIn("numeric", str(ctx.exception)) + def test_int_sum(self): + agg = _make("sum", "INT") + self.assertIsInstance(agg, FieldSumAgg) + + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 11) + self.assertEqual(agg.retract(10, 5), 5) + self.assertEqual(agg.retract(None, 5), -5) + + def test_byte_sum(self): + agg = _make("sum", "TINYINT") + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 11) + self.assertEqual(agg.retract(10, 5), 5) + self.assertEqual(agg.retract(None, 5), -5) + + def test_short_sum(self): + agg = _make("sum", "SMALLINT") + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 11) + self.assertEqual(agg.retract(10, 5), 5) + self.assertEqual(agg.retract(None, 5), -5) + + def test_long_sum(self): + agg = _make("sum", "BIGINT") + self.assertEqual(agg.agg(None, 10), 10) + self.assertEqual(agg.agg(1, 10), 11) + self.assertEqual(agg.retract(10, 5), 5) + self.assertEqual(agg.retract(None, 5), -5) + + def test_byte_overflow(self): + agg = _make("sum", "TINYINT") + + with self.assertRaises(ArithmeticError): + agg.agg(127, 1) + + with self.assertRaises(ArithmeticError): + agg.agg(-128, -1) + + def test_short_overflow(self): + agg = _make("sum", "SMALLINT") + + with self.assertRaises(ArithmeticError): + agg.agg(32767, 1) + + with self.assertRaises(ArithmeticError): + agg.agg(-32768, -1) + + def test_int_overflow(self): + agg = _make("sum", "INT") + + with self.assertRaises(ArithmeticError): + agg.agg(2147483647, 1) + + with self.assertRaises(ArithmeticError): + agg.agg(-2147483648, -1) + + def test_long_overflow(self): + agg = _make("sum", "BIGINT") + + with self.assertRaises(ArithmeticError): + agg.agg(9223372036854775807, 1) + + with self.assertRaises(ArithmeticError): + agg.agg(-9223372036854775808, -1) + + def test_byte_retract_overflow(self): + agg = _make("sum", "TINYINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-128, 1) + + with self.assertRaises(ArithmeticError): + agg.retract(127, -1) + + with self.assertRaises(ArithmeticError): + agg.retract(None, -128) + + def test_short_retract_overflow(self): + agg = _make("sum", "SMALLINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-32768, 1) + + with self.assertRaises(ArithmeticError): + agg.retract(32767, -1) + + with self.assertRaises(ArithmeticError): + agg.retract(None, -32768) + + def test_int_retract_overflow(self): + agg = _make("sum", "INT") + + with self.assertRaises(ArithmeticError): + agg.retract(-2147483648, 1) + + with self.assertRaises(ArithmeticError): + agg.retract(2147483647, -1) + + with self.assertRaises(ArithmeticError): + agg.retract(None, -2147483648) + + def test_long_retract_overflow(self): + agg = _make("sum", "BIGINT") + + with self.assertRaises(ArithmeticError): + agg.retract(-9223372036854775808, 1) + + with self.assertRaises(ArithmeticError): + agg.retract(9223372036854775807, -1) + + with self.assertRaises(ArithmeticError): + agg.retract(None, -9223372036854775808) + + def test_float_sum(self): + agg = _make("sum", "FLOAT") + + self.assertEqual(agg.agg(None, 10.0), 10.0) + self.assertEqual(agg.agg(1.0, 10.0), 11.0) + self.assertEqual(agg.retract(10.0, 5.0), 5.0) + self.assertEqual(agg.retract(None, 5.0), -5.0) + + def test_double_sum(self): + agg = _make("sum", "DOUBLE") + + self.assertEqual(agg.agg(1.5, 2.25), 3.75) + self.assertEqual(agg.agg(None, 10.0), 10.0) + self.assertEqual(agg.agg(1.0, 10.0), 11.0) + self.assertEqual(agg.retract(10.0, 5.0), 5.0) + self.assertEqual(agg.retract(None, 5.0), -5.0) + + def test_decimal_sum(self): + agg = _make("sum", "DECIMAL(10,2)") + + self.assertEqual(agg.agg(None, BigDecimal("10.00")), BigDecimal("10.00")) + self.assertEqual(agg.agg(BigDecimal("1.23"), BigDecimal("4.56")), BigDecimal("5.79")) + self.assertEqual(agg.retract(BigDecimal("10.00"), BigDecimal("5.00")), BigDecimal("5.00")) + self.assertEqual(agg.retract(None, BigDecimal("5.00")), BigDecimal("-5.00")) + class FieldProductAggTest(unittest.TestCase): From b23530e685d7f621517937a20b876d803bf2c407 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Mon, 20 Jul 2026 21:39:54 +0800 Subject: [PATCH 09/12] [python] Align Decimal implementation with Java semantics. --- paimon-python/pypaimon/data/decimal.py | 47 ++++++++++++++++++-------- 1 file changed, 33 insertions(+), 14 deletions(-) diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py index 0609a70263c5..126c2ae05d5c 100644 --- a/paimon-python/pypaimon/data/decimal.py +++ b/paimon-python/pypaimon/data/decimal.py @@ -19,6 +19,7 @@ import re from decimal import Decimal as BigDecimal from decimal import ROUND_HALF_UP +from decimal import localcontext from typing import Optional, Tuple _DECIMAL_PATTERN = re.compile( @@ -84,9 +85,7 @@ def to_big_decimal(self) -> BigDecimal: Converts this Decimal into a decimal.Decimal instance. """ if self.decimal_val is None: - self.decimal_val = ( - BigDecimal(self.long_val) / BigDecimal(self.POW10[self.scale]) - ) + self.decimal_val = BigDecimal(self.long_val).scaleb(-self.scale) return self.decimal_val def to_unscaled_long(self) -> int: @@ -100,20 +99,29 @@ def to_unscaled_long(self) -> int: if self.is_compact(): return self.long_val - value = self.to_big_decimal() * self.POW10[self.scale] + value = self.to_big_decimal().scaleb(self.scale) if value != value.to_integral_exact(): raise ArithmeticError("Decimal does not exactly fit in long.") - return int(value) + int_val = int(value) + if not (-(1 << 63) <= int_val <= (1 << 63) - 1): + raise ArithmeticError("BigInteger out of long range") + + return int_val def to_unscaled_bytes(self) -> bytes: """ Returns the unscaled value encoded as a signed byte array. """ - value = self.to_unscaled_long() - length = max(1, (value.bit_length() + 8) // 8) - return value.to_bytes(length, byteorder="big", signed=True) + value = self.to_big_decimal().scaleb(self.scale) + + if value != value.to_integral_exact(): + raise ArithmeticError("Decimal does not exactly fit in long.") + + unscaled = int(value) + length = max(1, (unscaled.bit_length() + 8) // 8) + return unscaled.to_bytes(length, byteorder="big", signed=True) def is_compact(self) -> bool: """ @@ -217,14 +225,25 @@ def from_big_decimal( The value is rounded to the requested scale using ROUND_HALF_UP. If the resulting precision exceeds the specified precision, None is returned. + + Note: + Java ``BigDecimal`` provides arbitrary precision. Paimon supports + DECIMAL precision up to 38 digits, so a temporary context with + precision >= 38 is used here to match Java semantics without + modifying the global decimal context. """ quant = BigDecimal(1).scaleb(-scale) - value = value.quantize( - quant, - rounding=ROUND_HALF_UP, - ) + with localcontext() as ctx: + # Java BigDecimal is arbitrary precision. + # Paimon DECIMAL supports precision up to 38. + ctx.prec = max(precision, 38) + + value = value.quantize( + quant, + rounding=ROUND_HALF_UP, + ) digits = len(value.as_tuple().digits) @@ -234,7 +253,7 @@ def from_big_decimal( long_val = -1 if precision <= cls.MAX_COMPACT_PRECISION: - unscaled = value * cls.POW10[scale] + unscaled = value.scaleb(scale) if unscaled != unscaled.to_integral_exact(): raise ArithmeticError( @@ -290,7 +309,7 @@ def from_unscaled_bytes( signed=True, ) - bd = BigDecimal(value) / BigDecimal(cls.POW10[scale]) + bd = BigDecimal(value).scaleb(-scale) return cls.from_big_decimal( bd, From 380912b9cdbfa09c07e0a80ce7c181859085b6a9 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Tue, 21 Jul 2026 09:14:50 +0800 Subject: [PATCH 10/12] [python] fix conflicts. --- paimon-python/pypaimon/tests/test_field_aggregators.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py b/paimon-python/pypaimon/tests/test_field_aggregators.py index f25287589687..82e27afda1f2 100644 --- a/paimon-python/pypaimon/tests/test_field_aggregators.py +++ b/paimon-python/pypaimon/tests/test_field_aggregators.py @@ -44,13 +44,10 @@ FieldMinAgg, FieldPrimaryKeyAgg, FieldSumAgg, + FieldProductAgg, FieldListaggAgg, FieldNestedUpdateAgg, -<<<<<<< HEAD - FieldProductAgg, -======= FieldCollectAgg, ->>>>>>> origin/master ) from pypaimon.schema.data_types import AtomicType, DataField, RowType, ArrayType, MapType from pypaimon.table.row.generic_row import GenericRow From a830ac4b363d9f38eb739ebb25f9a8998a2c2622 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Tue, 21 Jul 2026 17:38:12 +0800 Subject: [PATCH 11/12] [python] Fix Decimal.from_big_decimal precision handling --- paimon-python/pypaimon/data/decimal.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/data/decimal.py b/paimon-python/pypaimon/data/decimal.py index 126c2ae05d5c..05a5c3da0cc2 100644 --- a/paimon-python/pypaimon/data/decimal.py +++ b/paimon-python/pypaimon/data/decimal.py @@ -238,7 +238,11 @@ def from_big_decimal( with localcontext() as ctx: # Java BigDecimal is arbitrary precision. # Paimon DECIMAL supports precision up to 38. - ctx.prec = max(precision, 38) + ctx.prec = max( + precision + scale, + len(value.as_tuple().digits) + abs(value.as_tuple().exponent) + scale, + 38, + ) value = value.quantize( quant, From f04117849d2a7cf53fd502745e18fedc67cc44d9 Mon Sep 17 00:00:00 2001 From: AuroraVoyage Date: Tue, 21 Jul 2026 17:55:48 +0800 Subject: [PATCH 12/12] [python] Fix FieldSumAgg decimal precision handling --- .../read/reader/aggregate/aggregators.py | 26 ++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py index 77c6c34f424d..a02b9925f4d3 100644 --- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py +++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py @@ -32,7 +32,7 @@ the registry will report them as unsupported so users see a clear error rather than a silent fallback. """ - +from decimal import localcontext from typing import Any, List, Dict, Optional, Tuple, Union from pypaimon.common.options import CoreOptions @@ -383,8 +383,18 @@ def agg(self, accumulator: Any, input_field: Any) -> Any: return accumulator if input_field is None else input_field if self._base_type in _DECIMAL_TYPES: + with localcontext() as ctx: + ctx.prec = max( + len(accumulator.as_tuple().digits), + len(input_field.as_tuple().digits), + self._precision, + 38, + ) + 1 + + result = accumulator + input_field + value = Decimal.from_big_decimal( - accumulator + input_field, + result, self._precision, self._scale ) @@ -434,8 +444,18 @@ def retract(self, accumulator: Any, retract_field: Any) -> Any: return self._negative(retract_field) if accumulator is None else accumulator if self._base_type in _DECIMAL_TYPES: + with localcontext() as ctx: + ctx.prec = max( + len(accumulator.as_tuple().digits), + len(retract_field.as_tuple().digits), + self._precision, + 38, + ) + 1 + + result = accumulator - retract_field + value = Decimal.from_big_decimal( - accumulator - retract_field, + result, self._precision, self._scale, )