parthchandra commented on a change in pull request #35262:
URL: https://github.com/apache/spark/pull/35262#discussion_r828306457



##########
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:
       Fixed

##########
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 as well. @sunchao given that this is from your patch, is it ok 
to change the name to say `getByteBuffer` ?

##########
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:
       We could, but it would incur an additional function call in performance 
sensitive code. @sunchao ?

##########
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:
       Ok

##########
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:
       OK

##########
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:
       OK

##########
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:
       Fixed

##########
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:
       Added comment. Detailed explanation is in the comment in 
`VectorizedDeltaByteArrayReader.setPreviousValue`

##########
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:
       OK

##########
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:
       Added the comment to `getTotalValueCount` as well.

##########
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:
       Yes, it would. Changed. (Sorry that number sneaked in after I tested 
something else and forgot to undo it).

##########
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:
       Ok




-- 
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: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]



---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to