From 6e5af02bb392665eba3e9c72a6ec3b3450c75c66 Mon Sep 17 00:00:00 2001 From: tsreaper Date: Wed, 16 Sep 2026 12:18:33 +0800 Subject: [PATCH 1/2] [core] Add option to disable overflow exception in sum agg --- .../merge-engine/aggregation.mdx | 3 + .../java/org/apache/paimon/CoreOptions.java | 7 ++ .../compact/aggregate/FieldSumAgg.java | 77 ++++++++++--------- .../aggregate/factory/FieldSumAggFactory.java | 10 ++- ...okupChangelogMergeFunctionWrapperTest.java | 3 +- .../FieldAggregatorRetractNullTest.java | 2 +- .../aggregate/FieldAggregatorTest.java | 53 +++++++++---- 7 files changed, 99 insertions(+), 56 deletions(-) diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx b/docs/docs/primary-key-table/merge-engine/aggregation.mdx index fd2d639700f7..4b5987a6d724 100644 --- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx +++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx @@ -96,6 +96,9 @@ Check [Retraction](#retraction) before accepting `UPDATE_BEFORE` or `DELETE` rec The sum function aggregates the values across multiple rows. It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types. + For TINYINT, SMALLINT, INTEGER, and BIGINT, overflow throws an exception by default, including + during retraction. Set `fields..sum.fail-on-overflow` to `false` to disable exception. + ### product The product function can compute product values across multiple lines. It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types. diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 68215e18e4f0..889b86f28cc2 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -3366,6 +3366,13 @@ public boolean fieldAggIgnoreRetract(String fieldName) { .defaultValue(false)); } + public boolean fieldSumAggFailOnOverflow(String fieldName) { + return options.get( + key(FIELDS_PREFIX + "." + fieldName + ".sum.fail-on-overflow") + .booleanType() + .defaultValue(true)); + } + public List fieldNestedUpdateAggNestedKey(String fieldName) { String keyString = options.get( diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java index e9d868e0825f..fce3e2a61a45 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java @@ -27,8 +27,11 @@ public class FieldSumAgg extends FieldAggregator { private static final long serialVersionUID = 1L; - public FieldSumAgg(String name, DataType dataType) { + private final boolean failOnOverflow; + + public FieldSumAgg(String name, DataType dataType, boolean failOnOverflow) { super(name, dataType); + this.failOnOverflow = failOnOverflow; } @Override @@ -55,16 +58,16 @@ public Object agg(Object accumulator, Object inputField) { mergeFieldDD.scale()); break; case TINYINT: - sum = addExactByte((byte) accumulator, (byte) inputField); + sum = addByte((byte) accumulator, (byte) inputField); break; case SMALLINT: - sum = addExactShort((short) accumulator, (short) inputField); + sum = addShort((short) accumulator, (short) inputField); break; case INTEGER: - sum = addExactInt((int) accumulator, (int) inputField); + sum = addInt((int) accumulator, (int) inputField); break; case BIGINT: - sum = addExactLong((long) accumulator, (long) inputField); + sum = addLong((long) accumulator, (long) inputField); break; case FLOAT: sum = (float) accumulator + (float) inputField; @@ -105,16 +108,16 @@ public Object retract(Object accumulator, Object inputField) { mergeFieldDD.scale()); break; case TINYINT: - sum = subtractExactByte((byte) accumulator, (byte) inputField); + sum = subtractByte((byte) accumulator, (byte) inputField); break; case SMALLINT: - sum = subtractExactShort((short) accumulator, (short) inputField); + sum = subtractShort((short) accumulator, (short) inputField); break; case INTEGER: - sum = subtractExactInt((int) accumulator, (int) inputField); + sum = subtractInt((int) accumulator, (int) inputField); break; case BIGINT: - sum = subtractExactLong((long) accumulator, (long) inputField); + sum = subtractLong((long) accumulator, (long) inputField); break; case FLOAT: sum = (float) accumulator - (float) inputField; @@ -142,13 +145,13 @@ private Object negative(Object value) { return Decimal.fromBigDecimal( decimal.toBigDecimal().negate(), decimal.precision(), decimal.scale()); case TINYINT: - return negateExactByte((byte) value); + return negateByte((byte) value); case SMALLINT: - return negateExactShort((short) value); + return negateShort((short) value); case INTEGER: - return negateExactInt((int) value); + return negateInt((int) value); case BIGINT: - return negateExactLong((long) value); + return negateLong((long) value); case FLOAT: return -((float) value); case DOUBLE: @@ -162,101 +165,101 @@ private Object negative(Object value) { } } - private static byte addExactByte(byte a, byte b) { + private byte addByte(byte a, byte b) { int value = a + b; - if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) { + if (failOnOverflow && (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE)) { throw new ArithmeticException( String.format("byte overflow: %d + %d = %d", a, b, value)); } return (byte) value; } - private static short addExactShort(short a, short b) { + private short addShort(short a, short b) { int value = a + b; - if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) { + if (failOnOverflow && (value > Short.MAX_VALUE || value < Short.MIN_VALUE)) { throw new ArithmeticException( String.format("short overflow: %d + %d = %d", a, b, value)); } return (short) value; } - private static int addExactInt(int a, int b) { + private int addInt(int a, int b) { try { - return Math.addExact(a, b); + return failOnOverflow ? Math.addExact(a, b) : a + b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("int overflow: %d + %d", a, b)); } } - private static long addExactLong(long a, long b) { + private long addLong(long a, long b) { try { - return Math.addExact(a, b); + return failOnOverflow ? Math.addExact(a, b) : a + b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("long overflow: %d + %d", a, b)); } } - private static byte subtractExactByte(byte a, byte b) { + private byte subtractByte(byte a, byte b) { int value = a - b; - if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) { + if (failOnOverflow && (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE)) { throw new ArithmeticException( String.format("byte overflow: %d - %d = %d", a, b, value)); } return (byte) value; } - private static short subtractExactShort(short a, short b) { + private short subtractShort(short a, short b) { int value = a - b; - if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) { + if (failOnOverflow && (value > Short.MAX_VALUE || value < Short.MIN_VALUE)) { throw new ArithmeticException( String.format("short overflow: %d - %d = %d", a, b, value)); } return (short) value; } - private static int subtractExactInt(int a, int b) { + private int subtractInt(int a, int b) { try { - return Math.subtractExact(a, b); + return failOnOverflow ? Math.subtractExact(a, b) : a - b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("int overflow: %d - %d", a, b)); } } - private static long subtractExactLong(long a, long b) { + private long subtractLong(long a, long b) { try { - return Math.subtractExact(a, b); + return failOnOverflow ? Math.subtractExact(a, b) : a - b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("long overflow: %d - %d", a, b)); } } - private static byte negateExactByte(byte a) { + private byte negateByte(byte a) { int value = -a; - if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) { + if (failOnOverflow && (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE)) { throw new ArithmeticException(String.format("byte overflow: -%d = %d", a, value)); } return (byte) value; } - private static short negateExactShort(short a) { + private short negateShort(short a) { int value = -a; - if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) { + if (failOnOverflow && (value > Short.MAX_VALUE || value < Short.MIN_VALUE)) { throw new ArithmeticException(String.format("short overflow: -%d = %d", a, value)); } return (short) value; } - private static int negateExactInt(int a) { + private int negateInt(int a) { try { - return Math.negateExact(a); + return failOnOverflow ? Math.negateExact(a) : -a; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("int overflow: -%d", a)); } } - private static long negateExactLong(long a) { + private long negateLong(long a) { try { - return Math.negateExact(a); + return failOnOverflow ? Math.negateExact(a) : -a; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("long overflow: -%d", a)); } diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java index 5343f67b6ad7..c1d565623748 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java @@ -19,10 +19,13 @@ package org.apache.paimon.mergetree.compact.aggregate.factory; import org.apache.paimon.CoreOptions; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.mergetree.compact.aggregate.FieldSumAgg; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeFamily; +import java.util.Collections; + import static org.apache.paimon.utils.Preconditions.checkArgument; /** Factory for #{@link FieldSumAgg}. */ @@ -30,13 +33,18 @@ public class FieldSumAggFactory implements FieldAggregatorFactory { public static final String NAME = "sum"; + @VisibleForTesting + public FieldSumAgg create(DataType fieldType) { + return create(fieldType, CoreOptions.fromMap(Collections.emptyMap()), null); + } + @Override public FieldSumAgg create(DataType fieldType, CoreOptions options, String field) { checkArgument( fieldType.getTypeRoot().getFamilies().contains(DataTypeFamily.NUMERIC), "Data type for sum column must be 'NumericType' but was '%s'.", fieldType); - return new FieldSumAgg(identifier(), fieldType); + return new FieldSumAgg(identifier(), fieldType, options.fieldSumAggFailOnOverflow(field)); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java index 57d99557ca5f..db063462fb09 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java @@ -286,8 +286,7 @@ public void testSum(boolean changelogRowDeduplicate) { row -> row.isNullAt(0) ? null : row.getInt(0) }, new FieldAggregator[] { - new FieldSumAggFactory() - .create(DataTypes.INT(), null, null) + new FieldSumAggFactory().create(DataTypes.INT()) }, false, null), diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java index cc6f5dfbd366..45cfd53dfeb9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java @@ -75,7 +75,7 @@ public void testFieldFirstNonNullValueAgg() { @Test public void testFieldSumAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(DataTypes.INT(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(DataTypes.INT()); assertThat(fieldSumAgg.retract(1, 1)).isNotNull(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java index 6cc3c73a01c6..fd42aa3aaa34 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java @@ -622,7 +622,7 @@ public void testFieldMaxMinAggWithIncomparableTypeShouldFail() { @Test public void testFieldSumIntAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType()); assertThat(fieldSumAgg.agg(null, 10)).isEqualTo(10); assertThat(fieldSumAgg.agg(1, 10)).isEqualTo(11); assertThat(fieldSumAgg.retract(10, 5)).isEqualTo(5); @@ -641,7 +641,7 @@ public void testFieldProductIntAgg() { @Test public void testFieldSumByteAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType()); assertThat(fieldSumAgg.agg(null, (byte) 10)).isEqualTo((byte) 10); assertThat(fieldSumAgg.agg((byte) 1, (byte) 10)).isEqualTo((byte) 11); assertThat(fieldSumAgg.retract((byte) 10, (byte) 5)).isEqualTo((byte) 5); @@ -670,7 +670,7 @@ public void testFieldProductShortAgg() { @Test public void testFieldSumShortAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType()); assertThat(fieldSumAgg.agg(null, (short) 10)).isEqualTo((short) 10); assertThat(fieldSumAgg.agg((short) 1, (short) 10)).isEqualTo((short) 11); assertThat(fieldSumAgg.retract((short) 10, (short) 5)).isEqualTo((short) 5); @@ -679,7 +679,7 @@ public void testFieldSumShortAgg() { @Test public void testFieldSumLongAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType()); assertThat(fieldSumAgg.agg(null, 10L)).isEqualTo(10L); assertThat(fieldSumAgg.agg(1L, 10L)).isEqualTo(11L); assertThat(fieldSumAgg.retract(10L, 5L)).isEqualTo(5L); @@ -770,16 +770,39 @@ public void testFieldProductLongRetractOverflow() { @Test public void testFieldSumByteOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MAX_VALUE, (byte) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MIN_VALUE, (byte) -1)) .isInstanceOf(ArithmeticException.class); } + @Test + public void testFieldSumOverflowDisabled() { + assertSumOverflowDisabled( + new TinyIntType(), Byte.MIN_VALUE, Byte.MAX_VALUE, (byte) 1, (byte) -1); + assertSumOverflowDisabled( + new SmallIntType(), Short.MIN_VALUE, Short.MAX_VALUE, (short) 1, (short) -1); + assertSumOverflowDisabled(new IntType(), Integer.MIN_VALUE, Integer.MAX_VALUE, 1, -1); + assertSumOverflowDisabled(new BigIntType(), Long.MIN_VALUE, Long.MAX_VALUE, 1L, -1L); + } + + private void assertSumOverflowDisabled( + DataType type, Object min, Object max, Object one, Object minusOne) { + CoreOptions coreOptions = + CoreOptions.fromMap( + Collections.singletonMap("fields.f.sum.fail-on-overflow", "false")); + FieldSumAgg agg = new FieldSumAggFactory().create(type, coreOptions, "f"); + assertThat(agg.agg(max, one)).isEqualTo(min); + assertThat(agg.agg(min, minusOne)).isEqualTo(max); + assertThat(agg.retract(min, one)).isEqualTo(max); + assertThat(agg.retract(max, minusOne)).isEqualTo(min); + assertThat(agg.retract(null, min)).isEqualTo(min); + } + @Test public void testFieldSumShortOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Short.MAX_VALUE, (short) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Short.MIN_VALUE, (short) -1)) @@ -788,7 +811,7 @@ public void testFieldSumShortOverflow() { @Test public void testFieldSumIntOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MAX_VALUE, 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MIN_VALUE, -1)) @@ -797,7 +820,7 @@ public void testFieldSumIntOverflow() { @Test public void testFieldSumLongOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Long.MAX_VALUE, 1L)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Long.MIN_VALUE, -1L)) @@ -806,7 +829,7 @@ public void testFieldSumLongOverflow() { @Test public void testFieldSumByteRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MIN_VALUE, (byte) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MAX_VALUE, (byte) -1)) @@ -818,7 +841,7 @@ public void testFieldSumByteRetractOverflow() { @Test public void testFieldSumShortRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Short.MIN_VALUE, (short) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Short.MAX_VALUE, (short) -1)) @@ -829,7 +852,7 @@ public void testFieldSumShortRetractOverflow() { @Test public void testFieldSumIntRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MIN_VALUE, 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MAX_VALUE, -1)) @@ -840,7 +863,7 @@ public void testFieldSumIntRetractOverflow() { @Test public void testFieldSumLongRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Long.MIN_VALUE, 1L)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Long.MAX_VALUE, -1L)) @@ -861,7 +884,7 @@ public void testFieldProductFloatAgg() { @Test public void testFieldSumFloatAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new FloatType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new FloatType()); assertThat(fieldSumAgg.agg(null, (float) 10)).isEqualTo((float) 10); assertThat(fieldSumAgg.agg((float) 1, (float) 10)).isEqualTo((float) 11); assertThat(fieldSumAgg.retract((float) 10, (float) 5)).isEqualTo((float) 5); @@ -880,7 +903,7 @@ public void testFieldProductDoubleAgg() { @Test public void testFieldSumDoubleAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new DoubleType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new DoubleType()); assertThat(fieldSumAgg.agg(null, (double) 10)).isEqualTo((double) 10); assertThat(fieldSumAgg.agg((double) 1, (double) 10)).isEqualTo((double) 11); assertThat(fieldSumAgg.retract((double) 10, (double) 5)).isEqualTo((double) 5); @@ -899,7 +922,7 @@ public void testFieldProductDecimalAgg() { @Test public void testFieldSumDecimalAgg() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new DecimalType(), null, null); + FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new DecimalType()); assertThat(fieldSumAgg.agg(null, toDecimal(10))).isEqualTo(toDecimal(10)); assertThat(fieldSumAgg.agg(toDecimal(1), toDecimal(10))).isEqualTo(toDecimal(11)); assertThat(fieldSumAgg.retract(toDecimal(10), toDecimal(5))).isEqualTo(toDecimal(5)); From 67f62b669dc60438f46a01411faafd3162210460 Mon Sep 17 00:00:00 2001 From: tsreaper Date: Thu, 17 Sep 2026 13:52:13 +0800 Subject: [PATCH 2/2] [fix] Also fix product overflow --- .../merge-engine/aggregation.mdx | 7 +- .../java/org/apache/paimon/CoreOptions.java | 9 +- .../compact/aggregate/FieldProductAgg.java | 53 +++---- .../factory/FieldProductAggFactory.java | 11 +- .../FieldAggregatorRetractNullTest.java | 3 +- .../aggregate/FieldAggregatorTest.java | 138 +++++++++++------- 6 files changed, 140 insertions(+), 81 deletions(-) diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx b/docs/docs/primary-key-table/merge-engine/aggregation.mdx index 4b5987a6d724..3e81d1d52d7e 100644 --- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx +++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx @@ -96,13 +96,16 @@ Check [Retraction](#retraction) before accepting `UPDATE_BEFORE` or `DELETE` rec The sum function aggregates the values across multiple rows. It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types. - For TINYINT, SMALLINT, INTEGER, and BIGINT, overflow throws an exception by default, including - during retraction. Set `fields..sum.fail-on-overflow` to `false` to disable exception. + For TINYINT, SMALLINT, INTEGER, and BIGINT, overflows are ignored by default. + Set `fields..sum.fail-on-overflow` to `true` to throw an exception on overflow. ### product The product function can compute product values across multiple lines. It supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE data types. + For TINYINT, SMALLINT, INTEGER, and BIGINT, overflows are ignored by default. + Set `fields..product.fail-on-overflow` to `true` to throw an exception on overflow. + ### Counting with sum {#count} There is no `count` value for `fields..aggregate-function`. To count contributions, diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 889b86f28cc2..065b1c449630 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -3370,7 +3370,14 @@ public boolean fieldSumAggFailOnOverflow(String fieldName) { return options.get( key(FIELDS_PREFIX + "." + fieldName + ".sum.fail-on-overflow") .booleanType() - .defaultValue(true)); + .defaultValue(false)); + } + + public boolean fieldProductAggFailOnOverflow(String fieldName) { + return options.get( + key(FIELDS_PREFIX + "." + fieldName + ".product.fail-on-overflow") + .booleanType() + .defaultValue(false)); } public List fieldNestedUpdateAggNestedKey(String fieldName) { diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java index dc14b2917094..51fa0dade2d7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java @@ -30,8 +30,11 @@ public class FieldProductAgg extends FieldAggregator { private static final long serialVersionUID = 1L; - public FieldProductAgg(String name, DataType dataType) { + private final boolean failOnOverflow; + + public FieldProductAgg(String name, DataType dataType, boolean failOnOverflow) { super(name, dataType); + this.failOnOverflow = failOnOverflow; } @Override @@ -57,16 +60,16 @@ public Object agg(Object accumulator, Object inputField) { product = fromBigDecimal(mul, mergeFieldDD.precision(), mergeFieldDD.scale()); break; case TINYINT: - product = multiplyExactByte((byte) accumulator, (byte) inputField); + product = multiplyByte((byte) accumulator, (byte) inputField); break; case SMALLINT: - product = multiplyExactShort((short) accumulator, (short) inputField); + product = multiplyShort((short) accumulator, (short) inputField); break; case INTEGER: - product = multiplyExactInt((int) accumulator, (int) inputField); + product = multiplyInt((int) accumulator, (int) inputField); break; case BIGINT: - product = multiplyExactLong((long) accumulator, (long) inputField); + product = multiplyLong((long) accumulator, (long) inputField); break; case FLOAT: product = (float) accumulator * (float) inputField; @@ -84,67 +87,67 @@ public Object agg(Object accumulator, Object inputField) { return product; } - private static byte multiplyExactByte(byte a, byte b) { + private byte multiplyByte(byte a, byte b) { int value = a * b; - if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) { + if (failOnOverflow && (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE)) { throw new ArithmeticException( String.format("byte overflow: %d * %d = %d", a, b, value)); } return (byte) value; } - private static short multiplyExactShort(short a, short b) { + private short multiplyShort(short a, short b) { int value = a * b; - if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) { + if (failOnOverflow && (value > Short.MAX_VALUE || value < Short.MIN_VALUE)) { throw new ArithmeticException( String.format("short overflow: %d * %d = %d", a, b, value)); } return (short) value; } - private static int multiplyExactInt(int a, int b) { + private int multiplyInt(int a, int b) { try { - return Math.multiplyExact(a, b); + return failOnOverflow ? Math.multiplyExact(a, b) : a * b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("int overflow: %d * %d", a, b)); } } - private static long multiplyExactLong(long a, long b) { + private long multiplyLong(long a, long b) { try { - return Math.multiplyExact(a, b); + return failOnOverflow ? Math.multiplyExact(a, b) : a * b; } catch (ArithmeticException e) { throw new ArithmeticException(String.format("long overflow: %d * %d", a, b)); } } - private static byte divideExactByte(byte a, byte b) { + private byte divideByte(byte a, byte b) { int value = a / b; - if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) { + if (failOnOverflow && (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE)) { throw new ArithmeticException( String.format("byte overflow: %d / %d = %d", a, b, value)); } return (byte) value; } - private static short divideExactShort(short a, short b) { + private short divideShort(short a, short b) { int value = a / b; - if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) { + if (failOnOverflow && (value > Short.MAX_VALUE || value < Short.MIN_VALUE)) { throw new ArithmeticException( String.format("short overflow: %d / %d = %d", a, b, value)); } return (short) value; } - private static int divideExactInt(int a, int b) { - if (a == Integer.MIN_VALUE && b == -1) { + private int divideInt(int a, int b) { + if (failOnOverflow && a == Integer.MIN_VALUE && b == -1) { throw new ArithmeticException(String.format("int overflow: %d / %d", a, b)); } return a / b; } - private static long divideExactLong(long a, long b) { - if (a == Long.MIN_VALUE && b == -1L) { + private long divideLong(long a, long b) { + if (failOnOverflow && a == Long.MIN_VALUE && b == -1L) { throw new ArithmeticException(String.format("long overflow: %d / %d", a, b)); } return a / b; @@ -171,16 +174,16 @@ public Object retract(Object accumulator, Object inputField) { product = fromBigDecimal(div, mergeFieldDD.precision(), mergeFieldDD.scale()); break; case TINYINT: - product = divideExactByte((byte) accumulator, (byte) inputField); + product = divideByte((byte) accumulator, (byte) inputField); break; case SMALLINT: - product = divideExactShort((short) accumulator, (short) inputField); + product = divideShort((short) accumulator, (short) inputField); break; case INTEGER: - product = divideExactInt((int) accumulator, (int) inputField); + product = divideInt((int) accumulator, (int) inputField); break; case BIGINT: - product = divideExactLong((long) accumulator, (long) inputField); + product = divideLong((long) accumulator, (long) inputField); break; case FLOAT: product = (float) accumulator / (float) inputField; diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java index 7dbdd9f5af5a..786f65e55ff1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java @@ -19,10 +19,13 @@ package org.apache.paimon.mergetree.compact.aggregate.factory; import org.apache.paimon.CoreOptions; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.mergetree.compact.aggregate.FieldProductAgg; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypeFamily; +import java.util.Collections; + import static org.apache.paimon.utils.Preconditions.checkArgument; /** Factory for #{@link FieldProductAgg}. */ @@ -30,13 +33,19 @@ public class FieldProductAggFactory implements FieldAggregatorFactory { public static final String NAME = "product"; + @VisibleForTesting + public FieldProductAgg create(DataType fieldType) { + return create(fieldType, CoreOptions.fromMap(Collections.emptyMap()), null); + } + @Override public FieldProductAgg create(DataType fieldType, CoreOptions options, String field) { checkArgument( fieldType.getTypeRoot().getFamilies().contains(DataTypeFamily.NUMERIC), "Data type for product column must be 'NumericType' but was '%s'.", fieldType); - return new FieldProductAgg(identifier(), fieldType); + return new FieldProductAgg( + identifier(), fieldType, options.fieldProductAggFailOnOverflow(field)); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java index 45cfd53dfeb9..794a7edce740 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java @@ -185,8 +185,7 @@ public void testLastNonNullValueAgg() { @Test public void testFieldProductAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(DataTypes.INT(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(DataTypes.INT()); assertThat(fieldProductAgg.retract(1, 1)).isNotNull(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java index fd42aa3aaa34..47f0811b30e4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java @@ -631,8 +631,7 @@ public void testFieldSumIntAgg() { @Test public void testFieldProductIntAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new IntType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new IntType()); assertThat(fieldProductAgg.agg(null, 10)).isEqualTo(10); assertThat(fieldProductAgg.agg(1, 10)).isEqualTo(10); assertThat(fieldProductAgg.retract(10, 5)).isEqualTo(2); @@ -650,8 +649,7 @@ public void testFieldSumByteAgg() { @Test public void testFieldProductByteAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new TinyIntType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new TinyIntType()); assertThat(fieldProductAgg.agg(null, (byte) 10)).isEqualTo((byte) 10); assertThat(fieldProductAgg.agg((byte) 1, (byte) 10)).isEqualTo((byte) 10); assertThat(fieldProductAgg.retract((byte) 10, (byte) 5)).isEqualTo((byte) 2); @@ -660,8 +658,7 @@ public void testFieldProductByteAgg() { @Test public void testFieldProductShortAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new SmallIntType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new SmallIntType()); assertThat(fieldProductAgg.agg(null, (short) 10)).isEqualTo((short) 10); assertThat(fieldProductAgg.agg((short) 1, (short) 10)).isEqualTo((short) 10); assertThat(fieldProductAgg.retract((short) 10, (short) 5)).isEqualTo((short) 2); @@ -688,18 +685,45 @@ public void testFieldSumLongAgg() { @Test public void testFieldProductLongAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new BigIntType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new BigIntType()); assertThat(fieldProductAgg.agg(null, 10L)).isEqualTo(10L); assertThat(fieldProductAgg.agg(1L, 10L)).isEqualTo(10L); assertThat(fieldProductAgg.retract(10L, 5L)).isEqualTo(2L); assertThat(fieldProductAgg.retract(null, 5L)).isNull(); } + @Test + public void testFieldProductOverflowDisabled() { + assertProductOverflowDisabled( + new TinyIntType(), Byte.MIN_VALUE, (byte) 2, (byte) -1, (byte) 0); + assertProductOverflowDisabled( + new SmallIntType(), Short.MIN_VALUE, (short) 2, (short) -1, (short) 0); + assertProductOverflowDisabled(new IntType(), Integer.MIN_VALUE, 2, -1, 0); + assertProductOverflowDisabled(new BigIntType(), Long.MIN_VALUE, 2L, -1L, 0L); + } + + private void assertProductOverflowDisabled( + DataType type, Object min, Object two, Object minusOne, Object zero) { + CoreOptions coreOptions = + CoreOptions.fromMap( + Collections.singletonMap("fields.f.product.fail-on-overflow", "false")); + for (FieldProductAgg agg : + new FieldProductAgg[] { + new FieldProductAggFactory().create(type), + new FieldProductAggFactory().create(type, coreOptions, "f") + }) { + assertThat(agg.agg(min, two)).isEqualTo(zero); + assertThat(agg.agg(min, minusOne)).isEqualTo(min); + assertThat(agg.retract(min, minusOne)).isEqualTo(min); + assertThatThrownBy(() -> agg.retract(min, zero)) + .isInstanceOf(ArithmeticException.class) + .hasMessage("/ by zero"); + } + } + @Test public void testFieldProductByteOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new TinyIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new TinyIntType()); assertThatThrownBy(() -> fieldProductAgg.agg((byte) 64, (byte) 2)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldProductAgg.agg((byte) -64, (byte) 4)) @@ -708,8 +732,7 @@ public void testFieldProductByteOverflow() { @Test public void testFieldProductShortOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new SmallIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new SmallIntType()); assertThatThrownBy(() -> fieldProductAgg.agg((short) 1000, (short) 100)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldProductAgg.agg(Short.MIN_VALUE, (short) 2)) @@ -718,8 +741,7 @@ public void testFieldProductShortOverflow() { @Test public void testFieldProductIntOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new IntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new IntType()); assertThatThrownBy(() -> fieldProductAgg.agg(100_000, 100_000)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldProductAgg.agg(Integer.MIN_VALUE, -1)) @@ -728,8 +750,7 @@ public void testFieldProductIntOverflow() { @Test public void testFieldProductLongOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new BigIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new BigIntType()); assertThatThrownBy(() -> fieldProductAgg.agg(Long.MAX_VALUE, 2L)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldProductAgg.agg(Long.MIN_VALUE, -1L)) @@ -738,43 +759,40 @@ public void testFieldProductLongOverflow() { @Test public void testFieldProductByteRetractOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new TinyIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new TinyIntType()); assertThatThrownBy(() -> fieldProductAgg.retract(Byte.MIN_VALUE, (byte) -1)) .isInstanceOf(ArithmeticException.class); } @Test public void testFieldProductShortRetractOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new SmallIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new SmallIntType()); assertThatThrownBy(() -> fieldProductAgg.retract(Short.MIN_VALUE, (short) -1)) .isInstanceOf(ArithmeticException.class); } @Test public void testFieldProductIntRetractOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new IntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new IntType()); assertThatThrownBy(() -> fieldProductAgg.retract(Integer.MIN_VALUE, -1)) .isInstanceOf(ArithmeticException.class); } @Test public void testFieldProductLongRetractOverflow() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new BigIntType(), null, null); + FieldProductAgg fieldProductAgg = createProductWithOverflowCheck(new BigIntType()); assertThatThrownBy(() -> fieldProductAgg.retract(Long.MIN_VALUE, -1L)) .isInstanceOf(ArithmeticException.class); } - @Test - public void testFieldSumByteOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType()); - assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MAX_VALUE, (byte) 1)) - .isInstanceOf(ArithmeticException.class); - assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MIN_VALUE, (byte) -1)) - .isInstanceOf(ArithmeticException.class); + private FieldProductAgg createProductWithOverflowCheck(DataType type) { + return new FieldProductAggFactory() + .create( + type, + CoreOptions.fromMap( + Collections.singletonMap( + "fields.f.product.fail-on-overflow", "true")), + "f"); } @Test @@ -792,17 +810,31 @@ private void assertSumOverflowDisabled( CoreOptions coreOptions = CoreOptions.fromMap( Collections.singletonMap("fields.f.sum.fail-on-overflow", "false")); - FieldSumAgg agg = new FieldSumAggFactory().create(type, coreOptions, "f"); - assertThat(agg.agg(max, one)).isEqualTo(min); - assertThat(agg.agg(min, minusOne)).isEqualTo(max); - assertThat(agg.retract(min, one)).isEqualTo(max); - assertThat(agg.retract(max, minusOne)).isEqualTo(min); - assertThat(agg.retract(null, min)).isEqualTo(min); + for (FieldSumAgg agg : + new FieldSumAgg[] { + new FieldSumAggFactory().create(type), + new FieldSumAggFactory().create(type, coreOptions, "f") + }) { + assertThat(agg.agg(max, one)).isEqualTo(min); + assertThat(agg.agg(min, minusOne)).isEqualTo(max); + assertThat(agg.retract(min, one)).isEqualTo(max); + assertThat(agg.retract(max, minusOne)).isEqualTo(min); + assertThat(agg.retract(null, min)).isEqualTo(min); + } + } + + @Test + public void testFieldSumByteOverflow() { + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new TinyIntType()); + assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MAX_VALUE, (byte) 1)) + .isInstanceOf(ArithmeticException.class); + assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MIN_VALUE, (byte) -1)) + .isInstanceOf(ArithmeticException.class); } @Test public void testFieldSumShortOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new SmallIntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Short.MAX_VALUE, (short) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Short.MIN_VALUE, (short) -1)) @@ -811,7 +843,7 @@ public void testFieldSumShortOverflow() { @Test public void testFieldSumIntOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new IntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MAX_VALUE, 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MIN_VALUE, -1)) @@ -820,7 +852,7 @@ public void testFieldSumIntOverflow() { @Test public void testFieldSumLongOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new BigIntType()); assertThatThrownBy(() -> fieldSumAgg.agg(Long.MAX_VALUE, 1L)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.agg(Long.MIN_VALUE, -1L)) @@ -829,7 +861,7 @@ public void testFieldSumLongOverflow() { @Test public void testFieldSumByteRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new TinyIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MIN_VALUE, (byte) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MAX_VALUE, (byte) -1)) @@ -841,7 +873,7 @@ public void testFieldSumByteRetractOverflow() { @Test public void testFieldSumShortRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new SmallIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Short.MIN_VALUE, (short) 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Short.MAX_VALUE, (short) -1)) @@ -852,7 +884,7 @@ public void testFieldSumShortRetractOverflow() { @Test public void testFieldSumIntRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new IntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MIN_VALUE, 1)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MAX_VALUE, -1)) @@ -863,7 +895,7 @@ public void testFieldSumIntRetractOverflow() { @Test public void testFieldSumLongRetractOverflow() { - FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType()); + FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new BigIntType()); assertThatThrownBy(() -> fieldSumAgg.retract(Long.MIN_VALUE, 1L)) .isInstanceOf(ArithmeticException.class); assertThatThrownBy(() -> fieldSumAgg.retract(Long.MAX_VALUE, -1L)) @@ -872,10 +904,18 @@ public void testFieldSumLongRetractOverflow() { .isInstanceOf(ArithmeticException.class); } + private FieldSumAgg createSumWithOverflowCheck(DataType type) { + return new FieldSumAggFactory() + .create( + type, + CoreOptions.fromMap( + Collections.singletonMap("fields.f.sum.fail-on-overflow", "true")), + "f"); + } + @Test public void testFieldProductFloatAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new FloatType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new FloatType()); assertThat(fieldProductAgg.agg(null, (float) 10)).isEqualTo((float) 10); assertThat(fieldProductAgg.agg((float) 1, (float) 10)).isEqualTo((float) 10); assertThat(fieldProductAgg.retract((float) 10, (float) 5)).isEqualTo((float) 2); @@ -893,8 +933,7 @@ public void testFieldSumFloatAgg() { @Test public void testFieldProductDoubleAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new DoubleType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new DoubleType()); assertThat(fieldProductAgg.agg(null, (double) 10)).isEqualTo((double) 10); assertThat(fieldProductAgg.agg((double) 1, (double) 10)).isEqualTo((double) 10); assertThat(fieldProductAgg.retract((double) 10, (double) 5)).isEqualTo((double) 2); @@ -912,8 +951,7 @@ public void testFieldSumDoubleAgg() { @Test public void testFieldProductDecimalAgg() { - FieldProductAgg fieldProductAgg = - new FieldProductAggFactory().create(new DecimalType(), null, null); + FieldProductAgg fieldProductAgg = new FieldProductAggFactory().create(new DecimalType()); assertThat(fieldProductAgg.agg(null, toDecimal(10))).isEqualTo(toDecimal(10)); assertThat(fieldProductAgg.agg(toDecimal(1), toDecimal(10))).isEqualTo(toDecimal(10)); assertThat(fieldProductAgg.retract(toDecimal(10), toDecimal(5))).isEqualTo(toDecimal(2));