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) {