From 15dc3f22649c3e5834fc84fd971443e85facfc8f Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Sat, 22 Aug 2026 20:32:03 +0530 Subject: [PATCH] update --- .../reader/ParquetVectorUpdaterFactory.java | 12 +- ...VectorizedByteStreamSplitValuesReader.java | 254 ++++++++++++++++++ .../reader/VectorizedColumnReader.java | 24 ++ .../reader/VectorizedPlainValuesReader.java | 51 ++++ .../reader/VectorizedValuesReader.java | 31 +++ .../format/parquet/ParquetReadWriteTest.java | 72 +++++ ...orizedByteStreamSplitValuesReaderTest.java | 195 ++++++++++++++ 7 files changed, 630 insertions(+), 9 deletions(-) create mode 100644 paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReader.java create mode 100644 paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReaderTest.java diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java index c0445b3f136d..d1ef82933a50 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ParquetVectorUpdaterFactory.java @@ -611,9 +611,7 @@ public void readValues( int offset, WritableLongVector values, VectorizedValuesReader valuesReader) { - for (int i = 0; i < total; i++) { - values.setLong(offset + i, valuesReader.readInteger()); - } + valuesReader.readIntegersAsLongs(total, values, offset); } @Override @@ -999,9 +997,7 @@ public void readValues( int offset, WritableDoubleVector values, VectorizedValuesReader valuesReader) { - for (int i = 0; i < total; i++) { - values.setDouble(offset + i, valuesReader.readFloat()); - } + valuesReader.readFloatsAsDoubles(total, values, offset); } @Override @@ -1071,9 +1067,7 @@ public void readValues( int offset, WritableBytesVector values, VectorizedValuesReader valuesReader) { - for (int i = 0; i < total; i++) { - readValue(offset + i, values, valuesReader); - } + valuesReader.readFixedLenByteArray(total, arrayLen, values, offset); } @Override diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReader.java new file mode 100644 index 000000000000..c3d181905e7b --- /dev/null +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReader.java @@ -0,0 +1,254 @@ +/* + * 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.paimon.format.parquet.reader; + +import org.apache.paimon.data.columnar.writable.WritableByteVector; +import org.apache.paimon.data.columnar.writable.WritableBytesVector; +import org.apache.paimon.data.columnar.writable.WritableDoubleVector; +import org.apache.paimon.data.columnar.writable.WritableFloatVector; +import org.apache.paimon.data.columnar.writable.WritableIntVector; +import org.apache.paimon.data.columnar.writable.WritableLongVector; +import org.apache.paimon.data.columnar.writable.WritableShortVector; + +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.io.api.Binary; + +import java.io.IOException; +import java.nio.ByteBuffer; + +/** Vectorized reader for Parquet byte stream split encoding. */ +public class VectorizedByteStreamSplitValuesReader extends VectorizedReaderBase { + + private final int typeWidth; + + private int valueCount; + + private byte[] pageData; + + private int offset; + + public VectorizedByteStreamSplitValuesReader(int typeWidth) { + this.typeWidth = typeWidth; + } + + @Override + public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException { + int totalBytes = in.available(); + this.valueCount = totalBytes / typeWidth; + this.offset = 0; + this.pageData = new byte[totalBytes]; + ByteBuffer buf = in.slice(totalBytes); + buf.get(pageData, 0, totalBytes); + } + + private int assembleInt(int idx) { + return (pageData[idx] & 0xFF) + | ((pageData[valueCount + idx] & 0xFF) << 8) + | ((pageData[2 * valueCount + idx] & 0xFF) << 16) + | ((pageData[3 * valueCount + idx] & 0xFF) << 24); + } + + private long assembleLong(int idx) { + return (pageData[idx] & 0xFFL) + | ((pageData[valueCount + idx] & 0xFFL) << 8) + | ((pageData[2 * valueCount + idx] & 0xFFL) << 16) + | ((pageData[3 * valueCount + idx] & 0xFFL) << 24) + | ((pageData[4 * valueCount + idx] & 0xFFL) << 32) + | ((pageData[5 * valueCount + idx] & 0xFFL) << 40) + | ((pageData[6 * valueCount + idx] & 0xFFL) << 48) + | ((pageData[7 * valueCount + idx] & 0xFFL) << 56); + } + + @Override + public byte readByte() { + return (byte) readInteger(); + } + + @Override + public short readShort() { + return (short) readInteger(); + } + + @Override + public int readInteger() { + return assembleInt(offset++); + } + + @Override + public long readLong() { + return assembleLong(offset++); + } + + @Override + public float readFloat() { + return Float.intBitsToFloat(assembleInt(offset++)); + } + + @Override + public double readDouble() { + return Double.longBitsToDouble(assembleLong(offset++)); + } + + @Override + public Binary readBinary(int len) { + byte[] result = new byte[len]; + for (int b = 0; b < len; b++) { + result[b] = pageData[b * valueCount + offset]; + } + offset++; + return Binary.fromConstantByteArray(result); + } + + @Override + public void readBytes(int total, WritableByteVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setByte(rowId + i, (byte) assembleInt(offset + i)); + } + offset += total; + } + + @Override + public void readShorts(int total, WritableShortVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setShort(rowId + i, (short) assembleInt(offset + i)); + } + offset += total; + } + + @Override + public void readIntegers(int total, WritableIntVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setInt(rowId + i, assembleInt(offset + i)); + } + offset += total; + } + + @Override + public void readIntegersAsLongs(int total, WritableLongVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setLong(rowId + i, assembleInt(offset + i)); + } + offset += total; + } + + @Override + public void readIntegersAsDoubles(int total, WritableDoubleVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, (double) assembleInt(offset + i)); + } + offset += total; + } + + @Override + public void readLongs(int total, WritableLongVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setLong(rowId + i, assembleLong(offset + i)); + } + offset += total; + } + + @Override + public void readLongsAsInts(int total, WritableIntVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setInt(rowId + i, (int) assembleLong(offset + i)); + } + offset += total; + } + + @Override + public void readFloats(int total, WritableFloatVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setFloat(rowId + i, Float.intBitsToFloat(assembleInt(offset + i))); + } + offset += total; + } + + @Override + public void readFloatsAsDoubles(int total, WritableDoubleVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, (double) Float.intBitsToFloat(assembleInt(offset + i))); + } + offset += total; + } + + @Override + public void readDoubles(int total, WritableDoubleVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, Double.longBitsToDouble(assembleLong(offset + i))); + } + offset += total; + } + + @Override + public void readBinary(int total, WritableBytesVector c, int rowId) { + byte[] scratch = new byte[typeWidth]; + for (int i = 0; i < total; i++) { + for (int b = 0; b < typeWidth; b++) { + scratch[b] = pageData[b * valueCount + offset]; + } + c.putByteArray(rowId + i, scratch, 0, typeWidth); + offset++; + } + } + + @Override + public void readFixedLenByteArray(int total, int len, WritableBytesVector c, int rowId) { + readBinary(total, c, rowId); + } + + @Override + public void skipBytes(int total) { + offset += total; + } + + @Override + public void skipShorts(int total) { + offset += total; + } + + @Override + public void skipIntegers(int total) { + offset += total; + } + + @Override + public void skipLongs(int total) { + offset += total; + } + + @Override + public void skipFloats(int total) { + offset += total; + } + + @Override + public void skipDoubles(int total) { + offset += total; + } + + @Override + public void skipBinary(int total) { + offset += total; + } + + @Override + public void skipFixedLenByteArray(int total, int len) { + offset += total; + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedColumnReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedColumnReader.java index 53b11138da0f..2b4e4d769df9 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedColumnReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedColumnReader.java @@ -317,6 +317,30 @@ private ValuesReader getValuesReader(Encoding encoding) { return new VectorizedDeltaLengthByteArrayReader(); case DELTA_BINARY_PACKED: return new VectorizedDeltaBinaryPackedReader(); + case BYTE_STREAM_SPLIT: + { + PrimitiveType.PrimitiveTypeName typeName = + this.descriptor.getPrimitiveType().getPrimitiveTypeName(); + int typeWidth; + switch (typeName) { + case FLOAT: + case INT32: + typeWidth = 4; + break; + case DOUBLE: + case INT64: + typeWidth = 8; + break; + case FIXED_LEN_BYTE_ARRAY: + typeWidth = this.descriptor.getPrimitiveType().getTypeLength(); + break; + default: + throw new RuntimeException( + "error: _LEGACY_ERROR_TEMP_3190, typeName: " + + typeName.toString()); + } + return new VectorizedByteStreamSplitValuesReader(typeWidth); + } case RLE: { PrimitiveType.PrimitiveTypeName typeName = diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedPlainValuesReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedPlainValuesReader.java index 0db3dd7b0d4a..9d28ef2467d5 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedPlainValuesReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedPlainValuesReader.java @@ -128,6 +128,42 @@ public final void readIntegers(int total, WritableIntVector c, int rowId) { } } + @Override + public final void readIntegersAsLongs(int total, WritableLongVector c, int rowId) { + int requiredBytes = total * 4; + ByteBuffer buffer = getBuffer(requiredBytes); + for (int i = 0; i < total; i++) { + c.setLong(rowId + i, buffer.getInt()); + } + } + + @Override + public final void readIntegersAsDoubles(int total, WritableDoubleVector c, int rowId) { + int requiredBytes = total * 4; + ByteBuffer buffer = getBuffer(requiredBytes); + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, buffer.getInt()); + } + } + + @Override + public final void readFloatsAsDoubles(int total, WritableDoubleVector c, int rowId) { + int requiredBytes = total * 4; + ByteBuffer buffer = getBuffer(requiredBytes); + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, buffer.getFloat()); + } + } + + @Override + public final void readLongsAsInts(int total, WritableIntVector c, int rowId) { + int requiredBytes = total * 8; + ByteBuffer buffer = getBuffer(requiredBytes); + for (int i = 0; i < total; i++) { + c.setInt(rowId + i, (int) buffer.getLong()); + } + } + @Override public void skipIntegers(int total) { in.skip(total * 4L); @@ -316,6 +352,21 @@ public final Binary readBinary(int len) { } } + @Override + public final void readFixedLenByteArray(int total, int len, WritableBytesVector v, int rowId) { + for (int i = 0; i < total; i++) { + ByteBuffer buffer = getBuffer(len); + if (buffer.hasArray()) { + v.putByteArray( + rowId + i, buffer.array(), buffer.arrayOffset() + buffer.position(), len); + } else { + byte[] bytes = new byte[len]; + buffer.get(bytes); + v.putByteArray(rowId + i, bytes, 0, len); + } + } + } + @Override public void skipFixedLenByteArray(int total, int len) { in.skip(total * (long) len); diff --git a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedValuesReader.java b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedValuesReader.java index 0de7a5d1ba71..1a9d048080f0 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedValuesReader.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/VectorizedValuesReader.java @@ -67,8 +67,39 @@ public interface VectorizedValuesReader { void readDoubles(int total, WritableDoubleVector c, int rowId); + default void readIntegersAsLongs(int total, WritableLongVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setLong(rowId + i, readInteger()); + } + } + + default void readIntegersAsDoubles(int total, WritableDoubleVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, readInteger()); + } + } + + default void readFloatsAsDoubles(int total, WritableDoubleVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setDouble(rowId + i, readFloat()); + } + } + + default void readLongsAsInts(int total, WritableIntVector c, int rowId) { + for (int i = 0; i < total; i++) { + c.setInt(rowId + i, (int) readLong()); + } + } + void readBinary(int total, WritableBytesVector c, int rowId); + default void readFixedLenByteArray(int total, int len, WritableBytesVector c, int rowId) { + for (int i = 0; i < total; i++) { + byte[] bytes = readBinary(len).getBytesUnsafe(); + c.putByteArray(rowId + i, bytes, 0, bytes.length); + } + } + /* * Skips `total` values */ diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java index ccb793c6a68b..0150b497c603 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetReadWriteTest.java @@ -838,6 +838,78 @@ public void testReadBinaryWrittenByParquet() throws Exception { }); } + @Test + public void testReadByteStreamSplitWrittenByParquet() throws Exception { + Path path = new Path(folder.getPath(), UUID.randomUUID().toString()); + Configuration conf = new Configuration(); + MessageType schema = + new MessageType( + "origin-parquet", + Types.required(PrimitiveType.PrimitiveTypeName.FLOAT).named("f0").withId(0), + Types.required(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("f1") + .withId(1), + Types.optional(PrimitiveType.PrimitiveTypeName.FLOAT).named("f2").withId(2), + Types.optional(PrimitiveType.PrimitiveTypeName.DOUBLE) + .named("f3") + .withId(3)); + + int size = 8193; + try (ParquetWriter writer = + ExampleParquetWriter.builder( + HadoopOutputFile.fromPath( + new org.apache.hadoop.fs.Path(path.toString()), conf)) + .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) + .withConf(conf) + .withType(schema) + .withDictionaryEncoding(false) + .withByteStreamSplitEncoding(true) + .build()) { + SimpleGroupFactory groups = new SimpleGroupFactory(schema); + for (int i = 1; i <= size; i++) { + Group row = groups.newGroup().append("f0", i * 0.1f).append("f1", i * 0.01d); + if (i % 3 != 0) { + row.append("f2", i * 0.5f); + } + if (i % 5 != 0) { + row.append("f3", i * 0.25d); + } + writer.write(row); + } + } + + RowType rowType = + RowType.of(new FloatType(), new DoubleType(), new FloatType(), new DoubleType()); + ParquetReaderFactory format = new ParquetReaderFactory(new Options(), rowType, 500, null); + AtomicInteger count = new AtomicInteger(); + try (RecordReader reader = + format.createReader( + new FormatReaderContext( + new LocalFileIO(), + path, + new LocalFileIO().getFileSize(path), + null, + null))) { + reader.forEachRemaining( + row -> { + int i = count.incrementAndGet(); + assertThat(row.getFloat(0)).isEqualTo(i * 0.1f); + assertThat(row.getDouble(1)).isEqualTo(i * 0.01d); + if (i % 3 == 0) { + assertThat(row.isNullAt(2)).isTrue(); + } else { + assertThat(row.getFloat(2)).isEqualTo(i * 0.5f); + } + if (i % 5 == 0) { + assertThat(row.isNullAt(3)).isTrue(); + } else { + assertThat(row.getDouble(3)).isEqualTo(i * 0.25d); + } + }); + } + assertThat(count.get()).isEqualTo(size); + } + @Test public void testReadTimestampNanosWrittenByParquet() throws Exception { Path path = new Path(folder.getPath(), UUID.randomUUID().toString()); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReaderTest.java b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReaderTest.java new file mode 100644 index 000000000000..5794269bd579 --- /dev/null +++ b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/VectorizedByteStreamSplitValuesReaderTest.java @@ -0,0 +1,195 @@ +/* + * 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.paimon.format.parquet.reader; + +import org.apache.paimon.data.columnar.heap.HeapBytesVector; +import org.apache.paimon.data.columnar.heap.HeapDoubleVector; +import org.apache.paimon.data.columnar.heap.HeapFloatVector; +import org.apache.paimon.data.columnar.heap.HeapIntVector; +import org.apache.paimon.data.columnar.heap.HeapLongVector; + +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link VectorizedByteStreamSplitValuesReader}. */ +public class VectorizedByteStreamSplitValuesReaderTest { + + @Test + public void testIntegers() throws IOException { + int[] values = new int[] {Integer.MIN_VALUE, -1, 0, 42, Integer.MAX_VALUE}; + VectorizedByteStreamSplitValuesReader reader = reader(4, encodeIntegers(values)); + + assertThat(reader.readInteger()).isEqualTo(values[0]); + reader.skipIntegers(1); + HeapIntVector vector = new HeapIntVector(3); + reader.readIntegers(3, vector, 0); + assertThat(vector.getInt(0)).isEqualTo(values[2]); + assertThat(vector.getInt(1)).isEqualTo(values[3]); + assertThat(vector.getInt(2)).isEqualTo(values[4]); + + reader = reader(4, encodeIntegers(values)); + HeapLongVector longVector = new HeapLongVector(values.length); + reader.readIntegersAsLongs(values.length, longVector, 0); + for (int i = 0; i < values.length; i++) { + assertThat(longVector.getLong(i)).isEqualTo(values[i]); + } + } + + @Test + public void testLongs() throws IOException { + long[] values = new long[] {Long.MIN_VALUE, -1L, 0L, 42L, Long.MAX_VALUE}; + VectorizedByteStreamSplitValuesReader reader = reader(8, encodeLongs(values)); + + assertThat(reader.readLong()).isEqualTo(values[0]); + reader.skipLongs(1); + HeapLongVector vector = new HeapLongVector(3); + reader.readLongs(3, vector, 0); + assertThat(vector.getLong(0)).isEqualTo(values[2]); + assertThat(vector.getLong(1)).isEqualTo(values[3]); + assertThat(vector.getLong(2)).isEqualTo(values[4]); + } + + @Test + public void testFloats() throws IOException { + float[] values = + new float[] { + Float.NEGATIVE_INFINITY, + -0.0f, + 1.25f, + Float.POSITIVE_INFINITY, + Float.intBitsToFloat(0x7fc00001) + }; + VectorizedByteStreamSplitValuesReader reader = reader(4, encodeFloats(values)); + HeapFloatVector vector = new HeapFloatVector(values.length); + reader.readFloats(values.length, vector, 0); + for (int i = 0; i < values.length; i++) { + assertThat(Float.floatToRawIntBits(vector.getFloat(i))) + .isEqualTo(Float.floatToRawIntBits(values[i])); + } + + reader = reader(4, encodeFloats(values)); + HeapDoubleVector doubleVector = new HeapDoubleVector(values.length); + reader.readFloatsAsDoubles(values.length, doubleVector, 0); + for (int i = 0; i < values.length; i++) { + assertThat(Double.doubleToLongBits(doubleVector.getDouble(i))) + .isEqualTo(Double.doubleToLongBits((double) values[i])); + } + } + + @Test + public void testDoubles() throws IOException { + double[] values = + new double[] { + Double.NEGATIVE_INFINITY, + -0.0d, + 1.25d, + Double.POSITIVE_INFINITY, + Double.longBitsToDouble(0x7ff8000000000001L) + }; + VectorizedByteStreamSplitValuesReader reader = reader(8, encodeDoubles(values)); + HeapDoubleVector vector = new HeapDoubleVector(values.length); + reader.readDoubles(values.length, vector, 0); + for (int i = 0; i < values.length; i++) { + assertThat(Double.doubleToRawLongBits(vector.getDouble(i))) + .isEqualTo(Double.doubleToRawLongBits(values[i])); + } + } + + @Test + public void testFixedLenByteArrays() throws IOException { + byte[][] values = + new byte[][] { + new byte[] {0, 1, 2}, + new byte[] {3, 4, 5}, + new byte[] {6, 7, 8}, + new byte[] {9, 10, 11} + }; + VectorizedByteStreamSplitValuesReader reader = reader(3, encode(values)); + + assertThat(reader.readBinary(3).getBytes()).isEqualTo(values[0]); + reader.skipFixedLenByteArray(1, 3); + HeapBytesVector vector = new HeapBytesVector(2); + reader.readFixedLenByteArray(2, 3, vector, 0); + assertThat(vector.getBytes(0).getBytes()).isEqualTo(values[2]); + assertThat(vector.getBytes(1).getBytes()).isEqualTo(values[3]); + } + + private static VectorizedByteStreamSplitValuesReader reader(int width, byte[] encoded) + throws IOException { + VectorizedByteStreamSplitValuesReader reader = + new VectorizedByteStreamSplitValuesReader(width); + reader.initFromPage( + encoded.length / width, ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded))); + return reader; + } + + private static byte[] encodeIntegers(int[] values) { + byte[][] bytes = new byte[values.length][]; + for (int i = 0; i < values.length; i++) { + bytes[i] = + ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN).putInt(values[i]).array(); + } + return encode(bytes); + } + + private static byte[] encodeLongs(long[] values) { + byte[][] bytes = new byte[values.length][]; + for (int i = 0; i < values.length; i++) { + bytes[i] = + ByteBuffer.allocate(8) + .order(ByteOrder.LITTLE_ENDIAN) + .putLong(values[i]) + .array(); + } + return encode(bytes); + } + + private static byte[] encodeFloats(float[] values) { + int[] bits = new int[values.length]; + for (int i = 0; i < values.length; i++) { + bits[i] = Float.floatToRawIntBits(values[i]); + } + return encodeIntegers(bits); + } + + private static byte[] encodeDoubles(double[] values) { + long[] bits = new long[values.length]; + for (int i = 0; i < values.length; i++) { + bits[i] = Double.doubleToRawLongBits(values[i]); + } + return encodeLongs(bits); + } + + private static byte[] encode(byte[][] values) { + int width = values[0].length; + byte[] encoded = new byte[width * values.length]; + for (int b = 0; b < width; b++) { + for (int i = 0; i < values.length; i++) { + encoded[b * values.length + i] = values[i][b]; + } + } + return encoded; + } +}