From e69e066cbfa9fa190bd42f7f2c00953eb80b8d50 Mon Sep 17 00:00:00 2001
From: Rebecca Schlussel <rschlussel@fb.com>
Date: Thu, 5 Nov 2020 11:55:08 -0500
Subject: [PATCH] Fix int overflow error when spilling large page

Page serialization requires page size to fin in an integer, so larger
pages could hit an int overflow error.
---
 .../presto/spiller/FileSingleStreamSpiller.java | 17 +++++++++++------
 1 file changed, 11 insertions(+), 6 deletions(-)

diff --git a/presto-main/src/main/java/com/facebook/presto/spiller/FileSingleStreamSpiller.java b/presto-main/src/main/java/com/facebook/presto/spiller/FileSingleStreamSpiller.java
index 5fba1f9c94588..ae57aa589d6a4 100644
--- a/presto-main/src/main/java/com/facebook/presto/spiller/FileSingleStreamSpiller.java
+++ b/presto-main/src/main/java/com/facebook/presto/spiller/FileSingleStreamSpiller.java
@@ -19,7 +19,6 @@
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.page.PagesSerde;
 import com.facebook.presto.spi.page.PagesSerdeUtil;
-import com.facebook.presto.spi.page.SerializedPage;
 import com.facebook.presto.spi.spiller.SpillCipher;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.AbstractIterator;
@@ -44,6 +43,8 @@
 import java.util.List;
 import java.util.Optional;
 
+import static com.facebook.presto.common.block.PageBuilderStatus.DEFAULT_MAX_PAGE_SIZE_IN_BYTES;
+import static com.facebook.presto.execution.buffer.PageSplitterUtil.splitPage;
 import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static com.facebook.presto.spi.page.PagesSerdeUtil.writeSerializedPage;
 import static com.facebook.presto.spiller.FileSingleStreamSpillerFactory.SPILL_FILE_PREFIX;
@@ -144,11 +145,15 @@ private void writePages(Iterator<Page> pageIterator)
             while (pageIterator.hasNext()) {
                 Page page = pageIterator.next();
                 spilledPagesInMemorySize += page.getSizeInBytes();
-                SerializedPage serializedPage = serde.serialize(page);
-                long pageSize = serializedPage.getSizeInBytes();
-                localSpillContext.updateBytes(pageSize);
-                spillerStats.addToTotalSpilledBytes(pageSize);
-                writeSerializedPage(output, serializedPage);
+                // page serialization requires  page.getSizeInBytes() + Integer.BYTES to fit in an integer
+                splitPage(page, DEFAULT_MAX_PAGE_SIZE_IN_BYTES).stream()
+                        .map(serde::serialize)
+                        .forEach(serializedPage -> {
+                            long pageSize = serializedPage.getSizeInBytes();
+                            localSpillContext.updateBytes(pageSize);
+                            spillerStats.addToTotalSpilledBytes(pageSize);
+                            writeSerializedPage(output, serializedPage);
+                        });
             }
         }
         catch (UncheckedIOException | IOException e) {