Skip to content

[core] Add option to disable overflow exception in sum 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

Open
tsreaper wants to merge 1 commit into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 3 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,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.<field-name>.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.
Expand Down
7 changes: 7 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,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));

@JingsongLi JingsongLi Sep 16, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe default value should be false?

}

public List<String> fieldNestedUpdateAggNestedKey(String fieldName) {
String keyString =
options.get(
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
10 changes: 9 additions & 1 deletion ...c/main/java/org/apache/paimon/mergetree/compact/aggregate/factory/FieldSumAggFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,24 +19,32 @@
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}. */
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
Expand Down
3 changes: 1 addition & 2 deletions ...est/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
2 changes: 1 addition & 1 deletion ...st/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorRetractNullTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}

Expand Down
Loading
Loading