You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "jbewing (via GitHub)" <gi...@apache.org> on 2026/02/20 04:08:56 UTC
[PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
jbewing opened a new pull request, #15373:
URL: https://github.com/apache/iceberg/pull/15373
⚠️ ⚠️ This PR depends on https://github.com/apache/iceberg/pull/15362 being merged first and this PR being rebased on top of that ⚠️ ⚠️
### What
This PR adds vectorized read support to Iceberg for [BYTE_STREAM_SPLIT encoding](https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9) of the Apache Parquet v2 specification (see https://github.com/apache/iceberg/issues/7162). This builds on top of the existing support for reading [DELTA_BINARY_PACKED](https://parquet.apache.org/docs/file-format/data-pages/encodings/#delta-encoding-delta_binary_packed--5) implemented by @eric-maynard in https://github.com/apache/iceberg/pull/13391 with:
- Implementing vectorized read support for [BYTE_STREAM_SPLIT encoding](https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9)
- Bolstering golden files test coverage to cover each of the paths above. In addition, I added golden file tests that include rows with null values for each data type as well to ensure our handling of those is correct
~~This is split out from https://github.com/apache/iceberg/pull/14800 to make reviewing the changes easier / facilitate tighter feedback cycles per some feedback on the [Iceberg Developer Slack Community](https://apache-iceberg.slack.com/archives/C03LG1D563F/p1765466132290829).~~
This is built on top of the v2 resurrection of the vectorized parquet read effort namely https://github.com/apache/iceberg/pull/15362. This in particular revives https://github.com/apache/iceberg/pull/14852 which implements BYTE_STREAM_SPLIT vectorized read support for float/double data types (which are the defaults for the parquet v2 writer. Additional data types are supported like int, long, and fixed length byte array, but support for those is forthcoming in this PR although not crazy to add).
### Background
This solves a longstanding issue of: the reference Apache Iceberg Spark implementation with the default settings enabled (e.g. `spark.sql.iceberg.vectorization.enabled` = `true`) isn't able to read iceberg tables that may have been written by other compute engines that utilize the not so new anymore Apache Parquet v2 writer specification. It's a widely known workaround to need to disable the vectorized reader in Spark if you need to interop with other compute engines or adjust all compute engines to use the Apache Parquet v1 writer specification when writing parquet files. With disabling the vectorization flag, clients take a performance hit that we've anecdotally measured is _quite_ large in some cases/workloads. If forcing all writers of an iceberg table to write in Apache Parquet v1 format, clients are incurring additional performance and storage penalties (files written with parquet v2 tend to be smaller than those written with the v1 spec as the new encodings tend to save spa
ce and are often faster to read/write). So really, it's a lose-lose for performance & data size in the current setup with the additional papercut of Apache Iceberg not being super portable across engines in it's default setup. This PR seeks to solve that by finishing the swing on implementing vectorized parquet read support for the v2 format. In the future, we may also consider allowing clients to write Apache Parquet v2 files natively gated via a setting from Apache Iceberg. Even longer down that road, we may even consider changing that to be the "default" setting.
### Previous Work / Thanks
This PR is a revival + extension to the work that @eric-maynard was doing in https://github.com/apache/iceberg/pull/13709. That PR had been active for a little while, so I literally started from where Eric left off. Thank you for the great work here @eric-maynard, you made implementing the rest of the changes required for vectorized read support _way_ easier!
### Note to Reviewers
This is a split of the `BYTE_SPLIT_STREAM` encoding support from https://github.com/apache/iceberg/pull/14800, and revived from https://github.com/apache/iceberg/pull/14852 & based off of https://github.com/apache/iceberg/pull/15362.
### Testing
I've tested this on a fork of Spark 3.5 & Iceberg 1.10.0 and verified that a Spark job is able to read a table written with Parquet V2 writer without issues.
Issue: https://github.com/apache/iceberg/issues/7162
Split out from: https://github.com/apache/iceberg/pull/14800, https://github.com/apache/iceberg/pull/15362
Successor To: https://github.com/apache/iceberg/pull/14852
cc @nastra
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2835939509
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
Review Comment:
After looking at the Parquet implementation, I don't think it's going to add significant complexity so I would prefer to add the other types in this PR. The other reason for having it in this PR is that we would still have a gap if those types are being encoded and we'd have to fix it anyway but I'd like to add everything into the same release
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831604958
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
Review Comment:
```suggestion
this.dataStream = in;
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra merged PR #15373:
URL: https://github.com/apache/iceberg/pull/15373
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831624644
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
Review Comment:
```suggestion
public void initFromPage(int ignoredValueCount, ByteBufferInputStream in) {
```
there's no exception being thrown here
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2850865943
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java:
##########
@@ -86,15 +86,19 @@ public class TestParquetVectorizedReads extends AvroDataTestBase {
"RLE_DICTIONARY",
"DELTA_BINARY_PACKED",
"DELTA_LENGTH_BYTE_ARRAY",
- "DELTA_BYTE_ARRAY");
+ "DELTA_BYTE_ARRAY",
+ "BYTE_STREAM_SPLIT");
private static final Map<String, PrimitiveType> GOLDEN_FILE_TYPES =
- ImmutableMap.of(
- "string", Types.StringType.get(),
- "float", Types.FloatType.get(),
- "int32", Types.IntegerType.get(),
- "int64", Types.LongType.get(),
- "binary", Types.BinaryType.get(),
- "boolean", Types.BooleanType.get());
+ ImmutableMap.<String, PrimitiveType>builder()
Review Comment:
The golden test uses `typeName + ".parquet"` (no _with_nulls). Are the `*_with_nulls.parquet` fixtures used anywhere?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2849358202
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private final int elementSizeInBytes;
+ private int totalBytesInStream;
+ private ByteBufferInputStream dataStream;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader(int elementSizeInBytes) {
+ this.elementSizeInBytes = elementSizeInBytes;
+ }
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream in) {
+ this.totalBytesInStream = in.available();
+ this.dataStream = in;
+ }
+
+ @Override
+ public int readInteger() {
+ ensureDecoded();
+ return decodedDataStream.getInt();
+ }
+
+ @Override
+ public long readLong() {
+ ensureDecoded();
+ return decodedDataStream.getLong();
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded();
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded();
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public Binary readBinary(int len) {
+ ensureDecoded();
+ byte[] bytes = new byte[len];
+ decodedDataStream.get(bytes);
+ return Binary.fromConstantByteArray(bytes);
+ }
+
+ @Override
+ public void readIntegers(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readLongs(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int total, FieldVector vec, int rowId) {
+ ensureDecoded();
+ int bytesToRead = total * elementSizeInBytes;
+ long destOffset = (long) rowId * elementSizeInBytes;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded() {
+ if (decodedDataStream == null) {
+ this.decodedDataStream = decode(totalBytesInStream / elementSizeInBytes);
+ }
+ }
+
+ private ByteBuffer decode(int valuesCount) {
+ ByteBuffer encoded;
+ try {
+ encoded = dataStream.slice(totalBytesInStream).slice();
+ } catch (EOFException e) {
+ throw new UncheckedIOException("Failed to read bytes from stream", e);
+ }
+ byte[] decoded = new byte[encoded.limit()];
+ int destByteIndex = 0;
+ for (int srcValueIndex = 0; srcValueIndex < valuesCount; srcValueIndex++) {
+ for (int stream = 0; stream < elementSizeInBytes; stream++, destByteIndex++) {
+ decoded[destByteIndex] = encoded.get(srcValueIndex + stream * valuesCount);
+ }
+ }
+ return ByteBuffer.wrap(decoded).order(ByteOrder.LITTLE_ENDIAN);
Review Comment:
do we wanna add checkstate or assertions here like https://github.com/apache/parquet-java/blob/master/parquet-column/src/main/java/org/apache/parquet/column/values/bytestreamsplit/ByteStreamSplitValuesReader.java#L62
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#issuecomment-3960420836
thanks @jbewing for the fix and thanks to @singhpk234 and @huaxingao for the additional reviews
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831595453
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
Review Comment:
I think this also needs to implement `readInteger()` and `readBinary()` because that encoding supports `
INT32, INT64, FLOAT, DOUBLE, FIXED_LEN_BYTE_ARRAY`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831602521
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
Review Comment:
```suggestion
public void initFromPage(int ignoredValueCount, ByteBufferInputStream in)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831616357
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded(DOUBLE_SIZE);
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(FLOAT_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(DOUBLE_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int typeWidth, int total, FieldVector vec, int rowId) {
+ ensureDecoded(typeWidth);
+ int bytesToRead = total * typeWidth;
+ long destOffset = 1L * rowId * typeWidth;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded(int elementSizeInBytes) {
+ if (decodedDataStream == null) {
+ decodedDataStream = decode(totalBytesInStream / elementSizeInBytes, elementSizeInBytes);
Review Comment:
```suggestion
this.decodedDataStream = decode(totalBytesInStream / elementSizeInBytes, elementSizeInBytes);
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831603919
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
Review Comment:
```suggestion
this.totalBytesInStream = inputStream.available();
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "huaxingao (via GitHub)" <gi...@apache.org>.
huaxingao commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2854002641
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java:
##########
@@ -86,15 +86,19 @@ public class TestParquetVectorizedReads extends AvroDataTestBase {
"RLE_DICTIONARY",
"DELTA_BINARY_PACKED",
"DELTA_LENGTH_BYTE_ARRAY",
- "DELTA_BYTE_ARRAY");
+ "DELTA_BYTE_ARRAY",
+ "BYTE_STREAM_SPLIT");
private static final Map<String, PrimitiveType> GOLDEN_FILE_TYPES =
- ImmutableMap.of(
- "string", Types.StringType.get(),
- "float", Types.FloatType.get(),
- "int32", Types.IntegerType.get(),
- "int64", Types.LongType.get(),
- "binary", Types.BinaryType.get(),
- "boolean", Types.BooleanType.get());
+ ImmutableMap.<String, PrimitiveType>builder()
Review Comment:
Thanks for the pointer: I missed that one.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#issuecomment-3937445242
👋 @nastra, I've addressed all your feedback in https://github.com/apache/iceberg/pull/15373/commits/8cf3f3e8ab03b844019b7671e8ba8ac82b79063e.
You can hold off on reviewing for now though as I'm going to rebase this on master once the dependent PR gets in, and then things will be ready for a re-review.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2838953816
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
Review Comment:
I too agree that it's not of significant complexity. I do plan on punting it to a follow up PR though just to keep these small and focused. My intention with these PRs was to support vectorized reads over Parquet v2 writers. This scope creeps into Parquet v2.1 which isn't as well adopted on the writer side. I'll tag you into that PR for review as well
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "singhpk234 (via GitHub)" <gi...@apache.org>.
singhpk234 commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2850631855
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private final int elementSizeInBytes;
+ private int totalBytesInStream;
+ private ByteBufferInputStream dataStream;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader(int elementSizeInBytes) {
+ this.elementSizeInBytes = elementSizeInBytes;
+ }
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream in) {
+ this.totalBytesInStream = in.available();
+ this.dataStream = in;
+ }
+
+ @Override
+ public int readInteger() {
+ ensureDecoded();
+ return decodedDataStream.getInt();
+ }
+
+ @Override
+ public long readLong() {
+ ensureDecoded();
+ return decodedDataStream.getLong();
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded();
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded();
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public Binary readBinary(int len) {
+ ensureDecoded();
+ byte[] bytes = new byte[len];
+ decodedDataStream.get(bytes);
+ return Binary.fromConstantByteArray(bytes);
+ }
+
+ @Override
+ public void readIntegers(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readLongs(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int total, FieldVector vec, int rowId) {
+ ensureDecoded();
+ int bytesToRead = total * elementSizeInBytes;
+ long destOffset = (long) rowId * elementSizeInBytes;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded() {
+ if (decodedDataStream == null) {
+ this.decodedDataStream = decode(totalBytesInStream / elementSizeInBytes);
+ }
+ }
+
+ private ByteBuffer decode(int valuesCount) {
+ ByteBuffer encoded;
+ try {
+ encoded = dataStream.slice(totalBytesInStream).slice();
+ } catch (EOFException e) {
+ throw new UncheckedIOException("Failed to read bytes from stream", e);
+ }
+ byte[] decoded = new byte[encoded.limit()];
+ int destByteIndex = 0;
+ for (int srcValueIndex = 0; srcValueIndex < valuesCount; srcValueIndex++) {
+ for (int stream = 0; stream < elementSizeInBytes; stream++, destByteIndex++) {
+ decoded[destByteIndex] = encoded.get(srcValueIndex + stream * valuesCount);
+ }
+ }
+ return ByteBuffer.wrap(decoded).order(ByteOrder.LITTLE_ENDIAN);
Review Comment:
may be add a Preconditions.checkState or something
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2835259813
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
Review Comment:
Mind if I get to that in a follow up? I noted this in the PR description:
> This in particular revives https://github.com/apache/iceberg/pull/14852 which implements BYTE_STREAM_SPLIT vectorized read support for float/double data types (which are the defaults for the parquet v2 writer. Additional data types are supported like int, long, and fixed length byte array, but support for those is forthcoming in this PR although not crazy to add).
But the int, long, and fixed length binary array support are supported in the parquet v2 spec, but they were adopted for this encoding a bit later (see [PARQUET-2414](https://issues.apache.org/jira/browse/PARQUET-2414)).
It also remains the case that BYTE_STREAM_SPLIT is the default encoding used for floating point types, however, not for int, long, or fixed length byte array at this point (so the writer intentionally has to opt in even if they're on parquet v2).
All to say: you're totally right, but I'd like to punt to a follow up as just supporting float + double gets us in the door for parquet v2 support
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2853316851
##########
spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java:
##########
@@ -86,15 +86,19 @@ public class TestParquetVectorizedReads extends AvroDataTestBase {
"RLE_DICTIONARY",
"DELTA_BINARY_PACKED",
"DELTA_LENGTH_BYTE_ARRAY",
- "DELTA_BYTE_ARRAY");
+ "DELTA_BYTE_ARRAY",
+ "BYTE_STREAM_SPLIT");
private static final Map<String, PrimitiveType> GOLDEN_FILE_TYPES =
- ImmutableMap.of(
- "string", Types.StringType.get(),
- "float", Types.FloatType.get(),
- "int32", Types.IntegerType.get(),
- "int64", Types.LongType.get(),
- "binary", Types.BinaryType.get(),
- "boolean", Types.BooleanType.get());
+ ImmutableMap.<String, PrimitiveType>builder()
Review Comment:
Yup! All golden files are expected to have a "_with_nulls" variant that gets picked up here: https://github.com/apache/iceberg/blob/d2fbe427ecec298f1600260f40fcab1b7ab2e695/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetVectorizedReads.java#L501-L510
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2850616822
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private final int elementSizeInBytes;
+ private int totalBytesInStream;
+ private ByteBufferInputStream dataStream;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader(int elementSizeInBytes) {
+ this.elementSizeInBytes = elementSizeInBytes;
+ }
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream in) {
+ this.totalBytesInStream = in.available();
+ this.dataStream = in;
+ }
+
+ @Override
+ public int readInteger() {
+ ensureDecoded();
+ return decodedDataStream.getInt();
+ }
+
+ @Override
+ public long readLong() {
+ ensureDecoded();
+ return decodedDataStream.getLong();
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded();
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded();
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public Binary readBinary(int len) {
+ ensureDecoded();
+ byte[] bytes = new byte[len];
+ decodedDataStream.get(bytes);
+ return Binary.fromConstantByteArray(bytes);
+ }
+
+ @Override
+ public void readIntegers(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readLongs(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int total, FieldVector vec, int rowId) {
+ ensureDecoded();
+ int bytesToRead = total * elementSizeInBytes;
+ long destOffset = (long) rowId * elementSizeInBytes;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded() {
+ if (decodedDataStream == null) {
+ this.decodedDataStream = decode(totalBytesInStream / elementSizeInBytes);
+ }
+ }
+
+ private ByteBuffer decode(int valuesCount) {
+ ByteBuffer encoded;
+ try {
+ encoded = dataStream.slice(totalBytesInStream).slice();
+ } catch (EOFException e) {
+ throw new UncheckedIOException("Failed to read bytes from stream", e);
+ }
+ byte[] decoded = new byte[encoded.limit()];
+ int destByteIndex = 0;
+ for (int srcValueIndex = 0; srcValueIndex < valuesCount; srcValueIndex++) {
+ for (int stream = 0; stream < elementSizeInBytes; stream++, destByteIndex++) {
+ decoded[destByteIndex] = encoded.get(srcValueIndex + stream * valuesCount);
+ }
+ }
+ return ByteBuffer.wrap(decoded).order(ByteOrder.LITTLE_ENDIAN);
Review Comment:
Good callout there. To my knowledge though, this would be the _first_ usage of the Java assert keyword in this codebase and I'm not seeing any compile flags or run modes that enables these at runtime which makes me think that while it may be useful for documenting invariants, it would be effectively dead code.
Here's the ripgrep query that I ran from the iceberg root:
```bash
❯ rg -n --pcre2 --type java -C 3 '^(?!\s*//)(?!\s*\*)(?:[^/]|/(?!/))*\bassert\s' .
./spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
814- sql(
815- "SELECT id, `naMe`, dept, subdept from %s WHERE `naMe` = 'John Doe' ORDER BY id",
816- sourceTableName))
817: .as("If this assert breaks it means that Spark has fixed the pushdown issue")
818- .isEmpty();
819-
820- // Pushdown works for iceberg
./spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
812- sql(
813- "SELECT id, `naMe`, dept, subdept from %s WHERE `naMe` = 'John Doe' ORDER BY id",
814- sourceTableName))
815: .as("If this assert breaks it means that Spark has fixed the pushdown issue")
816- .isEmpty();
817-
818- // Pushdown works for iceberg
./spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
812- sql(
813- "SELECT id, `naMe`, dept, subdept from %s WHERE `naMe` = 'John Doe' ORDER BY id",
814- sourceTableName))
815: .as("If this assert breaks it means that Spark has fixed the pushdown issue")
816- .isEmpty();
817-
818- // Pushdown works for iceberg
./spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAddFilesProcedure.java
814- sql(
815- "SELECT id, `naMe`, dept, subdept from %s WHERE `naMe` = 'John Doe' ORDER BY id",
816- sourceTableName))
817: .as("If this assert breaks it means that Spark has fixed the pushdown issue")
818- .isEmpty();
819-
820- // Pushdown works for iceberg
```
Given that we can't really get any valuable results from the keyword, the runtime checking here would strictly introduce a performance regression.
All to say: if you think it's valuable enough to include those, we can. It will come with runtime cost though and what we have here isn't less safe than what the parquet-java impl has since I can't imagine anyone will be running with `-ea` in the wild
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831602521
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
Review Comment:
```suggestion
public void initFromPage(int ignoredValueCount, ByteBufferInputStream in)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#issuecomment-3942419079
I've updated this PR from main @nastra. I'm working on the follow up for the parquet v2.1 support that you pointed out and also getting the boolean RLE PR updated against the latest master (which yields _full_ parquet v2.0 writer support).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831629876
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded(DOUBLE_SIZE);
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(FLOAT_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(DOUBLE_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int typeWidth, int total, FieldVector vec, int rowId) {
+ ensureDecoded(typeWidth);
+ int bytesToRead = total * typeWidth;
+ long destOffset = 1L * rowId * typeWidth;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded(int elementSizeInBytes) {
+ if (decodedDataStream == null) {
+ decodedDataStream = decode(totalBytesInStream / elementSizeInBytes, elementSizeInBytes);
+ }
+ }
+
+ private ByteBuffer decode(int valuesCount, int elementSizeInBytes) {
+ ByteBuffer encoded;
+ try {
+ encoded = in.slice(totalBytesInStream).slice();
+ } catch (EOFException e) {
+ throw new UncheckedIOException("Failed to read bytes from stream", e);
+ }
+ byte[] decoded = new byte[encoded.limit()];
+ int destByteIndex = 0;
+ for (int srcValueIndex = 0; srcValueIndex < valuesCount; ++srcValueIndex) {
Review Comment:
why not `srcValueIndex++` here and in the other for loop?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "nastra (via GitHub)" <gi...@apache.org>.
nastra commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2831632936
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private int totalBytesInStream;
+ private ByteBufferInputStream in;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader() {}
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream inputStream)
+ throws IOException {
+ totalBytesInStream = inputStream.available();
+ this.in = inputStream;
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded(FLOAT_SIZE);
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded(DOUBLE_SIZE);
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(FLOAT_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(DOUBLE_SIZE, total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int typeWidth, int total, FieldVector vec, int rowId) {
+ ensureDecoded(typeWidth);
+ int bytesToRead = total * typeWidth;
+ long destOffset = 1L * rowId * typeWidth;
Review Comment:
```suggestion
long destOffset = (long) rowId * typeWidth;
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark, Arrow, Parquet: Add vectorized read support for parquet BYTE_STREAM_SPLIT encoding [iceberg]
Posted by "jbewing (via GitHub)" <gi...@apache.org>.
jbewing commented on code in PR #15373:
URL: https://github.com/apache/iceberg/pull/15373#discussion_r2850711724
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedByteStreamSplitValuesReader.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.iceberg.arrow.vectorized.parquet;
+
+import java.io.EOFException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.api.Binary;
+
+/**
+ * A {@link VectorizedValuesReader} implementation for the encoding type BYTE_STREAM_SPLIT. This is
+ * adapted from Parquet's ByteStreamSplitValuesReader.
+ *
+ * @see <a
+ * href="https://parquet.apache.org/docs/file-format/data-pages/encodings/#byte-stream-split-byte_stream_split--9">
+ * Parquet format encodings: BYTE_STREAM_SPLIT</a>
+ */
+public class VectorizedByteStreamSplitValuesReader extends ValuesReader
+ implements VectorizedValuesReader {
+
+ private final int elementSizeInBytes;
+ private int totalBytesInStream;
+ private ByteBufferInputStream dataStream;
+ private ByteBuffer decodedDataStream;
+
+ public VectorizedByteStreamSplitValuesReader(int elementSizeInBytes) {
+ this.elementSizeInBytes = elementSizeInBytes;
+ }
+
+ @Override
+ public void initFromPage(int ignoredValueCount, ByteBufferInputStream in) {
+ this.totalBytesInStream = in.available();
+ this.dataStream = in;
+ }
+
+ @Override
+ public int readInteger() {
+ ensureDecoded();
+ return decodedDataStream.getInt();
+ }
+
+ @Override
+ public long readLong() {
+ ensureDecoded();
+ return decodedDataStream.getLong();
+ }
+
+ @Override
+ public float readFloat() {
+ ensureDecoded();
+ return decodedDataStream.getFloat();
+ }
+
+ @Override
+ public double readDouble() {
+ ensureDecoded();
+ return decodedDataStream.getDouble();
+ }
+
+ @Override
+ public Binary readBinary(int len) {
+ ensureDecoded();
+ byte[] bytes = new byte[len];
+ decodedDataStream.get(bytes);
+ return Binary.fromConstantByteArray(bytes);
+ }
+
+ @Override
+ public void readIntegers(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readLongs(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readFloats(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void readDoubles(int total, FieldVector vec, int rowId) {
+ readBatch(total, vec, rowId);
+ }
+
+ @Override
+ public void skip() {
+ throw new UnsupportedOperationException("skip is not supported");
+ }
+
+ private void readBatch(int total, FieldVector vec, int rowId) {
+ ensureDecoded();
+ int bytesToRead = total * elementSizeInBytes;
+ long destOffset = (long) rowId * elementSizeInBytes;
+ ByteBuffer slice = decodedDataStream.slice();
+ slice.limit(bytesToRead);
+ vec.getDataBuffer().setBytes(destOffset, slice);
+ decodedDataStream.position(decodedDataStream.position() + bytesToRead);
+ }
+
+ private void ensureDecoded() {
+ if (decodedDataStream == null) {
+ this.decodedDataStream = decode(totalBytesInStream / elementSizeInBytes);
+ }
+ }
+
+ private ByteBuffer decode(int valuesCount) {
+ ByteBuffer encoded;
+ try {
+ encoded = dataStream.slice(totalBytesInStream).slice();
+ } catch (EOFException e) {
+ throw new UncheckedIOException("Failed to read bytes from stream", e);
+ }
+ byte[] decoded = new byte[encoded.limit()];
+ int destByteIndex = 0;
+ for (int srcValueIndex = 0; srcValueIndex < valuesCount; srcValueIndex++) {
+ for (int stream = 0; stream < elementSizeInBytes; stream++, destByteIndex++) {
+ decoded[destByteIndex] = encoded.get(srcValueIndex + stream * valuesCount);
+ }
+ }
+ return ByteBuffer.wrap(decoded).order(ByteOrder.LITTLE_ENDIAN);
Review Comment:
Done in https://github.com/apache/iceberg/pull/15373/commits/90f5be1000caaa4023c92d09f18a2a9c9ae1f450. I believe that this single check effectively logically covers the rest
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org