Use exact math in flat hash estimated size

core/trino-main/src/main/java/io/trino/operator/FlatGroupByHash.java

@@ -32,6 +32,7 @@
 import static com.google.common.base.Verify.verify;
 import static io.airlift.slice.SizeOf.instanceSize;
 import static io.airlift.slice.SizeOf.sizeOf;
+import static io.trino.operator.FlatHash.sumExact;
 import static java.lang.Math.min;
 import static java.lang.Math.multiplyExact;
 
@@ -93,10 +94,11 @@ public long getRawHash(int groupId)
     @Override
     public long getEstimatedSize()
     {
-        return INSTANCE_SIZE +
-                flatHash.getEstimatedSize() +
-                currentPageSizeInBytes +
-                (dictionaryLookBack != null ? dictionaryLookBack.getRetainedSizeInBytes() : 0);
+        return sumExact(
+                INSTANCE_SIZE,
+                flatHash.getEstimatedSize(),
+                currentPageSizeInBytes,
+                (dictionaryLookBack != null ? dictionaryLookBack.getRetainedSizeInBytes() : 0));
     }
 
     @Override
@@ -279,9 +281,10 @@ public void setProcessed(int position, int groupId)
 
         public long getRetainedSizeInBytes()
         {
-            return INSTANCE_SIZE +
-                    sizeOf(processed) +
-                    dictionary.getRetainedSizeInBytes();
+            return sumExact(
+                    INSTANCE_SIZE,
+                    sizeOf(processed),
+                    dictionary.getRetainedSizeInBytes());
         }
     }
 

core/trino-main/src/main/java/io/trino/operator/FlatHash.java

@@ -30,6 +30,7 @@
 import static io.trino.operator.VariableWidthData.POINTER_SIZE;
 import static io.trino.spi.StandardErrorCode.GENERIC_INSUFFICIENT_RESOURCES;
 import static io.trino.spi.type.BigintType.BIGINT;
+import static java.lang.Math.addExact;
 import static java.lang.Math.max;
 import static java.lang.Math.multiplyExact;
 import static java.lang.Math.toIntExact;
@@ -121,12 +122,13 @@ private static byte[][] createRecordGroups(int capacity, int recordSize)
 
     public long getEstimatedSize()
     {
-        return INSTANCE_SIZE +
-                sizeOf(control) +
-                (sizeOf(recordGroups[0]) * recordGroups.length) +
-                (variableWidthData == null ? 0 : variableWidthData.getRetainedSizeBytes()) +
-                sizeOf(groupRecordIndex) +
-                rehashMemoryReservation;
+        return sumExact(
+                INSTANCE_SIZE,
+                sizeOf(control),
+                multiplyExact(sizeOf(recordGroups[0]), recordGroups.length),
+                (variableWidthData == null ? 0 : variableWidthData.getRetainedSizeBytes()),
+                sizeOf(groupRecordIndex),
+                rehashMemoryReservation);
     }
 
     public int size()
@@ -469,4 +471,13 @@ public int getPhysicalPosition(int groupId)
     {
         return groupRecordIndex[groupId];
     }
+
+    public static long sumExact(long... values)
+    {
+        long result = 0;
+        for (long value : values) {
+            result = addExact(result, value);
+        }
+        return result;
+    }
 }

core/trino-main/src/main/java/io/trino/operator/VariableWidthData.java

@@ -27,8 +27,12 @@
 import static io.airlift.slice.SizeOf.SIZE_OF_INT;
 import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
 import static io.airlift.slice.SizeOf.instanceSize;
+import static io.airlift.slice.SizeOf.sizeOf;
 import static io.airlift.slice.SizeOf.sizeOfObjectArray;
+import static io.trino.operator.FlatHash.sumExact;
+import static java.lang.Math.addExact;
 import static java.lang.Math.max;
+import static java.lang.Math.subtractExact;
 import static java.util.Objects.checkIndex;
 
 public final class VariableWidthData
@@ -70,14 +74,17 @@ public VariableWidthData(List<byte[]> chunks, int openChunkOffset)
     {
         this.chunks.addAll(chunks);
         this.openChunkOffset = openChunkOffset;
-        this.chunksRetainedSizeInBytes = chunks.stream().mapToLong(SizeOf::sizeOf).sum();
+        this.chunksRetainedSizeInBytes = chunks.stream().mapToLong(SizeOf::sizeOf).reduce(0L, Math::addExact);
         this.allocatedBytes = chunks.stream().mapToLong(chunk -> chunk.length).sum();
         this.freeBytes = 0;
     }
 
     public long getRetainedSizeBytes()
     {
-        return INSTANCE_SIZE + chunksRetainedSizeInBytes + sizeOfObjectArray(chunks.size());
+        return sumExact(
+                INSTANCE_SIZE,
+                chunksRetainedSizeInBytes,
+                sizeOfObjectArray(chunks.size()));
     }
 
     public List<byte[]> getAllChunks()
@@ -116,7 +123,7 @@ public byte[] allocate(byte[] pointer, int pointerOffset, int size)
             openChunk = new byte[newSize];
             chunks.add(openChunk);
             allocatedBytes += newSize;
-            chunksRetainedSizeInBytes += SizeOf.sizeOf(openChunk);
+            chunksRetainedSizeInBytes = addExact(chunksRetainedSizeInBytes, sizeOf(openChunk));
             openChunkOffset = 0;
         }
 
@@ -152,7 +159,7 @@ public void free(byte[] pointer, int pointerOffset)
         // if this is the only value written to the chunk, we can simply replace the chunk with the empty chunk
         if (valueLength == valueChunk.length) {
             chunks.set(valueChunkIndex, EMPTY_CHUNK);
-            chunksRetainedSizeInBytes -= SizeOf.sizeOf(valueChunk);
+            chunksRetainedSizeInBytes = subtractExact(chunksRetainedSizeInBytes, sizeOf(valueChunk));
             allocatedBytes -= valueChunk.length;
             return;
         }