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));