feat(aggregate): support product aggregate function

docs/source/user_guide/primary_key_table.rst

@@ -80,3 +80,75 @@ merged according to the user-specified merge engine and the timestamp of each re
 New records written into the LSM tree will be first buffered in memory. When the
 memory buffer is full, all records in memory will be sorted and flushed to disk.
 A new sorted run is now created.
+
+
+Merge Engines
+-------------
+When Paimon sink receives two or more records with the same primary key, it
+merges them into one record to keep the primary key unique. The
+``merge-engine`` table option decides how. Paimon C++ supports ``deduplicate``
+(the default, which keeps the last record), ``partial-update``, ``first-row``
+and ``aggregation``.
+
+Aggregation
+~~~~~~~~~~~~~~
+The ``aggregation`` merge engine aggregates each value field across the records
+sharing a primary key, using the function configured for that field. Fields
+without a configured function fall back to
+``fields.default-aggregate-function``, and then to ``last_non_null_value``.
+
+.. code-block:: text
+
+   merge-engine = aggregation
+   fields.<field-name>.aggregate-function = <function>
+   fields.default-aggregate-function = <function>
+   fields.<field-name>.ignore-retract = true
+
+The available functions are ``sum``, ``product``, ``min``, ``max``,
+``first_value``, ``last_value``, ``first_non_null_value``,
+``last_non_null_value``, ``bool_and``, ``bool_or``, ``listagg``, ``collect``,
+``merge_map``, ``nested_update``, ``hll_sketch`` and ``theta_sketch``. Each one
+accepts only the field types it is defined for. An unknown function, or a
+function configured on an unsupported field type, is rejected when aggregation
+logic is initialized. Read paths that bypass aggregation may skip this
+validation.
+
+Not every function can process a retraction, that is a record whose row kind is
+``DELETE`` or ``UPDATE_BEFORE``. A function that cannot returns an error, unless
+``fields.<field-name>.ignore-retract`` is set, which drops the retraction
+instead.
+
+product
+^^^^^^^^^^^^^^
+Multiplies the values of a field. It accepts ``TINYINT``, ``SMALLINT``,
+``INT``, ``BIGINT``, ``FLOAT``, ``DOUBLE`` and ``DECIMAL``, and it supports
+retraction by dividing the accumulated value.
+
+.. code-block:: text
+
+   merge-engine = aggregation
+   fields.price.aggregate-function = product
+
+Null values are skipped, so aggregating a null into a field leaves the field
+unchanged, and the first non-null value seeds the product. Retracting into a
+field that is still null leaves it null rather than producing a reciprocal.
+
+Integer arithmetic is exact. A product or a quotient outside the range of the
+field type, a division by zero, and dividing the smallest value of the type by
+``-1`` all fail with an error instead of wrapping around. An integer quotient
+truncates towards zero, so retracting ``3`` from ``10`` yields ``3``.
+
+``FLOAT`` and ``DOUBLE`` follow IEEE 754 instead of reporting an error.
+Retracting a zero from a finite non-zero value yields an infinity carrying the
+sign of both operands, and retracting a zero from a zero yields ``NaN``.
+
+``DECIMAL`` has two boundary behaviors worth knowing, both matching Paimon
+Java:
+
+- The product and the quotient are rounded half up to the scale of the field.
+  A result that no longer fits the precision of the field aggregates to
+  ``NULL`` rather than failing, so a ``DECIMAL(4, 2)`` field holding
+  ``99.99 * 99.99`` becomes ``NULL``.
+- Retraction only accepts a quotient with a finite decimal expansion, and
+  returns an error otherwise. ``1.00 / 8.00`` is accepted and rounds to
+  ``0.13``, while ``2.00 / 3.00`` is rejected.

src/paimon/CMakeLists.txt

@@ -325,6 +325,7 @@ set(PAIMON_CORE_SRCS
     core/mergetree/compact/aggregate/field_collect_agg.cpp
     core/mergetree/compact/aggregate/field_merge_map_agg.cpp
     core/mergetree/compact/aggregate/field_nested_update_agg.cpp
+    core/mergetree/compact/aggregate/field_product_agg.cpp
     core/mergetree/compact/aggregate/field_sketch_agg.cpp
     core/mergetree/compact/aggregate/field_sum_agg.cpp
     core/mergetree/compact/interval_partition.cpp
@@ -846,6 +847,7 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
                     core/mergetree/compact/aggregate/field_nested_update_agg_test.cpp
                     core/mergetree/compact/aggregate/field_primary_key_agg_test.cpp
+                    core/mergetree/compact/aggregate/field_product_agg_test.cpp
                     core/mergetree/compact/aggregate/field_sketch_agg_test.cpp
                     core/mergetree/compact/aggregate/field_sum_agg_test.cpp
                     core/mergetree/compact/deduplicate_merge_function_test.cpp

src/paimon/core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp

@@ -19,6 +19,7 @@
 #include "paimon/core/mergetree/compact/aggregate/aggregate_merge_function.h"
 
 #include <map>
+#include <optional>
 #include <variant>
 
 #include "arrow/api.h"
@@ -80,6 +81,59 @@ TEST(AggregateMergeFunctionTest, TestGetAggFuncName) {
         ASSERT_EQ(FieldLastValueAgg::NAME, str_agg);
     }
 }
