You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2019/08/31 00:32:01 UTC
[incubator-iceberg] branch vectorized-read updated: Support for all
primitive data types: required and optional (#435)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch vectorized-read
in repository https://gitbox.apache.org/repos/asf/incubator-iceberg.git
The following commit(s) were added to refs/heads/vectorized-read by this push:
new bdb5d0d Support for all primitive data types: required and optional (#435)
bdb5d0d is described below
commit bdb5d0d06be5f1b9fe57bded9a5613ab922256bb
Author: Anjali Norwood <52...@users.noreply.github.com>
AuthorDate: Fri Aug 30 17:31:56 2019 -0700
Support for all primitive data types: required and optional (#435)
This commit provides optimal/near-optimal implementations for all data types as follows:
+ INT32, INT64, Float, Double, Date, Timestamp (non-decimal numeric data types):
The implementation reads bytes from Parquet for batches of contiguous values as indicated
by the definition level and writes them into the underlying data buffer of the ArrowVector.
It sets validity buffers to handle optional data types.
+ Decimal data type (backed by INT32, INT64 and fixed length byte array):
Arrow stores all decimals as 16 bytes and assumes that the decimals are stored in little
endian. Vectorized decimal read implementation pads the bytes read from Parquet as necessary
and stores the decimals in the expected little endian format.
+ Fixed width binary (e.g. BYTE[7]):
Spark does not support fixed width binary data type. The data is read as fixed number of
bytes from Parquet and stored as VarBinary in Arrow and exposed to Spark as such.
+ String data type (ENUM, JSON, UTF8, BSON) and Boolean data type:
Value reader implementations are used to read the string and boolean data types.
+ UUID data type is not supported.
Co-authored-by: Samarth Jain <sa...@apache.org>
---
.../org/apache/iceberg/arrow/ArrowSchemaUtil.java | 5 +-
gradle.properties | 2 +-
.../iceberg/parquet/BatchedColumnIterator.java | 99 ++-
.../iceberg/parquet/BatchedPageIterator.java | 815 +++++++++------------
.../org/apache/iceberg/parquet/BatchedReader.java | 3 +
.../org/apache/iceberg/parquet/BytesReader.java | 56 ++
.../iceberg/parquet/ColumnarBatchReader.java | 9 +-
.../org/apache/iceberg/parquet/ParquetUtil.java | 56 ++
.../iceberg/parquet/ParquetValueReaders.java | 91 +--
.../org/apache/iceberg/parquet/VectorReader.java | 183 ++++-
.../spark/source/IcebergSourceBenchmark.java | 4 +
.../source/IcebergSourceFlatDataBenchmark.java | 39 +-
.../IcebergSourceFlatParquetDataReadBenchmark.java | 182 ++---
.../data/vector/VectorizedSparkParquetReaders.java | 4 +-
.../apache/iceberg/spark/data/AvroDataTest.java | 26 +-
.../org/apache/iceberg/spark/data/RandomData.java | 2 +-
.../org/apache/iceberg/spark/data/TestHelpers.java | 2 +-
.../data/TestSparkParquetVectorizedReader.java | 11 +-
18 files changed, 880 insertions(+), 709 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java b/core/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
index 492af61..6e982e7 100644
--- a/core/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
+++ b/core/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java
@@ -73,10 +73,9 @@ public class ArrowSchemaUtil {
switch (field.type().typeId()) {
case BINARY:
- arrowType = ArrowType.Binary.INSTANCE;
- break;
+ // Spark doesn't support BYTE(fixed_size) type, so cast it to VarBinary
case FIXED:
- arrowType = new ArrowType.FixedSizeBinary(((Types.FixedType) field.type()).length());
+ arrowType = ArrowType.Binary.INSTANCE;
break;
case BOOLEAN:
arrowType = ArrowType.Bool.INSTANCE;
diff --git a/gradle.properties b/gradle.properties
index 90a909a..53b1ce7 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -3,4 +3,4 @@ jmhIncludeRegex=.*
artifactory_contextUrl=https://artifactory.corp.adobe.com/artifactory
artifactory_user_p=
-artifactory_key_p=
+artifactory_key_p=
\ No newline at end of file
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedColumnIterator.java b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedColumnIterator.java
index e427b3e..85d3d50 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedColumnIterator.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedColumnIterator.java
@@ -20,6 +20,7 @@
package org.apache.iceberg.parquet;
import java.io.IOException;
+
import org.apache.arrow.vector.FieldVector;
import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.column.Dictionary;
@@ -36,8 +37,8 @@ public class BatchedColumnIterator {
// state reset for each row group
private PageReader pageSource = null;
- private long triplesCount = 0L;
- private long triplesRead = 0L;
+ private long totalValuesCount = 0L;
+ private long valuesRead = 0L;
private long advanceNextPageCount = 0L;
private final int batchSize;
@@ -49,8 +50,8 @@ public class BatchedColumnIterator {
public void setPageSource(PageReadStore store) {
this.pageSource = store.getPageReader(desc);
- this.triplesCount = pageSource.getTotalValueCount();
- this.triplesRead = 0L;
+ this.totalValuesCount = pageSource.getTotalValueCount();
+ this.valuesRead = 0L;
this.advanceNextPageCount = 0L;
this.batchedPageIterator.reset();
this.batchedPageIterator.setDictionary(readDictionary(desc, pageSource));
@@ -58,8 +59,8 @@ public class BatchedColumnIterator {
}
private void advance() {
- if (triplesRead >= advanceNextPageCount) {
- // A parquet page could be empty (no values) -> confirm with Ryan, anjali
+ if (valuesRead >= advanceNextPageCount) {
+ // A parquet page may be empty i.e. contains no values
while (!batchedPageIterator.hasNext()) {
DataPage page = pageSource.readPage();
if (page != null) {
@@ -73,20 +74,98 @@ public class BatchedColumnIterator {
}
public boolean hasNext() {
- return triplesRead < triplesCount;
+ return valuesRead < totalValuesCount;
+ }
+
+ /**
+ * Method for reading a batch of non-decimal numeric data types (INT32, INT64, FLOAT, DOUBLE, DATE, TIMESTAMP)
+ */
+ public void nextBatchNumericNonDecimal(FieldVector fieldVector, int typeWidth) {
+ int rowsReadSoFar = 0;
+ while (rowsReadSoFar < batchSize && hasNext()) {
+ advance();
+ int rowsInThisBatch = batchedPageIterator.nextBatchNumericNonDecimal(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar, typeWidth);
+ rowsReadSoFar += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
+ }
+ }
+
+ /**
+ * Method for reading a batch of decimals backed by INT32 and INT64 parquet data types.
+ */
+ public void nextBatchIntLongBackedDecimal(FieldVector fieldVector, int typeWidth) {
+ int rowsReadSoFar = 0;
+ while (rowsReadSoFar < batchSize && hasNext()) {
+ advance();
+ int rowsInThisBatch = batchedPageIterator.nextBatchIntLongBackedDecimal(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar, typeWidth);
+ rowsReadSoFar += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
+ }
+ }
+
+ /**
+ * Method for reading a batch of decimals backed by fixed length byte array parquet data type.
+ */
+ public void nextBatchFixedLengthDecimal(FieldVector fieldVector, int typeWidth) {
+ int rowsReadSoFar = 0;
+ while (rowsReadSoFar < batchSize && hasNext()) {
+ advance();
+ int rowsInThisBatch = batchedPageIterator.nextBatchFixedLengthDecimal(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar, typeWidth);
+ rowsReadSoFar += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
+ }
}
- public void nextBatch(FieldVector fieldVector) {
+ /**
+ * Method for reading a batch of variable width data type (ENUM, JSON, UTF8, BSON).
+ */
+ public void nextBatchVarWidthType(FieldVector fieldVector) {
int rowsReadSoFar = 0;
while (rowsReadSoFar < batchSize && hasNext()) {
advance();
- int rowsInThisBatch = batchedPageIterator.nextBatch(fieldVector, batchSize - rowsReadSoFar);
+ int rowsInThisBatch = batchedPageIterator.nextBatchVarWidthType(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar);
rowsReadSoFar += rowsInThisBatch;
- this.triplesRead += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
}
}
+ /**
+ * Method for reading batches of fixed width binary type (e.g. BYTE[7]).
+ */
+ public void nextBatchFixedWidthBinary(FieldVector fieldVector, int typeWidth) {
+ int rowsReadSoFar = 0;
+ while (rowsReadSoFar < batchSize && hasNext()) {
+ advance();
+ int rowsInThisBatch = batchedPageIterator.nextBatchFixedWidthBinary(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar, typeWidth);
+ rowsReadSoFar += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
+ }
+ }
+ /**
+ * Method for reading batches of booleans.
+ */
+ public void nextBatchBoolean(FieldVector fieldVector) {
+ int rowsReadSoFar = 0;
+ while (rowsReadSoFar < batchSize && hasNext()) {
+ advance();
+ int rowsInThisBatch = batchedPageIterator.nextBatchBoolean(fieldVector, batchSize - rowsReadSoFar,
+ rowsReadSoFar);
+ rowsReadSoFar += rowsInThisBatch;
+ this.valuesRead += rowsInThisBatch;
+ fieldVector.setValueCount(rowsReadSoFar);
+ }
+ }
private static Dictionary readDictionary(ColumnDescriptor desc, PageReader pageSource) {
DictionaryPage dictionaryPage = pageSource.readDictionaryPage();
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedPageIterator.java b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedPageIterator.java
index e192e76..fac9937 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedPageIterator.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedPageIterator.java
@@ -20,10 +20,7 @@
package org.apache.iceberg.parquet;
import com.google.common.base.Preconditions;
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-
+import io.netty.buffer.ArrowBuf;
import org.apache.arrow.vector.*;
import org.apache.parquet.CorruptDeltaByteArrays;
import org.apache.parquet.bytes.ByteBufferInputStream;
@@ -39,18 +36,14 @@ import org.apache.parquet.column.values.RequiresPreviousReader;
import org.apache.parquet.column.values.ValuesReader;
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
import org.apache.parquet.io.ParquetDecodingException;
-import org.apache.parquet.io.api.Binary;
-import org.apache.parquet.schema.DecimalMetadata;
-import org.apache.parquet.schema.PrimitiveType;
-import org.apache.spark.sql.types.VarcharType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.spark.sql.types.Decimal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
import static java.lang.String.format;
-import static org.apache.parquet.column.ValuesType.DEFINITION_LEVEL;
-import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
-import static org.apache.parquet.column.ValuesType.VALUES;
+import static org.apache.parquet.column.ValuesType.*;
public class BatchedPageIterator {
private static final Logger LOG = LoggerFactory.getLogger(BatchedPageIterator.class);
@@ -79,7 +72,8 @@ public class BatchedPageIterator {
private Encoding valueEncoding = null;
private IntIterator definitionLevels = null;
private IntIterator repetitionLevels = null;
- private ValuesReader values = null;
+ private BytesReader bytesReader = null;
+ private ValuesReader valuesReader = null;
public void setPage(DataPage page) {
@@ -112,7 +106,7 @@ public class BatchedPageIterator {
this.triplesRead = 0;
this.definitionLevels = null;
this.repetitionLevels = null;
- this.values = null;
+ this.bytesReader = null;
this.hasNext = false;
}
@@ -124,453 +118,358 @@ public class BatchedPageIterator {
return hasNext;
}
- public int nextBatch(FieldVector vector, int numValsToRead) {
- final int valsToRead = Math.min(numValsToRead, triplesCount - triplesRead);
+ /**
+ * Method for reading a batch of non-decimal numeric data types (INT32, INT64, FLOAT, DOUBLE, DATE, TIMESTAMP)
+ * This method reads batches of bytes from Parquet and writes them into the data buffer underneath the Arrow
+ * vector. It appropriately sets the validity buffer in the Arrow vector.
+ */
+ public int nextBatchNumericNonDecimal(final FieldVector vector, final int expectedBatchSize, final int numValsInVector,
+ final int typeWidth) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
int valsRead = 0;
- int ordinal = vector.getValueCount();
- int numVals = 0;
- int numNulls = 0;
- while (valsRead < valsToRead) {
- numVals = 0;
- while (valsRead < valsToRead && definitionLevels.nextInt() == desc.getMaxDefinitionLevel()) {
- numVals++;
+ int numNonNulls = 0;
+ int startWriteValIdx = numValsInVector;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ ArrowBuf validityBuffer = vector.getValidityBuffer();
+ ArrowBuf dataBuffer = vector.getDataBuffer();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ BitVectorHelper.setValidityBitToOne(validityBuffer, ordinal);
+ numNonNulls++;
valsRead++;
+ ordinal++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+ if (numNonNulls > 0) {
+ ByteBuffer buffer = bytesReader.getBuffer(numNonNulls * typeWidth);
+ dataBuffer.setBytes(startWriteValIdx * typeWidth, buffer);
+ startWriteValIdx += numNonNulls;
}
- PrimitiveType primitive = desc.getPrimitiveType();
-
- if (primitive.getOriginalType() != null) {
- switch (desc.getPrimitiveType().getOriginalType()) {
- case ENUM:
- case JSON:
- case UTF8:
- //valsRead = nextBatchedString(vector);
- // Read next valsToRead values and put them in vector
- for (int i = 0; i < numVals; i++) {
- try {
- ((VarCharVector) vector).setSafe(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT_8:
- case INT_16:
- case INT_32:
- for (int i = 0; i < numVals; i++) {
- try {
- ((IntVector) vector).setSafe(ordinal, values.readInteger());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
-
- break;
- case DATE:
- for (int i = 0; i < numVals; i++) {
- try {
- ((DateDayVector) vector).setSafe(ordinal, values.readInteger());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT_64:
- for (int i = 0; i < numVals; i++) {
- try {
- ((BigIntVector) vector).setSafe(ordinal, values.readLong());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case TIMESTAMP_MICROS:
- for (int i = 0; i < numVals; i++) {
- try {
- ((TimeStampMicroTZVector) vector).setSafe(ordinal, values.readLong());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case TIMESTAMP_MILLIS:
- for (int i = 0; i < numVals; i++) {
- try {
- ((BigIntVector) vector).setSafe(ordinal, 1000 * values.readLong());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case DECIMAL:
- DecimalMetadata decimal = primitive.getDecimalMetadata();
- switch (primitive.getPrimitiveTypeName()) {
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- for (int i = 0; i < numVals; i++) {
- try {
- /*Binary binaryValue = values.readBytes();
-
- Decimal decimalValue = Decimal.fromDecimal(new BigDecimal(new BigInteger(binaryValue.getBytes()), decimal.getScale()));
- ((DecimalVector) vector).setSafe(ordinal, decimalValue.toJavaBigDecimal());*/
- ((DecimalVector) vector).setBigEndian(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
-
- break;
- case INT64:
- for (int i = 0; i < numVals; i++) {
- try {
- /*long decimalLongValue = values.readLong();
- Decimal decimalValue = Decimal.apply(decimalLongValue, precision, scale);
-
- ((DecimalVector) vector).setSafe(ordinal, decimalValue.toJavaBigDecimal()); anjali??*/
- ((DecimalVector) vector).setBigEndian(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT32:
- for (int i = 0; i < numVals; i++) {
- try {
- /* int decimalIntValue = values.readInteger();
- Decimal decimalValue = Decimal.apply(decimalIntValue, precision, scale);
-
- ((DecimalVector) vector).setSafe(ordinal, decimalValue.toJavaBigDecimal()); anjali??*/
- ((DecimalVector) vector).setBigEndian(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
-
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
- }
- break;
- case BSON:
- for (int i = 0; i < numVals; i++) {
- try {
- ((VarCharVector) vector).setSafe(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported logical type: " + primitive.getOriginalType());
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
+ BitVectorHelper.setValidityBit(validityBuffer, ordinal, 0);
+ valsRead++;
+ startWriteValIdx++;
+ ordinal++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
}
- } else {
- switch (primitive.getPrimitiveTypeName()) {
- case FIXED_LEN_BYTE_ARRAY:
- case BINARY:
- for (int i = 0; i < numVals; i++) {
- try {
- ((VarCharVector) vector).setSafe(ordinal, values.readBytes().getBytesUnsafe());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT32:
- for (int i = 0; i < numVals; i++) {
- try {
- ((IntVector) vector).setSafe(ordinal, values.readInteger());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case FLOAT:
- for (int i = 0; i < numVals; i++) {
- try {
- ((Float4Vector) vector).setSafe(ordinal, values.readFloat());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- // if (expected != null && expected.typeId() == org.apache.iceberg.types.Type.TypeID.DOUBLE) {
- // return new ParquetValueReaders.FloatAsDoubleReader(desc);
- // } else {
- // return new ParquetValueReaders.UnboxedReader<>(desc);
- // }
- case BOOLEAN:
- for (int i = 0; i < numVals; i++) {
- try {
- ((BitVector) vector).setSafe(ordinal, values.readBoolean() ? 1 : 0);
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT64:
- for (int i = 0; i < numVals; i++) {
- try {
- ((BigIntVector) vector).setSafe(ordinal, values.readLong());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case DOUBLE:
- for (int i = 0; i < numVals; i++) {
- try {
- ((Float8Vector) vector).setSafe(ordinal, values.readDouble());
- ordinal++;
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- default:
- throw new UnsupportedOperationException("Unsupported type: " + primitive);
+ }
+ }
+ triplesRead += valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
+ }
+
+ /**
+ * Method for reading a batch of decimals backed by INT32 and INT64 parquet data types.
+ * Arrow stores all decimals in 16 bytes. This method provides the necessary padding to the decimals read.
+ */
+ public int nextBatchIntLongBackedDecimal(final FieldVector vector, final int expectedBatchSize, final int numValsInVector,
+ final int typeWidth) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
+ int valsRead = 0;
+ int numNonNulls = 0;
+ int startWriteValIdx = numValsInVector;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ ArrowBuf validityBuffer = vector.getValidityBuffer();
+ ArrowBuf dataBuffer = vector.getDataBuffer();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ BitVectorHelper.setValidityBitToOne(validityBuffer, ordinal);
+ numNonNulls++;
+ valsRead++;
+ ordinal++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
}
}
+ for (int i = 0; i < numNonNulls; i++) {
+ try {
+ byte[] byteArray = new byte[DecimalVector.TYPE_WIDTH];
+ bytesReader.getBuffer(typeWidth).get(byteArray, 0, typeWidth);
+ dataBuffer.setBytes(startWriteValIdx * DecimalVector.TYPE_WIDTH, byteArray);
+ startWriteValIdx++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
+ BitVectorHelper.setValidityBit(validityBuffer, ordinal, 0);
+ valsRead++;
+ startWriteValIdx++;
+ ordinal++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+ }
+ triplesRead += valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
+ }
+
+ /**
+ * Method for reading a batch of decimals backed by fixed length byte array parquet data type.
+ * Arrow stores all decimals in 16 bytes. This method provides the necessary padding to the decimals read.
+ * Moreover, Arrow interprets the decimals in Arrow buffer as little endian. Parquet stores fixed length decimals
+ * as big endian. So, this method uses {@link DecimalVector#setBigEndian(int, byte[])} method so that the data in
+ * Arrow vector is indeed little endian.
+ */
+ public int nextBatchFixedLengthDecimal(final FieldVector vector, final int expectedBatchSize, final int numValsInVector,
+ final int typeWidth) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
+ int valsRead = 0;
+ int numNonNulls = 0;
+ int numNulls = 0;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
numNulls = 0;
- // Definition level found a null, read contiguous nulls
- while (valsRead < valsToRead && definitionLevels.nextInt() < desc.getMaxDefinitionLevel()) {
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ numNonNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+
+ for (int i = 0; i < numNonNulls; i++) {
+ try {
+ byte[] byteArray = new byte[DecimalVector.TYPE_WIDTH];
+ //bytesReader.getBuffer(typeWidth).get(byteArray, 0, typeWidth);
+ bytesReader.getBuffer(typeWidth).get(byteArray, DecimalVector.TYPE_WIDTH - typeWidth, typeWidth);
+ ((DecimalVector) vector).setBigEndian(ordinal, byteArray);
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+
+
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
numNulls++;
valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
}
+ if (numNulls > 0) {
+ for (int i = 0; i < numNulls; i++) {
+ try {
+ ((DecimalVector) vector).setNull(ordinal);
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+ }
+ }
+ triplesRead += valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
+ }
+ /**
+ * Method for reading a batch of variable width data type (ENUM, JSON, UTF8, BSON).
+ */
+ public int nextBatchVarWidthType(final FieldVector vector, final int expectedBatchSize, final int numValsInVector) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
+ int valsRead = 0;
+ int numNonNulls = 0;
+ int numNulls = 0;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
+ numNulls = 0;
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ numNonNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
- if (primitive.getOriginalType() != null) {
- switch (desc.getPrimitiveType().getOriginalType()) {
- case ENUM:
- case JSON:
- case UTF8:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((VarCharVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT_8:
- case INT_16:
- case INT_32:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((IntVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case DATE:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((DateDayVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT_64:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((BigIntVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case TIMESTAMP_MICROS:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((TimeStampMicroTZVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case TIMESTAMP_MILLIS:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((BigIntVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case DECIMAL:
- DecimalMetadata decimal = primitive.getDecimalMetadata();
- switch (primitive.getPrimitiveTypeName()) {
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((DecimalVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT64:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((DecimalVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT32:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((DecimalVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
- }
- break;
- case BSON:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((VarCharVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported logical type: " + primitive.getOriginalType());
+ for (int i = 0; i < numNonNulls; i++) {
+ try {
+ ((BaseVariableWidthVector) vector).setSafe(ordinal, valuesReader.readBytes().getBytesUnsafe());
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
}
- } else {
- switch (primitive.getPrimitiveTypeName()) {
- case FIXED_LEN_BYTE_ARRAY:
- case BINARY:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((VarCharVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT32:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((IntVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case FLOAT:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((Float4Vector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- // if (expected != null && expected.typeId() == org.apache.iceberg.types.Type.TypeID.DOUBLE) {
- // return new ParquetValueReaders.FloatAsDoubleReader(desc);
- // } else {
- // return new ParquetValueReaders.UnboxedReader<>(desc);
- // }
- case BOOLEAN:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((BitVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case INT64:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((BigIntVector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- case DOUBLE:
- for (int i = 0; i < numNulls; i++) {
- try {
- ((Float8Vector) vector).setNull(ordinal);
- ordinal++;
-
- } catch (RuntimeException e) {
- throw handleRuntimeException(e);
- }
- }
- break;
- default:
- throw new UnsupportedOperationException("Unsupported type: " + primitive);
+ }
+
+
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
+ numNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+ if (numNulls > 0) {
+ for (int i = 0; i < numNulls; i++) {
+ try {
+ ((BaseVariableWidthVector) vector).setNull(ordinal);
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
}
}
}
- vector.setValueCount(vector.getValueCount() + valsRead);
triplesRead += valsRead;
- advance();
- return valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
}
+ /**
+ * Method for reading batches of fixed width binary type (e.g. BYTE[7]).
+ * Spark does not support fixed width binary data type. To work around this limitation, the data is read as
+ * fixed width binary from parquet and stored in a {@link VarBinaryVector} in Arrow.
+ */
+ public int nextBatchFixedWidthBinary(final FieldVector vector, final int expectedBatchSize, final int numValsInVector,
+ final int typeWidth) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
+ int valsRead = 0;
+ int numNonNulls = 0;
+ int numNulls = 0;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
+ numNulls = 0;
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ numNonNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+
+ for (int i = 0; i < numNonNulls; i++) {
+ try {
+ byte[] byteArray = new byte[typeWidth];
+ bytesReader.getBuffer(typeWidth).get(byteArray);
+ ((VarBinaryVector) vector).setSafe(ordinal, byteArray);
+ ordinal++;
+ } catch (RuntimeException e) {
+ System.out.println("Ordinal: " + ordinal);
+ throw handleRuntimeException(e);
+ }
+ }
+
+
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
+ numNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+ if (numNulls > 0) {
+ for (int i = 0; i < numNulls; i++) {
+ try {
+ ((VarBinaryVector) vector).setNull(ordinal);
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+ }
+ }
+ triplesRead += valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
+ }
+
+ /**
+ * Method for reading batches of booleans.
+ */
+ public int nextBatchBoolean(final FieldVector vector, final int expectedBatchSize, final int numValsInVector) {
+ final int actualBatchSize = Math.min(expectedBatchSize, triplesCount - triplesRead);
+ if (actualBatchSize <= 0) {
+ return 0;
+ }
+ int ordinal = numValsInVector;
+ int valsRead = 0;
+ int numNonNulls = 0;
+ int numNulls = 0;
+ int maxDefLevel = desc.getMaxDefinitionLevel();
+ int defLevel = definitionLevels.nextInt();
+ while (valsRead < actualBatchSize) {
+ numNonNulls = 0;
+ numNulls = 0;
+ while (valsRead < actualBatchSize && defLevel == maxDefLevel) {
+ numNonNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+
+ for (int i = 0; i < numNonNulls; i++) {
+ try {
+ ((BitVector) vector).setSafe(ordinal, ((valuesReader.readBoolean() == false) ? 0 : 1));
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+
+
+ while (valsRead < actualBatchSize && defLevel < maxDefLevel) {
+ numNulls++;
+ valsRead++;
+ if (valsRead < actualBatchSize) {
+ defLevel = definitionLevels.nextInt();
+ }
+ }
+ if (numNulls > 0) {
+ for (int i = 0; i < numNulls; i++) {
+ try {
+ ((BitVector) vector).setNull(ordinal);
+ ordinal++;
+ } catch (RuntimeException e) {
+ throw handleRuntimeException(e);
+ }
+ }
+ }
+ }
+ triplesRead += valsRead;
+ this.hasNext = triplesRead < triplesCount;
+ return actualBatchSize;
+ }
private void advance() {
if (triplesRead < triplesCount) {
@@ -601,7 +500,7 @@ public class BatchedPageIterator {
}
private void initDataReader(Encoding dataEncoding, ByteBufferInputStream in, int valueCount) {
- ValuesReader previousReader = values;
+ ValuesReader previousReader = bytesReader;
this.valueEncoding = dataEncoding;
@@ -609,31 +508,34 @@ public class BatchedPageIterator {
// For dictionary columns, this class could rely on wrappers to correctly handle dictionaries
// This isn't currently possible because RLE must be read by getDictionaryBasedValuesReader
if (dataEncoding.usesDictionary()) {
- if (dict == null) {
+ /*if (dict == null) {
throw new ParquetDecodingException(
"could not read page in col " + desc + " as the dictionary was missing for encoding " + dataEncoding);
}
- this.values = dataEncoding.getDictionaryBasedValuesReader(desc, VALUES, dict);
+ this.bytesReader = dataEncoding.getDictionaryBasedValuesReader(desc, VALUES, dict); */
} else {
- this.values = dataEncoding.getValuesReader(desc, VALUES);
- }
-
-// if (dataEncoding.usesDictionary() && converter.hasDictionarySupport()) {
-// bindToDictionary(dictionary);
-// } else {
-// bind(path.getType());
-// }
-
- try {
- values.initFromPage(valueCount, in);
- } catch (IOException e) {
- throw new ParquetDecodingException("could not read page in col " + desc, e);
+ if (ParquetUtil.isVarWidthType(desc) || ParquetUtil.isBooleanType(desc)) {
+ this.valuesReader = dataEncoding.getValuesReader(desc, VALUES);
+ try {
+ valuesReader.initFromPage(valueCount, in);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("could not read page in col " + desc, e);
+ }
+ } else {
+ this.bytesReader = new BytesReader();
+ bytesReader.initFromPage(valueCount, in);
+ }
}
+ // if (dataEncoding.usesDictionary() && converter.hasDictionarySupport()) {
+ // bindToDictionary(dictionary);
+ // } else {
+ // bind(path.getType());
+ // }
if (CorruptDeltaByteArrays.requiresSequentialReads(writerVersion, dataEncoding) &&
previousReader != null && previousReader instanceof RequiresPreviousReader) {
// previous reader can only be set if reading sequentially
- ((RequiresPreviousReader) values).setPreviousReader(previousReader);
+ ((RequiresPreviousReader) bytesReader).setPreviousReader(previousReader);
}
}
@@ -726,4 +628,5 @@ public class BatchedPageIterator {
return 0;
}
}
+
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedReader.java b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedReader.java
index 635ce90..335636e 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/BatchedReader.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/BatchedReader.java
@@ -1,4 +1,7 @@
package org.apache.iceberg.parquet;
+/**
+ * Marker interface for vectorized Iceberg readers.
+ */
public interface BatchedReader {
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/BytesReader.java b/parquet/src/main/java/org/apache/iceberg/parquet/BytesReader.java
new file mode 100644
index 0000000..5df8a3c
--- /dev/null
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/BytesReader.java
@@ -0,0 +1,56 @@
+/*
+ * 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 KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.iceberg.parquet;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.ParquetDecodingException;
+
+/**
+ * Implements a {@link ValuesReader} specifically to read given number of bytes from the underlying
+ * {@link ByteBufferInputStream}.
+ */
+public class BytesReader extends ValuesReader {
+ private ByteBufferInputStream in = null;
+
+ public BytesReader() {
+ }
+
+ @Override
+ public void initFromPage(int valueCount, ByteBufferInputStream in) {
+ this.in = in;
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException();
+ }
+
+ public ByteBuffer getBuffer(int length) {
+ try {
+ return in.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+}
+
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ColumnarBatchReader.java b/parquet/src/main/java/org/apache/iceberg/parquet/ColumnarBatchReader.java
index 50a57b7..b3ecbee 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ColumnarBatchReader.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ColumnarBatchReader.java
@@ -4,6 +4,7 @@ import java.lang.reflect.Array;
import java.util.List;
import org.apache.arrow.vector.FieldVector;
import org.apache.iceberg.types.Types;
+import org.apache.parquet.Preconditions;
import org.apache.parquet.column.page.PageReadStore;
import org.apache.parquet.schema.Type;
import org.apache.spark.sql.vectorized.ArrowColumnVector;
@@ -13,16 +14,11 @@ import org.slf4j.LoggerFactory;
public class ColumnarBatchReader implements BatchedReader{
private static final Logger LOG = LoggerFactory.getLogger(ColumnarBatchReader.class);
- private final int numFields;
- private final Types.StructType iceExpectedFields;
private final VectorReader[] readers;
public ColumnarBatchReader(List<Type> types,
Types.StructType icebergExpectedFields,
List<BatchedReader> readers) {
-
- this.numFields = readers.size();
- this.iceExpectedFields = icebergExpectedFields;
this.readers = (VectorReader[]) Array.newInstance(
VectorReader.class, readers.size());
int i = 0;
@@ -49,6 +45,9 @@ public class ColumnarBatchReader implements BatchedReader{
FieldVector vec = readers[i].read();
arrowVectorArr[i] = new ArrowColumnVector(vec);
+ Preconditions.checkState(i > 0 && numRows == vec.getValueCount(),
+ "Different number of values returned by readers for columns: " +
+ readers[i-1] + " and " + readers[i]);
numRows = vec.getValueCount();
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
index 58e093d..7267259 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetUtil.java
@@ -41,6 +41,7 @@ import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.column.statistics.Statistics;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
@@ -48,12 +49,18 @@ import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
import org.apache.parquet.hadoop.metadata.ColumnPath;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.spark.sql.catalog.Column;
import static org.apache.iceberg.parquet.ParquetConversions.fromParquetPrimitive;
import static org.apache.iceberg.util.BinaryUtil.truncateBinaryMax;
import static org.apache.iceberg.util.BinaryUtil.truncateBinaryMin;
import static org.apache.iceberg.util.UnicodeUtil.truncateStringMax;
import static org.apache.iceberg.util.UnicodeUtil.truncateStringMin;
+import static org.apache.parquet.schema.OriginalType.*;
+import static org.apache.parquet.schema.OriginalType.BSON;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.*;
public class ParquetUtil {
// not meant to be instantiated
@@ -228,4 +235,53 @@ public class ParquetUtil {
}
return bufferMap;
}
+
+ public static boolean isFixedLengthDecimal(ColumnDescriptor desc) {
+ PrimitiveType primitive = desc.getPrimitiveType();
+ return (primitive.getOriginalType() != null &&
+ primitive.getOriginalType() == DECIMAL &&
+ (primitive.getPrimitiveTypeName() == FIXED_LEN_BYTE_ARRAY ||
+ primitive.getPrimitiveTypeName() == BINARY));
+ }
+
+ public static boolean isIntLongBackedDecimal(ColumnDescriptor desc) {
+ PrimitiveType primitive = desc.getPrimitiveType();
+ return (primitive.getOriginalType() != null &&
+ primitive.getOriginalType() == DECIMAL &&
+ (primitive.getPrimitiveTypeName() == INT64 ||
+ primitive.getPrimitiveTypeName() == INT32));
+ }
+
+ public static boolean isVarWidthType(ColumnDescriptor desc) {
+ PrimitiveType primitive = desc.getPrimitiveType();
+ OriginalType originalType = primitive.getOriginalType();
+ if (originalType != null &&
+ originalType != DECIMAL &&
+ (originalType == ENUM ||
+ originalType == OriginalType.JSON ||
+ originalType == UTF8 ||
+ originalType == BSON)) {
+ return true;
+ }
+ if (originalType == null && primitive.getPrimitiveTypeName() == BINARY) {
+ return true;
+ }
+ return false;
+ }
+
+ public static boolean isBooleanType(ColumnDescriptor desc) {
+ PrimitiveType primitive = desc.getPrimitiveType();
+ OriginalType originalType = primitive.getOriginalType();
+ return (originalType == null && primitive.getPrimitiveTypeName() == BOOLEAN);
+ }
+
+ public static boolean isFixedWidthBinary(ColumnDescriptor desc) {
+ PrimitiveType primitive = desc.getPrimitiveType();
+ OriginalType originalType = primitive.getOriginalType();
+ if (originalType == null && primitive.getPrimitiveTypeName() == FIXED_LEN_BYTE_ARRAY) {
+ return true;
+ }
+ return false;
+ }
+
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java
index bf5df89..dce8861 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java
@@ -22,6 +22,11 @@ package org.apache.iceberg.parquet;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.Type;
+
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.math.BigInteger;
@@ -29,15 +34,6 @@ import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import org.apache.arrow.vector.FieldVector;
-import org.apache.iceberg.types.Types;
-import org.apache.parquet.column.ColumnDescriptor;
-import org.apache.parquet.column.page.PageReadStore;
-import org.apache.parquet.io.api.Binary;
-import org.apache.parquet.schema.Type;
-import org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader;
-import org.apache.spark.sql.vectorized.ArrowColumnVector;
-import org.apache.spark.sql.vectorized.ColumnarBatch;
import static java.util.Collections.emptyIterator;
@@ -581,83 +577,6 @@ public class ParquetValueReaders {
}
}
-
- /*public static class ColumnarBatchReader {
-
- private final int numFields;
- private final Types.StructType iceExpectedFields;
- private final VectorReader[] readers;
- private ColumnarBatch columnarBatch;
-
- @SuppressWarnings("unchecked")
- public ColumnarBatchReader(List<Type> types,
- Types.StructType icebergExpectedFields,
- List<ParquetValueReader<FieldVector>> readers) {
-
- this.numFields = readers.size();
- this.iceExpectedFields = icebergExpectedFields;
- this.readers = (ParquetValueReader<FieldVector>[]) Array.newInstance(
- ParquetValueReader.class, readers.size());
- this.columns = (TripleIterator<?>[]) Array.newInstance(TripleIterator.class, readers.size());
-
-
- ImmutableList.Builder<TripleIterator<?>> columnsBuilder = ImmutableList.builder();
- for (int i = 0; i < readers.size(); i += 1) {
- ParquetValueReader<FieldVector> reader = readers.get(i);
- this.readers[i] = readers.get(i);
- this.columns[i] = reader.column();
- columnsBuilder.addAll(reader.columns());
- }
-
- this.children = columnsBuilder.build();
- if (children.size() > 0) {
- this.column = children.get(0);
- } else {
- this.column = NullReader.NULL_COLUMN;
- }
-
- }
-
- @Override
- public final void setPageSource(PageReadStore pageStore) {
- for (int i = 0; i < readers.length; i += 1) {
- readers[i].setPageSource(pageStore);
- }
- }
-
- @Override
- public final TripleIterator<?> column() {
- return column;
- }
-
- @Override
- public List<TripleIterator<?>> columns() {
- return children;
- }
-
-
- @Override
- public final ColumnarBatch read(ColumnarBatch ignore) {
-
- ArrowColumnVector[] arrowVectorArr = (ArrowColumnVector[])Array.newInstance(ArrowColumnVector.class,
- readers.length);
-
- int numRows=0;
- for (int i = 0; i < readers.length; i += 1) {
-
- FieldVector vec = readers[i].read(null);
- arrowVectorArr[i] = new ArrowColumnVector(vec);
- numRows = vec.getValueCount();
- }
-
- ColumnarBatch batch = new ColumnarBatch(arrowVectorArr);
- batch.setNumRows(numRows);
-
- return batch;
- }
-
- }
-*/
public abstract static class StructReader<T, I> implements ParquetValueReader<T> {
private interface Setter<R> {
void set(R record, int pos, Object reuse);
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/VectorReader.java b/parquet/src/main/java/org/apache/iceberg/parquet/VectorReader.java
index c879867..0915e1d 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/VectorReader.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/VectorReader.java
@@ -20,21 +20,17 @@
package org.apache.iceberg.parquet;
import org.apache.arrow.memory.BufferAllocator;
-import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.*;
import org.apache.iceberg.arrow.ArrowSchemaUtil;
import org.apache.iceberg.types.Types;
import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.schema.DecimalMetadata;
+import org.apache.parquet.schema.PrimitiveType;
/***
- * Parquet Value Reader implementations for Vectorization.
- * Contains type-wise readers to read parquet data as vectors.
- * - Returns Arrow's Field Vector for each type.
- * - Null values are explicitly handled.
- * - Type serialization is done based on types in Arrow.
- * - Creates One Vector per RowGroup. So a Batch would have as many rows as there are in the underlying RowGroup.
- * - Mapping of Iceberg type to Arrow type is done in ArrowSchemaUtil.convert()
- * - Iceberg to Arrow Type mapping :
+ * Creates and allocates space for Arrow field vectors based on Iceberg data type mapped to Arrow type.
+ * Iceberg to Arrow Type mapping :
* icebergType : LONG - Field Vector Type : org.apache.arrow.vector.BigIntVector
* icebergType : STRING - Field Vector Type : org.apache.arrow.vector.VarCharVector
* icebergType : BOOLEAN - Field Vector Type : org.apache.arrow.vector.BitVector
@@ -48,39 +44,154 @@ import org.apache.parquet.column.page.PageReadStore;
* icebergField : DECIMAL - Field Vector Type : org.apache.arrow.vector.DecimalVector
*/
public class VectorReader implements BatchedReader {
- public static final int DEFAULT_NUM_ROWS_IN_BATCH = 10000;
- // private static final Logger LOG = LoggerFactory.getLogger(VectorReader.class);
+ public static final int DEFAULT_NUM_ROWS_IN_BATCH = 10000;
+ public static final int UNKNOWN_WIDTH = -1;
- private final FieldVector vec;
- private int rowsInBatch = DEFAULT_NUM_ROWS_IN_BATCH;
- private final BatchedColumnIterator batchedColumnIterator;
+ private final ColumnDescriptor columnDescriptor;
+ private FieldVector vec;
+ private final int rowsInBatch;
+ private final BatchedColumnIterator batchedColumnIterator;
+ private final int typeWidth;
+ private final boolean isFixedLengthDecimal;
+ private final boolean isVarWidthType;
+ private final boolean isFixedWidthBinary;
+ private final boolean isBooleanType;
+ private final boolean isPaddedDecimal;
- public VectorReader(ColumnDescriptor desc,
- Types.NestedField icebergField,
- BufferAllocator rootAlloc,
- int rowsInBatch) {
- this.rowsInBatch = (rowsInBatch == 0) ? DEFAULT_NUM_ROWS_IN_BATCH : rowsInBatch;
- this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
- this.batchedColumnIterator = new BatchedColumnIterator(desc, "", rowsInBatch); // what should be the writer version?
- // anjali
- // LOG.info("=> [VectorReader] rowsInBatch = " + this.rowsInBatch);
- }
+ public VectorReader(ColumnDescriptor desc,
+ Types.NestedField icebergField,
+ BufferAllocator rootAlloc,
+ int rowsInBatch) {
+ this.rowsInBatch = (rowsInBatch == 0) ? DEFAULT_NUM_ROWS_IN_BATCH : rowsInBatch;
+ this.columnDescriptor = desc;
+ this.typeWidth = allocateFieldVector(rootAlloc, icebergField, desc);
- public FieldVector read() {
- vec.reset();
- if (batchedColumnIterator.hasNext()) {
- batchedColumnIterator.nextBatch(vec);
+ isFixedLengthDecimal = ParquetUtil.isFixedLengthDecimal(desc);
+ isVarWidthType = ParquetUtil.isVarWidthType(desc);
+ isFixedWidthBinary = ParquetUtil.isFixedWidthBinary(desc);
+ isBooleanType = ParquetUtil.isBooleanType(desc);
+ isPaddedDecimal = ParquetUtil.isIntLongBackedDecimal(desc);
+
+ this.batchedColumnIterator = new BatchedColumnIterator(desc, "", rowsInBatch);
+ }
+
+ private int allocateFieldVector(BufferAllocator rootAlloc, Types.NestedField icebergField, ColumnDescriptor desc) {
+
+ PrimitiveType primitive = desc.getPrimitiveType();
+ if (primitive.getOriginalType() != null) {
+ switch (desc.getPrimitiveType().getOriginalType()) {
+ case ENUM:
+ case JSON:
+ case UTF8:
+ case BSON:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ return UNKNOWN_WIDTH;
+ case INT_8:
+ case INT_16:
+ case INT_32:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((IntVector) vec).allocateNew(rowsInBatch * IntVector.TYPE_WIDTH);
+ return IntVector.TYPE_WIDTH;
+ case DATE:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((DateDayVector) vec).allocateNew(rowsInBatch * IntVector.TYPE_WIDTH);
+ return IntVector.TYPE_WIDTH;
+ case INT_64:
+ case TIMESTAMP_MILLIS:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((BigIntVector) vec).allocateNew(rowsInBatch * BigIntVector.TYPE_WIDTH);
+ return BigIntVector.TYPE_WIDTH;
+ case TIMESTAMP_MICROS:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((TimeStampMicroTZVector) vec).allocateNew(rowsInBatch * BigIntVector.TYPE_WIDTH);
+ return BigIntVector.TYPE_WIDTH;
+ case DECIMAL:
+ DecimalMetadata decimal = primitive.getDecimalMetadata();
+ this.vec = new DecimalVector(icebergField.name(), rootAlloc, decimal.getPrecision(), decimal.getScale());
+ ((DecimalVector) vec).allocateNew(rowsInBatch * DecimalVector.TYPE_WIDTH);
+ switch (primitive.getPrimitiveTypeName()) {
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return primitive.getTypeLength();
+ case INT64:
+ return BigIntVector.TYPE_WIDTH;
+ case INT32:
+ return IntVector.TYPE_WIDTH;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported base type for decimal: " + primitive.getPrimitiveTypeName());
+ }
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported logical type: " + primitive.getOriginalType());
+ }
+ } else {
+ switch (primitive.getPrimitiveTypeName()) {
+ case FIXED_LEN_BYTE_ARRAY:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ int len = ((Types.FixedType) icebergField.type()).length();
+ vec.setInitialCapacity(rowsInBatch * len);
+ vec.allocateNew();
+ return len;
+ case BINARY:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ vec.setInitialCapacity(rowsInBatch);
+ vec.allocateNew();
+ return UNKNOWN_WIDTH;
+ case INT32:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((IntVector) vec).allocateNew(rowsInBatch * IntVector.TYPE_WIDTH);
+ return IntVector.TYPE_WIDTH;
+ case FLOAT:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((Float4Vector) vec).allocateNew(rowsInBatch * Float4Vector.TYPE_WIDTH);
+ return Float4Vector.TYPE_WIDTH;
+ case BOOLEAN:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((BitVector) vec).allocateNew(rowsInBatch);
+ return UNKNOWN_WIDTH;
+ case INT64:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((BigIntVector) vec).allocateNew(rowsInBatch * BigIntVector.TYPE_WIDTH);
+ return BigIntVector.TYPE_WIDTH;
+ case DOUBLE:
+ this.vec = ArrowSchemaUtil.convert(icebergField).createVector(rootAlloc);
+ ((Float8Vector) vec).allocateNew(rowsInBatch * Float8Vector.TYPE_WIDTH);
+ return Float8Vector.TYPE_WIDTH;
+ default:
+ throw new UnsupportedOperationException("Unsupported type: " + primitive);
+ }
+ }
}
- return vec;
- }
- public int getRowCount() {
- return vec.getValueCount();
- }
+ public FieldVector read() {
+ if (batchedColumnIterator.hasNext()) {
+ if (isFixedLengthDecimal) {
+ batchedColumnIterator.nextBatchFixedLengthDecimal(vec, typeWidth);
+ } else if (isVarWidthType) {
+ batchedColumnIterator.nextBatchVarWidthType(vec);
+ } else if (isFixedWidthBinary) {
+ vec.reset();
+ batchedColumnIterator.nextBatchFixedWidthBinary(vec, typeWidth);
+ } else if (isBooleanType) {
+ batchedColumnIterator.nextBatchBoolean(vec);
+ } else if (isPaddedDecimal) {
+ batchedColumnIterator.nextBatchIntLongBackedDecimal(vec, typeWidth);
+ } else {
+ batchedColumnIterator.nextBatchNumericNonDecimal(vec, typeWidth);
+ }
+ }
+ return vec;
+ }
+
+ public void setPageSource(PageReadStore source) {
+ batchedColumnIterator.setPageSource(source);
+ }
- public void setPageSource(PageReadStore source) {
- batchedColumnIterator.setPageSource(source);
- }
+ @Override
+ public String toString() {
+ return columnDescriptor.toString();
+ }
}
diff --git a/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java b/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java
index 1648ee7..db67521 100644
--- a/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java
+++ b/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java
@@ -43,6 +43,7 @@ import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Warmup;
+import static org.apache.iceberg.TableProperties.PARQUET_DICT_SIZE_BYTES;
import static org.apache.iceberg.TableProperties.WRITE_NEW_DATA_LOCATION;
@Fork(1)
@@ -95,6 +96,9 @@ public abstract class IcebergSourceBenchmark {
protected void setupSpark() {
spark = SparkSession.builder()
.config("spark.ui.enabled", false)
+ .config("parquet.enable.dictionary",false)
+ .config(PARQUET_DICT_SIZE_BYTES, "1")
+ .config("parquet.dictionary.page.size", "1")
.master("local")
.getOrCreate();
Configuration sparkHadoopConf = spark.sparkContext().hadoopConfiguration();
diff --git a/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceFlatDataBenchmark.java b/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceFlatDataBenchmark.java
index 9f9a070..ec0163a 100644
--- a/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceFlatDataBenchmark.java
+++ b/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceFlatDataBenchmark.java
@@ -19,18 +19,18 @@
package org.apache.iceberg.spark.source;
-import com.google.common.collect.Maps;
-import java.util.Map;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.iceberg.PartitionSpec;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.TableProperties;
-import org.apache.iceberg.hadoop.HadoopTables;
-import org.apache.iceberg.types.Types;
+ import com.google.common.collect.Maps;
+ import java.util.Map;
+ import org.apache.hadoop.conf.Configuration;
+ import org.apache.iceberg.PartitionSpec;
+ import org.apache.iceberg.Schema;
+ import org.apache.iceberg.Table;
+ import org.apache.iceberg.TableProperties;
+ import org.apache.iceberg.hadoop.HadoopTables;
+ import org.apache.iceberg.types.Types;
-import static org.apache.iceberg.types.Types.NestedField.optional;
-import static org.apache.iceberg.types.Types.NestedField.required;
+ import static org.apache.iceberg.types.Types.NestedField.optional;
+ import static org.apache.iceberg.types.Types.NestedField.required;
public abstract class IcebergSourceFlatDataBenchmark extends IcebergSourceBenchmark {
@@ -42,18 +42,19 @@ public abstract class IcebergSourceFlatDataBenchmark extends IcebergSourceBenchm
@Override
protected final Table initTable() {
Schema schema = new Schema(
- required(1, "longCol", Types.LongType.get()),
- required(2, "intCol", Types.IntegerType.get()),
- required(3, "floatCol", Types.FloatType.get()),
- optional(4, "doubleCol", Types.DoubleType.get()),
- optional(5, "decimalCol", Types.DecimalType.of(20, 5)),
- optional(6, "dateCol", Types.DateType.get()),
- optional(7, "timestampCol", Types.TimestampType.withZone()),
- optional(8, "stringCol", Types.StringType.get()));
+ required(1, "longCol", Types.LongType.get()),
+ required(2, "intCol", Types.IntegerType.get()),
+ required(3, "floatCol", Types.FloatType.get()),
+ optional(4, "doubleCol", Types.DoubleType.get()),
+ optional(5, "decimalCol", Types.DecimalType.of(20, 5)),
+ optional(6, "dateCol", Types.DateType.get()),
+ optional(7, "timestampCol", Types.TimestampType.withZone()),
+ optional(8, "stringCol", Types.StringType.get()));
PartitionSpec partitionSpec = PartitionSpec.unpartitioned();
HadoopTables tables = new HadoopTables(hadoopConf());
Map<String, String> properties = Maps.newHashMap();
properties.put(TableProperties.METADATA_COMPRESSION, "gzip");
+ properties.put(TableProperties.PARQUET_DICT_SIZE_BYTES, "1");
return tables.create(schema, partitionSpec, properties, newTableLocation());
}
}
diff --git a/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceFlatParquetDataReadBenchmark.java b/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceFlatParquetDataReadBenchmark.java
index 9e51081..01144ab 100644
--- a/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceFlatParquetDataReadBenchmark.java
+++ b/spark/src/jmh/java/org/apache/iceberg/spark/source/parquet/IcebergSourceFlatParquetDataReadBenchmark.java
@@ -50,7 +50,7 @@ import static org.apache.spark.sql.functions.expr;
public class IcebergSourceFlatParquetDataReadBenchmark extends IcebergSourceFlatDataBenchmark {
private static final int NUM_FILES = 10;
- private static final int NUM_ROWS = 1000000;
+ private static final int NUM_ROWS = 10000000;
@Setup
public void setupBenchmark() {
@@ -64,35 +64,49 @@ public class IcebergSourceFlatParquetDataReadBenchmark extends IcebergSourceFlat
cleanupFiles();
}
- @Benchmark
- @Threads(1)
- public void readIcebergVectorized100k() {
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
- withTableProperties(tableProperties, () -> {
- String tableLocation = table().location();
- Dataset<Row> df = spark().read().format("iceberg")
- .option("iceberg.read.numrecordsperbatch", "100000")
- .load(tableLocation);
- materialize(df);
- });
- }
+// @Benchmark
+// @Threads(1)
+// public void readIcebergVectorized100k() {
+// Map<String, String> tableProperties = Maps.newHashMap();
+// tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+// withTableProperties(tableProperties, () -> {
+// String tableLocation = table().location();
+// Dataset<Row> df = spark().read().format("iceberg")
+// .option("iceberg.read.numrecordsperbatch", "100000")
+// .load(tableLocation);
+// materialize(df);
+// });
+// }
+//
+// @Benchmark
+// @Threads(1)
+// public void readIcebergVectorized10k() {
+// Map<String, String> tableProperties = Maps.newHashMap();
+// tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+// withTableProperties(tableProperties, () -> {
+// String tableLocation = table().location();
+// Dataset<Row> df = spark().read().format("iceberg")
+// .option("iceberg.read.numrecordsperbatch", "10000")
+// .load(tableLocation);
+// materialize(df);
+// });
+// }
+
@Benchmark
@Threads(1)
- public void readIcebergVectorized10k() {
+ public void readIcebergVectorized100() {
Map<String, String> tableProperties = Maps.newHashMap();
tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
withTableProperties(tableProperties, () -> {
String tableLocation = table().location();
Dataset<Row> df = spark().read().format("iceberg")
- .option("iceberg.read.numrecordsperbatch", "10000")
- .load(tableLocation);
+ .option("iceberg.read.numrecordsperbatch", "100")
+ .load(tableLocation);
materialize(df);
});
}
-
@Benchmark
@Threads(1)
public void readIcebergVectorized5k() {
@@ -131,72 +145,72 @@ public class IcebergSourceFlatParquetDataReadBenchmark extends IcebergSourceFlat
});
}
- @Benchmark
- @Threads(1)
- public void readWithProjectionIcebergVectorized100k() {
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
- withTableProperties(tableProperties, () -> {
- String tableLocation = table().location();
- Dataset<Row> df = spark().read().format("iceberg")
- .option("iceberg.read.numrecordsperbatch", "100000")
- .load(tableLocation).select("longCol");
- materialize(df);
- });
- }
-
- @Benchmark
- @Threads(1)
- public void readWithProjectionIcebergVectorized10k() {
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
- withTableProperties(tableProperties, () -> {
- String tableLocation = table().location();
- Dataset<Row> df = spark().read().format("iceberg")
- .option("iceberg.read.numrecordsperbatch", "10000")
- .load(tableLocation).select("longCol");
- materialize(df);
- });
- }
-
-
- @Benchmark
- @Threads(1)
- public void readWithProjectionIcebergVectorized5k() {
- Map<String, String> tableProperties = Maps.newHashMap();
- tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
- withTableProperties(tableProperties, () -> {
- String tableLocation = table().location();
- Dataset<Row> df = spark().read().format("iceberg")
- .option("iceberg.read.numrecordsperbatch", "5000")
- .load(tableLocation).select("longCol");
- materialize(df);
- });
- }
-
- @Benchmark
- @Threads(1)
- public void readWithProjectionFileSourceVectorized() {
- Map<String, String> conf = Maps.newHashMap();
- conf.put(SQLConf.PARQUET_VECTORIZED_READER_ENABLED().key(), "true");
- conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
- withSQLConf(conf, () -> {
- Dataset<Row> df = spark().read().parquet(dataLocation()).select("longCol");
- materialize(df);
- });
- }
-
- @Benchmark
- @Threads(1)
- public void readWithProjectionFileSourceNonVectorized() {
- Map<String, String> conf = Maps.newHashMap();
- conf.put(SQLConf.PARQUET_VECTORIZED_READER_ENABLED().key(), "false");
- conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
- withSQLConf(conf, () -> {
- Dataset<Row> df = spark().read().parquet(dataLocation()).select("longCol");
- materialize(df);
- });
- }
+// @Benchmark
+// @Threads(1)
+// public void readWithProjectionIcebergVectorized100k() {
+// Map<String, String> tableProperties = Maps.newHashMap();
+// tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+// withTableProperties(tableProperties, () -> {
+// String tableLocation = table().location();
+// Dataset<Row> df = spark().read().format("iceberg")
+// .option("iceberg.read.numrecordsperbatch", "100000")
+// .load(tableLocation).select("longCol");
+// materialize(df);
+// });
+// }
+//
+// @Benchmark
+// @Threads(1)
+// public void readWithProjectionIcebergVectorized10k() {
+// Map<String, String> tableProperties = Maps.newHashMap();
+// tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+// withTableProperties(tableProperties, () -> {
+// String tableLocation = table().location();
+// Dataset<Row> df = spark().read().format("iceberg")
+// .option("iceberg.read.numrecordsperbatch", "10000")
+// .load(tableLocation).select("longCol");
+// materialize(df);
+// });
+// }
+
+
+// @Benchmark
+// @Threads(1)
+// public void readWithProjectionIcebergVectorized5k() {
+// Map<String, String> tableProperties = Maps.newHashMap();
+// tableProperties.put(SPLIT_OPEN_FILE_COST, Integer.toString(128 * 1024 * 1024));
+// withTableProperties(tableProperties, () -> {
+// String tableLocation = table().location();
+// Dataset<Row> df = spark().read().format("iceberg")
+// .option("iceberg.read.numrecordsperbatch", "5000")
+// .load(tableLocation).select("longCol");
+// materialize(df);
+// });
+// }
+//
+// @Benchmark
+// @Threads(1)
+// public void readWithProjectionFileSourceVectorized() {
+// Map<String, String> conf = Maps.newHashMap();
+// conf.put(SQLConf.PARQUET_VECTORIZED_READER_ENABLED().key(), "true");
+// conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+// withSQLConf(conf, () -> {
+// Dataset<Row> df = spark().read().parquet(dataLocation()).select("longCol");
+// materialize(df);
+// });
+// }
+
+// @Benchmark
+// @Threads(1)
+// public void readWithProjectionFileSourceNonVectorized() {
+// Map<String, String> conf = Maps.newHashMap();
+// conf.put(SQLConf.PARQUET_VECTORIZED_READER_ENABLED().key(), "false");
+// conf.put(SQLConf.FILES_OPEN_COST_IN_BYTES().key(), Integer.toString(128 * 1024 * 1024));
+// withSQLConf(conf, () -> {
+// Dataset<Row> df = spark().read().parquet(dataLocation()).select("longCol");
+// materialize(df);
+// });
+// }
@Benchmark
diff --git a/spark/src/main/java/org/apache/iceberg/spark/data/vector/VectorizedSparkParquetReaders.java b/spark/src/main/java/org/apache/iceberg/spark/data/vector/VectorizedSparkParquetReaders.java
index e1f8687..1696a6c 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/data/vector/VectorizedSparkParquetReaders.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/data/vector/VectorizedSparkParquetReaders.java
@@ -86,9 +86,7 @@ public class VectorizedSparkParquetReaders {
this.projectedIcebergSchema = projectedIcebergSchema;
this.arrowSchema = ArrowSchemaUtil.convert(projectedIcebergSchema);
this.recordsPerBatch = recordsPerBatch;
- // this.rootAllocator = new RootAllocator(Long.MAX_VALUE);
- this.rootAllocator = ArrowUtils.rootAllocator().newChildAllocator("VectorizedReadBuilder",
- 0, Long.MAX_VALUE);
+ this.rootAllocator = ArrowUtils.rootAllocator().newChildAllocator("VectorizedReadBuilder", 0, Long.MAX_VALUE);
LOG.info("=> [ReadBuilder] recordsPerBatch = {}", this.recordsPerBatch);
}
diff --git a/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTest.java b/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTest.java
index 86af9f4..31f6231 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTest.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTest.java
@@ -39,8 +39,32 @@ public abstract class AvroDataTest {
protected abstract void writeAndValidate(Schema schema) throws IOException;
protected static final StructType SUPPORTED_PRIMITIVES = StructType.of(
+
required(100, "id", LongType.get()),
- required(200, "id2", Types.DecimalType.of(20, 5)));
+// optional(101, "data", Types.StringType.get()),
+// required(102, "b", Types.BooleanType.get()),
+// optional(103, "i", Types.IntegerType.get()),
+// required(104, "l", LongType.get()),
+// optional(105, "f", Types.FloatType.get()),
+// required(106, "d", Types.DoubleType.get()),
+// optional(107, "date", Types.DateType.get()),
+// required(108, "ts", Types.TimestampType.withZone()),
+// required(110, "s", Types.StringType.get()),
+// //required(111, "uuid", Types.UUIDType.get()),
+// required(112, "fixed", Types.FixedType.ofLength(7)),
+// optional(113, "bytes", Types.BinaryType.get()),
+// required(114, "dec_9_0", Types.DecimalType.of(9, 0)),
+// required(115, "dec_11_2", Types.DecimalType.of(11, 2)),
+// required(116, "dec_38_10", Types.DecimalType.of(38, 10)),
+// required(117, "dec_38_0", Types.DecimalType.of(38, 0)));
+
+ required(113, "dec_28_0", Types.DecimalType.of(28, 0)),
+ required(114, "dec_9_0", Types.DecimalType.of(9, 0)),
+ required(115, "dec_11_2", Types.DecimalType.of(11, 2)),
+ required(116, "dec_38_10", Types.DecimalType.of(38, 10)),
+ required(117, "dec_38_0", Types.DecimalType.of(38, 0)),
+ required(118, "dec_9_5", Types.DecimalType.of(9, 5)),
+ required(119, "dec_20_5", Types.DecimalType.of(20, 5)));
@Rule
public TemporaryFolder temp = new TemporaryFolder();
diff --git a/spark/src/test/java/org/apache/iceberg/spark/data/RandomData.java b/spark/src/test/java/org/apache/iceberg/spark/data/RandomData.java
index 080c0d1..f254b43 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/data/RandomData.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/data/RandomData.java
@@ -56,7 +56,7 @@ public class RandomData {
List<Record> records = Lists.newArrayListWithExpectedSize(numRecords);
for (int i = 0; i < numRecords; i += 1) {
Record rec = (Record) TypeUtil.visit(schema, generator);
- // System.out.println("Add record "+rec);
+ //System.out.println("Add record "+rec);
records.add(rec);
}
diff --git a/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java b/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java
index f5ee124..100d89a 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/data/TestHelpers.java
@@ -225,7 +225,7 @@ public class TestHelpers {
// + " , expected:"+expectedValue);
if (actualRow.isNullAt(i)) {
- Assert.assertTrue("Expect null", expectedValue == null);
+ Assert.assertTrue("Expect null at " + r , expectedValue == null);
} else {
Object actualValue = actualRow.get(i, convert(fieldType));
assertEqualsUnsafe(fieldType, expectedValue, actualValue);
diff --git a/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkParquetVectorizedReader.java b/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkParquetVectorizedReader.java
index 2207131..f8f080c 100644
--- a/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkParquetVectorizedReader.java
+++ b/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkParquetVectorizedReader.java
@@ -38,6 +38,7 @@ import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
+import static org.apache.iceberg.TableProperties.PARQUET_DICT_SIZE_BYTES;
import static org.apache.iceberg.spark.data.TestHelpers.assertEqualsUnsafe;
public class TestSparkParquetVectorizedReader extends AvroDataTest {
@@ -56,9 +57,13 @@ public class TestSparkParquetVectorizedReader extends AvroDataTest {
Assert.assertTrue("Delete should succeed", testFile.delete());
try (FileAppender<GenericData.Record> writer = Parquet.write(Files.localOutput(testFile))
- .schema(schema)
- .named("test")
- .build()) {
+ .schema(schema)
+ .named("test")
+ /*.set("parquet.dictionary.page.size", "0")
+ .set("parquet.writer.encoding-override.double", "plain")
+ .set("parquet.enable.dictionary", "false")*/
+ .set(PARQUET_DICT_SIZE_BYTES, "1")
+ .build()) {
writer.addAll(expected);
}