You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ho...@apache.org on 2026/09/20 04:08:12 UTC
(paimon) 09/20: [core] Add option to disable overflow exception in sum/product agg (#9874)
This is an automated email from the ASF dual-hosted git repository.
XiaoHongbo-Hope pushed a commit to branch release-2.1
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit bcd1681e740466c04f825755ca7d60975bae2928
Author: tsreaper <ts...@gmail.com>
AuthorDate: Thu Sep 17 16:51:07 2026 +0800
[core] Add option to disable overflow exception in sum/product agg (#9874)
---
.../primary-key-table/merge-engine/aggregation.mdx | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 14 ++
.../compact/aggregate/FieldProductAgg.java | 53 ++++----
.../mergetree/compact/aggregate/FieldSumAgg.java | 77 ++++++-----
.../aggregate/factory/FieldProductAggFactory.java | 11 +-
.../aggregate/factory/FieldSumAggFactory.java | 10 +-
.../LookupChangelogMergeFunctionWrapperTest.java | 3 +-
.../aggregate/FieldAggregatorRetractNullTest.java | 5 +-
.../compact/aggregate/FieldAggregatorTest.java | 151 +++++++++++++++------
9 files changed, 216 insertions(+), 114 deletions(-)
diff --git a/docs/docs/primary-key-table/merge-engine/aggregation.mdx b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
index fd2d639700..3e81d1d52d 100644
--- a/docs/docs/primary-key-table/merge-engine/aggregation.mdx
+++ b/docs/docs/primary-key-table/merge-engine/aggregation.mdx
@@ -96,10 +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, overflows are ignored by default.
+ Set `fields.<field-name>.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.<field-name>.product.fail-on-overflow` to `true` to throw an exception on overflow.
+
### Counting with sum {#count}
There is no `count` value for `fields.<field-name>.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 637a8ba0e7..60605bc74c 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -3378,6 +3378,20 @@ public class CoreOptions implements Serializable {
.defaultValue(false));
}
+ public boolean fieldSumAggFailOnOverflow(String fieldName) {
+ return options.get(
+ key(FIELDS_PREFIX + "." + fieldName + ".sum.fail-on-overflow")
+ .booleanType()
+ .defaultValue(false));
+ }
+
+ public boolean fieldProductAggFailOnOverflow(String fieldName) {
+ return options.get(
+ key(FIELDS_PREFIX + "." + fieldName + ".product.fail-on-overflow")
+ .booleanType()
+ .defaultValue(false));
+ }
+
public List<String> fieldNestedUpdateAggNestedKey(String fieldName) {
String keyString =
options.get(
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 dc14b29170..51fa0dade2 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 class FieldProductAgg extends FieldAggregator {
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 class FieldProductAgg extends FieldAggregator {
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 class FieldProductAgg extends FieldAggregator {
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/FieldSumAgg.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
index e9d868e082..fce3e2a61a 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 class FieldSumAgg extends FieldAggregator {
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 class FieldSumAgg extends FieldAggregator {
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 @@ public class FieldSumAgg extends FieldAggregator {
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 @@ public class FieldSumAgg extends FieldAggregator {
}
}
- 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/FieldProductAggFactory.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldProductAggFactory.java
index 7dbdd9f5af..786f65e55f 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/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 5343f67b6a..c1d5656237 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 57d99557ca..db063462fb 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 class LookupChangelogMergeFunctionWrapperTest {
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 cc6f5dfbd3..794a7edce7 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 class FieldAggregatorRetractNullTest {
@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();
}
@@ -185,8 +185,7 @@ public class FieldAggregatorRetractNullTest {
@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 2dc2eb7890..820888d7a1 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 class FieldAggregatorTest {
@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);
@@ -631,8 +631,7 @@ public class FieldAggregatorTest {
@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);
@@ -641,7 +640,7 @@ public class FieldAggregatorTest {
@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);
@@ -650,8 +649,7 @@ public class FieldAggregatorTest {
@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 class FieldAggregatorTest {
@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);
@@ -670,7 +667,7 @@ public class FieldAggregatorTest {
@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 +676,7 @@ public class FieldAggregatorTest {
@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);
@@ -688,18 +685,45 @@ public class FieldAggregatorTest {
@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 class FieldAggregatorTest {
@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 class FieldAggregatorTest {
@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 class FieldAggregatorTest {
@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,39 +759,73 @@ public class FieldAggregatorTest {
@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);
}
+ private FieldProductAgg createProductWithOverflowCheck(DataType type) {
+ return new FieldProductAggFactory()
+ .create(
+ type,
+ CoreOptions.fromMap(
+ Collections.singletonMap(
+ "fields.f.product.fail-on-overflow", "true")),
+ "f");
+ }
+
+ @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"));
+ 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 = new FieldSumAggFactory().create(new TinyIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new TinyIntType());
assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MAX_VALUE, (byte) 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.agg(Byte.MIN_VALUE, (byte) -1))
@@ -779,7 +834,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumShortOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new SmallIntType());
assertThatThrownBy(() -> fieldSumAgg.agg(Short.MAX_VALUE, (short) 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.agg(Short.MIN_VALUE, (short) -1))
@@ -788,7 +843,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumIntOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new IntType());
assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MAX_VALUE, 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.agg(Integer.MIN_VALUE, -1))
@@ -797,7 +852,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumLongOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new BigIntType());
assertThatThrownBy(() -> fieldSumAgg.agg(Long.MAX_VALUE, 1L))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.agg(Long.MIN_VALUE, -1L))
@@ -806,7 +861,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumByteRetractOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new TinyIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new TinyIntType());
assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MIN_VALUE, (byte) 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.retract(Byte.MAX_VALUE, (byte) -1))
@@ -818,7 +873,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumShortRetractOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new SmallIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new SmallIntType());
assertThatThrownBy(() -> fieldSumAgg.retract(Short.MIN_VALUE, (short) 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.retract(Short.MAX_VALUE, (short) -1))
@@ -829,7 +884,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumIntRetractOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new IntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new IntType());
assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MIN_VALUE, 1))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.retract(Integer.MAX_VALUE, -1))
@@ -840,7 +895,7 @@ public class FieldAggregatorTest {
@Test
public void testFieldSumLongRetractOverflow() {
- FieldSumAgg fieldSumAgg = new FieldSumAggFactory().create(new BigIntType(), null, null);
+ FieldSumAgg fieldSumAgg = createSumWithOverflowCheck(new BigIntType());
assertThatThrownBy(() -> fieldSumAgg.retract(Long.MIN_VALUE, 1L))
.isInstanceOf(ArithmeticException.class);
assertThatThrownBy(() -> fieldSumAgg.retract(Long.MAX_VALUE, -1L))
@@ -849,10 +904,18 @@ public class FieldAggregatorTest {
.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);
@@ -861,7 +924,7 @@ public class FieldAggregatorTest {
@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);
@@ -870,8 +933,7 @@ public class FieldAggregatorTest {
@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);
@@ -880,7 +942,7 @@ public class FieldAggregatorTest {
@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);
@@ -889,8 +951,7 @@ public class FieldAggregatorTest {
@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));
@@ -899,7 +960,7 @@ public class FieldAggregatorTest {
@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));