Skip to content

PARQUET-1533: TestSnappy() throws OOM exception with Parquet-1485 change #622

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

Merged
gszadovszky merged 3 commits into from
Feb 25, 2019
Merged

PARQUET-1533: TestSnappy() throws OOM exception with Parquet-1485 change #622

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5a1bd29
PARQUET-1533: TestSnappy() throws OOM exception with Parquet-1485 change
gszadovszky Feb 20, 2019
938e8b6
PARQUET-1533: Remove unnecessary constructor from the compressor just…
gszadovszky Feb 20, 2019
6712686
PARQUET-1533: Adding clean to end()
gszadovszky Feb 22, 2019
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
18 changes: 2 additions & 16 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/SnappyCompressor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,13 +32,11 @@
* entire input in setInput and compresses it as one compressed block.
*/
public class SnappyCompressor implements Compressor {
private static final int initialBufferSize = 64 * 1024 * 1024;

// Buffer for compressed output. This buffer grows as necessary.
private ByteBuffer outputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
private ByteBuffer outputBuffer = ByteBuffer.allocateDirect(0);

// Buffer for uncompressed input. This buffer grows as necessary.
private ByteBuffer inputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
private ByteBuffer inputBuffer = ByteBuffer.allocateDirect(0);

private long bytesRead = 0L;
private long bytesWritten = 0L;
Expand Down Expand Up @@ -157,18 +155,6 @@ public void reinit(Configuration c) {

@Override
public synchronized void reset() {
if (inputBuffer.capacity() > initialBufferSize) {

@liupc liupc Feb 21, 2019 •
edited
Loading

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

May be we can keep these codes in the reset call and lower the initialBufferSize to 4MiB or else, as event without the preallocation of 128MiB, the inputBuffer and writeBuffer can grow very large if we just see it from the API itself, because the setInput can be called many times.

@gszadovszky gszadovszky Feb 21, 2019

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.

I cannot see why the buffers would grow that much. We compress/decompress one page at a time so it should not grow much larger than the largest page which is usually much smaller than 64M. By default it is 1M.

@shangxinli shangxinli Feb 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I still don't know 4M is a good number to be. If multiple compressors/decompressors are instantiated in parallel, the number can go up. Let's say 20 decompressors and each initialized with 2*4M, that results in 160M direct memory. That could be a problem for some applications. I agree with liupc that "keep the reset logic and remove the preallocation codes" unless we see a huge penalty for not doing that.

@gszadovszky gszadovszky Feb 21, 2019

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.

If you instantiates 20 decompressors then you probably tries to read 20 parquet files in the same time in the same jvm. It means you'll need tons of memory to keep 20 row-groups in the memory. Comparing to that 160M would not be too much I guess.
Meanwhile, I am not against keeping the reset logic but what would be the size to be reset to? 0?

@shangxinli shangxinli Feb 21, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

How about vectorized parquet reader? Does it instantiate the decompressor per column? I am not an expert of Parquet, so I could be wrong.

@liupc liupc Feb 22, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

@shangxinli
I think it's up to the implementation, either instantiating the decompressor per column or not can do this work. If you keep so many column compressor/decompressor in memory, then the memory consumed by the column data itself is much larger than 4MiB.
What's more, I suggest 4MiB here because the compress/decompress buffer size is 4MiB now. so I think it's ok.

@shangxinli shangxinli Feb 22, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

If the default page size is 1M(assume the typical usage is default), 4M still seems too large. It will be adjusted later anyway if it is smaller than needs. I would prefer low over high.

ByteBuffer oldBuffer = inputBuffer;
inputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
CleanUtil.clean(oldBuffer);
}

if (outputBuffer.capacity() > initialBufferSize) {
ByteBuffer oldBuffer = outputBuffer;
outputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
CleanUtil.clean(oldBuffer);
}

finishCalled = false;
bytesRead = bytesWritten = 0;
inputBuffer.rewind();
Expand Down
23 changes: 2 additions & 21 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/SnappyDecompressor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,21 +27,14 @@
import org.apache.parquet.Preconditions;

public class SnappyDecompressor implements Decompressor {
private static final int initialBufferSize = 64 * 1024 * 1024;

// Buffer for uncompressed output. This buffer grows as necessary.
private ByteBuffer outputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
private ByteBuffer outputBuffer = ByteBuffer.allocateDirect(0);

// Buffer for compressed input. This buffer grows as necessary.
private ByteBuffer inputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
private ByteBuffer inputBuffer = ByteBuffer.allocateDirect(0);

private boolean finished;

public SnappyDecompressor() {
inputBuffer.limit(0);
outputBuffer.limit(0);
}

/**
* Fills specified buffer with uncompressed data. Returns actual number
* of bytes of uncompressed data. A return value of 0 indicates that
Expand Down Expand Up @@ -142,18 +135,6 @@ public synchronized boolean needsInput() {

@Override
public synchronized void reset() {
if (inputBuffer.capacity() > initialBufferSize) {

@liupc liupc Feb 21, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Same as above comments.

ByteBuffer oldBuffer = inputBuffer;
inputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
CleanUtil.clean(oldBuffer);
}

if (outputBuffer.capacity() > initialBufferSize) {
ByteBuffer oldBuffer = outputBuffer;
outputBuffer = ByteBuffer.allocateDirect(initialBufferSize);
CleanUtil.clean(oldBuffer);
}

finished = false;
inputBuffer.rewind();
outputBuffer.rewind();
Expand Down