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..05a5c3da0cc2
--- /dev/null
+++ b/paimon-python/pypaimon/data/decimal.py
@@ -0,0 +1,350 @@
+################################################################################
+#  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 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 decimal.Decimal instance.
+        """
+        if self.decimal_val is None:
+            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
+
+        value = self.to_big_decimal().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.
+        """
+        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:
+        """
+        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.
+
+        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)
+
+        with localcontext() as ctx:
+            # Java BigDecimal is arbitrary precision.
+            # Paimon DECIMAL supports precision up to 38.
+            ctx.prec = max(
+                precision + scale,
+                len(value.as_tuple().digits) + abs(value.as_tuple().exponent) + scale,
+                38,
+            )
+
+            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.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,
+        )
+
+        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,
+        )
+
+    @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/read/merge_engine_support.py b/paimon-python/pypaimon/read/merge_engine_support.py
index ff551f9ab51e..258521193640 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -65,6 +65,7 @@
     "listagg",
     "nested_update",
     "collect",
+    "product",
     "merge_map_with_keytime",
 ])
 _FIELDS_PREFIX = "fields."
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index 321a70dacaee..a02b9925f4d3 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -32,11 +32,12 @@
 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
 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
@@ -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"
@@ -64,6 +66,16 @@
 NAME_MERGE_MAP_WITH_KEYTIME = "merge_map_with_keytime"
 
 
+# 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.
@@ -72,6 +84,16 @@
     "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`,
@@ -340,17 +362,306 @@ 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:
+            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(
+                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:
+            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(
+                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:
+            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, input_field: Any) -> Any:
+        if accumulator is None or input_field is None:
+            return accumulator
+
+        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 = 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):
@@ -913,6 +1224,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_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")
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py b/paimon-python/pypaimon/tests/test_field_aggregators.py
index c6a188d90af4..d823ebbd26ec 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,
@@ -44,6 +44,7 @@
     FieldMinAgg,
     FieldPrimaryKeyAgg,
     FieldSumAgg,
+    FieldProductAgg,
     FieldListaggAgg,
     FieldNestedUpdateAgg,
     FieldCollectAgg,
@@ -141,30 +142,322 @@ def test_reset_re_arms_first_non_null(self):
 
 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"))
+
+
+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))