+TEST(AggregateMergeFunctionTest, TestProduct) {
+    arrow::FieldVector fields = {arrow::field("k0", arrow::int32()),
+                                 arrow::field("v0", arrow::int32())};
+    auto value_schema = arrow::schema(fields);
+    ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
+                         CoreOptions::FromMap({{"fields.v0.aggregate-function", "product"}}));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<AggregateMergeFunction> merge_func,
+                         AggregateMergeFunction::Create(value_schema, /*primary_keys=*/{"k0"},
+                                                        core_options, GetDefaultPool()));
+
+    auto pool = GetDefaultPool();
+    // three rows sharing a key multiply into 2 * 3 * 5
+    KeyValue kv1(RowKind::Insert(), /*sequence_number=*/0, /*level=*/0, /*key=*/
+                 BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                 /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 2}, pool.get()));
+    KeyValue kv2(RowKind::Insert(), /*sequence_number=*/0, /*level=*/1, /*key=*/
+                 BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                 /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 3}, pool.get()));
+    KeyValue kv3(RowKind::Insert(), /*sequence_number=*/0, /*level=*/2, /*key=*/
+                 BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                 /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 5}, pool.get()));
+    ASSERT_OK(merge_func->Add(std::move(kv1)));
+    ASSERT_OK(merge_func->Add(std::move(kv2)));
+    ASSERT_OK(merge_func->Add(std::move(kv3)));
+    ASSERT_OK_AND_ASSIGN(std::optional<KeyValue> product_result, merge_func->GetResult());
+    ASSERT_TRUE(product_result.has_value());
+    KeyValue expected(RowKind::Insert(), /*sequence_number=*/0,
+                      /*level=*/KeyValue::UNKNOWN_LEVEL, /*key=*/
+                      BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                      /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 30}, pool.get()));
+    KeyValueChecker::CheckResult(expected, product_result.value(), /*key_arity=*/1,
+                                 /*value_arity=*/2);
+
+    // a delete retracts by dividing the accumulator back out
+    merge_func->Reset();
+    KeyValue kv4(RowKind::Insert(), /*sequence_number=*/0, /*level=*/0, /*key=*/
+                 BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                 /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 30}, pool.get()));
+    KeyValue kv5(RowKind::Delete(), /*sequence_number=*/0, /*level=*/1, /*key=*/
+                 BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                 /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 5}, pool.get()));
+    ASSERT_OK(merge_func->Add(std::move(kv4)));
+    ASSERT_OK(merge_func->Add(std::move(kv5)));
+    ASSERT_OK_AND_ASSIGN(std::optional<KeyValue> retract_result, merge_func->GetResult());
+    ASSERT_TRUE(retract_result.has_value());
+    KeyValue expected2(RowKind::Insert(), /*sequence_number=*/0,
+                       /*level=*/KeyValue::UNKNOWN_LEVEL, /*key=*/
+                       BinaryRowGenerator::GenerateRowPtr({10}, pool.get()),
+                       /*value=*/BinaryRowGenerator::GenerateRowPtr({10, 6}, pool.get()));
+    KeyValueChecker::CheckResult(expected2, retract_result.value(), /*key_arity=*/1,
+                                 /*value_arity=*/2);
+}
+
 TEST(AggregateMergeFunctionTest, TestSimple) {
     arrow::FieldVector fields = {arrow::field("k0", arrow::int32()),
                                  arrow::field("v0", arrow::int32())};

src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory.h

@@ -38,6 +38,7 @@
 #include "paimon/core/mergetree/compact/aggregate/field_min_agg.h"
 #include "paimon/core/mergetree/compact/aggregate/field_nested_update_agg.h"
 #include "paimon/core/mergetree/compact/aggregate/field_primary_key_agg.h"
+#include "paimon/core/mergetree/compact/aggregate/field_product_agg.h"
 #include "paimon/core/mergetree/compact/aggregate/field_sketch_agg.h"
 #include "paimon/core/mergetree/compact/aggregate/field_sum_agg.h"
 #include "paimon/result.h"
@@ -80,6 +81,8 @@ class FieldAggregatorFactory {
             field_aggregator = std::make_unique<FieldFirstValueAgg>(field_type, pool);
         } else if (str_agg == FieldSumAgg::NAME) {
             PAIMON_ASSIGN_OR_RAISE(field_aggregator, FieldSumAgg::Create(field_type, pool));
+        } else if (str_agg == FieldProductAgg::NAME) {
+            PAIMON_ASSIGN_OR_RAISE(field_aggregator, FieldProductAgg::Create(field_type, pool));
         } else if (str_agg == FieldMinAgg::NAME) {
             PAIMON_ASSIGN_OR_RAISE(field_aggregator, FieldMinAgg::Create(field_type, pool));
         } else if (str_agg == FieldMaxAgg::NAME) {

src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp

@@ -43,6 +43,13 @@ TEST(FieldAggregatorFactoryTest, TestSimple) {
                                  "f0", arrow::int32(), "sum", options, GetDefaultPool()));
         ASSERT_TRUE(dynamic_cast<FieldSumAgg*>(agg.get()));
     }
+    {
+        ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldAggregator> agg,
+                             FieldAggregatorFactory::CreateFieldAggregator(
+                                 "f0", arrow::int32(), "product", options, GetDefaultPool()));
+        ASSERT_TRUE(dynamic_cast<FieldProductAgg*>(agg.get()));
+    }
     {
         ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
         ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldAggregator> agg,