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 bce7744d93 [python] Align FieldSumAgg with Java implementation (#8719)
bce7744d93 is described below
commit bce7744d93a2a6002bfd49b97c3aff1f8786b279
Author: AuroraVoyage <[email protected]>
AuthorDate: Sun Jul 26 19:47:45 2026 +0800
[python] Align FieldSumAgg with Java implementation (#8719)
---
paimon-python/pypaimon/data/__init__.py | 3 +-
paimon-python/pypaimon/data/decimal.py | 507 +++++++++++++++++++++
.../pypaimon/read/merge_engine_support.py | 1 +
.../pypaimon/read/reader/aggregate/aggregators.py | 320 ++++++++++++-
paimon-python/pypaimon/tests/test_decimal.py | 255 +++++++++++
.../pypaimon/tests/test_field_aggregators.py | 359 ++++++++++++++-
6 files changed, 1426 insertions(+), 19 deletions(-)
diff --git a/paimon-python/pypaimon/data/__init__.py
b/paimon-python/pypaimon/data/__init__.py
index 6308bf139b..97f36f3d52 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 0000000000..30b5101bb4
--- /dev/null
+++ b/paimon-python/pypaimon/data/decimal.py
@@ -0,0 +1,507 @@
+################################################################################
+# 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 re
+from decimal import Decimal as BigDecimal
+from decimal import Inexact
+from decimal import ROUND_HALF_UP
+from decimal import localcontext
+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 python builtin-in decimal.Decimal
instance.
+ """
+ if self.decimal_val is None:
+ with localcontext() as ctx:
+ ctx.prec = max(self.precision + abs(self.scale), 38)
+ self.decimal_val =
BigDecimal(self.long_val).scaleb(-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
+
+ with localcontext() as ctx:
+ bd = self.to_big_decimal()
+ ctx.prec = max(
+ len(bd.as_tuple().digits) + abs(self.scale),
+ self.precision,
+ 38,
+ )
+ value = bd.scaleb(self.scale)
+
+ if value != value.to_integral_exact():
+ raise ArithmeticError("Decimal does not exactly fit in long.")
+
+ 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.
+ """
+
+ with localcontext() as ctx:
+ bd = self.to_big_decimal()
+ ctx.prec = max(
+ len(bd.as_tuple().digits) + abs(self.scale),
+ self.precision,
+ 38,
+ )
+ value = bd.scaleb(self.scale)
+
+ 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)
+
+ 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,
+ )
+
+ # ----------------------------------------------------------------------
+ # 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.
+
+ 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.
+ """
+
+ if not value.is_finite():
+ raise ArithmeticError("NaN and Infinity are not supported.")
+
+ with localcontext() as ctx:
+ t = value.as_tuple()
+ ctx.prec = max(
+ precision + scale,
+ len(t.digits) + abs(t.exponent) + scale,
+ 38,
+ )
+
+ value = value.quantize(
+ BigDecimal(1).scaleb(-scale),
+ 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.scaleb(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,
+ )
+
+ with localcontext() as ctx:
+ ctx.prec = max(
+ len(str(abs(value))) + abs(scale),
+ precision,
+ 38,
+ )
+ bd = BigDecimal(value).scaleb(-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,
+ )
+
+ # ----------------------------------------------------------------------
+ # Arithmetic Operations
+ # ----------------------------------------------------------------------
+
+ @staticmethod
+ def add(lhs: BigDecimal, rhs: BigDecimal, precision: int) -> BigDecimal:
+ """
+ Returns the sum of two decimal values without losing precision due to
+ Python's default decimal context.
+
+ The operation is performed in a temporary high-precision context to
+ emulate Java BigDecimal arithmetic.
+ """
+ assert isinstance(
+ lhs, BigDecimal
+ ), f"'lhs' must be a decimal.Decimal, got {type(lhs).__name__}"
+
+ assert isinstance(
+ rhs, BigDecimal
+ ), f"'rhs' must be a decimal.Decimal, got {type(rhs).__name__}"
+
+ with localcontext() as ctx:
+ ctx.prec = max(
+ len(lhs.as_tuple().digits),
+ len(rhs.as_tuple().digits),
+ precision,
+ 38,
+ ) + 1
+
+ return lhs + rhs
+
+ @staticmethod
+ def subtract(lhs: BigDecimal, rhs: BigDecimal, precision: int) ->
BigDecimal:
+ """
+ Returns the difference of two decimal values without losing precision
due
+ to Python's default decimal context.
+
+ The operation is performed in a temporary high-precision context to
+ emulate Java BigDecimal arithmetic.
+ """
+ assert isinstance(
+ lhs, BigDecimal
+ ), f"'lhs' must be a decimal.Decimal, got {type(lhs).__name__}"
+
+ assert isinstance(
+ rhs, BigDecimal
+ ), f"'rhs' must be a decimal.Decimal, got {type(rhs).__name__}"
+
+ with localcontext() as ctx:
+ ctx.prec = max(
+ len(lhs.as_tuple().digits),
+ len(rhs.as_tuple().digits),
+ precision,
+ 38,
+ ) + 1
+
+ return lhs - rhs
+
+ @staticmethod
+ def multiply(lhs: BigDecimal, rhs: BigDecimal, precision: int) ->
BigDecimal:
+ """
+ Returns the product of two decimal values without losing precision due
to
+ Python's default decimal context.
+
+ The operation is performed in a temporary high-precision context to
+ emulate Java BigDecimal arithmetic.
+ """
+ assert isinstance(
+ lhs, BigDecimal
+ ), f"'lhs' must be a decimal.Decimal, got {type(lhs).__name__}"
+
+ assert isinstance(
+ rhs, BigDecimal
+ ), f"'rhs' must be a decimal.Decimal, got {type(rhs).__name__}"
+
+ with localcontext() as ctx:
+ ctx.prec = max(
+ len(lhs.as_tuple().digits)
+ + len(rhs.as_tuple().digits),
+ precision,
+ 38,
+ )
+
+ return lhs * rhs
+
+ @staticmethod
+ def divide(lhs: BigDecimal, rhs: BigDecimal, precision: int) -> BigDecimal:
+ """
+ Returns the exact quotient of two decimal values.
+
+ Raises:
+ ArithmeticError:
+ If the division has a non-terminating decimal expansion,
+ matching the behavior of Java BigDecimal.divide(BigDecimal).
+ """
+ assert isinstance(
+ lhs, BigDecimal
+ ), f"'lhs' must be a decimal.Decimal, got {type(lhs).__name__}"
+
+ assert isinstance(
+ rhs, BigDecimal
+ ), f"'rhs' must be a decimal.Decimal, got {type(rhs).__name__}"
+
+ with localcontext() as ctx:
+ ctx.prec = max(precision, 38)
+ ctx.traps[Inexact] = True
+
+ try:
+ return lhs / rhs
+ except Inexact:
+ raise ArithmeticError(
+ "Non-terminating decimal expansion; "
+ "no exact representable decimal result."
+ )
+
+ # ----------------------------------------------------------------------
+ # Utility Methods
+ # ----------------------------------------------------------------------
+
+ @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
+
+ @staticmethod
+ def extract_decimal_precision_scale(type_str: str) -> Tuple[int, int]:
+ """
+ Extracts the precision and scale from a DECIMAL/NUMERIC/DEC type
string.
+
+ Returns: (precision, scale)
+
+ 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)
+
+ @staticmethod
+ def _get_compact_byte_length(val: int) -> int:
+ if val == 0:
+ return 1
+
+ if val > 0:
+ bits = val.bit_length()
+ else:
+ bits = (~val).bit_length()
+
+ return bits // 8 + 1
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index d6b38c20c6..0624bf1e65 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -65,6 +65,7 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"listagg",
"nested_update", "nested_partial_update",
"collect",
+ "product",
"merge_map_with_keytime",
"merge_map",
"theta_sketch",
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 4d06e303f9..0f7c4d4c31 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -32,13 +32,13 @@ the 9 most commonly-used value aggregators: ``primary_key``
/
the registry will report them as unsupported so users see a clear
error rather than a silent fallback.
"""
-
from typing import Any, List, Dict, Optional, Tuple, Union, Set
from _datasketches import compact_theta_sketch, theta_union
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, MapType
@@ -57,6 +57,7 @@ NAME_LAST_NON_NULL_VALUE = "last_non_null_value"
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"
@@ -70,6 +71,16 @@ NAME_MERGE_MAP = "merge_map"
NAME_THETA_SKETCH = "theta_sketch"
+# Integer range limits used for overflow checking.
+_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.
@@ -78,6 +89,16 @@ _NUMERIC_BASE_TYPES = frozenset([
"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"})
+
def _atomic_base_name(field_type: DataType):
"""Extract the bare SQL type name from an :class:`AtomicType`,
@@ -346,17 +367,301 @@ class FieldFirstNonNullValueAgg(FieldAggregator):
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:
+ result = Decimal.add(accumulator, input_field, self._precision)
+
+ value = Decimal.from_big_decimal(
+ result,
+ 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:
+ result = Decimal.subtract(accumulator, retract_field,
self._precision)
+
+ value = Decimal.from_big_decimal(
+ result,
+ 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):
+ """
+ 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:
+ mul = Decimal.multiply(accumulator, input_field, self._precision)
+
+ value = Decimal.from_big_decimal(
+ mul,
+ 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 accumulator
+
+ if self._base_type in _DECIMAL_TYPES:
+ div = Decimal.divide(accumulator, retract_field, self._precision)
+
+ value = Decimal.from_big_decimal(
+ div,
+ self._precision,
+ self._scale,
+ )
+ return None if value is None else value.to_big_decimal()
+
+ elif self._base_type == "TINYINT":
+ value = int(accumulator / retract_field)
+ if value > _BYTE_MAX or value < _BYTE_MIN:
+ raise ArithmeticError(
+ "byte overflow: {} / {} = {}".format(accumulator,
retract_field, value)
+ )
+ return value
+
+ elif self._base_type == "SMALLINT":
+ value = int(accumulator / retract_field)
+ if value > _SHORT_MAX or value < _SHORT_MIN:
+ raise ArithmeticError(
+ "short overflow: {} / {} = {}".format(accumulator,
retract_field, value)
+ )
+ return value
+
+ elif self._base_type in _INT_TYPES:
+ if accumulator == _INT_MIN and retract_field == -1:
+ raise ArithmeticError(
+ "int overflow: {} / {}".format(accumulator, retract_field)
+ )
+ return int(accumulator / retract_field)
+
+ elif self._base_type == "BIGINT":
+ if accumulator == _LONG_MIN and retract_field == -1:
+ raise ArithmeticError(
+ "long overflow: {} / {}".format(accumulator, retract_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) == (retract_field >= 0):
+ return abs(accumulator) // abs(retract_field)
+ else:
+ return -(abs(accumulator) // abs(retract_field))
+
+ elif self._base_type in _FLOAT_TYPES:
+ if retract_field == 0.0:
+ if accumulator == 0.0:
+ return float("nan")
+ elif accumulator > 0:
+ return float("inf")
+ else:
+ return float("-inf")
+ return accumulator / retract_field
+
+ raise ValueError(
+ "type {} not support in {}".format(self._base_type,
self.__class__.__name__)
+ )
class FieldMaxAgg(FieldAggregator):
@@ -1167,6 +1472,7 @@ register_aggregator(
_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_decimal.py
b/paimon-python/pypaimon/tests/test_decimal.py
new file mode 100644
index 0000000000..7853aecc0e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/test_decimal.py
@@ -0,0 +1,255 @@
+# 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")
+
+ def test_high_precision_unscaled_bytes_round_trip(self):
+ value = BigDecimal("12345678901234567890123456789012345678")
+
+ d = Decimal.from_big_decimal(value, precision=38, scale=0)
+
+ self.assertIsNotNone(d)
+
+ encoded = d.to_unscaled_bytes()
+
+ d2 = Decimal.from_unscaled_bytes(
+ encoded,
+ precision=38,
+ scale=0,
+ )
+
+ self.assertEqual(d2, d)
+ self.assertEqual(d2.to_big_decimal(), value)
+
+ def test_high_precision_scaled_round_trip(self):
+ value = BigDecimal("12345678901234567890.123456789012345678")
+
+ d = Decimal.from_big_decimal(
+ value,
+ precision=38,
+ scale=18,
+ )
+
+ self.assertIsNotNone(d)
+
+ d2 = Decimal.from_unscaled_bytes(
+ d.to_unscaled_bytes(),
+ precision=38,
+ scale=18,
+ )
+
+ self.assertEqual(d2.to_big_decimal(), value)
+
+ def test_to_unscaled_bytes_java_compatible(self):
+ cases = [
+ (-32769, "ff7fff"),
+ (-32768, "8000"),
+ (-129, "ff7f"),
+ (-128, "80"),
+ (-127, "81"),
+ (-2, "fe"),
+ (-1, "ff"),
+ (0, "00"),
+ (1, "01"),
+ (127, "7f"),
+ (128, "0080"),
+ (255, "00ff"),
+ (256, "0100"),
+ (32767, "7fff"),
+ (32768, "008000"),
+ ]
+
+ for unscaled, expected in cases:
+ with self.subTest(unscaled=unscaled):
+ d = Decimal.from_unscaled_long(
+ unscaled,
+ precision=18,
+ scale=0,
+ )
+
+ self.assertEqual(
+ d.to_unscaled_bytes().hex(),
+ expected,
+ )
+
+ def test_from_unscaled_bytes_java_compatible(self):
+ cases = [
+ ("80", -128),
+ ("ff7f", -129),
+ ("ff", -1),
+ ("00", 0),
+ ("7f", 127),
+ ("0080", 128),
+ ("00ff", 255),
+ ]
+
+ for encoded, expected in cases:
+ with self.subTest(encoded=encoded):
+ d = Decimal.from_unscaled_bytes(
+ bytes.fromhex(encoded),
+ precision=18,
+ scale=0,
+ )
+
+ self.assertEqual(
+ d.to_unscaled_long(),
+ expected,
+ )
+
+ def test_unscaled_bytes_round_trip(self):
+ values = [
+ -123456789012345678,
+ -1000,
+ -129,
+ -128,
+ -127,
+ -1,
+ 0,
+ 1,
+ 127,
+ 128,
+ 255,
+ 256,
+ 123456789012345678,
+ ]
+
+ for value in values:
+ with self.subTest(value=value):
+ d = Decimal.from_unscaled_long(
+ value,
+ precision=18,
+ scale=0,
+ )
+
+ d2 = Decimal.from_unscaled_bytes(
+ d.to_unscaled_bytes(),
+ precision=18,
+ scale=0,
+ )
+
+ self.assertEqual(d2, d)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index f04da32a11..6d548499ca 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -25,15 +25,16 @@ on real PK tables lives in ``test_aggregation_e2e.py``.
"""
import datetime
+import math
import unittest
-from decimal import Decimal
+from decimal import Decimal as BigDecimal
from functools import reduce
from typing import List
from _datasketches import update_theta_sketch
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 +47,7 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldMinAgg,
FieldPrimaryKeyAgg,
FieldSumAgg,
+ FieldProductAgg,
FieldListaggAgg,
FieldNestedUpdateAgg,
FieldNestedPartialUpdateAgg,
@@ -146,32 +148,367 @@ class FieldFirstNonNullValueAggTest(unittest.TestCase):
class FieldSumAggTest(unittest.TestCase):
+ def test_null_inputs_return_non_null_operand(self):
+ agg = _make("sum", "INT")
+ self.assertEqual(agg.agg(None, 5), 5)
+ self.assertEqual(agg.agg(5, None), 5)
+ self.assertIsNone(agg.agg(None, None))
+
+ def test_non_numeric_type_rejected_at_construction(self):
+ with self.assertRaises(ValueError) as ctx:
+ _make("sum", "VARCHAR")
+ self.assertIn("numeric", str(ctx.exception))
+
def test_int_sum(self):
- agg = _make("sum", "BIGINT")
+ agg = _make("sum", "INT")
self.assertIsInstance(agg, FieldSumAgg)
- self.assertEqual(agg.agg(None, 5), 5)
- self.assertEqual(agg.agg(5, 7), 12)
+
+ 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.assertAlmostEqual(agg.agg(1.5, 2.25), 3.75)
+
+ 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)")
- result = agg.agg(Decimal("1.23"), Decimal("4.56"))
- self.assertEqual(result, Decimal("5.79"))
- def test_null_inputs_return_non_null_operand(self):
- agg = _make("sum", "INT")
+ 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"))
+
+ def test_decimal_high_precision_sum(self):
+ agg = _make("sum", "DECIMAL(38,0)")
+
+ result = agg.agg(
+ BigDecimal("12345678901234567890123456789012345678"),
+ BigDecimal("1"),
+ )
+
+ self.assertEqual(
+ result,
+ BigDecimal("12345678901234567890123456789012345679"),
+ )
+
+
+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("sum", "VARCHAR")
+ _make("product", "VARCHAR")
+
self.assertIn("numeric", str(ctx.exception))
+ def test_decimal_high_precision_product(self):
+ agg = _make("product", "DECIMAL(38,0)")
+
+ left = BigDecimal("1234567890123456789")
+ right = BigDecimal("1000000000000000001")
+
+ result = agg.agg(left, right)
+
+ self.assertEqual(
+ result,
+ BigDecimal("1234567890123456790234567890123456789"),
+ )
+
+ def test_decimal_divide_requires_exact_result(self):
+ agg = _make("product", "DECIMAL(38,0)")
+
+ with self.assertRaises(ArithmeticError):
+ agg.retract(
+ BigDecimal("1"),
+ BigDecimal("3"),
+ )
+
+ def test_float_divide_by_zero(self):
+ agg = _make("product", "FLOAT")
+
+ self.assertTrue(math.isnan(agg.retract(0.0, 0.0)))
+ self.assertTrue(math.isinf(agg.retract(1.0, 0.0)))
+ self.assertEqual(agg.retract(1.0, 0.0), float("inf"))
+ self.assertEqual(agg.retract(-1.0, 0.0), float("-inf"))
+
class FieldMaxAggTest(unittest.TestCase):
