[core] Fix silent data corruption due to overflow in FieldProductAgg

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/aggregate/FieldProductAgg.java

@@ -57,16 +57,16 @@ public Object agg(Object accumulator, Object inputField) {
                 product = fromBigDecimal(mul, mergeFieldDD.precision(), mergeFieldDD.scale());
                 break;
             case TINYINT:
-                product = (byte) ((byte) accumulator * (byte) inputField);
+                product = toExactByte((byte) accumulator * (byte) inputField);
                 break;
             case SMALLINT:
-                product = (short) ((short) accumulator * (short) inputField);
+                product = toExactShort((short) accumulator * (short) inputField);
                 break;
             case INTEGER:
-                product = (int) accumulator * (int) inputField;
+                product = Math.multiplyExact((int) accumulator, (int) inputField);
                 break;
             case BIGINT:
-                product = (long) accumulator * (long) inputField;
+                product = Math.multiplyExact((long) accumulator, (long) inputField);
                 break;
             case FLOAT:
                 product = (float) accumulator * (float) inputField;
@@ -84,6 +84,20 @@ public Object agg(Object accumulator, Object inputField) {
         return product;
     }
 
+    private static byte toExactByte(int value) {
+        if (value > Byte.MAX_VALUE || value < Byte.MIN_VALUE) {
+            throw new ArithmeticException("byte overflow");
+        }
+        return (byte) value;
+    }
+
+    private static short toExactShort(int value) {
+        if (value > Short.MAX_VALUE || value < Short.MIN_VALUE) {
+            throw new ArithmeticException("short overflow");
+        }
+        return (short) value;
+    }
+
     @Override
     public Object retract(Object accumulator, Object inputField) {
         Object product;

paimon-core/src/test/java/org/apache/paimon/mergetree/compact/aggregate/FieldAggregatorTest.java

@@ -503,6 +503,46 @@ public void testFieldProductLongAgg() {
         assertThat(fieldProductAgg.retract(null, 5L)).isNull();
     }
 
+    @Test
+    public void testFieldProductByteOverflow() {
+        FieldProductAgg fieldProductAgg =
+                new FieldProductAggFactory().create(new TinyIntType(), null, null);
+        assertThatThrownBy(() -> fieldProductAgg.agg((byte) 64, (byte) 2))
+                .isInstanceOf(ArithmeticException.class);
+        assertThatThrownBy(() -> fieldProductAgg.agg((byte) -64, (byte) 4))
+                .isInstanceOf(ArithmeticException.class);
+    }
+
+    @Test
+    public void testFieldProductShortOverflow() {
+        FieldProductAgg fieldProductAgg =
+                new FieldProductAggFactory().create(new SmallIntType(), null, null);
+        assertThatThrownBy(() -> fieldProductAgg.agg((short) 1000, (short) 100))
+                .isInstanceOf(ArithmeticException.class);
+        assertThatThrownBy(() -> fieldProductAgg.agg(Short.MIN_VALUE, (short) 2))
+                .isInstanceOf(ArithmeticException.class);
+    }
+
+    @Test
+    public void testFieldProductIntOverflow() {
+        FieldProductAgg fieldProductAgg =
+                new FieldProductAggFactory().create(new IntType(), null, null);
+        assertThatThrownBy(() -> fieldProductAgg.agg(100_000, 100_000))
+                .isInstanceOf(ArithmeticException.class);
+        assertThatThrownBy(() -> fieldProductAgg.agg(Integer.MIN_VALUE, -1))
+                .isInstanceOf(ArithmeticException.class);
+    }
+
+    @Test
+    public void testFieldProductLongOverflow() {
+        FieldProductAgg fieldProductAgg =
+                new FieldProductAggFactory().create(new BigIntType(), null, null);
+        assertThatThrownBy(() -> fieldProductAgg.agg(Long.MAX_VALUE, 2L))
+                .isInstanceOf(ArithmeticException.class);
+        assertThatThrownBy(() -> fieldProductAgg.agg(Long.MIN_VALUE, -1L))
+                .isInstanceOf(ArithmeticException.class);
+    }
+
     @Test
     public void testFieldProductFloatAgg() {
         FieldProductAgg fieldProductAgg =