This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 83b2eec81d [format] Introduce support for Parquet byte stream split
encoding (#9359)
83b2eec81d is described below
commit 83b2eec81dd820a3529c410d83d3d387c909782f
Author: Arnav Balyan <[email protected]>
AuthorDate: Sun Aug 23 20:48:59 2026 +0530
[format] Introduce support for Parquet byte stream split encoding (#9359)
---
.../reader/ParquetVectorUpdaterFactory.java | 12 +-
.../VectorizedByteStreamSplitValuesReader.java | 254 +++++++++++++++++++++
.../parquet/reader/VectorizedColumnReader.java | 24 ++
.../reader/VectorizedPlainValuesReader.java | 51 +++++
.../parquet/reader/VectorizedValuesReader.java | 31 +++
.../format/parquet/ParquetReadWriteTest.java | 72 ++++++
.../VectorizedByteStreamSplitValuesReaderTest.java | 195 ++++++++++++++++
7 files changed, 630 insertions(+), 9 deletions(-)
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 c0445b3f13..d1ef82933a 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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 class ParquetVectorUpdaterFactory {
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 0000000000..c3d181905e
--- /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 53b11138da..2b4e4d769d 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 @@ public class VectorizedColumnReader {
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 0db3dd7b0d..9d28ef2467 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 class VectorizedPlainValuesReader extends
ValuesReader implements Vectori
}
}
+ @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 class VectorizedPlainValuesReader extends
ValuesReader implements Vectori
}
}
+ @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 0de7a5d1ba..1a9d048080 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 9d471df92b..fb03ac3778 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
@@ -900,6 +900,78 @@ public class ParquetReadWriteTest {
});
}
+ @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<Group> 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<InternalRow> 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 0000000000..5794269bd5
--- /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;
+ }
+}
