Skip to content

[core] Add option to disable overflow exception in sum/product agg #9874

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 2 commits into from
Sep 17, 2026
Merged

[core] Add option to disable overflow exception in sum/product agg #9874

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6e5af02
[core] Add option to disable overflow exception in sum agg
tsreaper Sep 16, 2026
67f62b6
[fix] Also fix product overflow
tsreaper Sep 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/docs/primary-key-table/merge-engine/aggregation.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
14 changes: 14 additions & 0 deletions paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3366,6 +3366,20 @@ 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(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(
Expand Down
53 changes: 28 additions & 25 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand Down
77 changes: 40 additions & 37 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldSumAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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:
Expand All @@ -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));
}
Expand Down
Loading
Loading