Optimize decoders for DELTA_BINARY_PACKED parquet encoding

lib/trino-parquet/src/main/java/io/trino/parquet/ParquetReaderUtils.java

@@ -83,6 +83,71 @@ public static int readUleb128Int(SimpleSliceInputStream input)
         return value | inputByte << 28;
     }
 
+    public static long readUleb128Long(SimpleSliceInputStream input)
+    {
+        byte[] inputBytes = input.getByteArray();
+        int offset = input.getByteArrayOffset();
+        // Manual loop unrolling shows improvements in BenchmarkReadUleb128Long
+        long inputByte = inputBytes[offset];
+        long value = inputByte & 0x7F;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(1);
+            return value;
+        }
+        inputByte = inputBytes[offset + 1];
+        value |= (inputByte & 0x7F) << 7;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(2);
+            return value;
+        }
+        inputByte = inputBytes[offset + 2];
+        value |= (inputByte & 0x7F) << 14;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(3);
+            return value;
+        }
+        inputByte = inputBytes[offset + 3];
+        value |= (inputByte & 0x7F) << 21;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(4);
+            return value;
+        }
+        inputByte = inputBytes[offset + 4];
+        value |= (inputByte & 0x7F) << 28;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(5);
+            return value;
+        }
+        inputByte = inputBytes[offset + 5];
+        value |= (inputByte & 0x7F) << 35;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(6);
+            return value;
+        }
+        inputByte = inputBytes[offset + 6];
+        value |= (inputByte & 0x7F) << 42;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(7);
+            return value;
+        }
+        inputByte = inputBytes[offset + 7];
+        value |= (inputByte & 0x7F) << 49;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(8);
+            return value;
+        }
+        inputByte = inputBytes[offset + 8];
+        value |= (inputByte & 0x7F) << 56;
+        if ((inputByte & 0x80) == 0) {
+            input.skip(9);
+            return value;
+        }
+        inputByte = inputBytes[offset + 9];
+        verify((inputByte & 0x80) == 0, "ULEB128 variable-width long should not be longer than 10 bytes");
+        input.skip(10);
+        return value | inputByte << 63;
+    }
+
     public static int readFixedWidthInt(SimpleSliceInputStream input, int bytesWidth)
     {
         return switch (bytesWidth) {
@@ -98,6 +163,27 @@ public static int readFixedWidthInt(SimpleSliceInputStream input, int bytesWidth
         };
     }
 
+    /**
+     * For storing signed values (not the deltas themselves) in DELTA_BINARY_PACKED encoding, zigzag encoding
+     * (<a href="https://developers.google.com/protocol-buffers/docs/encoding#signed-integers">...</a>)
+     * is used to map negative values to positive ones and then apply ULEB128 on the result.
+     */
+    public static long zigzagDecode(long value)
+    {
+        return (value >>> 1) ^ -(value & 1);
+    }
+
+    /**
+     * Returns the result of arguments division rounded up.
+     * <p>
+     * Works only for positive numbers.
+     * The sum of dividend and divisor cannot exceed Integer.MAX_VALUE
+     */
+    public static int ceilDiv(int dividend, int divisor)
+    {
+        return (dividend + divisor - 1) / divisor;
+    }
+
     /**
      * Propagate the sign bit in values that are shorter than 8 bytes.
      * <p>

lib/trino-parquet/src/main/java/io/trino/parquet/reader/ColumnReaderFactory.java

@@ -143,7 +143,7 @@ public static ColumnReader create(PrimitiveField field, DateTimeZone timeZone, A
                     throw unsupportedException(type, field);
                 }
                 if (primitiveType == INT32) {
-                    return new FlatColumnReader<>(field, ValueDecoders::getIntToLongDecoder, LONG_ADAPTER, memoryContext);
+                    return new FlatColumnReader<>(field, TransformingValueDecoders::getInt32ToLongDecoder, LONG_ADAPTER, memoryContext);
                 }
                 if (primitiveType == INT64) {
                     return new FlatColumnReader<>(field, ValueDecoders::getLongDecoder, LONG_ADAPTER, memoryContext);
@@ -218,7 +218,7 @@ && isIntegerOrDecimalPrimitive(primitiveType)) {
                 if (decimalType.getScale() == 0 && decimalType.getPrecision() >= MAX_INT_DIGITS
                         && primitiveType == INT32
                         && isIntegerAnnotation(annotation)) {
-                    return new FlatColumnReader<>(field, ValueDecoders::getIntToLongDecoder, LONG_ADAPTER, memoryContext);
+                    return new FlatColumnReader<>(field, TransformingValueDecoders::getInt32ToLongDecoder, LONG_ADAPTER, memoryContext);
                 }
                 if (!(annotation instanceof DecimalLogicalTypeAnnotation decimalAnnotation)) {
                     throw unsupportedException(type, field);

lib/trino-parquet/src/main/java/io/trino/parquet/reader/SimpleSliceInputStream.java

@@ -77,6 +77,12 @@ public byte[] readBytes()
         return bytes;
     }
 
+    public void readBytes(byte[] output, int outputOffset, int length)
+    {
+        slice.getBytes(offset, output, outputOffset, length);
+        offset += length;
+    }
+
     public void readBytes(Slice destination, int destinationIndex, int length)
     {
         slice.getBytes(offset, destination, destinationIndex, length);

lib/trino-parquet/src/main/java/io/trino/parquet/reader/decoders/ApacheParquetValueDecoders.java

@@ -38,8 +38,6 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static io.trino.parquet.ParquetReaderUtils.castToByte;
-import static io.trino.parquet.ParquetReaderUtils.toByteExact;
-import static io.trino.parquet.ParquetReaderUtils.toShortExact;
 import static io.trino.parquet.ParquetTimestampUtils.decodeInt96Timestamp;
 import static io.trino.parquet.ParquetTypeUtils.checkBytesFitInShortDecimal;
 import static io.trino.parquet.ParquetTypeUtils.getShortDecimalValue;
@@ -55,161 +53,6 @@ public class ApacheParquetValueDecoders
 {
     private ApacheParquetValueDecoders() {}
 
-    public static final class IntApacheParquetValueDecoder
-            implements ValueDecoder<int[]>
-    {
-        private final ValuesReader delegate;
-
-        public IntApacheParquetValueDecoder(ValuesReader delegate)
-        {
-            this.delegate = requireNonNull(delegate, "delegate is null");
-        }
-
-        @Override
-        public void init(SimpleSliceInputStream input)
-        {
-            initialize(input, delegate);
-        }
-
-        @Override
-        public void read(int[] values, int offset, int length)
-        {
-            for (int i = offset; i < offset + length; i++) {
-                values[i] = delegate.readInteger();
-            }
-        }
-
-        @Override
-        public void skip(int n)
-        {
-            delegate.skip(n);
-        }
-    }
-
-    public static final class ShortApacheParquetValueDecoder
-            implements ValueDecoder<short[]>
-    {
-        private final ValuesReader delegate;
-
-        public ShortApacheParquetValueDecoder(ValuesReader delegate)
-        {
-            this.delegate = requireNonNull(delegate, "delegate is null");
-        }
-
-        @Override
-        public void init(SimpleSliceInputStream input)
-        {
-            initialize(input, delegate);
-        }
-
-        @Override
-        public void read(short[] values, int offset, int length)
-        {
-            for (int i = offset; i < offset + length; i++) {
-                values[i] = toShortExact(delegate.readInteger());
-            }
-        }
-
-        @Override
-        public void skip(int n)
-        {
-            delegate.skip(n);
-        }
-    }
-
-    public static final class ByteApacheParquetValueDecoder
-            implements ValueDecoder<byte[]>
-    {
-        private final ValuesReader delegate;
-
-        public ByteApacheParquetValueDecoder(ValuesReader delegate)
-        {
-            this.delegate = requireNonNull(delegate, "delegate is null");
-        }
-
-        @Override
-        public void init(SimpleSliceInputStream input)
-        {
-            initialize(input, delegate);
-        }
-
-        @Override
-        public void read(byte[] values, int offset, int length)
-        {
-            for (int i = offset; i < offset + length; i++) {
-                values[i] = toByteExact(delegate.readInteger());
-            }
-        }
-
-        @Override
-        public void skip(int n)
-        {
-            delegate.skip(n);
-        }
-    }
-
-    public static final class IntToLongApacheParquetValueDecoder
-            implements ValueDecoder<long[]>
-    {
-        private final ValuesReader delegate;
-
-        public IntToLongApacheParquetValueDecoder(ValuesReader delegate)
-        {
-            this.delegate = requireNonNull(delegate, "delegate is null");
-        }
-
-        @Override
-        public void init(SimpleSliceInputStream input)
-        {
-            initialize(input, delegate);
-        }
-
-        @Override
-        public void read(long[] values, int offset, int length)
-        {
-            for (int i = offset; i < offset + length; i++) {
-                values[i] = delegate.readInteger();
-            }
-        }
-
-        @Override
-        public void skip(int n)
-        {
-            delegate.skip(n);
-        }
-    }
-
-    public static final class LongApacheParquetValueDecoder
-            implements ValueDecoder<long[]>
-    {
-        private final ValuesReader delegate;
-
-        public LongApacheParquetValueDecoder(ValuesReader delegate)
-        {
-            this.delegate = requireNonNull(delegate, "delegate is null");
-        }
-
-        @Override
-        public void init(SimpleSliceInputStream input)
-        {
-            initialize(input, delegate);
-        }
-
-        @Override
-        public void read(long[] values, int offset, int length)
-        {
-            for (int i = offset; i < offset + length; i++) {
-                values[i] = delegate.readLong();
-            }
-        }
-
-        @Override
-        public void skip(int n)
-        {
-            delegate.skip(n);
-        }
-    }
-
     public static final class BooleanApacheParquetValueDecoder
             implements ValueDecoder<byte[]>
     {

lib/trino-parquet/src/main/java/io/trino/parquet/reader/decoders/ByteBitUnpacker.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed 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 io.trino.parquet.reader.decoders;
+
+import io.trino.parquet.reader.SimpleSliceInputStream;
+
+public interface ByteBitUnpacker
+{
+    /**
+     * @param length must be a multiple of 32
+     */
+    void unpack(byte[] output, int outputOffset, SimpleSliceInputStream input, int length);
+}