Skip to content

PARQUET-869 Configurable min/max record counts for block size check #447

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

Open
pradeepg26 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

PARQUET-869 Configurable min/max record counts for block size check #447

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7187a3d
PARQUET-869 Configurable record counts for block size checks
Jan 10, 2018
7ba6473
Merge remote-tracking branch 'apache/master'
Jan 10, 2018
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
80 changes: 64 additions & 16 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
@@ -1,14 +1,14 @@
/*
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
Expand Down Expand Up @@ -48,8 +48,11 @@ public class ParquetProperties {
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;
public static final boolean DEFAULT_ESTIMATE_ROW_COUNT_FOR_BLOCK_SIZE_CHECK = true;
public static final int DEFAULT_MINIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK = 100;
public static final int DEFAULT_MAXIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK = 10000;
public static final int DEFAULT_MINIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK = 100;
Copy link
Contributor

@rdblue rdblue Jan 31, 2018

Choose a reason for hiding this comment

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

Instead of block, please use "row group". Block is an overused term that we are avoiding in the public API, except for those places where it already appears.

Copy link
Author

@pradeepg26 pradeepg26 Jan 31, 2018

Choose a reason for hiding this comment

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

Will do.

public static final int DEFAULT_MAXIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK = 10000;

public static final ValuesWriterFactory DEFAULT_VALUES_WRITER_FACTORY = new DefaultValuesWriterFactory();

Expand Down Expand Up @@ -83,12 +86,15 @@ public static WriterVersion fromString(String name) {
private final boolean enableDictionary;
private final int minRowCountForPageSizeCheck;
private final int maxRowCountForPageSizeCheck;
private final boolean estimateNextSizeCheck;
private final int minRowCountForBlockSizeCheck;
private final int maxRowCountForBlockSizeCheck;
private final boolean estimateNextPageSizeCheck;
private final boolean estimateNextBlockSizeCheck;
private final ByteBufferAllocator allocator;
private final ValuesWriterFactory valuesWriterFactory;

private ParquetProperties(WriterVersion writerVersion, int pageSize, int dictPageSize, boolean enableDict, int minRowCountForPageSizeCheck,
int maxRowCountForPageSizeCheck, boolean estimateNextSizeCheck, ByteBufferAllocator allocator,
int maxRowCountForPageSizeCheck, int minRowCountForBlockSizeCheck, int maxRowCountForBlockSizeCheck, boolean estimateNextPageSizeCheck, boolean estimateNextBlockSizeCheck, ByteBufferAllocator allocator,
ValuesWriterFactory writerFactory) {
this.pageSizeThreshold = pageSize;
this.initialSlabSize = CapacityByteArrayOutputStream
Expand All @@ -98,7 +104,10 @@ private ParquetProperties(WriterVersion writerVersion, int pageSize, int dictPag
this.enableDictionary = enableDict;
this.minRowCountForPageSizeCheck = minRowCountForPageSizeCheck;
this.maxRowCountForPageSizeCheck = maxRowCountForPageSizeCheck;
this.estimateNextSizeCheck = estimateNextSizeCheck;
this.minRowCountForBlockSizeCheck = minRowCountForBlockSizeCheck;
this.maxRowCountForBlockSizeCheck = maxRowCountForBlockSizeCheck;
this.estimateNextPageSizeCheck = estimateNextPageSizeCheck;
this.estimateNextBlockSizeCheck = estimateNextBlockSizeCheck;
this.allocator = allocator;

this.valuesWriterFactory = writerFactory;
Expand Down Expand Up @@ -182,12 +191,24 @@ public int getMaxRowCountForPageSizeCheck() {
return maxRowCountForPageSizeCheck;
}

public int getMinRowCountForBlockSizeCheck() {
return minRowCountForBlockSizeCheck;
}

public int getMaxRowCountForBlockSizeCheck() {
return maxRowCountForBlockSizeCheck;
}

public ValuesWriterFactory getValuesWriterFactory() {
return valuesWriterFactory;
}

public boolean estimateNextSizeCheck() {
return estimateNextSizeCheck;
public boolean estimateNextPageSizeCheck() {
return estimateNextPageSizeCheck;
}

public boolean estimateNextBlockSizeCheck() {
return estimateNextBlockSizeCheck;
}

public static Builder builder() {
Expand All @@ -203,9 +224,12 @@ public static class Builder {
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 int minRowCountForPageSizeCheck = DEFAULT_MINIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK;
private int maxRowCountForPageSizeCheck = DEFAULT_MAXIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK;
private boolean estimateNextPageSizeCheck = DEFAULT_ESTIMATE_ROW_COUNT_FOR_PAGE_SIZE_CHECK;
private int minRowCountForBlockSizeCheck = DEFAULT_MINIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK;
private int maxRowCountForBlockSizeCheck = DEFAULT_MAXIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK;
private boolean estimateNextBlockSizeCheck = DEFAULT_ESTIMATE_ROW_COUNT_FOR_BLOCK_SIZE_CHECK;
private ByteBufferAllocator allocator = new HeapByteBufferAllocator();
private ValuesWriterFactory valuesWriterFactory = DEFAULT_VALUES_WRITER_FACTORY;

Expand All @@ -218,7 +242,10 @@ private Builder(ParquetProperties toCopy) {
this.writerVersion = toCopy.writerVersion;
this.minRowCountForPageSizeCheck = toCopy.minRowCountForPageSizeCheck;
this.maxRowCountForPageSizeCheck = toCopy.maxRowCountForPageSizeCheck;
this.estimateNextSizeCheck = toCopy.estimateNextSizeCheck;
this.estimateNextPageSizeCheck = toCopy.estimateNextPageSizeCheck;
this.minRowCountForBlockSizeCheck = toCopy.minRowCountForBlockSizeCheck;
this.maxRowCountForBlockSizeCheck = toCopy.maxRowCountForBlockSizeCheck;
this.estimateNextBlockSizeCheck = toCopy.estimateNextBlockSizeCheck;
this.allocator = toCopy.allocator;
}

Expand Down Expand Up @@ -284,9 +311,28 @@ public Builder withMaxRowCountForPageSizeCheck(int max) {
return this;
}

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

public Builder withMaxRowCountForBlockSizeCheck(int max) {
Preconditions.checkArgument(max > 0,
"Invalid row count for block size check (negative): %s", max);
this.maxRowCountForBlockSizeCheck = 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) {
this.estimateNextSizeCheck = estimateNextSizeCheck;
this.estimateNextPageSizeCheck = estimateNextSizeCheck;
return this;
}

public Builder estimateRowCountForBlockSizeCheck(boolean estimateBlockSizeCheck) {
this.estimateNextBlockSizeCheck = estimateBlockSizeCheck;
return this;
}

Expand All @@ -306,7 +352,9 @@ public ParquetProperties build() {
ParquetProperties properties =
new ParquetProperties(writerVersion, pageSize, dictPageSize,
enableDict, minRowCountForPageSizeCheck, maxRowCountForPageSizeCheck,
estimateNextSizeCheck, allocator, valuesWriterFactory);
minRowCountForBlockSizeCheck, maxRowCountForBlockSizeCheck,
estimateNextPageSizeCheck, estimateNextBlockSizeCheck,
allocator, valuesWriterFactory);
// we pass a constructed but uninitialized factory to ParquetProperties above as currently
// creation of ValuesWriters is invoked from within ParquetProperties. In the future
// we'd like to decouple that and won't need to pass an object to properties and then pass the
Expand Down
8 changes: 4 additions & 4 deletions parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriteStoreV2.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
/*
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
Expand Down Expand Up @@ -161,7 +161,7 @@ private void sizeCheck() {
minRecordToWait = props.getMinRowCountForPageSizeCheck();
}

if(props.estimateNextSizeCheck()) {
if(props.estimateNextPageSizeCheck()) {
// will check again halfway if between min and max
rowCountForNextSizeCheck = rowCount +
min(
Expand Down
10 changes: 5 additions & 5 deletions parquet-column/src/main/java/org/apache/parquet/column/impl/ColumnWriterV1.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
/*
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
Expand Down Expand Up @@ -101,13 +101,13 @@ private void accountForValueWritten() {
+ dataColumn.getBufferedSize();
if (memSize > props.getPageSizeThreshold()) {
// we will write the current page and check again the size at the predicted middle of next page
if (props.estimateNextSizeCheck()) {
if (props.estimateNextPageSizeCheck()) {
valueCountForNextSizeCheck = valueCount / 2;
} else {
valueCountForNextSizeCheck = props.getMinRowCountForPageSizeCheck();
}
writePage();
} else if (props.estimateNextSizeCheck()) {
} else if (props.estimateNextPageSizeCheck()) {
// not reached the threshold, will check again midway
valueCountForNextSizeCheck = (int)(valueCount + ((float)valueCount * props.getPageSizeThreshold() / memSize)) / 2 + 1;
} else {
Expand Down
35 changes: 24 additions & 11 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,14 +42,11 @@
class InternalParquetRecordWriter<T> {
private static final Logger LOG = LoggerFactory.getLogger(InternalParquetRecordWriter.class);

private static final int MINIMUM_RECORD_COUNT_FOR_CHECK = 100;
private static final int MAXIMUM_RECORD_COUNT_FOR_CHECK = 10000;

private final ParquetFileWriter parquetFileWriter;
private final WriteSupport<T> writeSupport;
private final MessageType schema;
private final Map<String, String> extraMetaData;
private final long rowGroupSize;
private final boolean estimateNextBlockSizeCheck;
Copy link
Contributor

@rdblue rdblue Jan 31, 2018

Choose a reason for hiding this comment

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

Was rowGroupSize not used? Why was it removed?

Copy link
Author

@pradeepg26 pradeepg26 Jan 31, 2018

Choose a reason for hiding this comment

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

I wrote this code over a year ago... I don't remember, but I'll check. I think it was an unused variable.

private long rowGroupSizeThreshold;
private long nextRowGroupSize;
private final BytesCompressor compressor;
Expand All @@ -59,12 +56,14 @@ class InternalParquetRecordWriter<T> {
private boolean closed;

private long recordCount = 0;
private long recordCountForNextMemCheck = MINIMUM_RECORD_COUNT_FOR_CHECK;
private long recordCountForNextMemCheck;
private long lastRowGroupEndPos = 0;

private ColumnWriteStore columnStore;
private ColumnChunkPageWriteStore pageStore;
private RecordConsumer recordConsumer;
private int minRecordCountForBlockSizeCheck;
private int maxRecordCountForBlockSizeCheck;

/**
* @param parquetFileWriter the file to write to
Expand All @@ -87,12 +86,14 @@ public InternalParquetRecordWriter(
this.writeSupport = checkNotNull(writeSupport, "writeSupport");
this.schema = schema;
this.extraMetaData = extraMetaData;
this.rowGroupSize = rowGroupSize;
this.rowGroupSizeThreshold = rowGroupSize;
this.nextRowGroupSize = rowGroupSizeThreshold;
this.compressor = compressor;
this.validating = validating;
this.props = props;
this.minRecordCountForBlockSizeCheck = props.getMinRowCountForPageSizeCheck();
this.maxRecordCountForBlockSizeCheck = props.getMaxRowCountForPageSizeCheck();
this.estimateNextBlockSizeCheck = props.estimateNextBlockSizeCheck();
initStore();
}

Expand All @@ -102,6 +103,8 @@ private void initStore() {
MessageColumnIO columnIO = new ColumnIOFactory(validating).getColumnIO(schema);
this.recordConsumer = columnIO.getRecordWriter(columnStore);
writeSupport.prepareForWrite(recordConsumer);
System.out.println(String.format("Created ParquetWriter with [%d, %d] for block size checks. Estimation(%s). BlockSize(%d)",
Copy link
Contributor

@rdblue rdblue Jan 31, 2018

Choose a reason for hiding this comment

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

Please use SLF4J for logging. You can see examples in the rest of the library.

Copy link
Author

@pradeepg26 pradeepg26 Jan 31, 2018

Choose a reason for hiding this comment

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

Oops... missed removing this statement.

minRecordCountForBlockSizeCheck, maxRecordCountForBlockSizeCheck, estimateNextBlockSizeCheck, rowGroupSizeThreshold));
}

public void close() throws IOException, InterruptedException {
Expand Down Expand Up @@ -142,13 +145,23 @@ private void checkBlockSizeReached() throws IOException {
LOG.info("mem size {} > {}: flushing {} records to disk.", memSize, nextRowGroupSize, recordCount);
flushRowGroupToStore();
initStore();
recordCountForNextMemCheck = min(max(MINIMUM_RECORD_COUNT_FOR_CHECK, recordCount / 2), MAXIMUM_RECORD_COUNT_FOR_CHECK);
if (estimateNextBlockSizeCheck) {
Copy link
Contributor

@rdblue rdblue Jan 31, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

Do you need this setting, or is it included to match the page size checking? I don't think we would ever want to set it to true because it is important to be under the max row group size.

Copy link
Author

@pradeepg26 pradeepg26 Jan 31, 2018

Choose a reason for hiding this comment

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

Will do.

Copy link
Author

@pradeepg26 pradeepg26 Feb 2, 2018

Choose a reason for hiding this comment

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

Working on the comments @rdblue... do you think instead of a min/max record count for row group size checks, we should check every n rows?

Copy link
Author

@pradeepg26 pradeepg26 Feb 2, 2018

Choose a reason for hiding this comment

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

I've taken a closer look at this code. The only change with the estimateNextBlockSizeCheck is that it uses a static min number of records before performing the next size check if disabled.

If estimateNextBlockSizeCheck is enabled, then existing behavior is preserved where we try to estimate when to perform the next size check by performing the check at the halfway point of the predicted number of records (unless it's larger than the max).

So, I added the estimateNextBlockSizeCheck to keep it consistent with page size checking.

Copy link
Contributor

@rdblue rdblue Feb 2, 2018

Choose a reason for hiding this comment

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

Consistency here is fine. I'm not sure we will use it, but I can certainly think of valid use cases for it.

recordCountForNextMemCheck = min(max(minRecordCountForBlockSizeCheck, recordCount / 2),
maxRecordCountForBlockSizeCheck);
} else {
recordCountForNextMemCheck = minRecordCountForBlockSizeCheck;
}
this.lastRowGroupEndPos = parquetFileWriter.getPos();
} else {
recordCountForNextMemCheck = min(
max(MINIMUM_RECORD_COUNT_FOR_CHECK, (recordCount + (long)(nextRowGroupSize / ((float)recordSize))) / 2), // will check halfway
recordCount + MAXIMUM_RECORD_COUNT_FOR_CHECK // will not look more than max records ahead
);
if (estimateNextBlockSizeCheck) {
recordCountForNextMemCheck = min(
max(minRecordCountForBlockSizeCheck, (recordCount + (long) (nextRowGroupSize / ((float) recordSize))) / 2),
// will check halfway
recordCount + maxRecordCountForBlockSizeCheck // will not look more than max records ahead
);
} else {
recordCountForNextMemCheck += minRecordCountForBlockSizeCheck;
}
LOG.debug("Checked mem at {} will check again at: {}", recordCount, recordCountForNextMemCheck);
}
}
Expand Down
41 changes: 32 additions & 9 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,14 @@
/*
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
Expand Down Expand Up @@ -142,9 +142,12 @@ 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 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";
public static final String MIN_ROW_COUNT_FOR_PAGE_SIZE_CHECK = "parquet.page.size_row_check_min";
Copy link
Contributor

@rdblue rdblue Jan 31, 2018

Choose a reason for hiding this comment

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

The page properties should not change, and the row group properties should match the naming convention for pages.

Copy link
Author

@pradeepg26 pradeepg26 Jan 31, 2018

Choose a reason for hiding this comment

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

Oops... it was changed for internal reasons... forgot to change it back.

public static final String MAX_ROW_COUNT_FOR_PAGE_SIZE_CHECK = "parquet.page.size_row_check_max";
public static final String MIN_ROW_COUNT_FOR_BLOCK_SIZE_CHECK = "parquet.block.size_row_check_min";
public static final String MAX_ROW_COUNT_FOR_BLOCK_SIZE_CHECK = "parquet.block.size_row_check_max";
public static final String ESTIMATE_PAGE_SIZE_CHECK = "parquet.page.size_check_estimate";
public static final String ESTIMATE_BLOCK_SIZE_CHECK = "parquet.block.size_check_estimate";

public static JobSummaryLevel getJobSummaryLevel(Configuration conf) {
String level = conf.get(JOB_SUMMARY_LEVEL);
Expand Down Expand Up @@ -245,12 +248,27 @@ public static boolean getEnableDictionary(Configuration configuration) {

public static int getMinRowCountForPageSizeCheck(Configuration configuration) {
return configuration.getInt(MIN_ROW_COUNT_FOR_PAGE_SIZE_CHECK,
ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_CHECK);
ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK);
}

public static int getMaxRowCountForPageSizeCheck(Configuration configuration) {
return configuration.getInt(MAX_ROW_COUNT_FOR_PAGE_SIZE_CHECK,
ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_CHECK);
ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_PAGE_SIZE_CHECK);
}

public static int getMinRowCountForBlockSizeCheck(Configuration configuration) {
return configuration.getInt(MIN_ROW_COUNT_FOR_BLOCK_SIZE_CHECK,
ParquetProperties.DEFAULT_MINIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK);
}

public static int getMaxRowCountForBlockSizeCheck(Configuration configuration) {
return configuration.getInt(MAX_ROW_COUNT_FOR_BLOCK_SIZE_CHECK,
ParquetProperties.DEFAULT_MAXIMUM_RECORD_COUNT_FOR_BLOCK_SIZE_CHECK);
}

public static boolean getEstimateBlockSizeCheck(Configuration configuration) {
return configuration.getBoolean(ESTIMATE_BLOCK_SIZE_CHECK,
ParquetProperties.DEFAULT_ESTIMATE_ROW_COUNT_FOR_BLOCK_SIZE_CHECK);
}

public static boolean getEstimatePageSizeCheck(Configuration configuration) {
Expand Down Expand Up @@ -362,8 +380,11 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
.withDictionaryEncoding(getEnableDictionary(conf))
.withWriterVersion(getWriterVersion(conf))
.estimateRowCountForPageSizeCheck(getEstimatePageSizeCheck(conf))
.estimateRowCountForBlockSizeCheck(getEstimateBlockSizeCheck(conf))
.withMinRowCountForPageSizeCheck(getMinRowCountForPageSizeCheck(conf))
.withMaxRowCountForPageSizeCheck(getMaxRowCountForPageSizeCheck(conf))
.withMinRowCountForBlockSizeCheck(getMinRowCountForBlockSizeCheck(conf))
.withMaxRowCountForBlockSizeCheck(getMaxRowCountForBlockSizeCheck(conf))
.build();

long blockSize = getLongBlockSize(conf);
Expand All @@ -378,9 +399,11 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
LOG.info("Validation is {}", (validating ? "on" : "off"));
LOG.info("Writer version is: {}", props.getWriterVersion());
LOG.info("Maximum row group padding size is {} bytes", maxPaddingSize);
LOG.info("Page size checking is: {}", (props.estimateNextSizeCheck() ? "estimated" : "constant"));
LOG.info("Page size checking is: {}", (props.estimateNextPageSizeCheck() ? "estimated" : "constant"));
LOG.info("Min row count for page size check is: {}", props.getMinRowCountForPageSizeCheck());
LOG.info("Max row count for page size check is: {}", props.getMaxRowCountForPageSizeCheck());
LOG.info("Min row count for block size check is: {}", props.getMinRowCountForBlockSizeCheck());
LOG.info("Max row count for block size check is: {}", props.getMaxRowCountForBlockSizeCheck());
}

WriteContext init = writeSupport.init(conf);
Expand Down