Skip to content

[SPARK-23972][BUILD][SQL] Update Parquet to 1.10.0. #21070

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Closed
rdblue wants to merge 11 commits into from
Closed

[SPARK-23972][BUILD][SQL] Update Parquet to 1.10.0. #21070

Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
26336a9
SPARK-23972: Update Parquet to 1.10.0.
rdblue Dec 1, 2017
a599515
SPARK-23972: Update dependencies for Parquet 1.10.0.
rdblue Apr 16, 2018
3e1f84c
SPARK-23972: Fix method signatures for IOExceptions.
rdblue Apr 16, 2018
93b8ee8
SPAKR-23972: Bulk write to ColumnVector when Parquet data is on the h…
rdblue Apr 16, 2018
93f2a92
SPARK-23972: Fix test failures and review comments.
rdblue Apr 18, 2018
64907bc
SPARK-23972: Add lz4, brotli, and zstd compression for Parquet.
rdblue Apr 18, 2018
66df4b6
SPARK-23972: Update InMemoryColumnarQuerySuite for new Parquet version.
rdblue Apr 18, 2018
ee2ffdd
SPARK-23972: Regenerate golden outputs for Parquet size changes.
rdblue Apr 19, 2018
270540f
SPARK-23972: Update Hadoop 3.1 dependency profile.
rdblue May 7, 2018
6c9d47b
SPARK-23972: Update sql-programming-guide.md.
rdblue May 7, 2018
95ecde0
Fix buffer unpacking offset in VectorizedRleValuesReader.
rdblue May 8, 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
Prev Previous commit
Next Next commit
SPAKR-23972: Bulk write to ColumnVector when Parquet data is on the h… 
…eap.
  • Loading branch information
@rdblue
rdblue committed May 7, 2018
commit 93b8ee876f6b9fef41b47fc7d8d4d5af0e8083e5
37 changes: 29 additions & 8 deletions .../java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@

import org.apache.parquet.column.values.ValuesReader;
import org.apache.parquet.io.api.Binary;
import org.apache.spark.unsafe.Platform;

/**
* An implementation of the Parquet PLAIN decoder that supports the vectorized interface.
Expand Down Expand Up @@ -71,8 +72,13 @@ public final void readIntegers(int total, WritableColumnVector c, int rowId) {
int requiredBytes = total * 4;
ByteBuffer buffer = getBuffer(requiredBytes);

for (int i = 0; i < total; i += 1) {
c.putInt(rowId + i, buffer.getInt());
if (buffer.hasArray()) {
Copy link
Contributor

@cloud-fan cloud-fan May 7, 2018

Choose a reason for hiding this comment

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

shall we assert buffer.hasArray() is always true?

Copy link
Contributor Author

@rdblue rdblue May 7, 2018

Choose a reason for hiding this comment

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

No, there is no guarantee that the buffer from Parquet is on the heap.

int offset = buffer.arrayOffset() + buffer.position();
c.putIntsLittleEndian(rowId, total, buffer.array(), offset - Platform.BYTE_ARRAY_OFFSET);
} else {
for (int i = 0; i < total; i += 1) {
c.putInt(rowId + i, buffer.getInt());
}
}
}

Expand All @@ -81,8 +87,13 @@ public final void readLongs(int total, WritableColumnVector c, int rowId) {
int requiredBytes = total * 8;
ByteBuffer buffer = getBuffer(requiredBytes);

for (int i = 0; i < total; i += 1) {
c.putLong(rowId + i, buffer.getLong());
if (buffer.hasArray()) {
int offset = buffer.arrayOffset() + buffer.position();
c.putLongsLittleEndian(rowId, total, buffer.array(), offset - Platform.BYTE_ARRAY_OFFSET);
} else {
for (int i = 0; i < total; i += 1) {
c.putLong(rowId + i, buffer.getLong());
}
}
}

Expand All @@ -91,8 +102,13 @@ public final void readFloats(int total, WritableColumnVector c, int rowId) {
int requiredBytes = total * 4;
ByteBuffer buffer = getBuffer(requiredBytes);

for (int i = 0; i < total; i += 1) {
c.putFloat(rowId + i, buffer.getFloat());
if (buffer.hasArray()) {
int offset = buffer.arrayOffset() + buffer.position();
c.putFloats(rowId, total, buffer.array(), offset - Platform.BYTE_ARRAY_OFFSET);
} else {
for (int i = 0; i < total; i += 1) {
c.putFloat(rowId + i, buffer.getFloat());
}
}
}

Expand All @@ -101,8 +117,13 @@ public final void readDoubles(int total, WritableColumnVector c, int rowId) {
int requiredBytes = total * 8;
ByteBuffer buffer = getBuffer(requiredBytes);

for (int i = 0; i < total; i += 1) {
c.putDouble(rowId + i, buffer.getDouble());
if (buffer.hasArray()) {
int offset = buffer.arrayOffset() + buffer.position();
c.putDoubles(rowId, total, buffer.array(), offset - Platform.BYTE_ARRAY_OFFSET);
} else {
for (int i = 0; i < total; i += 1) {
c.putDouble(rowId + i, buffer.getDouble());
}
}
}

Expand Down