Improve OrcWriter by estimating page logical size
Calculate accurate logical size for each page chunk is expensive
and not necessary, estimate maximal number of rows a chunk can hold
based on page size estimated from sampled data.
@@ -68,6 +68,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Verify.verify;
import static io.airlift.slice.Slices.utf8Slice;
+import static java.lang.Integer.max;
import static java.lang.Integer.min;
import static java.lang.Math.toIntExact;
import static java.util.Objects.requireNonNull;
@@ -143,7 +144,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.chunkMaxLogicalBytes = max(1, stripeMaxBytes / 2);
this.stripeMaxRowCount = options.getStripeMaxRowCount();
this.rowGroupMaxRowCount = options.getRowGroupMaxRowCount();
recordValidation(validation -> validation.setRowGroupMaxRowCount(rowGroupMaxRowCount));
@@ -238,16 +239,18 @@ public class OrcWriter
validationBuilder.addPage(page);
}
+ // avoid chunk with huge logical size
+ int averageLogicalSizePerRow = estimateAverageLogicalSizePerRow(page);
+ int maxChunkRowCount = max(1, chunkMaxLogicalBytes / max(1, averageLogicalSizePerRow));
+
while (page != null) {
- // align page to row group boundaries
- int chunkRows = min(page.getPositionCount(), min(rowGroupMaxRowCount - rowGroupRowCount, stripeMaxRowCount - stripeRowCount));
- Page chunk = page.getRegion(0, chunkRows);
+ // logical size and row group boundaries
+ int chunkRows = min(maxChunkRowCount, min(rowGroupMaxRowCount - rowGroupRowCount, stripeMaxRowCount - stripeRowCount));
- // avoid chunk with huge logical size
- while (chunkRows > 1 && chunk.getLogicalSizeInBytes() > chunkMaxLogicalBytes) {
- chunkRows /= 2;
- chunk = chunk.getRegion(0, chunkRows);
- }
+ // align page to max size per chunk
+ chunkRows = min(page.getPositionCount(), chunkRows);
+
+ Page chunk = page.getRegion(0, chunkRows);
if (chunkRows < page.getPositionCount()) {
page = page.getRegion(chunkRows, page.getPositionCount() - chunkRows);
@@ -515,6 +518,14 @@ public class OrcWriter
orcEncoding);
}
+ private int estimateAverageLogicalSizePerRow(Page page)
+ {
+ checkArgument(page.getPositionCount() > 0, "page is empty");
+ // sample at most 100 rows to estimate average row logical size
+ Page chunk = page.getRegion(0, min(page.getPositionCount(), 100));
+ return toIntExact(chunk.getLogicalSizeInBytes() / chunk.getPositionCount());
+ }
+
private static <T> List<T> toDenseList(Map<Integer, T> data, int expectedSize)
{
checkArgument(data.size() == expectedSize);
