Skip to content

PARQUET-99: Add page size check properties #297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rdblue wants to merge 10 commits into from
Closed

PARQUET-99: Add page size check properties #297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e7cd54b
Added property to toggle page size check estimation and initial row s…
Jul 24, 2015
a057f46
Fixed inverted property logic
Jul 24, 2015
b49f03c
Fixed reset of nextSizeCheck
Jul 29, 2015
68794f0
Merge branch 'master' into page_size_check
Nov 30, 2015
3f7870c
Rebase to resolve byte buffer conflicts
Nov 30, 2015
5d99072
Update page size checking for v2 writer
Dec 1, 2015
71336ee
Fixed param name
Dec 2, 2015
2090719
Update sizeCheck to write page properly if estimating is turned off
Dec 2, 2015
18f8d3a
Spacing
Dec 2, 2015
c93b73e
PARQUET-99: Use ParquetProperties to carry encoding config.
rdblue Dec 2, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
243 changes: 201 additions & 42 deletions parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import org.apache.parquet.Preconditions;
import org.apache.parquet.bytes.ByteBufferAllocator;
import org.apache.parquet.bytes.CapacityByteArrayOutputStream;
import org.apache.parquet.bytes.HeapByteBufferAllocator;

import static org.apache.parquet.bytes.BytesUtils.getWidthFromMaxInt;
Expand All @@ -44,6 +45,7 @@
import org.apache.parquet.column.values.plain.BooleanPlainValuesWriter;
import org.apache.parquet.column.values.plain.FixedLenByteArrayPlainValuesWriter;
import org.apache.parquet.column.values.plain.PlainValuesWriter;
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter;
import org.apache.parquet.schema.MessageType;

