feat(aggregate): support product aggregate function

[SteNicholas]

Purpose

Linked issue: close #317

product is one of the aggregate functions Java Paimon supports, through FieldProductAgg and FieldProductAggFactory, but FieldAggregatorFactory::CreateFieldAggregator has no branch for it. A primary-key table configured with 'merge-engine' = 'aggregation' and 'fields.<field-name>.aggregate-function' = 'product' is therefore unusable: the aggregators for every value field are built as a group, so the whole merge function fails with Invalid: Use unsupported aggregation product or spell aggregate function incorrectly! as soon as the aggregation logic is initialized — eagerly in FileStoreWrite::Create on the write path, and on the first merge on the read path. That blocks writes, merge reads and compaction of tables a Java writer already produces.

This PR ports FieldProductAgg, following the shape of the existing FieldSumAgg: the per-type arithmetic is resolved once in Create() into a std::function, so the merge path does not switch on the field type per row. Agg multiplies the accumulator by the input and Retract divides it, both keeping Java's null handling — Agg returns whichever side is non-null, while Retract returns the accumulator unchanged when either side is null, so retracting into a null accumulator stays null rather than producing a reciprocal the way sum produces a negation.

Supported types are Java's NUMERIC family: TINYINT, SMALLINT, INT, BIGINT, FLOAT, DOUBLE and DECIMAL. Anything else is rejected by Create(), matching the checkArgument in FieldProductAggFactory.

• Integer arithmetic is exact. __builtin_mul_overflow catches products that leave the field type, and division checks for a zero divisor and for MIN / -1; each returns Status::Invalid at the points where Java throws ArithmeticException. Every width computes in its own type, so 300 * 200 overflows SMALLINT instead of silently widening. TINYINT is read back through int8_t first, because the variant holds it as a plain char whose signedness follows the ABI.
• Decimal arithmetic runs through arrow::BasicDecimal256. A 128-bit intermediate is not enough: the unscaled product of two DECIMAL(38, 18) values reaches 1e76, well past int128. Multiplying doubles the scale, so the result is brought back down with ReduceScaleBy(scale, round=true), which rounds away from zero on ties and therefore matches Java's BigDecimal.setScale(scale, HALF_UP). Division cancels the scale out, so the dividend is scaled up first and the truncated quotient is then rounded half up from the remainder.

Java's two decimal edge cases are reproduced rather than smoothed over, so a table reads the same way from either engine:

• A result the field precision can no longer hold aggregates to null, because that is what Decimal.fromBigDecimal returns for it.
• Retraction rejects a quotient with no finite decimal expansion, because BigDecimal.divide without a rounding mode throws there. This is narrower than rejecting every inexact division: 1.00 / 8.00 terminates at 0.125 and is still rounded half up to 0.13, while 2.00 / 3.00 repeats and is refused. The test is whether the divisor, stripped of its factors of 2 and 5, divides the dividend — so 3.00 / 6.00 is accepted and a zero dividend always is.

Changes:

• field_product_agg.h / .cpp (new): FieldProductAgg, plus the file-local helpers it needs — ToDecimal256 / ToInt128 to move between paimon::Decimal and arrow::BasicDecimal256, MultiplyExact / DivideExact for the checked integer arithmetic, and QuotientTerminates for the finite-expansion test.
• field_aggregator_factory.h: register product next to sum.
• src/paimon/CMakeLists.txt: add the source and the test to their targets.

Cross-checked against Java: the expectations in ProductAggregationITCase reproduce exactly, including DECIMAL(5, 3) 1.01 * 1.10 * 10.00 = 11.110 and DECIMAL(4, 2) 1.01 * 1.10 * 10.00 = 11.10 where the intermediate rounds half up, and the product cases in FieldAggregatorTest (agg(null, 10) == 10, agg(1, 10) == 10, retract(10, 5) == 2, retract(null, 5) == null, and the overflow cases for all four integer widths).

Tests

New UT in field_product_agg_test.cpp:

• TestSimple, TestNull, TestSupportedTypes — multiply and divide over every accepted type, and the null combinations for both directions.
• TestIntegerRetractTruncatesTowardsZero — negative divisor, and 10 / 3 / -10 / 3 truncating towards zero rather than rounding or flooring.
• TestDecimalRoundsHalfUp — ties, below-tie, negative operands, a negative divisor, and the case where the truncated quotient is zero so the rounding step must take its sign from the operands.
• TestDecimalRejectsNonTerminatingQuotient — 2.00 / 3.00, 1.00 / 7.00, 0.01 / 1.50 rejected; 3.00 / 6.00 and 0.00 / 3.00 accepted.
• TestDecimalWithZeroScale — DECIMAL(10, 0), where the rescale steps are no-ops.
• TestDecimalWiderThanInt128Intermediate — DECIMAL(38, 18), whose unscaled intermediate does not fit an int128.
• TestIntegerArithmeticErrors — overflow, MIN / -1 and division by zero for all four integer widths.
• TestDecimalOverflowsToNull, TestDecimalDivisionByZero, TestFloatingPointDivisionByZero (signed infinities including a negative-zero divisor, and NaN for 0 / 0), TestInvalidType.

Also FieldAggregatorFactoryTest.TestSimple gains a product case, and AggregateMergeFunctionTest.TestProduct covers a multi-row merge (2 * 3 * 5) plus a DELETE retraction through the merge function.

New IT in write_and_read_inte_test.cpp: TestPKProductAggMergesAndSurvivesCompaction writes three commits with overlapping keys over an INT and a DECIMAL(10, 2) column, verifies the merge-on-read result, then full-compacts and verifies the same result. The decimal values are chosen so every intermediate product is exact at scale 2, since decimal rounding is not associative and the merge order differs between merge-on-read and compaction.

API and Format

No. No header under include/paimon/ is touched, and neither the storage format nor the protocol changes. product is already a legal value of the existing fields.<field-name>.aggregate-function option; it simply had no implementation behind it.

Documentation

Yes. docs/source/user_guide/primary_key_table.rst had no merge engine section at all, only a passing mention of "the user-specified merge engine" under LSM trees. This PR adds a Merge Engines section with an Aggregation subsection covering the options, the fallback order, the available function names and retraction handling, and a product subsection covering the accepted types, the null semantics, integer exactness and truncation, IEEE 754 behaviour for FLOAT and DOUBLE, and the two DECIMAL boundary behaviours above.

The other aggregate functions are listed by name only. Documenting their semantics was left out deliberately rather than written from their names, since this change did not verify them.

Generative AI tooling

Generated-by: Claude Code (claude-opus-5)

[lxy-9602]

+1

[lxy-9602]

Thank you for the contribution and for improving the documentation! 👍