From e7cd54b033cc5fd9c1693222f5ff1ac378092bb9 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Fri, 24 Jul 2015 16:44:46 -0700 Subject: [PATCH 1/8] Added property to toggle page size check estimation and initial row size checking --- .../parquet/column/ParquetProperties.java | 16 ++++++-- .../column/impl/ColumnWriteStoreV1.java | 13 +++++- .../parquet/column/impl/ColumnWriterV1.java | 17 +++++--- .../column/impl/TestColumnReaderImpl.java | 6 ++- .../hadoop/InternalParquetRecordWriter.java | 39 +++++++++++++++++- .../parquet/hadoop/ParquetOutputFormat.java | 17 ++++++++ .../parquet/hadoop/ParquetRecordWriter.java | 41 ++++++++++++++++++- 7 files changed, 135 insertions(+), 14 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index df44c4be4e..a54cf3959f 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -51,6 +51,9 @@ */ public class ParquetProperties { + public static final int INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK = 100; + public static final boolean DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK = true; + public enum WriterVersion { PARQUET_1_0 ("v1"), PARQUET_2_0 ("v2"); @@ -74,11 +77,15 @@ public static WriterVersion fromString(String name) { private final int dictionaryPageSizeThreshold; private final WriterVersion writerVersion; private final boolean enableDictionary; + private final int initialRowCountForSizeCheck; + private final boolean estimateNextSizeCheck; - public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict) { + public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck) { this.dictionaryPageSizeThreshold = dictPageSize; this.writerVersion = writerVersion; this.enableDictionary = enableDict; + this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; + this.estimateNextSizeCheck = estimateNextSizeCheck; } public static ValuesWriter getColumnDescriptorValuesWriter(int maxLevel, int initialSizePerCol, int pageSize) { @@ -228,13 +235,16 @@ public ColumnWriteStore newColumnWriteStore( pageStore, pageSize, dictionaryPageSizeThreshold, - enableDictionary, writerVersion); + enableDictionary, + initialRowCountForSizeCheck, + estimateNextSizeCheck, + writerVersion); case PARQUET_2_0: return new ColumnWriteStoreV2( schema, pageStore, pageSize, - new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary)); + new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary, initialRowCountForSizeCheck, estimateNextSizeCheck)); default: throw new IllegalArgumentException("unknown version " + writerVersion); } diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java index a72b6f7f08..ae6044918a 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java @@ -32,6 +32,9 @@ import org.apache.parquet.column.page.PageWriteStore; import org.apache.parquet.column.page.PageWriter; +import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; + public class ColumnWriteStoreV1 implements ColumnWriteStore { private final Map columns = new TreeMap(); @@ -39,14 +42,22 @@ public class ColumnWriteStoreV1 implements ColumnWriteStore { private final int pageSizeThreshold; private final int dictionaryPageSizeThreshold; private final boolean enableDictionary; + private final int initialRowCountForSizeCheck; + private final boolean estimateNextSizeCheck; private final WriterVersion writerVersion; public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, WriterVersion writerVersion) { + this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion); + } + + public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion) { super(); this.pageWriteStore = pageWriteStore; this.pageSizeThreshold = pageSizeThreshold; this.dictionaryPageSizeThreshold = dictionaryPageSizeThreshold; this.enableDictionary = enableDictionary; + this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; + this.estimateNextSizeCheck = estimateNextSizeCheck; this.writerVersion = writerVersion; } @@ -65,7 +76,7 @@ public Set getColumnDescriptors() { private ColumnWriterV1 newMemColumn(ColumnDescriptor path) { PageWriter pageWriter = pageWriteStore.getPageWriter(path); - return new ColumnWriterV1(path, pageWriter, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, writerVersion); + return new ColumnWriterV1(path, pageWriter, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, initialRowCountForSizeCheck, estimateNextSizeCheck, writerVersion); } @Override diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java index f4079c761c..929ac7d8c2 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java @@ -46,7 +46,6 @@ final class ColumnWriterV1 implements ColumnWriter { private static final Log LOG = Log.getLog(ColumnWriterV1.class); private static final boolean DEBUG = Log.DEBUG; - private static final int INITIAL_COUNT_FOR_SIZE_CHECK = 100; private static final int MIN_SLAB_SIZE = 64; private final ColumnDescriptor path; @@ -57,6 +56,7 @@ final class ColumnWriterV1 implements ColumnWriter { private ValuesWriter dataColumn; private int valueCount; private int valueCountForNextSizeCheck; + private boolean estimateNextSizeCheck; private Statistics statistics; @@ -66,15 +66,20 @@ public ColumnWriterV1( int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, + int initialRowCountForSizeCheck, + boolean estimateNextSizeCheck, WriterVersion writerVersion) { this.path = path; this.pageWriter = pageWriter; this.pageSizeThreshold = pageSizeThreshold; // initial check of memory usage. So that we have enough data to make an initial prediction - this.valueCountForNextSizeCheck = INITIAL_COUNT_FOR_SIZE_CHECK; + this.valueCountForNextSizeCheck = initialRowCountForSizeCheck; + // Do not attempt to predict next size check. Prevents issues with rows that vary significantly in size. + this.estimateNextSizeCheck = estimateNextSizeCheck; + resetStatistics(); - ParquetProperties parquetProps = new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary); + ParquetProperties parquetProps = new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary, initialRowCountForSizeCheck, estimateNextSizeCheck); this.repetitionLevelColumn = ParquetProperties.getColumnDescriptorValuesWriter(path.getMaxRepetitionLevel(), MIN_SLAB_SIZE, pageSizeThreshold); this.definitionLevelColumn = ParquetProperties.getColumnDescriptorValuesWriter(path.getMaxDefinitionLevel(), MIN_SLAB_SIZE, pageSizeThreshold); @@ -109,9 +114,11 @@ private void accountForValueWritten() { + dataColumn.getBufferedSize(); if (memSize > pageSizeThreshold) { // we will write the current page and check again the size at the predicted middle of next page - valueCountForNextSizeCheck = valueCount / 2; + if(!estimateNextSizeCheck) { + valueCountForNextSizeCheck = valueCount / 2; + } writePage(); - } else { + } else if (!estimateNextSizeCheck) { // not reached the threshold, will check again midway valueCountForNextSizeCheck = (int)(valueCount + ((float)valueCount * pageSizeThreshold / memSize)) / 2 + 1; } diff --git a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java index a1820e654e..bf5a1beb3d 100644 --- a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java +++ b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java @@ -19,7 +19,9 @@ package org.apache.parquet.column.impl; import static junit.framework.Assert.assertEquals; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; +import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; import java.util.List; @@ -58,7 +60,7 @@ public void test() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { required binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.write(Binary.fromString("bar" + i % 10), 0, 0); if ((i + 1) % 1000 == 0) { @@ -93,7 +95,7 @@ public void testOptional() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { optional binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.writeNull(0, 0); if ((i + 1) % 1000 == 0) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index 37e8db5b80..6def8c6512 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -65,6 +65,41 @@ class InternalParquetRecordWriter { private ColumnChunkPageWriteStore pageStore; + /** + * @param parquetFileWriter the file to write to + * @param writeSupport the class to convert incoming records + * @param schema the schema of the records + * @param extraMetaData extra meta data to write in the footer of the file + * @param rowGroupSize the size of a block in the file (this will be approximate) + * @param compressor the codec used to compress + */ + public InternalParquetRecordWriter( + ParquetFileWriter parquetFileWriter, + WriteSupport writeSupport, + MessageType schema, + Map extraMetaData, + long rowGroupSize, + int pageSize, + BytesCompressor compressor, + int dictionaryPageSize, + boolean enableDictionary, + boolean validating, + WriterVersion writerVersion) { + this(parquetFileWriter, + writeSupport, + schema, + extraMetaData, + rowGroupSize, + pageSize, + compressor, + dictionaryPageSize, + enableDictionary, + ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + false, + validating, + writerVersion); + } + /** * @param parquetFileWriter the file to write to * @param writeSupport the class to convert incoming records @@ -83,6 +118,8 @@ public InternalParquetRecordWriter( BytesCompressor compressor, int dictionaryPageSize, boolean enableDictionary, + int initialRowCountForSizeCheck, + boolean constantNextSizeCheck, boolean validating, WriterVersion writerVersion) { this.parquetFileWriter = parquetFileWriter; @@ -95,7 +132,7 @@ public InternalParquetRecordWriter( this.pageSize = pageSize; this.compressor = compressor; this.validating = validating; - this.parquetProperties = new ParquetProperties(dictionaryPageSize, writerVersion, enableDictionary); + this.parquetProperties = new ParquetProperties(dictionaryPageSize, writerVersion, enableDictionary, initialRowCountForSizeCheck, constantNextSizeCheck); initStore(); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index e075db38c4..35476c268a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -37,6 +37,7 @@ import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.parquet.Log; +import org.apache.parquet.column.ParquetProperties; import org.apache.parquet.column.ParquetProperties.WriterVersion; import org.apache.parquet.hadoop.ParquetFileWriter.Mode; import org.apache.parquet.hadoop.api.WriteSupport; @@ -115,6 +116,8 @@ public class ParquetOutputFormat extends FileOutputFormat { public static final String MEMORY_POOL_RATIO = "parquet.memory.pool.ratio"; public static final String MIN_MEMORY_ALLOCATION = "parquet.memory.min.chunk.size"; public static final String MAX_PADDING_BYTES = "parquet.writer.max-padding"; + public static final String INITIAL_ROW_COUNT_SIZE_CHECK = "parquet.page.size.initial.row.check"; + public static final String ESTIMATE_PAGE_SIZE_CHECK = "parquet.page.size.check.estimate"; // default to no padding for now private static final int DEFAULT_MAX_PADDING_SIZE = 0; @@ -192,6 +195,14 @@ public static boolean getEnableDictionary(Configuration configuration) { return configuration.getBoolean(ENABLE_DICTIONARY, true); } + public static int getInitialRowCountForPageSizeCheck(Configuration configuration) { + return configuration.getInt(INITIAL_ROW_COUNT_SIZE_CHECK, ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK); + } + + public static boolean getEstimatePageSizeCheck(Configuration configuration) { + return configuration.getBoolean(ESTIMATE_PAGE_SIZE_CHECK, ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK); + } + @Deprecated public static int getBlockSize(Configuration configuration) { return configuration.getInt(BLOCK_SIZE, DEFAULT_BLOCK_SIZE); @@ -307,6 +318,10 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp if (INFO) LOG.info("Writer version is: " + writerVersion); int maxPaddingSize = getMaxPaddingSize(conf); if (INFO) LOG.info("Maximum row group padding size is " + maxPaddingSize + " bytes"); + boolean estimateNextSizeCheck = getEstimatePageSizeCheck(conf); + if(INFO) LOG.info("Page size checking is: " + (estimateNextSizeCheck ? "estimated" : "constant")); + int initialRowCountForPageSizeCheck = getInitialRowCountForPageSizeCheck(conf); + if(INFO) LOG.info("Initial row count for page size check is: " + initialRowCountForPageSizeCheck); WriteContext init = writeSupport.init(conf); ParquetFileWriter w = new ParquetFileWriter( @@ -333,6 +348,8 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp codecFactory.getCompressor(codec, pageSize), dictionaryPageSize, enableDictionary, + initialRowCountForPageSizeCheck, + estimateNextSizeCheck, validating, writerVersion, memoryManager); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index 2449192205..f9d1bcb41a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -29,6 +29,8 @@ import org.apache.parquet.schema.MessageType; import static org.apache.parquet.Preconditions.checkNotNull; +import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; /** * Writes records to a Parquet file @@ -70,9 +72,41 @@ public ParquetRecordWriter( WriterVersion writerVersion) { internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, - validating, writerVersion); + 100, false, validating, writerVersion); } + /** + * + * @param w the file to write to + * @param writeSupport the class to convert incoming records + * @param schema the schema of the records + * @param extraMetaData extra meta data to write in the footer of the file + * @param blockSize the size of a block in the file (this will be approximate) + * @param compressor the compressor used to compress the pages + * @param dictionaryPageSize the threshold for dictionary size + * @param enableDictionary to enable the dictionary + * @param validating if schema validation should be turned on + */ + public ParquetRecordWriter( + ParquetFileWriter w, + WriteSupport writeSupport, + MessageType schema, + Map extraMetaData, + long blockSize, int pageSize, + BytesCompressor compressor, + int dictionaryPageSize, + boolean enableDictionary, + boolean validating, + WriterVersion writerVersion, + MemoryManager memoryManager) { + internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, + extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, + INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + validating, writerVersion); + this.memoryManager = checkNotNull(memoryManager, "memoryManager"); + memoryManager.addWriter(internalWriter, blockSize); + } + /** * * @param w the file to write to @@ -94,11 +128,14 @@ public ParquetRecordWriter( BytesCompressor compressor, int dictionaryPageSize, boolean enableDictionary, + int initialRowCountForSizeCheck, + boolean estimateNextPageSizeCheck, boolean validating, WriterVersion writerVersion, MemoryManager memoryManager) { internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, + extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, initialRowCountForSizeCheck, + estimateNextPageSizeCheck, validating, writerVersion); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); From a057f4616c045e98ea834208924527962d615358 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Fri, 24 Jul 2015 16:57:12 -0700 Subject: [PATCH 2/8] Fixed inverted property logic --- .../java/org/apache/parquet/column/impl/ColumnWriterV1.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java index 929ac7d8c2..8858fea841 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java @@ -114,11 +114,11 @@ private void accountForValueWritten() { + dataColumn.getBufferedSize(); if (memSize > pageSizeThreshold) { // we will write the current page and check again the size at the predicted middle of next page - if(!estimateNextSizeCheck) { + if(estimateNextSizeCheck) { valueCountForNextSizeCheck = valueCount / 2; } writePage(); - } else if (!estimateNextSizeCheck) { + } else if (estimateNextSizeCheck) { // not reached the threshold, will check again midway valueCountForNextSizeCheck = (int)(valueCount + ((float)valueCount * pageSizeThreshold / memSize)) / 2 + 1; } From b49f03c2581639854f3b89f3c3a1d3d0e0abc0c3 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Wed, 29 Jul 2015 13:50:35 -0700 Subject: [PATCH 3/8] Fixed reset of nextSizeCheck --- .../java/org/apache/parquet/column/impl/ColumnWriterV1.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java index 8858fea841..6c6c9c2467 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java @@ -55,6 +55,7 @@ final class ColumnWriterV1 implements ColumnWriter { private ValuesWriter definitionLevelColumn; private ValuesWriter dataColumn; private int valueCount; + private int initialRowCountForSizeCheck; private int valueCountForNextSizeCheck; private boolean estimateNextSizeCheck; @@ -73,6 +74,7 @@ public ColumnWriterV1( this.pageWriter = pageWriter; this.pageSizeThreshold = pageSizeThreshold; // initial check of memory usage. So that we have enough data to make an initial prediction + this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; this.valueCountForNextSizeCheck = initialRowCountForSizeCheck; // Do not attempt to predict next size check. Prevents issues with rows that vary significantly in size. this.estimateNextSizeCheck = estimateNextSizeCheck; @@ -116,11 +118,15 @@ private void accountForValueWritten() { // we will write the current page and check again the size at the predicted middle of next page if(estimateNextSizeCheck) { valueCountForNextSizeCheck = valueCount / 2; + } else { + valueCountForNextSizeCheck = initialRowCountForSizeCheck; } writePage(); } else if (estimateNextSizeCheck) { // not reached the threshold, will check again midway valueCountForNextSizeCheck = (int)(valueCount + ((float)valueCount * pageSizeThreshold / memSize)) / 2 + 1; + } else { + valueCountForNextSizeCheck += initialRowCountForSizeCheck; } } } From 3f7870c4e1783f1e4e31d5e71ef77096a43cf454 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Mon, 30 Nov 2015 11:27:01 -0800 Subject: [PATCH 4/8] Rebase to resolve byte buffer conflicts --- .../java/org/apache/parquet/column/ParquetProperties.java | 7 ++++++- .../org/apache/parquet/column/impl/ColumnWriteStoreV1.java | 6 +++--- .../apache/parquet/column/impl/TestColumnReaderImpl.java | 4 ++-- .../apache/parquet/hadoop/InternalParquetRecordWriter.java | 6 ++++-- .../org/apache/parquet/hadoop/ParquetRecordWriter.java | 2 +- 5 files changed, 16 insertions(+), 9 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index 7706d77ff8..20ddfaeb7a 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -90,6 +90,11 @@ public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean } public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, ByteBufferAllocator allocator) { + this(dictPageSize, writerVersion, enableDict, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, allocator); + } + + public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, ByteBufferAllocator allocator) { this.dictionaryPageSizeThreshold = dictPageSize; this.writerVersion = writerVersion; this.enableDictionary = enableDict; @@ -104,7 +109,7 @@ public ValuesWriter getColumnDescriptorValuesWriter(int maxLevel, int initialSiz return new DevNullValuesWriter(); } else { return new RunLengthBitPackingHybridValuesWriter( - getWidthFromMaxInt(maxLevel), initialSizePerCol, pageSize); + getWidthFromMaxInt(maxLevel), initialSizePerCol, pageSize, this.allocator); } } diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java index a6cabc8b5a..90ac9408ae 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java @@ -48,11 +48,11 @@ public class ColumnWriteStoreV1 implements ColumnWriteStore { private final WriterVersion writerVersion; private final ByteBufferAllocator allocator; - public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, WriterVersion writerVersion) { - this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion); + public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, WriterVersion writerVersion, ByteBufferAllocator allocator) { + this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion, allocator); } - public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion) { + public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { super(); this.pageWriteStore = pageWriteStore; this.pageSizeThreshold = pageSizeThreshold; diff --git a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java index 5c297325b3..caf3e961fd 100644 --- a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java +++ b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java @@ -61,7 +61,7 @@ public void test() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { required binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.write(Binary.fromString("bar" + i % 10), 0, 0); if ((i + 1) % 1000 == 0) { @@ -96,7 +96,7 @@ public void testOptional() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { optional binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.writeNull(0, 0); if ((i + 1) % 1000 == 0) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index 0b625ff1f7..92640402c4 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -87,7 +87,8 @@ public InternalParquetRecordWriter( int dictionaryPageSize, boolean enableDictionary, boolean validating, - WriterVersion writerVersion) { + WriterVersion writerVersion, + ByteBufferAllocator allocator) { this(parquetFileWriter, writeSupport, schema, @@ -100,7 +101,8 @@ public InternalParquetRecordWriter( ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, false, validating, - writerVersion); + writerVersion, + allocator); } /** diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index 62dda8dfdf..9cf0b73f84 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -138,7 +138,7 @@ public ParquetRecordWriter( internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, initialRowCountForSizeCheck, estimateNextPageSizeCheck, - validating, writerVersion); + validating, writerVersion, new HeapByteBufferAllocator()); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); } From 5d99072bea638118740247c408f4065288d70cdf Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Tue, 1 Dec 2015 10:23:04 -0800 Subject: [PATCH 5/8] Update page size checking for v2 writer --- .../parquet/column/ParquetProperties.java | 37 +++++++-- .../column/impl/ColumnWriteStoreV1.java | 12 +-- .../column/impl/ColumnWriteStoreV2.java | 22 ++++-- .../parquet/column/impl/ColumnWriterV1.java | 12 +-- .../column/impl/TestColumnReaderImpl.java | 11 ++- .../hadoop/InternalParquetRecordWriter.java | 9 ++- .../parquet/hadoop/ParquetOutputFormat.java | 20 +++-- .../parquet/hadoop/ParquetRecordWriter.java | 75 ++++++++++--------- 8 files changed, 124 insertions(+), 74 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index 20ddfaeb7a..5b670f0c63 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -55,8 +55,9 @@ */ public class ParquetProperties { - public static final int INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK = 100; 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; public enum WriterVersion { PARQUET_1_0 ("v1"), @@ -81,7 +82,8 @@ public static WriterVersion fromString(String name) { private final int dictionaryPageSizeThreshold; private final WriterVersion writerVersion; private final boolean enableDictionary; - private final int initialRowCountForSizeCheck; + private final int minRowCountForPageSizeCheck; + private final int maxRowCountForPageSizeCheck; private final boolean estimateNextSizeCheck; private final ByteBufferAllocator allocator; @@ -90,15 +92,17 @@ public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean } public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, ByteBufferAllocator allocator) { - this(dictPageSize, writerVersion, enableDict, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + this(dictPageSize, writerVersion, enableDict, DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, allocator); } - public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, ByteBufferAllocator allocator) { + public ParquetProperties(int dictPageSize, WriterVersion writerVersion, boolean enableDict, int minRowCountForPageSizeCheck, + int maxRowCountForPageSizeCheck, boolean estimateNextSizeCheck, ByteBufferAllocator allocator) { this.dictionaryPageSizeThreshold = dictPageSize; this.writerVersion = writerVersion; this.enableDictionary = enableDict; - this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; + this.minRowCountForPageSizeCheck = minRowCountForPageSizeCheck; + this.maxRowCountForPageSizeCheck = maxRowCountForPageSizeCheck; this.estimateNextSizeCheck = estimateNextSizeCheck; Preconditions.checkNotNull(allocator, "ByteBufferAllocator"); this.allocator = allocator; @@ -257,7 +261,7 @@ public ColumnWriteStore newColumnWriteStore( pageSize, dictionaryPageSizeThreshold, enableDictionary, - initialRowCountForSizeCheck, + minRowCountForPageSizeCheck, estimateNextSizeCheck, writerVersion, allocator); @@ -266,9 +270,28 @@ public ColumnWriteStore newColumnWriteStore( schema, pageStore, pageSize, - new ParquetProperties(dictionaryPageSizeThreshold, writerVersion, enableDictionary, initialRowCountForSizeCheck, estimateNextSizeCheck, allocator)); + new ParquetProperties( + dictionaryPageSizeThreshold, + writerVersion, + enableDictionary, + minRowCountForPageSizeCheck, + maxRowCountForPageSizeCheck, + estimateNextSizeCheck, + allocator)); default: throw new IllegalArgumentException("unknown version " + writerVersion); } } + + public int getMinRowCountForPageSizeCheck() { + return minRowCountForPageSizeCheck; + } + + public int getMaxRowCountForPageSizeCheck() { + return maxRowCountForPageSizeCheck; + } + + public boolean isEstimateNextSizeCheck() { + return estimateNextSizeCheck; + } } diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java index 90ac9408ae..6222b4380d 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java @@ -33,7 +33,7 @@ import org.apache.parquet.column.page.PageWriteStore; import org.apache.parquet.column.page.PageWriter; -import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK; import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; public class ColumnWriteStoreV1 implements ColumnWriteStore { @@ -43,22 +43,22 @@ public class ColumnWriteStoreV1 implements ColumnWriteStore { private final int pageSizeThreshold; private final int dictionaryPageSizeThreshold; private final boolean enableDictionary; - private final int initialRowCountForSizeCheck; + private final int initialRowCountForPageSizeCheck; private final boolean estimateNextSizeCheck; private final WriterVersion writerVersion; private final ByteBufferAllocator allocator; public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, WriterVersion writerVersion, ByteBufferAllocator allocator) { - this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion, allocator); + this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion, allocator); } - public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int initialRowCountForSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { + public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int maxRowCountForPageSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { super(); this.pageWriteStore = pageWriteStore; this.pageSizeThreshold = pageSizeThreshold; this.dictionaryPageSizeThreshold = dictionaryPageSizeThreshold; this.enableDictionary = enableDictionary; - this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; + this.initialRowCountForPageSizeCheck = maxRowCountForPageSizeCheck; this.estimateNextSizeCheck = estimateNextSizeCheck; this.writerVersion = writerVersion; this.allocator = allocator; @@ -79,7 +79,7 @@ public Set getColumnDescriptors() { private ColumnWriterV1 newMemColumn(ColumnDescriptor path) { PageWriter pageWriter = pageWriteStore.getPageWriter(path); - return new ColumnWriterV1(path, pageWriter, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, initialRowCountForSizeCheck, estimateNextSizeCheck, writerVersion, allocator); + return new ColumnWriterV1(path, pageWriter, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, initialRowCountForPageSizeCheck, estimateNextSizeCheck, writerVersion, allocator); } @Override diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java index 4126004109..64ea40abe0 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java @@ -40,16 +40,14 @@ public class ColumnWriteStoreV2 implements ColumnWriteStore { - // will wait for at least that many records before checking again - private static final int MINIMUM_RECORD_COUNT_FOR_CHECK = 100; - private static final int MAXIMUM_RECORD_COUNT_FOR_CHECK = 10000; // will flush even if size bellow the threshold by this much to facilitate page alignment private static final float THRESHOLD_TOLERANCE_RATIO = 0.1f; // 10 % private final Map columns; private final Collection writers; + private final ParquetProperties parquetProperties; private long rowCount; - private long rowCountForNextSizeCheck = MINIMUM_RECORD_COUNT_FOR_CHECK; + private long rowCountForNextSizeCheck; private final long thresholdTolerance; private final ByteBufferAllocator allocator; @@ -61,6 +59,7 @@ public ColumnWriteStoreV2( int pageSizeThreshold, ParquetProperties parquetProps) { super(); + this.parquetProperties = parquetProps; this.pageSizeThreshold = pageSizeThreshold; this.thresholdTolerance = (long)(pageSizeThreshold * THRESHOLD_TOLERANCE_RATIO); this.allocator = parquetProps.getAllocator(); @@ -71,6 +70,8 @@ public ColumnWriteStoreV2( } this.columns = unmodifiableMap(mcolumns); this.writers = this.columns.values(); + + this.rowCountForNextSizeCheck = parquetProperties.getMinRowCountForPageSizeCheck(); } public ColumnWriter getColumnWriter(ColumnDescriptor path) { @@ -147,6 +148,11 @@ public void endRecord() { } private void sizeCheck() { + if(!parquetProperties.isEstimateNextSizeCheck()) { + rowCountForNextSizeCheck = rowCount + parquetProperties.getMinRowCountForPageSizeCheck(); + return; + } + long minRecordToWait = Long.MAX_VALUE; for (ColumnWriterV2 writer : writers) { long usedMem = writer.getCurrentPageBufferedSize(); @@ -158,20 +164,20 @@ private void sizeCheck() { } long rowsToFillPage = usedMem == 0 ? - MAXIMUM_RECORD_COUNT_FOR_CHECK + parquetProperties.getMaxRowCountForPageSizeCheck() : (long)((float)rows) / usedMem * remainingMem; if (rowsToFillPage < minRecordToWait) { minRecordToWait = rowsToFillPage; } } if (minRecordToWait == Long.MAX_VALUE) { - minRecordToWait = MINIMUM_RECORD_COUNT_FOR_CHECK; + minRecordToWait = parquetProperties.getMinRowCountForPageSizeCheck(); } // will check again halfway rowCountForNextSizeCheck = rowCount + min( - max(minRecordToWait / 2, MINIMUM_RECORD_COUNT_FOR_CHECK), // no less than MINIMUM_RECORD_COUNT_FOR_CHECK - MAXIMUM_RECORD_COUNT_FOR_CHECK); // no more than MAXIMUM_RECORD_COUNT_FOR_CHECK + max(minRecordToWait / 2, parquetProperties.getMinRowCountForPageSizeCheck()), // no less than MINIMUM_RECORD_COUNT_FOR_CHECK + parquetProperties.getMaxRowCountForPageSizeCheck()); // no more than MAXIMUM_RECORD_COUNT_FOR_CHECK } } diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java index 6c24bc5a40..d2ec54b4e2 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java @@ -56,7 +56,7 @@ final class ColumnWriterV1 implements ColumnWriter { private ValuesWriter definitionLevelColumn; private ValuesWriter dataColumn; private int valueCount; - private int initialRowCountForSizeCheck; + private int initialRowCountForPageSizeCheck; private int valueCountForNextSizeCheck; private boolean estimateNextSizeCheck; @@ -68,7 +68,7 @@ public ColumnWriterV1( int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, - int initialRowCountForSizeCheck, + int initialRowCountForPageSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { @@ -76,8 +76,8 @@ public ColumnWriterV1( this.pageWriter = pageWriter; this.pageSizeThreshold = pageSizeThreshold; // initial check of memory usage. So that we have enough data to make an initial prediction - this.initialRowCountForSizeCheck = initialRowCountForSizeCheck; - this.valueCountForNextSizeCheck = initialRowCountForSizeCheck; + this.initialRowCountForPageSizeCheck = initialRowCountForPageSizeCheck; + this.valueCountForNextSizeCheck = initialRowCountForPageSizeCheck; // Do not attempt to predict next size check. Prevents issues with rows that vary significantly in size. this.estimateNextSizeCheck = estimateNextSizeCheck; @@ -121,14 +121,14 @@ private void accountForValueWritten() { if(estimateNextSizeCheck) { valueCountForNextSizeCheck = valueCount / 2; } else { - valueCountForNextSizeCheck = initialRowCountForSizeCheck; + valueCountForNextSizeCheck = initialRowCountForPageSizeCheck; } writePage(); } else if (estimateNextSizeCheck) { // not reached the threshold, will check again midway valueCountForNextSizeCheck = (int)(valueCount + ((float)valueCount * pageSizeThreshold / memSize)) / 2 + 1; } else { - valueCountForNextSizeCheck += initialRowCountForSizeCheck; + valueCountForNextSizeCheck += initialRowCountForPageSizeCheck; } } } diff --git a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java index caf3e961fd..521b668c4a 100644 --- a/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java +++ b/parquet-column/src/test/java/org/apache/parquet/column/impl/TestColumnReaderImpl.java @@ -21,7 +21,8 @@ import static junit.framework.Assert.assertEquals; import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; import static org.apache.parquet.column.ParquetProperties.WriterVersion.PARQUET_2_0; -import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK; import java.util.List; @@ -61,7 +62,9 @@ public void test() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { required binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, + DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, + DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.write(Binary.fromString("bar" + i % 10), 0, 0); if ((i + 1) % 1000 == 0) { @@ -96,7 +99,9 @@ public void testOptional() throws Exception { MessageType schema = MessageTypeParser.parseMessageType("message test { optional binary foo; }"); ColumnDescriptor col = schema.getColumns().get(0); MemPageWriter pageWriter = new MemPageWriter(); - ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); + ColumnWriterV2 columnWriterV2 = new ColumnWriterV2(col, pageWriter, new ParquetProperties(1024, PARQUET_2_0, true, + DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, + DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, new HeapByteBufferAllocator()), 2048); for (int i = 0; i < rows; i++) { columnWriterV2.writeNull(0, 0); if ((i + 1) % 1000 == 0) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index 92640402c4..0730f2a578 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -98,7 +98,8 @@ public InternalParquetRecordWriter( compressor, dictionaryPageSize, enableDictionary, - ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, + ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, false, validating, writerVersion, @@ -123,7 +124,8 @@ public InternalParquetRecordWriter( BytesCompressor compressor, int dictionaryPageSize, boolean enableDictionary, - int initialRowCountForSizeCheck, + int minRowCountForSizeCheck, + int maxRowCountForSizeCheck, boolean constantNextSizeCheck, boolean validating, WriterVersion writerVersion, @@ -138,7 +140,8 @@ public InternalParquetRecordWriter( this.pageSize = pageSize; this.compressor = compressor; this.validating = validating; - this.parquetProperties = new ParquetProperties(dictionaryPageSize, writerVersion, enableDictionary, initialRowCountForSizeCheck, constantNextSizeCheck, allocator); + this.parquetProperties = new ParquetProperties(dictionaryPageSize, writerVersion, enableDictionary, + minRowCountForSizeCheck, maxRowCountForSizeCheck, constantNextSizeCheck, allocator); initStore(); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index e33793d740..e9d701dc60 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -142,7 +142,8 @@ public static enum JobSummaryLevel { public static final String MEMORY_POOL_RATIO = "parquet.memory.pool.ratio"; public static final String MIN_MEMORY_ALLOCATION = "parquet.memory.min.chunk.size"; public static final String MAX_PADDING_BYTES = "parquet.writer.max-padding"; - public static final String INITIAL_ROW_COUNT_SIZE_CHECK = "parquet.page.size.initial.row.check"; + public static final String MIN_ROW_COUNT_FOR_PAGE_SIZE_CHECK = "parquet.page.size.row.check.min"; + public static final String MAX_ROW_COUNT_FOR_PAGE_SIZE_CHECK = "parquet.page.size.row.check.max"; public static final String ESTIMATE_PAGE_SIZE_CHECK = "parquet.page.size.check.estimate"; // default to no padding for now @@ -244,8 +245,12 @@ public static boolean getEnableDictionary(Configuration configuration) { return configuration.getBoolean(ENABLE_DICTIONARY, true); } - public static int getInitialRowCountForPageSizeCheck(Configuration configuration) { - return configuration.getInt(INITIAL_ROW_COUNT_SIZE_CHECK, ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK); + public static int getMinRowCountForPageSizeCheck(Configuration configuration) { + return configuration.getInt(MIN_ROW_COUNT_FOR_PAGE_SIZE_CHECK, ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK); + } + + public static int getMaxRowCountForPageSizeCheck(Configuration configuration) { + return configuration.getInt(MAX_ROW_COUNT_FOR_PAGE_SIZE_CHECK, ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK); } public static boolean getEstimatePageSizeCheck(Configuration configuration) { @@ -368,8 +373,10 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp if (INFO) LOG.info("Maximum row group padding size is " + maxPaddingSize + " bytes"); boolean estimateNextSizeCheck = getEstimatePageSizeCheck(conf); if(INFO) LOG.info("Page size checking is: " + (estimateNextSizeCheck ? "estimated" : "constant")); - int initialRowCountForPageSizeCheck = getInitialRowCountForPageSizeCheck(conf); - if(INFO) LOG.info("Initial row count for page size check is: " + initialRowCountForPageSizeCheck); + int minRowCountForPageSizeCheck = getMinRowCountForPageSizeCheck(conf); + if(INFO) LOG.info("Min row count for page size check is: " + minRowCountForPageSizeCheck); + int maxRowCountForPageSizeCheck = getMaxRowCountForPageSizeCheck(conf); + if(INFO) LOG.info("Min row count for page size check is: " + maxRowCountForPageSizeCheck); CodecFactory codecFactory = new CodecFactory(conf, pageSize); @@ -398,7 +405,8 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp codecFactory.getCompressor(codec), dictionaryPageSize, enableDictionary, - initialRowCountForPageSizeCheck, + minRowCountForPageSizeCheck, + maxRowCountForPageSizeCheck, estimateNextSizeCheck, validating, writerVersion, diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index 9cf0b73f84..3b08e53b4a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -31,7 +31,8 @@ import org.apache.parquet.schema.MessageType; import static org.apache.parquet.Preconditions.checkNotNull; -import static org.apache.parquet.column.ParquetProperties.INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK; +import static org.apache.parquet.column.ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK; import static org.apache.parquet.column.ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK; /** @@ -74,39 +75,39 @@ public ParquetRecordWriter( WriterVersion writerVersion) { internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, - 100, false, validating, writerVersion, new HeapByteBufferAllocator()); + DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, + DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, validating, writerVersion, new HeapByteBufferAllocator()); } - /** - * - * @param w the file to write to - * @param writeSupport the class to convert incoming records - * @param schema the schema of the records - * @param extraMetaData extra meta data to write in the footer of the file - * @param blockSize the size of a block in the file (this will be approximate) - * @param compressor the compressor used to compress the pages - * @param dictionaryPageSize the threshold for dictionary size - * @param enableDictionary to enable the dictionary - * @param validating if schema validation should be turned on - */ - public ParquetRecordWriter( - ParquetFileWriter w, - WriteSupport writeSupport, - MessageType schema, - Map extraMetaData, - long blockSize, int pageSize, - BytesCompressor compressor, - int dictionaryPageSize, - boolean enableDictionary, - boolean validating, - WriterVersion writerVersion, - MemoryManager memoryManager) { - internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, - INITIAL_ROW_COUNT_FOR_PAGE_SIZE_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, - validating, writerVersion, new HeapByteBufferAllocator()); - this.memoryManager = checkNotNull(memoryManager, "memoryManager"); - memoryManager.addWriter(internalWriter, blockSize); + /** + * + * @param w the file to write to + * @param writeSupport the class to convert incoming records + * @param schema the schema of the records + * @param extraMetaData extra meta data to write in the footer of the file + * @param blockSize the size of a block in the file (this will be approximate) + * @param compressor the compressor used to compress the pages + * @param dictionaryPageSize the threshold for dictionary size + * @param enableDictionary to enable the dictionary + * @param validating if schema validation should be turned on + */ + public ParquetRecordWriter( + ParquetFileWriter w, + WriteSupport writeSupport, + MessageType schema, + Map extraMetaData, + long blockSize, int pageSize, + BytesCompressor compressor, + int dictionaryPageSize, + boolean enableDictionary, + boolean validating, + WriterVersion writerVersion, + MemoryManager memoryManager) { + this(w, writeSupport, schema, + extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, + DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK, + DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, + validating, writerVersion, memoryManager); } /** @@ -119,6 +120,9 @@ public ParquetRecordWriter( * @param compressor the compressor used to compress the pages * @param dictionaryPageSize the threshold for dictionary size * @param enableDictionary to enable the dictionary + * @param minRowCountForSizeCheck min number of rows before page size check + * @param maxRowCountForSizeCheck max number of rows before page size check + * @param estimateNextPageSizeCheck dynamically estimate next page size check (min will be used if false) * @param validating if schema validation should be turned on */ public ParquetRecordWriter( @@ -130,14 +134,15 @@ public ParquetRecordWriter( BytesCompressor compressor, int dictionaryPageSize, boolean enableDictionary, - int initialRowCountForSizeCheck, + int minRowCountForSizeCheck, + int maxRowCountForSizeCheck, boolean estimateNextPageSizeCheck, boolean validating, WriterVersion writerVersion, MemoryManager memoryManager) { internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, initialRowCountForSizeCheck, - estimateNextPageSizeCheck, + extraMetaData, blockSize, pageSize, compressor, dictionaryPageSize, enableDictionary, minRowCountForSizeCheck, + maxRowCountForSizeCheck, estimateNextPageSizeCheck, validating, writerVersion, new HeapByteBufferAllocator()); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); From 71336eefde68b60f198e762a87605e3dafe45fa8 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Wed, 2 Dec 2015 10:00:24 -0800 Subject: [PATCH 6/8] Fixed param name --- .../org/apache/parquet/column/impl/ColumnWriteStoreV1.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java index 6222b4380d..ad9946c6fc 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV1.java @@ -52,13 +52,13 @@ public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, this(pageWriteStore, pageSizeThreshold, dictionaryPageSizeThreshold, enableDictionary, DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK, DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK, writerVersion, allocator); } - public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int maxRowCountForPageSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { + public ColumnWriteStoreV1(PageWriteStore pageWriteStore, int pageSizeThreshold, int dictionaryPageSizeThreshold, boolean enableDictionary, int initialRowCountForPageSizeCheck, boolean estimateNextSizeCheck, WriterVersion writerVersion, ByteBufferAllocator allocator) { super(); this.pageWriteStore = pageWriteStore; this.pageSizeThreshold = pageSizeThreshold; this.dictionaryPageSizeThreshold = dictionaryPageSizeThreshold; this.enableDictionary = enableDictionary; - this.initialRowCountForPageSizeCheck = maxRowCountForPageSizeCheck; + this.initialRowCountForPageSizeCheck = initialRowCountForPageSizeCheck; this.estimateNextSizeCheck = estimateNextSizeCheck; this.writerVersion = writerVersion; this.allocator = allocator; From 2090719c76eaa6ca66cb8fd2ef62a5744f00baff Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Wed, 2 Dec 2015 10:10:58 -0800 Subject: [PATCH 7/8] Update sizeCheck to write page properly if estimating is turned off --- .../column/impl/ColumnWriteStoreV2.java | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java index 64ea40abe0..197bc95e0e 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java @@ -148,11 +148,6 @@ public void endRecord() { } private void sizeCheck() { - if(!parquetProperties.isEstimateNextSizeCheck()) { - rowCountForNextSizeCheck = rowCount + parquetProperties.getMinRowCountForPageSizeCheck(); - return; - } - long minRecordToWait = Long.MAX_VALUE; for (ColumnWriterV2 writer : writers) { long usedMem = writer.getCurrentPageBufferedSize(); @@ -173,11 +168,16 @@ private void sizeCheck() { if (minRecordToWait == Long.MAX_VALUE) { minRecordToWait = parquetProperties.getMinRowCountForPageSizeCheck(); } - // will check again halfway - rowCountForNextSizeCheck = rowCount + - min( - max(minRecordToWait / 2, parquetProperties.getMinRowCountForPageSizeCheck()), // no less than MINIMUM_RECORD_COUNT_FOR_CHECK - parquetProperties.getMaxRowCountForPageSizeCheck()); // no more than MAXIMUM_RECORD_COUNT_FOR_CHECK + + if(parquetProperties.isEstimateNextSizeCheck()) { + // will check again halfway + rowCountForNextSizeCheck = rowCount + + min( + max(minRecordToWait / 2, parquetProperties.getMinRowCountForPageSizeCheck()), // no less than MINIMUM_RECORD_COUNT_FOR_CHECK + parquetProperties.getMaxRowCountForPageSizeCheck()); // no more than MAXIMUM_RECORD_COUNT_FOR_CHECK + } else { + rowCountForNextSizeCheck = rowCount + parquetProperties.getMinRowCountForPageSizeCheck(); + } } } From 18f8d3aa72b6058cebbe9a5f1f4d70e131cc0223 Mon Sep 17 00:00:00 2001 From: Daniel Weeks Date: Wed, 2 Dec 2015 10:14:13 -0800 Subject: [PATCH 8/8] Spacing --- .../java/org/apache/parquet/hadoop/ParquetOutputFormat.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index e9d701dc60..6e28266753 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -372,11 +372,11 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp int maxPaddingSize = getMaxPaddingSize(conf); if (INFO) LOG.info("Maximum row group padding size is " + maxPaddingSize + " bytes"); boolean estimateNextSizeCheck = getEstimatePageSizeCheck(conf); - if(INFO) LOG.info("Page size checking is: " + (estimateNextSizeCheck ? "estimated" : "constant")); + if (INFO) LOG.info("Page size checking is: " + (estimateNextSizeCheck ? "estimated" : "constant")); int minRowCountForPageSizeCheck = getMinRowCountForPageSizeCheck(conf); - if(INFO) LOG.info("Min row count for page size check is: " + minRowCountForPageSizeCheck); + if (INFO) LOG.info("Min row count for page size check is: " + minRowCountForPageSizeCheck); int maxRowCountForPageSizeCheck = getMaxRowCountForPageSizeCheck(conf); - if(INFO) LOG.info("Min row count for page size check is: " + maxRowCountForPageSizeCheck); + if (INFO) LOG.info("Min row count for page size check is: " + maxRowCountForPageSizeCheck); CodecFactory codecFactory = new CodecFactory(conf, pageSize);