This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new b9274202 feat(aggregate): support product aggregate function (#320)
b9274202 is described below
commit b92742024759cea77007fc44319a5834c25aa1ff
Author: Nicholas Jiang <[email protected]>
AuthorDate: Thu Sep 10 16:08:09 2026 +0800
feat(aggregate): support product aggregate function (#320)
---
docs/source/user_guide/primary_key_table.rst | 72 ++++
src/paimon/CMakeLists.txt | 2 +
.../aggregate/aggregate_merge_function_test.cpp | 54 +++
.../compact/aggregate/field_aggregator_factory.h | 3 +
.../aggregate/field_aggregator_factory_test.cpp | 7 +
.../compact/aggregate/field_product_agg.cpp | 313 ++++++++++++++++
.../compact/aggregate/field_product_agg.h | 89 +++++
.../compact/aggregate/field_product_agg_test.cpp | 410 +++++++++++++++++++++
test/inte/write_and_read_inte_test.cpp | 59 +++
9 files changed, 1009 insertions(+)
diff --git a/docs/source/user_guide/primary_key_table.rst
b/docs/source/user_guide/primary_key_table.rst
index fcbe198e..e9cdc10b 100644
--- a/docs/source/user_guide/primary_key_table.rst
+++ b/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.
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index cb083e99..a25397b9 100644
--- a/src/paimon/CMakeLists.txt
+++ b/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
@@ -847,6 +848,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
diff --git
a/src/paimon/core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp
b/src/paimon/core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp
index 5d083b14..6695b2f8 100644
---
a/src/paimon/core/mergetree/compact/aggregate/aggregate_merge_function_test.cpp
+++
b/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())};
diff --git
a/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory.h
b/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory.h
index e64b9e77..fafded4f 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory.h
+++ b/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) {
diff --git
a/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp
b/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp
index b5b626c0..f876ff3d 100644
---
a/src/paimon/core/mergetree/compact/aggregate/field_aggregator_factory_test.cpp
+++
b/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,
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_product_agg.cpp
b/src/paimon/core/mergetree/compact/aggregate/field_product_agg.cpp
new file mode 100644
index 00000000..5c1cf7c2
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/aggregate/field_product_agg.cpp
@@ -0,0 +1,313 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/mergetree/compact/aggregate/field_product_agg.h"
+
+#include <array>
+#include <cassert>
+#include <cstdint>
+#include <limits>
+
+#include "arrow/type.h"
+#include "arrow/util/basic_decimal.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/checked_cast.h"
+#include "paimon/common/utils/decimal_utils.h"
+#include "paimon/data/decimal.h"
+#include "paimon/status.h"
+
+namespace paimon {
+namespace {
+arrow::BasicDecimal256 ToDecimal256(const Decimal& value) {
+ return arrow::BasicDecimal256(
+ arrow::BasicDecimal128(static_cast<int64_t>(value.HighBits()),
value.LowBits()));
+}
+
+// @warning Only call this for values already known to fit into a decimal of
precision 38 or less,
+// the higher 128 bits are dropped.
+Decimal::int128_t ToInt128(const arrow::BasicDecimal256& value) {
+ std::array<uint64_t, 4> words = value.little_endian_array();
+ return
static_cast<Decimal::int128_t>((static_cast<Decimal::uint128_t>(words[1]) <<
64) |
+ words[0]);
+}
+
+template <typename T>
+Result<T> MultiplyExact(T left, T right, const char* type_name) {
+ T product = 0;
+ if (__builtin_mul_overflow(left, right, &product)) {
+ return Status::Invalid(fmt::format("{} overflow: {} * {}", type_name,
+ static_cast<int64_t>(left),
+ static_cast<int64_t>(right)));
+ }
+ return product;
+}
+
+// Rejects the two divisions Java rejects: a zero divisor, and the one
quotient the type cannot
+// represent, its minimum divided by -1. The quotient itself still truncates
towards zero, the way
+// integer division does in Java.
+template <typename T>
+Result<T> DivideExact(T left, T right, const char* type_name) {
+ if (right == 0) {
+ return Status::Invalid(fmt::format("{} division by zero: {} / {}",
type_name,
+ static_cast<int64_t>(left),
+ static_cast<int64_t>(right)));
+ }
+ if (left == std::numeric_limits<T>::min() && right == -1) {
+ return Status::Invalid(fmt::format("{} overflow: {} / {}", type_name,
+ static_cast<int64_t>(left),
+ static_cast<int64_t>(right)));
+ }
+ return static_cast<T>(left / right);
+}
+
+// Negating through the unsigned type rather than the signed one keeps the
type's minimum, whose
+// magnitude a signed negation cannot represent, well defined.
+Decimal::uint128_t Magnitude(Decimal::int128_t value) {
+ auto magnitude = static_cast<Decimal::uint128_t>(value);
+ return value < 0 ? -magnitude : magnitude;
+}
+
+// Whether the exact quotient of two decimals sharing a scale has a finite
decimal expansion, which
+// is what Java's BigDecimal.divide() requires before it hands the quotient to
fromBigDecimal().
+//
+// The scales cancel out, so the exact quotient is the ratio of the unscaled
values. Reduced to
+// lowest terms it terminates exactly when its denominator has no prime factor
besides 2 and 5,
+// which holds when the divisor stripped of those two factors divides the
dividend.
+//
+// @warning The divisor must be non-zero; stripping the factors of a zero
never terminates.
+bool QuotientTerminates(Decimal::int128_t dividend, Decimal::int128_t divisor)
{
+ Decimal::uint128_t coprime_divisor = Magnitude(divisor);
+ while (coprime_divisor % 2 == 0) {
+ coprime_divisor /= 2;
+ }
+ while (coprime_divisor % 5 == 0) {
+ coprime_divisor /= 5;
+ }
+ return Magnitude(dividend) % coprime_divisor == 0;
+}
+
+Result<VariantType> MultiplyDecimal(const Decimal& accumulator, const Decimal&
input_field) {
+ arrow::BasicDecimal256 product = ToDecimal256(accumulator);
+ product *= ToDecimal256(input_field);
+ // Multiplying two values sharing the field scale doubles that scale, so
drop the extra digits
+ // again, rounding half up like Java's BigDecimal.setScale(scale, HALF_UP)
does.
+ product = product.ReduceScaleBy(accumulator.Scale(), /*round=*/true);
+ if (!product.FitsInPrecision(accumulator.Precision())) {
+ // Java's Decimal.fromBigDecimal() returns null for a result the field
precision cannot
+ // hold, which leaves the aggregated field null.
+ return VariantType(NullType());
+ }
+ return VariantType(Decimal(accumulator.Precision(), accumulator.Scale(),
ToInt128(product)));
+}
+
+Result<VariantType> DivideDecimal(const Decimal& accumulator, const Decimal&
input_field) {
+ arrow::BasicDecimal256 divisor = ToDecimal256(input_field);
+ // Dividing two values sharing the field scale cancels that scale out, so
scale the dividend up
+ // first to keep the field scale in the quotient.
+ arrow::BasicDecimal256 dividend =
+ ToDecimal256(accumulator).IncreaseScaleBy(accumulator.Scale());
+ arrow::BasicDecimal256 quotient;
+ arrow::BasicDecimal256 remainder;
+ if (dividend.Divide(divisor, "ient, &remainder) !=
arrow::DecimalStatus::kSuccess) {
+ return Status::Invalid(fmt::format("decimal division by zero: {} / {}",
+ accumulator.ToString(),
input_field.ToString()));
+ }
+ if (!QuotientTerminates(accumulator.Value(), input_field.Value())) {
+ return Status::Invalid(fmt::format("decimal {} / {} has no finite
decimal expansion",
+ accumulator.ToString(),
input_field.ToString()));
+ }
+ // Divide() drops the fractional part towards zero, round it half up
instead. A quotient that
+ // terminates can still carry more digits than the field scale, so this is
still needed.
+ arrow::BasicDecimal256 abs_remainder =
arrow::BasicDecimal256::Abs(remainder);
+ arrow::BasicDecimal256 twice_remainder = abs_remainder;
+ twice_remainder += abs_remainder;
+ if (twice_remainder >= arrow::BasicDecimal256::Abs(divisor)) {
+ quotient += arrow::BasicDecimal256(
+ dividend.IsNegative() == divisor.IsNegative() ? int64_t{1} :
int64_t{-1});
+ }
+ if (!quotient.FitsInPrecision(accumulator.Precision())) {
+ // Same as the multiplication, Java's fromBigDecimal() nulls out a
quotient the field
+ // precision cannot hold.
+ return VariantType(NullType());
+ }
+ return VariantType(Decimal(accumulator.Precision(), accumulator.Scale(),
ToInt128(quotient)));
+}
+} // namespace
+
+Result<std::unique_ptr<FieldProductAgg>> FieldProductAgg::Create(
+ const std::shared_ptr<arrow::DataType>& field_type, const
std::shared_ptr<MemoryPool>& pool) {
+ if (field_type->id() == arrow::Type::type::DECIMAL128) {
+ // Both directions rescale by the field scale, so reject up front the
decimal types the
+ // rescaling helpers cannot handle.
+ PAIMON_RETURN_NOT_OK(DecimalUtils::CheckDecimalType(*field_type));
+ const auto* decimal_type = checked_cast<const
arrow::Decimal128Type*>(field_type.get());
+ if (decimal_type->scale() < 0) {
+ return Status::Invalid(
+ fmt::format("Invalid decimal type {}, scale must >= 0",
field_type->ToString()));
+ }
+ }
+ PAIMON_ASSIGN_OR_RAISE(FieldArithmeticFunc multiply_func,
CreateMultiplyFunc(field_type));
+ PAIMON_ASSIGN_OR_RAISE(FieldArithmeticFunc divide_func,
CreateDivideFunc(field_type));
+ return std::unique_ptr<FieldProductAgg>(
+ new FieldProductAgg(field_type, multiply_func, divide_func, pool));
+}
+
+Result<FieldProductAgg::FieldArithmeticFunc>
FieldProductAgg::CreateMultiplyFunc(
+ const std::shared_ptr<arrow::DataType>& field_type) {
+ arrow::Type::type type = field_type->id();
+ switch (type) {
+ case arrow::Type::type::INT8:
+ // The variant holds TINYINT as a plain char, whose signedness
follows the ABI, so
+ // multiply the signed value it stands for.
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ auto accumulator_value =
+
static_cast<int8_t>(DataDefine::GetVariantValue<char>(accumulator));
+ auto input_value =
+
static_cast<int8_t>(DataDefine::GetVariantValue<char>(input_field));
+ PAIMON_ASSIGN_OR_RAISE(int8_t product,
+ MultiplyExact(accumulator_value,
input_value, "int8"));
+ return VariantType(static_cast<char>(product));
+ });
+ case arrow::Type::type::INT16:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int16_t product,
+
MultiplyExact(DataDefine::GetVariantValue<int16_t>(accumulator),
+
DataDefine::GetVariantValue<int16_t>(input_field), "int16"));
+ return VariantType(product);
+ });
+ case arrow::Type::type::INT32:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t product,
+
MultiplyExact(DataDefine::GetVariantValue<int32_t>(accumulator),
+
DataDefine::GetVariantValue<int32_t>(input_field), "int32"));
+ return VariantType(product);
+ });
+ case arrow::Type::type::INT64:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int64_t product,
+
MultiplyExact(DataDefine::GetVariantValue<int64_t>(accumulator),
+
DataDefine::GetVariantValue<int64_t>(input_field), "int64"));
+ return VariantType(product);
+ });
+ case arrow::Type::type::FLOAT:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ float product =
DataDefine::GetVariantValue<float>(accumulator) *
+
DataDefine::GetVariantValue<float>(input_field);
+ return VariantType(product);
+ });
+ case arrow::Type::type::DOUBLE:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ double product =
DataDefine::GetVariantValue<double>(accumulator) *
+
DataDefine::GetVariantValue<double>(input_field);
+ return VariantType(product);
+ });
+ case arrow::Type::type::DECIMAL128:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ auto accumulator_value =
DataDefine::GetVariantValue<Decimal>(accumulator);
+ auto input_value =
DataDefine::GetVariantValue<Decimal>(input_field);
+ assert(accumulator_value.Precision() ==
input_value.Precision() &&
+ accumulator_value.Scale() == input_value.Scale());
+ return MultiplyDecimal(accumulator_value, input_value);
+ });
+ default:
+ return Status::Invalid(
+ fmt::format("type {} not support in FieldProductAgg",
field_type->ToString()));
+ }
+}
+
+Result<FieldProductAgg::FieldArithmeticFunc> FieldProductAgg::CreateDivideFunc(
+ const std::shared_ptr<arrow::DataType>& field_type) {
+ arrow::Type::type type = field_type->id();
+ switch (type) {
+ case arrow::Type::type::INT8:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ auto accumulator_value =
+
static_cast<int8_t>(DataDefine::GetVariantValue<char>(accumulator));
+ auto input_value =
+
static_cast<int8_t>(DataDefine::GetVariantValue<char>(input_field));
+ PAIMON_ASSIGN_OR_RAISE(int8_t quotient,
+ DivideExact(accumulator_value,
input_value, "int8"));
+ return VariantType(static_cast<char>(quotient));
+ });
+ case arrow::Type::type::INT16:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int16_t quotient,
+
DivideExact(DataDefine::GetVariantValue<int16_t>(accumulator),
+
DataDefine::GetVariantValue<int16_t>(input_field), "int16"));
+ return VariantType(quotient);
+ });
+ case arrow::Type::type::INT32:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int32_t quotient,
+
DivideExact(DataDefine::GetVariantValue<int32_t>(accumulator),
+
DataDefine::GetVariantValue<int32_t>(input_field), "int32"));
+ return VariantType(quotient);
+ });
+ case arrow::Type::type::INT64:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ PAIMON_ASSIGN_OR_RAISE(
+ int64_t quotient,
+
DivideExact(DataDefine::GetVariantValue<int64_t>(accumulator),
+
DataDefine::GetVariantValue<int64_t>(input_field), "int64"));
+ return VariantType(quotient);
+ });
+ case arrow::Type::type::FLOAT:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ float quotient =
DataDefine::GetVariantValue<float>(accumulator) /
+
DataDefine::GetVariantValue<float>(input_field);
+ return VariantType(quotient);
+ });
+ case arrow::Type::type::DOUBLE:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ double quotient =
DataDefine::GetVariantValue<double>(accumulator) /
+
DataDefine::GetVariantValue<double>(input_field);
+ return VariantType(quotient);
+ });
+ case arrow::Type::type::DECIMAL128:
+ return FieldArithmeticFunc([](const VariantType& accumulator,
+ const VariantType& input_field) ->
Result<VariantType> {
+ auto accumulator_value =
DataDefine::GetVariantValue<Decimal>(accumulator);
+ auto input_value =
DataDefine::GetVariantValue<Decimal>(input_field);
+ assert(accumulator_value.Precision() ==
input_value.Precision() &&
+ accumulator_value.Scale() == input_value.Scale());
+ return DivideDecimal(accumulator_value, input_value);
+ });
+ default:
+ return Status::Invalid(
+ fmt::format("type {} not support in FieldProductAgg",
field_type->ToString()));
+ }
+}
+} // namespace paimon
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_product_agg.h
b/src/paimon/core/mergetree/compact/aggregate/field_product_agg.h
new file mode 100644
index 00000000..4fb9046b
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/aggregate/field_product_agg.h
@@ -0,0 +1,89 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <string>
+
+#include "paimon/common/data/data_define.h"
+#include "paimon/core/mergetree/compact/aggregate/field_aggregator.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+/// Multiplies non-null values of a field across rows.
+///
+/// Retraction divides the accumulator by the retracted value. Everything Java
rejects is rejected
+/// here as an error rather than by wrapping around: an integer product or
quotient outside the
+/// field type, an integer division by zero, and a decimal quotient with no
finite decimal
+/// expansion. A decimal result that no longer fits the field precision
aggregates to null, which is
+/// what Java's Decimal.fromBigDecimal() yields for it.
+class FieldProductAgg : public FieldAggregator {
+ public:
+ static Result<std::unique_ptr<FieldProductAgg>> Create(
+ const std::shared_ptr<arrow::DataType>& field_type,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ Result<VariantType> Agg(const VariantType& accumulator,
+ const VariantType& input_field) override {
+ bool accumulator_null = DataDefine::IsVariantNull(accumulator);
+ bool input_null = DataDefine::IsVariantNull(input_field);
+ if (accumulator_null || input_null) {
+ return accumulator_null ? input_field : accumulator;
+ }
+ return multiply_func_(accumulator, input_field);
+ }
+
+ Result<VariantType> Retract(const VariantType& accumulator,
+ const VariantType& input_field) const override
{
+ if (DataDefine::IsVariantNull(accumulator) ||
DataDefine::IsVariantNull(input_field)) {
+ return accumulator;
+ }
+ return divide_func_(accumulator, input_field);
+ }
+
+ public:
+ static constexpr char NAME[] = "product";
+
+ private:
+ using FieldArithmeticFunc = std::function<Result<VariantType>(const
VariantType& accumulator,
+ const
VariantType& input_field)>;
+
+ FieldProductAgg(const std::shared_ptr<arrow::DataType>& field_type,
+ const FieldArithmeticFunc& multiply_func,
+ const FieldArithmeticFunc& divide_func, const
std::shared_ptr<MemoryPool>& pool)
+ : FieldAggregator(std::string(NAME), field_type, pool),
+ multiply_func_(multiply_func),
+ divide_func_(divide_func) {}
+
+ static Result<FieldArithmeticFunc> CreateMultiplyFunc(
+ const std::shared_ptr<arrow::DataType>& field_type);
+
+ static Result<FieldArithmeticFunc> CreateDivideFunc(
+ const std::shared_ptr<arrow::DataType>& field_type);
+
+ private:
+ FieldArithmeticFunc multiply_func_;
+ FieldArithmeticFunc divide_func_;
+};
+} // namespace paimon
diff --git
a/src/paimon/core/mergetree/compact/aggregate/field_product_agg_test.cpp
b/src/paimon/core/mergetree/compact/aggregate/field_product_agg_test.cpp
new file mode 100644
index 00000000..f0bd9fdf
--- /dev/null
+++ b/src/paimon/core/mergetree/compact/aggregate/field_product_agg_test.cpp
@@ -0,0 +1,410 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/mergetree/compact/aggregate/field_product_agg.h"
+
+#include <cmath>
+#include <cstdint>
+#include <limits>
+#include <memory>
+#include <string>
+
+#include "arrow/type_fwd.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/decimal_utils.h"
+#include "paimon/data/decimal.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+namespace {
+Decimal MakeDecimal(int32_t precision, int32_t scale, const std::string&
unscaled) {
+ return Decimal(precision, scale,
DecimalUtils::StrToInt128(unscaled).value());
+}
+} // namespace
+
+TEST(FieldProductAggTest, TestSimple) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::int32(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret, field_product_agg->Agg(5, 10));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(agg_ret), 50);
+
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
field_product_agg->Retract(50, 10));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(retract_ret), 5);
+}
+
+TEST(FieldProductAggTest, TestNull) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::int32(),
GetDefaultPool()));
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret, field_product_agg->Agg(5,
NullType()));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(agg_ret), 5);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(NullType(), 10));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(agg_ret), 10);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(NullType(), NullType()));
+ ASSERT_TRUE(DataDefine::IsVariantNull(agg_ret));
+ }
+
+ // retraction keeps the accumulator whenever either side is null
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
field_product_agg->Retract(5, NullType()));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(retract_ret), 5);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
field_product_agg->Retract(NullType(), 10));
+ ASSERT_TRUE(DataDefine::IsVariantNull(retract_ret));
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
+ field_product_agg->Retract(NullType(),
NullType()));
+ ASSERT_TRUE(DataDefine::IsVariantNull(retract_ret));
+ }
+}
+
+TEST(FieldProductAggTest, TestSupportedTypes) {
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int8(),
GetDefaultPool()));
+ // The variant holds TINYINT as a plain char, so read back the signed
value it stands for
+ // rather than comparing a char that is unsigned under some ABIs
against a negative literal.
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
+ field_product_agg->Agg(static_cast<char>(-5),
static_cast<char>(20)));
+
ASSERT_EQ(static_cast<int8_t>(DataDefine::GetVariantValue<char>(agg_ret)),
-100);
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(static_cast<char>(-100),
static_cast<char>(20)));
+
ASSERT_EQ(static_cast<int8_t>(DataDefine::GetVariantValue<char>(retract_ret)),
-5);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int16(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(static_cast<int16_t>(100),
+
static_cast<int16_t>(15)));
+ ASSERT_EQ(DataDefine::GetVariantValue<int16_t>(agg_ret), 1500);
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(static_cast<int16_t>(1500),
static_cast<int16_t>(15)));
+ ASSERT_EQ(DataDefine::GetVariantValue<int16_t>(retract_ret), 100);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int64(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(static_cast<int64_t>(100),
+
static_cast<int64_t>(15)));
+ ASSERT_EQ(DataDefine::GetVariantValue<int64_t>(agg_ret), 1500);
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(static_cast<int64_t>(1500),
static_cast<int64_t>(15)));
+ ASSERT_EQ(DataDefine::GetVariantValue<int64_t>(retract_ret), 100);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::float32(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(static_cast<float>(1.5),
+
static_cast<float>(2.5)));
+ ASSERT_NEAR(DataDefine::GetVariantValue<float>(agg_ret), 3.75, 0.0001);
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(static_cast<float>(3.75),
static_cast<float>(2.5)));
+ ASSERT_NEAR(DataDefine::GetVariantValue<float>(retract_ret), 1.5,
0.0001);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::float64(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret, field_product_agg->Agg(1.5,
2.5));
+ ASSERT_NEAR(DataDefine::GetVariantValue<double>(agg_ret), 3.75,
0.0001);
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
field_product_agg->Retract(3.75, 2.5));
+ ASSERT_NEAR(DataDefine::GetVariantValue<double>(retract_ret), 1.5,
0.0001);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 2),
GetDefaultPool()));
+ // 1.50 * 2.00 = 3.00
+ ASSERT_OK_AND_ASSIGN(
+ VariantType agg_ret,
+ field_product_agg->Agg(MakeDecimal(10, 2, "150"), MakeDecimal(10,
2, "200")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(agg_ret),
MakeDecimal(10, 2, "300"));
+ // 3.00 / 2.00 = 1.50
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "300"),
MakeDecimal(10, 2, "200")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(retract_ret),
MakeDecimal(10, 2, "150"));
+ }
+}
+
+TEST(FieldProductAggTest, TestIntegerRetractTruncatesTowardsZero) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::int32(),
GetDefaultPool()));
+ // a negative divisor flips the sign of the quotient
+ ASSERT_OK_AND_ASSIGN(VariantType negative_divisor_ret,
field_product_agg->Retract(10, -5));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(negative_divisor_ret), -2);
+ // 10 / 3 drops the fraction rather than rounding it, unlike the decimal
quotients
+ ASSERT_OK_AND_ASSIGN(VariantType inexact_ret,
field_product_agg->Retract(10, 3));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(inexact_ret), 3);
+ // -10 / 3 truncates towards zero to -3 rather than flooring to -4
+ ASSERT_OK_AND_ASSIGN(VariantType negative_inexact_ret,
field_product_agg->Retract(-10, 3));
+ ASSERT_EQ(DataDefine::GetVariantValue<int32_t>(negative_inexact_ret), -3);
+}
+
+TEST(FieldProductAggTest, TestDecimalRoundsHalfUp) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 2),
GetDefaultPool()));
+ // 1.05 * 1.10 = 1.1550, rounded half up to 1.16
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
field_product_agg->Agg(MakeDecimal(10, 2, "105"),
+
MakeDecimal(10, 2, "110")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(agg_ret), MakeDecimal(10,
2, "116"));
+ // 1.01 * 1.01 = 1.0201, the dropped digits stay below half so the result
truncates to 1.02
+ ASSERT_OK_AND_ASSIGN(
+ VariantType truncated_agg_ret,
+ field_product_agg->Agg(MakeDecimal(10, 2, "101"), MakeDecimal(10, 2,
"101")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(truncated_agg_ret),
MakeDecimal(10, 2, "102"));
+ // -1.05 * 1.10 = -1.1550, rounded half up (away from zero) to -1.16
+ ASSERT_OK_AND_ASSIGN(
+ VariantType negative_agg_ret,
+ field_product_agg->Agg(MakeDecimal(10, 2, "-105"), MakeDecimal(10, 2,
"110")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(negative_agg_ret),
MakeDecimal(10, 2, "-116"));
+
+ // 1.00 / 8.00 = 0.125 exactly, landing on a tie that rounds away from
zero to 0.13
+ ASSERT_OK_AND_ASSIGN(
+ VariantType tie_retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "100"), MakeDecimal(10,
2, "800")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(tie_retract_ret),
MakeDecimal(10, 2, "13"));
+ // -1.00 / 8.00 = -0.125, the tie rounds away from zero to -0.13
+ ASSERT_OK_AND_ASSIGN(
+ VariantType negative_tie_retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "-100"), MakeDecimal(10,
2, "800")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(negative_tie_retract_ret),
+ MakeDecimal(10, 2, "-13"));
+ // 1.00 / -8.00 = -0.125, a negative divisor rounds away from zero the
same way
+ ASSERT_OK_AND_ASSIGN(
+ VariantType negative_divisor_retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "100"), MakeDecimal(10,
2, "-800")));
+
ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(negative_divisor_retract_ret),
+ MakeDecimal(10, 2, "-13"));
+ // 1.00 / 16.00 = 0.0625, below the tie so it rounds down to 0.06
+ ASSERT_OK_AND_ASSIGN(
+ VariantType below_tie_retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "100"), MakeDecimal(10,
2, "1600")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(below_tie_retract_ret),
MakeDecimal(10, 2, "6"));
+
+ // 0.01 / 1.25 = 0.008, rounded half up to 0.01. The truncated quotient is
zero here, so the
+ // rounding step has to take the sign from the operands rather than from
the quotient.
+ ASSERT_OK_AND_ASSIGN(
+ VariantType rounded_up_from_zero_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "1"), MakeDecimal(10, 2,
"125")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(rounded_up_from_zero_ret),
+ MakeDecimal(10, 2, "1"));
+ // -0.01 / 1.25 = -0.008, rounded away from zero to -0.01
+ ASSERT_OK_AND_ASSIGN(
+ VariantType rounded_down_from_zero_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "-1"), MakeDecimal(10,
2, "125")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(rounded_down_from_zero_ret),
+ MakeDecimal(10, 2, "-1"));
+}
+
+TEST(FieldProductAggTest, TestDecimalRejectsNonTerminatingQuotient) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 2),
GetDefaultPool()));
+ // 2.00 / 3.00 repeats forever, which is what Java's BigDecimal.divide()
refuses
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(MakeDecimal(10, 2, "200"), MakeDecimal(10,
2, "300")),
+ "decimal 2.00 / 3.00 has no finite decimal expansion");
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(MakeDecimal(10, 2, "100"), MakeDecimal(10,
2, "700")),
+ "has no finite decimal expansion");
+ // 0.01 / 1.50 = 0.00666..., rejected even though it would round to a
representable 0.01
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(MakeDecimal(10, 2, "1"), MakeDecimal(10, 2,
"150")),
+ "has no finite decimal expansion");
+
+ // A divisor that is not a power of two or five is fine as long as the
quotient still
+ // terminates: 3.00 / 6.00 = 0.5 reduces to 1/2.
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "300"), MakeDecimal(10,
2, "600")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(retract_ret),
MakeDecimal(10, 2, "50"));
+ // Zero divided by the very divisor rejected above still terminates, at
zero.
+ ASSERT_OK_AND_ASSIGN(
+ VariantType zero_retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "0"), MakeDecimal(10, 2,
"300")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(zero_retract_ret),
MakeDecimal(10, 2, "0"));
+}
+
+TEST(FieldProductAggTest, TestDecimalWithZeroScale) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 0),
GetDefaultPool()));
+ // a zero scale rescales by nothing in either direction, 3 * 4 = 12
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
+ field_product_agg->Agg(MakeDecimal(10, 0, "3"),
MakeDecimal(10, 0, "4")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(agg_ret), MakeDecimal(10,
0, "12"));
+ // 10 / 4 = 2.5, a tie that rounds away from zero to 3 once the scale
leaves no decimals
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 0, "10"), MakeDecimal(10,
0, "4")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(retract_ret),
MakeDecimal(10, 0, "3"));
+}
+
+TEST(FieldProductAggTest, TestDecimalWiderThanInt128Intermediate) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(38, 18),
GetDefaultPool()));
+ // 20 * 30 = 600, the unscaled intermediate 6e38 does not fit into an
int128
+ ASSERT_OK_AND_ASSIGN(VariantType agg_ret,
+ field_product_agg->Agg(MakeDecimal(38, 18,
"20000000000000000000"),
+ MakeDecimal(38, 18,
"30000000000000000000")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(agg_ret),
+ MakeDecimal(38, 18, "600000000000000000000"));
+
+ ASSERT_OK_AND_ASSIGN(VariantType retract_ret,
+ field_product_agg->Retract(MakeDecimal(38, 18,
"600000000000000000000"),
+ MakeDecimal(38, 18,
"30000000000000000000")));
+ ASSERT_EQ(DataDefine::GetVariantValue<Decimal>(retract_ret),
+ MakeDecimal(38, 18, "20000000000000000000"));
+}
+
+TEST(FieldProductAggTest, TestIntegerArithmeticErrors) {
+ // every width computes in its own type, so a product that only fits a
wider type overflows
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int8(),
GetDefaultPool()));
+ ASSERT_NOK_WITH_MSG(field_product_agg->Agg(static_cast<char>(100),
static_cast<char>(2)),
+ "int8 overflow: 100 * 2");
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(static_cast<char>(-128),
static_cast<char>(-1)),
+ "int8 overflow: -128 / -1");
+ ASSERT_NOK_WITH_MSG(field_product_agg->Retract(static_cast<char>(10),
static_cast<char>(0)),
+ "int8 division by zero: 10 / 0");
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int16(),
GetDefaultPool()));
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Agg(static_cast<int16_t>(300),
static_cast<int16_t>(200)),
+ "int16 overflow: 300 * 200");
+
ASSERT_NOK_WITH_MSG(field_product_agg->Retract(std::numeric_limits<int16_t>::min(),
+
static_cast<int16_t>(-1)),
+ "int16 overflow: -32768 / -1");
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(static_cast<int16_t>(10),
static_cast<int16_t>(0)),
+ "int16 division by zero: 10 / 0");
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int32(),
GetDefaultPool()));
+
ASSERT_NOK_WITH_MSG(field_product_agg->Agg(std::numeric_limits<int32_t>::max(),
2),
+ "int32 overflow: 2147483647 * 2");
+
ASSERT_NOK_WITH_MSG(field_product_agg->Retract(std::numeric_limits<int32_t>::min(),
-1),
+ "int32 overflow: -2147483648 / -1");
+ ASSERT_NOK_WITH_MSG(field_product_agg->Retract(5, 0), "int32 division
by zero: 5 / 0");
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::int64(),
GetDefaultPool()));
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Agg(std::numeric_limits<int64_t>::max(),
static_cast<int64_t>(2)),
+ "int64 overflow: 9223372036854775807 * 2");
+
ASSERT_NOK_WITH_MSG(field_product_agg->Retract(std::numeric_limits<int64_t>::min(),
+
static_cast<int64_t>(-1)),
+ "int64 overflow: -9223372036854775808 / -1");
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(static_cast<int64_t>(10),
static_cast<int64_t>(0)),
+ "int64 division by zero: 10 / 0");
+ }
+}
+
+TEST(FieldProductAggTest, TestDecimalOverflowsToNull) {
+ {
+ // 99.99 * 99.99 = 9998.0001, which rounds to 9998.00 and still needs
a precision of 6
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(4, 2),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(
+ VariantType agg_ret,
+ field_product_agg->Agg(MakeDecimal(4, 2, "9999"), MakeDecimal(4,
2, "9999")));
+ ASSERT_TRUE(DataDefine::IsVariantNull(agg_ret));
+ }
+ {
+ // 99999999.99 / 0.01 needs 12 digits
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 2),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(
+ VariantType retract_ret,
+ field_product_agg->Retract(MakeDecimal(10, 2, "9999999999"),
MakeDecimal(10, 2, "1")));
+ ASSERT_TRUE(DataDefine::IsVariantNull(retract_ret));
+ }
+}
+
+TEST(FieldProductAggTest, TestDecimalDivisionByZero) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg> field_product_agg,
+ FieldProductAgg::Create(arrow::decimal128(10, 2),
GetDefaultPool()));
+ ASSERT_NOK_WITH_MSG(
+ field_product_agg->Retract(MakeDecimal(10, 2, "150"), MakeDecimal(10,
2, "0")),
+ "decimal division by zero: 1.50 / 0.00");
+}
+
+TEST(FieldProductAggTest, TestFloatingPointDivisionByZero) {
+ // unlike the integer types, the floating point types divide by zero the
way IEEE 754 says to
+ // rather than reporting an error
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::float64(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(VariantType positive_ret,
field_product_agg->Retract(3.75, 0.0));
+ ASSERT_EQ(DataDefine::GetVariantValue<double>(positive_ret),
+ std::numeric_limits<double>::infinity());
+ ASSERT_OK_AND_ASSIGN(VariantType negative_ret,
field_product_agg->Retract(-3.75, 0.0));
+ ASSERT_EQ(DataDefine::GetVariantValue<double>(negative_ret),
+ -std::numeric_limits<double>::infinity());
+ // A negative zero flips the sign just as a negative dividend does,
and product itself
+ // produces one whenever a negative value is multiplied by zero.
+ ASSERT_OK_AND_ASSIGN(VariantType negative_zero_ret,
field_product_agg->Retract(3.75, -0.0));
+ ASSERT_EQ(DataDefine::GetVariantValue<double>(negative_zero_ret),
+ -std::numeric_limits<double>::infinity());
+ // zero over zero is the one case that is not an infinity
+ ASSERT_OK_AND_ASSIGN(VariantType nan_ret,
field_product_agg->Retract(0.0, 0.0));
+ ASSERT_TRUE(std::isnan(DataDefine::GetVariantValue<double>(nan_ret)));
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldProductAgg>
field_product_agg,
+ FieldProductAgg::Create(arrow::float32(),
GetDefaultPool()));
+ ASSERT_OK_AND_ASSIGN(
+ VariantType positive_ret,
+ field_product_agg->Retract(static_cast<float>(3.75),
static_cast<float>(0.0)));
+ ASSERT_EQ(DataDefine::GetVariantValue<float>(positive_ret),
+ std::numeric_limits<float>::infinity());
+ ASSERT_OK_AND_ASSIGN(
+ VariantType nan_ret,
+ field_product_agg->Retract(static_cast<float>(0.0),
static_cast<float>(0.0)));
+ ASSERT_TRUE(std::isnan(DataDefine::GetVariantValue<float>(nan_ret)));
+ }
+}
+
+TEST(FieldProductAggTest, TestInvalidType) {
+ ASSERT_NOK_WITH_MSG(FieldProductAgg::Create(arrow::boolean(),
GetDefaultPool()),
+ "type bool not support in FieldProductAgg");
+ ASSERT_NOK_WITH_MSG(FieldProductAgg::Create(arrow::utf8(),
GetDefaultPool()),
+ "not support in FieldProductAgg");
+ // rescaling by the field scale needs a scale within [0, precision]
+ ASSERT_NOK_WITH_MSG(FieldProductAgg::Create(arrow::decimal128(20, 22),
GetDefaultPool()),
+ "precision must >= scale");
+ ASSERT_NOK_WITH_MSG(FieldProductAgg::Create(arrow::decimal128(10, -2),
GetDefaultPool()),
+ "scale must >= 0");
+}
+} // namespace paimon::test
diff --git a/test/inte/write_and_read_inte_test.cpp
b/test/inte/write_and_read_inte_test.cpp
index 7ce6f2e6..cca4f786 100644
--- a/test/inte/write_and_read_inte_test.cpp
+++ b/test/inte/write_and_read_inte_test.cpp
@@ -680,6 +680,65 @@ TEST_P(WriteAndReadInteTest, TestPKSimple) {
ASSERT_TRUE(success);
}
+TEST_P(WriteAndReadInteTest, TestPKProductAggMergesAndSurvivesCompaction) {
+ arrow::FieldVector fields = {arrow::field("pk", arrow::utf8()),
+ arrow::field("qty", arrow::int32()),
+ arrow::field("rate", arrow::decimal128(10,
2))};
+ auto [file_format, file_system] = GetParam();
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::TARGET_FILE_SIZE, "1024"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {"fields.qty.aggregate-function", "product"},
+ {"fields.rate.aggregate-function", "product"},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<TestHelper> helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ // Every decimal product below is exact at scale 2, so the result does not
depend on the order
+ // the files happen to be merged in.
+ const char* batches[] = {R"([["a", 2, "1.50"], ["b", 2, "2.00"]])",
+ R"([["a", 3, "2.00"], ["b", 1, "0.50"]])",
+ R"([["a", 5, "1.00"], ["b", 4, "1.00"]])"};
+ for (int64_t i = 0; i < 3; i++) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
+
TestHelper::MakeRecordBatch(arrow::struct_(fields), batches[i],
+ /*partition_map=*/{},
/*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(batch),
/*commit_identifier=*/i,
+
/*expected_commit_messages=*/std::nullopt));
+ }
+
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND",
arrow::int8()));
+ // a: 2 * 3 * 5 = 30 and 1.50 * 2.00 * 1.00 = 3.00
+ // b: 2 * 1 * 4 = 8 and 2.00 * 0.50 * 1.00 = 1.00
+ const char* expected = R"([[0, "a", 30, "3.00"], [0, "b", 8, "1.00"]])";
+
+ // merge on read across the three level-0 files
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(bool success,
helper->ReadAndCheckResult(arrow::struct_(result_fields),
+ data_splits,
expected));
+ ASSERT_TRUE(success);
+
+ // the same aggregation performed by a full compaction
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ ASSERT_OK(CompactAndCommit(table_path, options, /*commit_identifier=*/3));
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> compacted_splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt));
+ ASSERT_OK_AND_ASSIGN(
+ bool compacted_success,
+ helper->ReadAndCheckResult(arrow::struct_(result_fields),
compacted_splits, expected));
+ ASSERT_TRUE(compacted_success);
+}
+
TEST_P(WriteAndReadInteTest, TestPKListAggPreservesResultsAcrossKeys) {
arrow::FieldVector fields = {arrow::field("pk", arrow::utf8()),
arrow::field("value", arrow::utf8())};
