Limit the size of page chunk for OrcWriter
Presto can produce huge pages (e.g. dictionary blocks produced in
JoinOperator). Writing a large page at one time for ORC Writer will
cause excessively huge stripe.
@@ -94,6 +94,7 @@ public class OrcWriter
private final CompressionKind compression;
private final int stripeMinBytes;
private final int stripeMaxBytes;
+ private final int chunkMaxLogicalBytes;
private final int stripeMaxRowCount;
private final int rowGroupMaxRowCount;
private final int maxCompressionBufferSize;
@@ -142,6 +143,7 @@ public class OrcWriter
checkArgument(options.getStripeMaxSize().compareTo(options.getStripeMinSize()) >= 0, "stripeMaxSize must be greater than stripeMinSize");
this.stripeMinBytes = toIntExact(requireNonNull(options.getStripeMinSize(), "stripeMinSize is null").toBytes());
this.stripeMaxBytes = toIntExact(requireNonNull(options.getStripeMaxSize(), "stripeMaxSize is null").toBytes());
+ this.chunkMaxLogicalBytes = Math.max(1, stripeMaxBytes / 2);
this.stripeMaxRowCount = options.getStripeMaxRowCount();
this.rowGroupMaxRowCount = options.getRowGroupMaxRowCount();
recordValidation(validation -> validation.setRowGroupMaxRowCount(rowGroupMaxRowCount));
@@ -238,16 +240,22 @@ public class OrcWriter
while (page != null) {
// align page to row group boundaries
- Page chunk;
- if (rowGroupRowCount + page.getPositionCount() > rowGroupMaxRowCount || stripeRowCount + page.getPositionCount() > stripeMaxRowCount) {
- int chunkRows = min(rowGroupMaxRowCount - rowGroupRowCount, stripeMaxRowCount - stripeRowCount);
- chunk = page.getRegion(0, chunkRows);
+ int chunkRows = min(page.getPositionCount(), min(rowGroupMaxRowCount - rowGroupRowCount, stripeMaxRowCount - stripeRowCount));
+ Page chunk = page.getRegion(0, chunkRows);
+
+ // avoid chunk with huge logical size
+ while (chunkRows > 1 && chunk.getLogicalSizeInBytes() > chunkMaxLogicalBytes) {
+ chunkRows /= 2;
+ chunk = chunk.getRegion(0, chunkRows);
+ }
+
+ if (chunkRows < page.getPositionCount()) {
page = page.getRegion(chunkRows, page.getPositionCount() - chunkRows);
}
else {
- chunk = page;
page = null;
}
+
writeChunk(chunk);
}
