PARQUET-2159: Vectorized BytePacker decoder using Java VectorAPI

README.md

@@ -83,6 +83,20 @@ Parquet is a very active project, and new features are being added quickly. Here
 * Column stats
 * Delta encoding
 * Index pages
+* Java Vector API support
+
+## Java Vector API support
+Parquet-MR has supported Java Vector API to speed up reading, to enable this feature:
+* Java 17+, 64-bit
+* Requiring the CPU to support instruction sets:
+  * avx512vbmi
+  * avx512_vbmi2
+* To build the jars: `mvn clean package -P vector-plugins`
+* For Apache Spark to enable this feature:
+  * Build parquet and replace the parquet-encoding-{VERSION}.jar on the spark jars folder
+  * Build parquet-encoding-vector and copy parquet-encoding-vector-{VERSION}.jar to the spark jars folder
+  * Edit spark class#VectorizedRleValuesReader, function#readNextGroup refer to parquet class#ParquetReadRouter, function#readBatchUsing512Vector
+  * Build spark with maven and replace spark-sql_2.12-{VERSION}.jar on the spark jars folder
 
 ## Map/Reduce integration
 

parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/BytePacker.java

@@ -31,6 +31,13 @@ public abstract class BytePacker {
 
   private final int bitWidth;
 
+  /**
+   * Number of integer values to be unpacked at a time.
+   * unpackCount is a multiple of 8
+   * For AVX512, the register is 512 bits, so output values at a time maybe different when different bitWidth.
+   */
+  protected int unpackCount;
+
   BytePacker(int bitWidth) {
     this.bitWidth = bitWidth;
   }
@@ -42,6 +49,10 @@ public final int getBitWidth() {
     return bitWidth;
   }
 
+  public int getUnpackCount() {
+    throw new RuntimeException("getUnpackCount must be implemented by subclass!");
+  }
+
   /**
    * pack 8 values from input at inPos into bitWidth bytes in output at outPos.
    * nextPosition: inPos += 8; outPos += getBitWidth()
@@ -105,4 +116,26 @@ public void unpack8Values(final byte[] input, final int inPos, final int[] outpu
   public void unpack32Values(byte[] input, int inPos, int[] output, int outPos) {
     unpack32Values(ByteBuffer.wrap(input), inPos, output, outPos);
   }
+
+  /**
+   * unpack bitWidth bytes from input at inPos into {unpackCount} values in output at outPos using Java Vector API.
+   * @param input the input bytes
+   * @param inPos where to read from in input
+   * @param output the output values
+   * @param outPos where to write to in output
+   */
+  public void unpackValuesUsingVector(final byte[] input, final int inPos, final int[] output, final int outPos) {
+    throw new RuntimeException("unpackValuesUsingVector must be implemented by subclass!");
+  }
+
+  /**
+   * unpack bitWidth bytes from input at inPos into {unpackCount} values in output at outPos using Java Vector API.
+   * @param input the input bytes
+   * @param inPos where to read from in input
+   * @param output the output values
+   * @param outPos where to write to in output
+   */
+  public void unpackValuesUsingVector(final ByteBuffer input, final int inPos, final int[] output, final int outPos) {
+    throw new RuntimeException("unpackValuesUsingVector must be implemented by subclass!");
+  }
 }

parquet-encoding/src/main/java/org/apache/parquet/column/values/bitpacking/Packer.java

@@ -36,6 +36,12 @@ public IntPacker newIntPacker(int width) {
     public BytePacker newBytePacker(int width) {
       return beBytePackerFactory.newBytePacker(width);
     }
+
+    @Override
+    public BytePacker newBytePackerVector(int width) {
+      throw new RuntimeException("Not currently supported!");
+    }
+
     @Override
     public BytePackerForLong newBytePackerForLong(int width) {
       return beBytePackerForLongFactory.newBytePackerForLong(width);
@@ -55,6 +61,22 @@ public IntPacker newIntPacker(int width) {
     public BytePacker newBytePacker(int width) {
       return leBytePackerFactory.newBytePacker(width);
     }
+
+    @Override
+    public BytePacker newBytePackerVector(int width) {
+      if (leBytePacker512VectorFactory == null) {
+        synchronized (Packer.class) {
+          if (leBytePacker512VectorFactory == null) {
+            leBytePacker512VectorFactory = getBytePackerFactory("ByteBitPacking512VectorLE");
+          }
+        }
+      }
+      if (leBytePacker512VectorFactory == null) {
+        throw new RuntimeException("No enable java vector plugin on little endian architectures");
+      }
+      return leBytePacker512VectorFactory.newBytePacker(width);
+    }
+
     @Override
     public BytePackerForLong newBytePackerForLong(int width) {
       return leBytePackerForLongFactory.newBytePackerForLong(width);
@@ -86,6 +108,8 @@ private static Object getStaticField(String className, String fieldName) {
   static IntPackerFactory leIntPackerFactory = getIntPackerFactory("LemireBitPackingLE");
   static BytePackerFactory beBytePackerFactory = getBytePackerFactory("ByteBitPackingBE");
   static BytePackerFactory leBytePackerFactory = getBytePackerFactory("ByteBitPackingLE");
+  // ByteBitPacking512VectorLE is not enabled default, so leBytePacker512VectorFactory cannot be initialized as a static property
+  static BytePackerFactory leBytePacker512VectorFactory = null;
   static BytePackerForLongFactory beBytePackerForLongFactory = getBytePackerForLongFactory("ByteBitPackingForLongBE");
   static BytePackerForLongFactory leBytePackerForLongFactory = getBytePackerForLongFactory("ByteBitPackingForLongLE");
 
@@ -101,6 +125,10 @@ private static Object getStaticField(String className, String fieldName) {
    */
   public abstract BytePacker newBytePacker(int width);
 
+  public BytePacker newBytePackerVector(int width) {
+    throw new RuntimeException("newBytePackerVector must be implemented by subclasses!");
+  }
+
   /**
    * @param width the width in bits of the packed values
    * @return a byte based packer for INT64

parquet-plugins/parquet-encoding-vector/pom.xml

@@ -0,0 +1,127 @@
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <parent>
+    <groupId>org.apache.parquet</groupId>
+    <artifactId>parquet</artifactId>
+    <version>1.13.0-SNAPSHOT</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <artifactId>parquet-encoding-vector</artifactId>
+  <packaging>jar</packaging>
+
+  <name>Apache Parquet Encodings Vector</name>
+  <url>https://parquet.apache.org</url>
+
+  <properties>
+    <extraJavaVectorArgs>
+      --add-modules=jdk.incubator.vector
+    </extraJavaVectorArgs>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.parquet</groupId>
+      <artifactId>parquet-encoding</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>${slf4j.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>${slf4j.version}</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <release>17</release>
+          <compilerArgs combine.children="append">
+            <compilerArg>${extraJavaVectorArgs}</compilerArg>
+          </compilerArgs>
+        </configuration>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-failsafe-plugin</artifactId>
+        <configuration>
+          <argLine>${extraJavaVectorArgs}</argLine>
+        </configuration>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <argLine>${extraJavaVectorArgs}</argLine>
+        </configuration>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>3.4.1</version>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <version>3.3.0</version>
+        <executions>
+          <execution>
+            <goals>
+              <goal>analyze-only</goal>
+            </goals>
+            <configuration>
+              <failOnWarning>true</failOnWarning>
+              <ignoreNonCompile>true</ignoreNonCompile>
+              <ignoredNonTestScopedDependencies>
+                <ignoredNonTestScopedDependency>*</ignoredNonTestScopedDependency>
+              </ignoredNonTestScopedDependencies>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>