Split large pages in TempSingleStreamSpiller
This prevents int overflow when serializing pages, since page
serialization requires the page size in bytes to fit in an integer.
Was already added for FileSingleStreamSpiller at commit
@@ -22,7 +22,6 @@ import com.facebook.presto.spi.PrestoException;
import com.facebook.presto.spi.page.PageDataOutput;
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.facebook.presto.spi.storage.TempDataOperationContext;
import com.facebook.presto.spi.storage.TempDataSink;
@@ -47,6 +46,8 @@ import java.util.Iterator;
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.google.common.base.Preconditions.checkState;
import static java.util.Objects.requireNonNull;
@@ -149,17 +150,20 @@ public class TempStorageSingleStreamSpiller
while (pageIterator.hasNext()) {
Page page = pageIterator.next();
spilledPagesInMemorySize += page.getSizeInBytes();
- SerializedPage serializedPage = serde.serialize(page);
+ // 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);
-
PageDataOutput pageDataOutput = new PageDataOutput(serializedPage);
bufferedBytes += pageDataOutput.size();
bufferedPages.add(pageDataOutput);
if (bufferedBytes > maxBufferSizeInBytes) {
flushBufferedPages();
}
+ });
}
memoryContext.setBytes(bufferedBytes + dataSink.getRetainedSizeInBytes());
