You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/03/16 04:30:03 UTC
[GitHub] [spark] sadikovi commented on a change in pull request #35262: [SPARK-37974][SQL] Implement vectorized DELTA_BYTE_ARRAY and DELTA_LENGTH_BYTE_ARRAY encodings for Parquet V2 support
sadikovi commented on a change in pull request #35262:
URL: https://github.com/apache/spark/pull/35262#discussion_r827612677
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -283,6 +290,11 @@ private void initDataReader(
} catch (IOException e) {
throw new IOException("could not read page in col " + descriptor, e);
}
+ if (CorruptDeltaByteArrays.requiresSequentialReads(writerVersion, dataEncoding) &&
+ previousReader instanceof RequiresPreviousReader) {
+ // previous reader can only be set if reading sequentially
Review comment:
nit: [P]revious.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java
##########
@@ -90,13 +90,18 @@ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOExce
Preconditions.checkArgument(miniSize % 8 == 0,
"miniBlockSize must be multiple of 8, but it's " + miniSize);
this.miniBlockSizeInValues = (int) miniSize;
+ // True value count. May be less than valueCount because of nulls
Review comment:
I think it would be more useful to annotate the method `getTotalValueCount` instead of here.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -283,6 +290,11 @@ private void initDataReader(
} catch (IOException e) {
throw new IOException("could not read page in col " + descriptor, e);
}
+ if (CorruptDeltaByteArrays.requiresSequentialReads(writerVersion, dataEncoding) &&
Review comment:
When does this happen? Can you add a comment on why we need this?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
##########
@@ -16,50 +16,127 @@
*/
package org.apache.spark.sql.execution.datasources.parquet;
+import static org.apache.spark.sql.types.DataTypes.BinaryType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+
import org.apache.parquet.bytes.ByteBufferInputStream;
-import org.apache.parquet.column.values.deltastrings.DeltaByteArrayReader;
+import org.apache.parquet.column.values.RequiresPreviousReader;
+import org.apache.parquet.column.values.ValuesReader;
import org.apache.parquet.io.api.Binary;
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
import java.io.IOException;
import java.nio.ByteBuffer;
/**
- * An implementation of the Parquet DELTA_BYTE_ARRAY decoder that supports the vectorized interface.
+ * An implementation of the Parquet DELTA_BYTE_ARRAY decoder that supports the vectorized
+ * interface.
*/
-public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase {
- private final DeltaByteArrayReader deltaByteArrayReader = new DeltaByteArrayReader();
+public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
+ implements VectorizedValuesReader, RequiresPreviousReader {
+
+ private final VectorizedDeltaBinaryPackedReader prefixLengthReader;
+ private final VectorizedDeltaLengthByteArrayReader suffixReader;
+ private WritableColumnVector prefixLengthVector;
+ private ByteBuffer previous;
+ private int currentRow = 0;
+
+ // temporary variable used by readBinary
+ private final WritableColumnVector binaryValVector;
+ // temporary variable used by skipBinary
Review comment:
nit: Upper case.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
##########
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.nio.ByteBuffer;
Review comment:
nit: new line after the import.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetDeltaByteArrayEncodingSuite.scala
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet
+
+import org.apache.parquet.bytes.DirectByteBufferAllocator
+import org.apache.parquet.column.values.Utils
+import org.apache.parquet.column.values.deltastrings.DeltaByteArrayWriter
+
+import org.apache.spark.sql.execution.vectorized.{OnHeapColumnVector, WritableColumnVector}
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, StringType}
+
+/**
+ * Read tests for vectorized Delta byte array reader.
+ * Translated from * org.apache.parquet.column.values.delta.TestDeltaByteArray
+ */
+class ParquetDeltaByteArrayEncodingSuite extends ParquetCompatibilityTest with SharedSparkSession {
+ val values: Array[String] = Array("parquet-mr", "parquet", "parquet-format");
+ val randvalues: Array[String] = Utils.getRandomStringSamples(10000, 32)
+
+ var writer: DeltaByteArrayWriter = _
+ var reader: VectorizedDeltaByteArrayReader = _
+ private var writableColumnVector: WritableColumnVector = _
+
+ protected override def beforeEach(): Unit = {
+ writer = new DeltaByteArrayWriter(64 * 1024, 64 * 1024, new DirectByteBufferAllocator)
+ reader = new VectorizedDeltaByteArrayReader()
+ super.beforeAll()
+ }
+
+ test("test Serialization") {
+ assertReadWrite(writer, reader, values)
+ }
+
+ test("random strings") {
+ assertReadWrite(writer, reader, randvalues)
+ }
+
+ test("random strings with skip") {
+ assertReadWriteWithSkip(writer, reader, randvalues)
+ }
+
+ test("random strings with skipN") {
+ assertReadWriteWithSkipN(writer, reader, randvalues)
+ }
+
+ test("test lengths") {
+ var reader = new VectorizedDeltaBinaryPackedReader
+ Utils.writeData(writer, values)
+ val data = writer.getBytes.toInputStream
+ val length = values.length
+ writableColumnVector = new OnHeapColumnVector(length, IntegerType)
+ reader.initFromPage(length, data)
+ reader.readIntegers(length, writableColumnVector, 0)
+ // test prefix lengths
+ assert(0 == writableColumnVector.getInt(0))
+ assert(7 == writableColumnVector.getInt(1))
+ assert(7 == writableColumnVector.getInt(2))
+
+ reader = new VectorizedDeltaBinaryPackedReader
+ writableColumnVector = new OnHeapColumnVector(length, IntegerType)
+ reader.initFromPage(length, data)
+ reader.readIntegers(length, writableColumnVector, 0)
+ // test suffix lengths
+ assert(10 == writableColumnVector.getInt(0))
+ assert(0 == writableColumnVector.getInt(1))
+ assert(7 == writableColumnVector.getInt(2))
+ }
+
+ private def assertReadWrite(
+ writer: DeltaByteArrayWriter,
+ reader: VectorizedDeltaByteArrayReader,
+ vals: Array[String]): Unit = {
+ Utils.writeData(writer, vals)
+ val length = vals.length
+ val is = writer.getBytes.toInputStream
+
+ writableColumnVector = new OnHeapColumnVector(length, StringType)
+
+ reader.initFromPage(length, is)
+ reader.readBinary(length, writableColumnVector, 0)
+
+ for (i <- 0 until length) {
+ assert(vals(i).getBytes() sameElements writableColumnVector.getBinary(i))
+ }
+ }
+
+ private def assertReadWriteWithSkip(
+ writer: DeltaByteArrayWriter,
+ reader: VectorizedDeltaByteArrayReader,
+ vals: Array[String]): Unit = {
+ Utils.writeData(writer, vals)
+ val length = vals.length
+ val is = writer.getBytes.toInputStream
+ writableColumnVector = new OnHeapColumnVector(length, StringType)
+ reader.initFromPage(length, is)
+ var i = 0
+ while ( {
+ i < vals.length
+ }) {
+ reader.readBinary(1, writableColumnVector, i)
+ assert(vals(i).getBytes() sameElements writableColumnVector.getBinary(i))
+ reader.skipBinary(1)
+ i += 2
+ }
+ }
+
+ private def assertReadWriteWithSkipN(
+ writer: DeltaByteArrayWriter,
+ reader: VectorizedDeltaByteArrayReader,
+ vals: Array[String]): Unit = {
+ Utils.writeData(writer, vals)
+ val length = vals.length
+ val is = writer.getBytes.toInputStream
+ writableColumnVector = new OnHeapColumnVector(length, StringType)
+ reader.initFromPage(length, is)
+ var skipCount = 0
+ var i = 0
+ while ( {
+ i < vals.length
+ }) {
+ skipCount = (vals.length - i) / 2
+ reader.readBinary(1, writableColumnVector, i)
+ assert(vals(i).getBytes() sameElements writableColumnVector.getBinary(i))
+ reader.skipBinary(skipCount)
+ i += skipCount + 1
+ }
+ }
+
Review comment:
ditto.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/WritableColumnVector.java
##########
@@ -443,6 +444,8 @@ public UTF8String getUTF8String(int rowId) {
}
}
+ public abstract ByteBuffer getBytesUnsafe(int rowId, int count);
Review comment:
I agree, the method is misleading since there is a memory copy involved, it is just does not call System.arraycopy in OnHeapColumnVector.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java
##########
@@ -16,50 +16,127 @@
*/
package org.apache.spark.sql.execution.datasources.parquet;
+import static org.apache.spark.sql.types.DataTypes.BinaryType;
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+
import org.apache.parquet.bytes.ByteBufferInputStream;
-import org.apache.parquet.column.values.deltastrings.DeltaByteArrayReader;
+import org.apache.parquet.column.values.RequiresPreviousReader;
+import org.apache.parquet.column.values.ValuesReader;
import org.apache.parquet.io.api.Binary;
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
import java.io.IOException;
import java.nio.ByteBuffer;
/**
- * An implementation of the Parquet DELTA_BYTE_ARRAY decoder that supports the vectorized interface.
+ * An implementation of the Parquet DELTA_BYTE_ARRAY decoder that supports the vectorized
+ * interface.
*/
-public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase {
- private final DeltaByteArrayReader deltaByteArrayReader = new DeltaByteArrayReader();
+public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase
+ implements VectorizedValuesReader, RequiresPreviousReader {
+
+ private final VectorizedDeltaBinaryPackedReader prefixLengthReader;
+ private final VectorizedDeltaLengthByteArrayReader suffixReader;
+ private WritableColumnVector prefixLengthVector;
+ private ByteBuffer previous;
+ private int currentRow = 0;
+
+ // temporary variable used by readBinary
Review comment:
nit: Upper case.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetEncodingSuite.scala
##########
@@ -141,45 +149,54 @@ class ParquetEncodingSuite extends ParquetCompatibilityTest with SharedSparkSess
)
val hadoopConf = spark.sessionState.newHadoopConfWithOptions(extraOptions)
- withSQLConf(
- SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
- ParquetOutputFormat.JOB_SUMMARY_LEVEL -> "ALL") {
- withTempPath { dir =>
- val path = s"${dir.getCanonicalPath}/test.parquet"
-
- val data = (1 to 3).map { i =>
- ( i, i.toLong, i.toShort, Array[Byte](i.toByte), s"test_${i}",
- DateTimeUtils.fromJavaDate(Date.valueOf(s"2021-11-0" + i)),
- DateTimeUtils.fromJavaTimestamp(Timestamp.valueOf(s"2020-11-01 12:00:0" + i)),
- Period.of(1, i, 0), Duration.ofMillis(i * 100),
- new BigDecimal(java.lang.Long.toUnsignedString(i*100000))
- )
+ withMemoryModes { offHeapMode =>
+ withSQLConf(
+ SQLConf.COLUMN_VECTOR_OFFHEAP_ENABLED.key -> offHeapMode,
+ SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true",
+ ParquetOutputFormat.JOB_SUMMARY_LEVEL -> "ALL") {
+ withTempPath { dir =>
+ val path = s"${dir.getCanonicalPath}/test.parquet"
+ // Have more than 2 * 4096 records (so we have multiple tasks and each task
+ // reads at least twice from the reader). This will catch any issues with state
+ // maintained by the reader(s)
+ // Add at least one string with a null
+ val data = (1 to 81971).map { i =>
Review comment:
I don't quite understand how this number was chosen. Can you elaborate? Can we make it `2 * 4096 + 1` - would it work as well?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java
##########
@@ -86,4 +87,20 @@ void readLongsWithRebase(
void write(WritableColumnVector outputColumnVector, int rowId, long val);
}
+ @FunctionalInterface
+ interface ByteBufferOutputWriter {
+ void write(WritableColumnVector c, int rowId, ByteBuffer val, int length);
+
+ static void writeArrayByteBuffer(WritableColumnVector c, int rowId, ByteBuffer val,
+ int length) {
+ c.putByteArray(rowId,
+ val.array(),
+ val.arrayOffset() + val.position(),
+ length);
+ }
+
+ static void skipWrite(WritableColumnVector c, int rowId, ByteBuffer val, int length) { }
+
+ }
+
Review comment:
nit: new line.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java
##########
@@ -367,7 +367,9 @@ private void checkEndOfRowGroup() throws IOException {
datetimeRebaseMode,
datetimeRebaseTz,
int96RebaseMode,
- int96RebaseTz);
+ int96RebaseTz,
+ writerVersion
+ );
Review comment:
nit: should be on the previous line.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
##########
@@ -221,6 +221,13 @@ protected UTF8String getBytesAsUTF8String(int rowId, int count) {
return UTF8String.fromAddress(null, data + rowId, count);
}
+ @Override
+ public ByteBuffer getBytesUnsafe(int rowId, int count) {
Review comment:
Can we replace it with:
```java
return ByteBuffer.wrap(getBytes(rowId, count));
```
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaLengthByteArrayReader.java
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import static org.apache.spark.sql.types.DataTypes.IntegerType;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.spark.sql.execution.vectorized.OnHeapColumnVector;
+import org.apache.spark.sql.execution.vectorized.WritableColumnVector;
+
+/**
+ * An implementation of the Parquet DELTA_LENGTH_BYTE_ARRAY decoder that supports the vectorized
+ * interface.
+ */
+public class VectorizedDeltaLengthByteArrayReader extends VectorizedReaderBase implements
+ VectorizedValuesReader {
+
+ private final VectorizedDeltaBinaryPackedReader lengthReader;
+ private ByteBufferInputStream in;
+ private WritableColumnVector lengthsVector;
+ private int currentRow = 0;
+
+ VectorizedDeltaLengthByteArrayReader() {
+ lengthReader = new VectorizedDeltaBinaryPackedReader();
+ }
+
+ @Override
+ public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException {
+ lengthsVector = new OnHeapColumnVector(valueCount, IntegerType);
+ lengthReader.initFromPage(valueCount, in);
+ lengthReader.readIntegers(lengthReader.getTotalValueCount(), lengthsVector, 0);
+ this.in = in.remainingStream();
+ }
+
+ @Override
+ public void readBinary(int total, WritableColumnVector c, int rowId) {
+ ByteBuffer buffer;
+ ByteBufferOutputWriter outputWriter = ByteBufferOutputWriter::writeArrayByteBuffer;
+ int length;
+ for (int i = 0; i < total; i++) {
+ length = lengthsVector.getInt(rowId + i);
+ try {
+ buffer = in.slice(length);
+ } catch (EOFException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes");
+ }
+ outputWriter.write(c, rowId + i, buffer, length);
+ }
+ currentRow += total;
+ }
+
+ public ByteBuffer getBytes(int rowId) {
+ int length = lengthsVector.getInt(rowId);
+ try {
+ return in.slice(length);
+ } catch (EOFException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes");
+ }
+ }
+
+ @Override
+ public void skipBinary(int total) {
+ for (int i = 0; i < total; i++) {
+ int remaining = lengthsVector.getInt(currentRow + i);
+ while (remaining > 0) {
+ remaining -= in.skip(remaining);
+ }
+ }
+ currentRow += total;
+ }
+
Review comment:
nit: new line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org