Expand All @@ -55,6 +57,16 @@
*/
public class ParquetProperties {

@julienledem julienledem Dec 7, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The name is a little outdated now that you can pass an allocator.
ParquetConfiguration?

@rdblue rdblue Dec 8, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this class defines ParquetProperties.WriterVersion that is exposed in the public API, I think it would be a breaking change to rename it. This class is otherwise not part of the public API (and I've tried to keep it that way here) so we should fix this when we separate the API into well-defined packages.


public static final int DEFAULT_PAGE_SIZE = 1024 * 1024;
public static final int DEFAULT_DICTIONARY_PAGE_SIZE = DEFAULT_PAGE_SIZE;
public static final boolean DEFAULT_IS_DICTIONARY_ENABLED = true;
public static final WriterVersion DEFAULT_WRITER_VERSION = WriterVersion.PARQUET_1_0;
public static final boolean DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK = true;
public static final int DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK = 100;
public static final int DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK = 10000;

private static final int MIN_SLAB_SIZE = 64;

public enum WriterVersion {
PARQUET_1_0 ("v1"),
PARQUET_2_0 ("v2");
Expand All @@ -75,53 +87,83 @@ public static WriterVersion fromString(String name) {
return WriterVersion.valueOf(name);
}
}

private final int pageSizeThreshold;
private final int dictionaryPageSizeThreshold;
private final WriterVersion writerVersion;
private final boolean enableDictionary;
private final int minRowCountForPageSizeCheck;
private final int maxRowCountForPageSizeCheck;
private final boolean estimateNextSizeCheck;
private final ByteBufferAllocator allocator;

public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict) {
this(dictPageSize, writerVersion, enableDict, new HeapByteBufferAllocator());
}
private final int initialSlabSize;

public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, ByteBufferAllocator allocator) {
private ParquetProperties(WriterVersion writerVersion, int pageSize, int dictPageSize, boolean enableDict, int minRowCountForPageSizeCheck,
int maxRowCountForPageSizeCheck, boolean estimateNextSizeCheck, ByteBufferAllocator allocator) {
this.pageSizeThreshold = pageSize;
this.initialSlabSize = CapacityByteArrayOutputStream
.initialSlabSizeHeuristic(MIN_SLAB_SIZE, pageSizeThreshold, 10);
this.dictionaryPageSizeThreshold = dictPageSize;
this.writerVersion = writerVersion;
this.enableDictionary = enableDict;
Preconditions.checkNotNull(allocator, "ByteBufferAllocator");
this.minRowCountForPageSizeCheck = minRowCountForPageSizeCheck;
this.maxRowCountForPageSizeCheck = maxRowCountForPageSizeCheck;
this.estimateNextSizeCheck = estimateNextSizeCheck;
this.allocator = allocator;
}

public ValuesWriter getColumnDescriptorValuesWriter(int maxLevel, int initialSizePerCol, int pageSize) {
public ValuesWriter newRepetitionLevelWriter(ColumnDescriptor path) {
return newColumnDescriptorValuesWriter(path.getMaxRepetitionLevel());
}

public ValuesWriter newDefinitionLevelWriter(ColumnDescriptor path) {
return newColumnDescriptorValuesWriter(path.getMaxDefinitionLevel());
}

private ValuesWriter newColumnDescriptorValuesWriter(int maxLevel) {
if (maxLevel == 0) {
return new DevNullValuesWriter();
} else {
return new RunLengthBitPackingHybridValuesWriter(
getWidthFromMaxInt(maxLevel), initialSizePerCol, pageSize, this.allocator
);
getWidthFromMaxInt(maxLevel), MIN_SLAB_SIZE, pageSizeThreshold, allocator);
}
}

private ValuesWriter plainWriter(ColumnDescriptor path, int initialSizePerCol, int pageSize) {
public RunLengthBitPackingHybridEncoder newRepetitionLevelEncoder(ColumnDescriptor path) {
return newLevelEncoder(path.getMaxRepetitionLevel());
}

public RunLengthBitPackingHybridEncoder newDefinitionLevelEncoder(ColumnDescriptor path) {
return newLevelEncoder(path.getMaxDefinitionLevel());
}

private RunLengthBitPackingHybridEncoder newLevelEncoder(int maxLevel) {
return new RunLengthBitPackingHybridEncoder(
getWidthFromMaxInt(maxLevel), MIN_SLAB_SIZE, pageSizeThreshold, allocator);
}

private ValuesWriter plainWriter(ColumnDescriptor path) {
switch (path.getType()) {
case BOOLEAN:
return new BooleanPlainValuesWriter();
case INT96:
return new FixedLenByteArrayPlainValuesWriter(12, initialSizePerCol, pageSize, this.allocator);
return new FixedLenByteArrayPlainValuesWriter(12, initialSlabSize, pageSizeThreshold, allocator);
case FIXED_LEN_BYTE_ARRAY:
return new FixedLenByteArrayPlainValuesWriter(path.getTypeLength(), initialSizePerCol, pageSize, this.allocator);
return new FixedLenByteArrayPlainValuesWriter(path.getTypeLength(), initialSlabSize, pageSizeThreshold, allocator);
case BINARY:
case INT32:
case INT64:
case DOUBLE:
case FLOAT:
return new PlainValuesWriter(initialSizePerCol, pageSize, this.allocator);
return new PlainValuesWriter(initialSlabSize, pageSizeThreshold, allocator);
default:
throw new IllegalArgumentException("Unknown type " + path.getType());
}
}

private DictionaryValuesWriter dictionaryWriter(ColumnDescriptor path, int initialSizePerCol) {
@SuppressWarnings("deprecation")
private DictionaryValuesWriter dictionaryWriter(ColumnDescriptor path) {
Encoding encodingForDataPage;
Encoding encodingForDictionaryPage;
switch(writerVersion) {
Expand Down Expand Up @@ -158,24 +200,24 @@ private DictionaryValuesWriter dictionaryWriter(ColumnDescriptor path, int initi
}
}

private ValuesWriter writerToFallbackTo(ColumnDescriptor path, int initialSizePerCol, int pageSize) {
private ValuesWriter writerToFallbackTo(ColumnDescriptor path) {
switch(writerVersion) {
case PARQUET_1_0:
return plainWriter(path, initialSizePerCol, pageSize);
return plainWriter(path);
case PARQUET_2_0:
switch (path.getType()) {
case BOOLEAN:
return new RunLengthBitPackingHybridValuesWriter(1, initialSizePerCol, pageSize, this.allocator);
return new RunLengthBitPackingHybridValuesWriter(1, initialSlabSize, pageSizeThreshold, allocator);
case BINARY:
case FIXED_LEN_BYTE_ARRAY:
return new DeltaByteArrayWriter(initialSizePerCol, pageSize,this.allocator);
return new DeltaByteArrayWriter(initialSlabSize, pageSizeThreshold, allocator);
case INT32:
return new DeltaBinaryPackingValuesWriter(initialSizePerCol, pageSize, this.allocator);
return new DeltaBinaryPackingValuesWriter(initialSlabSize, pageSizeThreshold, allocator);
case INT96:
case INT64:
case DOUBLE:
case FLOAT:
return plainWriter(path, initialSizePerCol, pageSize);
return plainWriter(path);
default:
throw new IllegalArgumentException("Unknown type " + path.getType());
}
Expand All @@ -184,40 +226,44 @@ private ValuesWriter writerToFallbackTo(ColumnDescriptor path, int initialSizePe
}
}

private ValuesWriter dictWriterWithFallBack(ColumnDescriptor path, int initialSizePerCol, int pageSize) {
ValuesWriter writerToFallBackTo = writerToFallbackTo(path, initialSizePerCol, pageSize);
private ValuesWriter dictWriterWithFallBack(ColumnDescriptor path) {
ValuesWriter writerToFallBackTo = writerToFallbackTo(path);
if (enableDictionary) {
return FallbackValuesWriter.of(
dictionaryWriter(path, initialSizePerCol),
dictionaryWriter(path),
writerToFallBackTo);
} else {
return writerToFallBackTo;
}
}

public ValuesWriter getValuesWriter(ColumnDescriptor path, int initialSizePerCol, int pageSize) {
public ValuesWriter newValuesWriter(ColumnDescriptor path) {
switch (path.getType()) {
case BOOLEAN: // no dictionary encoding for boolean
return writerToFallbackTo(path, initialSizePerCol, pageSize);
return writerToFallbackTo(path);
case FIXED_LEN_BYTE_ARRAY:
// dictionary encoding for that type was not enabled in PARQUET 1.0
if (writerVersion == WriterVersion.PARQUET_2_0) {
return dictWriterWithFallBack(path, initialSizePerCol, pageSize);
return dictWriterWithFallBack(path);
} else {
return writerToFallbackTo(path, initialSizePerCol, pageSize);
return writerToFallbackTo(path);
}
case BINARY:
case INT32:
case INT64:
case INT96:
case DOUBLE:
case FLOAT:
return dictWriterWithFallBack(path, initialSizePerCol, pageSize);
return dictWriterWithFallBack(path);
default:
throw new IllegalArgumentException("Unknown type " + path.getType());
}
}

public int getPageSizeThreshold() {
return pageSizeThreshold;
}

public int getDictionaryPageSizeThreshold() {
return dictionaryPageSizeThreshold;
}
Expand All @@ -234,26 +280,139 @@ public ByteBufferAllocator getAllocator() {
return allocator;
}

public ColumnWriteStore newColumnWriteStore(
MessageType schema,
PageWriteStore pageStore,
int pageSize,
ByteBufferAllocator allocator) {
public ColumnWriteStore newColumnWriteStore(MessageType schema,
PageWriteStore pageStore) {
switch (writerVersion) {
case PARQUET_1_0:
return new ColumnWriteStoreV1(
pageStore,
pageSize,
dictionaryPageSizeThreshold,
enableDictionary, writerVersion, allocator);
return new ColumnWriteStoreV1(pageStore, this);
case PARQUET_2_0:
return new ColumnWriteStoreV2(
schema,
pageStore,
pageSize,
new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary, allocator));
return new ColumnWriteStoreV2(schema, pageStore, this);
default:
throw new IllegalArgumentException("unknown version " + writerVersion);
}
}

public int getMinRowCountForPageSizeCheck() {
return minRowCountForPageSizeCheck;
}

public int getMaxRowCountForPageSizeCheck() {
return maxRowCountForPageSizeCheck;
}

public boolean estimateNextSizeCheck() {
return estimateNextSizeCheck;
}

public static Builder builder() {
return new Builder();
}

public static Builder copy(ParquetProperties toCopy) {
return new Builder(toCopy);
}

public static class Builder {
private int pageSize = DEFAULT_PAGE_SIZE;
private int dictPageSize = DEFAULT_DICTIONARY_PAGE_SIZE;
private boolean enableDict = DEFAULT_IS_DICTIONARY_ENABLED;
private WriterVersion writerVersion = DEFAULT_WRITER_VERSION;
private int minRowCountForPageSizeCheck = DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK;
private int maxRowCountForPageSizeCheck = DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK;
private boolean estimateNextSizeCheck = DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK;
private ByteBufferAllocator allocator = new HeapByteBufferAllocator();

private Builder() {
}

private Builder(ParquetProperties toCopy) {
this.enableDict = toCopy.enableDictionary;
this.dictPageSize = toCopy.dictionaryPageSizeThreshold;
this.writerVersion = toCopy.writerVersion;
this.minRowCountForPageSizeCheck = toCopy.minRowCountForPageSizeCheck;
this.maxRowCountForPageSizeCheck = toCopy.maxRowCountForPageSizeCheck;
this.estimateNextSizeCheck = toCopy.estimateNextSizeCheck;
this.allocator = toCopy.allocator;
}

/**
* Set the Parquet format page size.
*
* @param pageSize an integer size in bytes
* @return this builder for method chaining.
*/
public Builder withPageSize(int pageSize) {
Preconditions.checkArgument(pageSize > 0,
"Invalid page size (negative): %s", pageSize);
this.pageSize = pageSize;
return this;
}

/**
* Enable or disable dictionary encoding.
*
* @param enableDictionary whether dictionary encoding should be enabled
* @return this builder for method chaining.
*/
public Builder withDictionaryEncoding(boolean enableDictionary) {
this.enableDict = enableDictionary;
return this;
}

/**
* Set the Parquet format dictionary page size.
*
* @param dictionaryPageSize an integer size in bytes
* @return this builder for method chaining.
*/
public Builder withDictionaryPageSize(int dictionaryPageSize) {
Preconditions.checkArgument(dictionaryPageSize > 0,
"Invalid dictionary page size (negative): %s", dictionaryPageSize);
this.dictPageSize = dictionaryPageSize;
return this;
}

/**
* Set the {@link WriterVersion format version}.
*
* @param version a {@code WriterVersion}
* @return this builder for method chaining.
*/
public Builder withWriterVersion(WriterVersion version) {
this.writerVersion = version;
return this;
}

public Builder withMinRowCountForPageSizeCheck(int min) {
Preconditions.checkArgument(min > 0,
"Invalid row count for page size check (negative): %s", min);
this.minRowCountForPageSizeCheck = min;
return this;
}

public Builder withMaxRowCountForPageSizeCheck(int max) {
Preconditions.checkArgument(max > 0,
"Invalid row count for page size check (negative): %s", max);
this.maxRowCountForPageSizeCheck = max;
return this;
}

// Do not attempt to predict next size check. Prevents issues with rows that vary significantly in size.
public Builder estimateRowCountForPageSizeCheck(boolean estimateNextSizeCheck) {

@julienledem julienledem Dec 7, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with?

@rdblue rdblue Dec 8, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

withEstimate implies that you're setting an estimate because it makes estimate a noun. Using estimate as the verb makes the boolean parameter make sense: estimateRowCountForPageSizeCheck(false)

this.estimateNextSizeCheck = estimateNextSizeCheck;
return this;
}

public Builder withAllocator(ByteBufferAllocator allocator) {
Preconditions.checkNotNull(allocator, "ByteBufferAllocator");
this.allocator = allocator;
return this;
}

public ParquetProperties build() {
return new ParquetProperties(writerVersion, pageSize, dictPageSize,
enableDict, minRowCountForPageSizeCheck, maxRowCountForPageSizeCheck,
estimateNextSizeCheck, allocator);
}
}
}
Loading