parthchandra commented on a change in pull request #34471: URL: https://github.com/apache/spark/pull/34471#discussion_r751550718
########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java ########## @@ -0,0 +1,318 @@ +/* + * 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.spark.sql.execution.datasources.parquet; + +import java.io.IOException; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.Arrays; + +import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.column.values.bitpacking.BytePackerForLong; +import org.apache.parquet.column.values.bitpacking.Packer; +import org.apache.parquet.io.ParquetDecodingException; +import org.apache.spark.sql.catalyst.util.RebaseDateTime; +import org.apache.spark.sql.execution.datasources.DataSourceUtils; +import org.apache.spark.sql.execution.vectorized.WritableColumnVector; + + +/** + * An implementation of the Parquet DELTA_BINARY_PACKED decoder that supports the vectorized + * interface. DELTA_BINARY_PACKED is a delta encoding for integer and long types that stores values + * as a delta between consecutive values. Delta values are themselves bit packed. Similar to RLE but + * is more effective in the case of large variation of values in the encoded column. <br/> + * DELTA_BINARY_PACKED is the default encoding for integer and long columns in Parquet V2. <br/> + * Supported Types: INT32, INT64 <br/> + * + * @see <a href="https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-encoding-delta_binary_packed--5";> + * Parquet format encodings: DELTA_BINARY_PACKED</a> + */ +public class VectorizedDeltaBinaryPackedReader extends VectorizedReaderBase { + + // header data + private int blockSizeInValues; + private int miniBlockNumInABlock; + private int totalValueCount; + private long firstValue; + + private int miniBlockSizeInValues; + + // values read by the caller + private int valuesRead = 0; + + // variables to keep state of the current block and miniblock + private long lastValueRead; // needed to compute the next value + private long minDeltaInCurrentBlock; // needed to compute the next value + private int currentMiniBlock = 0; // keep track of the mini block within the current block that we + // we read and decoded most recently. Only used as an index into + // bitWidths array + private int[] bitWidths; // bit widths for each miniBlock in the current block + private int remainingInBlock = 0; // values in current block still to be read + private int remainingInMiniBlock = 0; // values in current mini block still to be read + private long[] unpackedValuesBuffer; + + private ByteBufferInputStream in; + + // temporary buffers used by readByte, readShort, readInteger, and readLong + byte byteVal; + short shortVal; + int intVal; + long longVal; + + @SuppressWarnings("unused") Review comment: Removed ########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java ########## @@ -0,0 +1,319 @@ +/* + * 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.spark.sql.execution.datasources.parquet; + +import java.nio.ByteBuffer; +import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.column.values.ValuesReader; +import org.apache.parquet.column.values.bitpacking.BytePackerForLong; +import org.apache.parquet.column.values.bitpacking.Packer; +import org.apache.parquet.io.ParquetDecodingException; +import org.apache.parquet.io.api.Binary; +import org.apache.spark.sql.execution.vectorized.WritableColumnVector; + +import java.io.IOException; + +/** + * An implementation of the Parquet DELTA_BINARY_PACKED decoder that supports the vectorized + * interface. + */ +public class VectorizedDeltaBinaryPackedReader extends ValuesReader + implements VectorizedValuesReader { + + // header data + private int blockSizeInValues; + private int miniBlockNumInABlock; + private int totalValueCount; + private long firstValue; + + private int miniBlockSizeInValues; + + // values read by the caller + private int valuesRead = 0; + + //variables to keep state of the current block and miniblock + private long lastValueRead; + private long minDeltaInCurrentBlock; + private int currentMiniBlock = 0; + private int[] bitWidths; // bit widths for each miniblock in the current block + private int remainingInBlock = 0; // values in current block still to be read + private int remainingInMiniBlock = 0; // values in current mini block still to be read + private long[] unpackedValuesBuffer; + + private ByteBufferInputStream in; + + @SuppressWarnings("unused") + @Override + public void initFromPage(/*unused*/int valueCount, ByteBufferInputStream in) throws IOException { + Preconditions.checkArgument(valueCount >= 1, + "Page must have at least one value, but it has " + valueCount); + this.in = in; + + // Read the header + this.blockSizeInValues = BytesUtils.readUnsignedVarInt(in); + this.miniBlockNumInABlock = BytesUtils.readUnsignedVarInt(in); + double miniSize = (double) blockSizeInValues / miniBlockNumInABlock; + Preconditions.checkArgument(miniSize % 8 == 0, + "miniBlockSize must be multiple of 8, but it's " + miniSize); + this.miniBlockSizeInValues = (int) miniSize; + this.totalValueCount = BytesUtils.readUnsignedVarInt(in); + this.bitWidths = new int[miniBlockNumInABlock]; + + // read the first value + firstValue = BytesUtils.readZigZagVarLong(in); + + } + + @Override + public void skip() { + throw new UnsupportedOperationException(); + } + + @Override + public byte readByte() { + throw new UnsupportedOperationException(); + } + + @Override + public short readShort() { + throw new UnsupportedOperationException(); + } + + @Override + public Binary readBinary(int len) { + throw new UnsupportedOperationException(); + } + + @Override + public void readBooleans(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readBytes(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readShorts(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readIntegers(int total, WritableColumnVector c, int rowId) { + readValues(total, c, rowId, (w, r, v) -> { + c.putInt(r, (int) v); + }); + } + + @Override + public void readIntegersWithRebase(int total, WritableColumnVector c, int rowId, + boolean failIfRebase) { + throw new UnsupportedOperationException("Only readIntegers is valid."); + } + + @Override + public void readUnsignedIntegers(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readUnsignedLongs(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readLongs(int total, WritableColumnVector c, int rowId) { + readValues(total, c, rowId, WritableColumnVector::putLong); + } + + @Override + public void readLongsWithRebase(int total, WritableColumnVector c, int rowId, + boolean failIfRebase) { + throw new UnsupportedOperationException(); + } + + @Override + public void readFloats(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readDoubles(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void readBinary(int total, WritableColumnVector c, int rowId) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipBooleans(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipBytes(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipShorts(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipIntegers(int total) { + // Read the values but don't write them out (the writer output method is a no-op) + readValues(total, null, -1, (w, r, v) -> { + }); + } + + @Override + public void skipLongs(int total) { + // Read the values but don't write them out (the writer output method is a no-op) + readValues(total, null, -1, (w, r, v) -> { + }); + } + + @Override + public void skipFloats(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipDoubles(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipBinary(int total) { + throw new UnsupportedOperationException(); + } + + @Override + public void skipFixedLenByteArray(int total, int len) { + throw new UnsupportedOperationException(); + } + + private void readValues(int total, WritableColumnVector c, int rowId, + IntegerOutputWriter outputWriter) { + int remaining = total; + if (valuesRead + total > totalValueCount) { + throw new ParquetDecodingException( + "no more values to read, total value count is " + valuesRead); + } + // First value + if (valuesRead == 0) { + //c.putInt(rowId, (int)firstValue); + outputWriter.write(c, rowId, firstValue); + lastValueRead = firstValue; + rowId++; + remaining--; + } + while (remaining > 0) { + int n; + try { + n = loadMiniBlockToOutput(remaining, c, rowId, outputWriter); + } catch (IOException e) { + throw new ParquetDecodingException("Error reading mini block.", e); + } + rowId += n; + remaining -= n; + } + valuesRead = total - remaining; + } + + + /** + * Read from a mini block. Read at most 'remaining' values into output. + * + * @return the number of values read into output + */ + private int loadMiniBlockToOutput(int remaining, WritableColumnVector c, int rowId, + IntegerOutputWriter outputWriter) throws IOException { + + // new block; read the block header + if (remainingInBlock == 0) { + readBlockHeader(); + } + + // new miniblock, unpack the miniblock + if (remainingInMiniBlock == 0) { + unpackMiniBlock(); + } + + //read values from miniblock + int valuesRead = 0; + for (int i = miniBlockSizeInValues - remainingInMiniBlock; + i < miniBlockSizeInValues && valuesRead < remaining; i++) { + //calculate values from deltas unpacked for current block + long outValue = lastValueRead + minDeltaInCurrentBlock + unpackedValuesBuffer[i]; + lastValueRead = outValue; + outputWriter.write(c, rowId + valuesRead, outValue); + remaining--; Review comment: Sorry, my previous comment was for a different part of the code. You're right, remaining does not need to be decremented. Doing so was just shortening the loop so loadMiniBlockToOutput was being called more times than necessary. Nice catch. Thank you! ########## File path: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetTest.scala ########## @@ -168,6 +169,16 @@ private[sql] trait ParquetTest extends FileBasedDataSourceTest { withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "true")(code) } + def withAllParquetWriters(code: => Unit): Unit = { + ParquetOutputFormat.WRITER_VERSION -> ParquetProperties.WriterVersion.PARQUET_2_0.toString Review comment: Oops. Cut-paste error. Removed. ########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedValuesReader.java ########## @@ -63,4 +64,36 @@ void skipDoubles(int total); void skipBinary(int total); void skipFixedLenByteArray(int total, int len); + + /** + * An interface to write columnar output in various ways + */ + @FunctionalInterface + interface IntegerOutputWriter { + void write(WritableColumnVector c, int rowId, long val); + } + + @FunctionalInterface + interface ByteBufferOutputWriter { Review comment: I just knew you would say that :). Done. ########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedReaderBase.java ########## @@ -0,0 +1,156 @@ +/* + * 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.spark.sql.execution.datasources.parquet; + +import org.apache.parquet.column.values.ValuesReader; +import org.apache.parquet.io.api.Binary; +import org.apache.spark.sql.execution.vectorized.WritableColumnVector; + +/** + * Base class for implementations of VectorizedValuesReader. Mainly to avoid duplication + * of methods that are not supported by concrete implementations + */ +public class VectorizedReaderBase extends ValuesReader implements VectorizedValuesReader { Review comment: Yeah, I thought about that too. But then I left it so as not to clutter up the PR. ########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaByteArrayReader.java ########## @@ -0,0 +1,66 @@ +/* + * 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.spark.sql.execution.datasources.parquet; + +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.column.values.deltastrings.DeltaByteArrayReader; +import org.apache.parquet.io.api.Binary; +import org.apache.spark.sql.execution.vectorized.WritableColumnVector; + +import java.io.IOException; +import java.nio.ByteBuffer; + +/** + * An implementation of the Parquet DELTA_BYTE_ARRAY decoder that supports the vectorized interface. + */ +public class VectorizedDeltaByteArrayReader extends VectorizedReaderBase { + private final DeltaByteArrayReader deltaByteArrayReader = new DeltaByteArrayReader(); + + @Override + public void initFromPage(int valueCount, ByteBufferInputStream in) throws IOException { + deltaByteArrayReader.initFromPage(valueCount, in); + } + + @Override + public Binary readBinary(int len) { + return deltaByteArrayReader.readBytes(); + } + + @Override + public void readBinary(int total, WritableColumnVector c, int rowId) { + for (int i = 0; i < total; i++) { + Binary binary = deltaByteArrayReader.readBytes(); + ByteBuffer buffer = binary.toByteBuffer(); + if (buffer.hasArray()) { + c.putByteArray(rowId + i, buffer.array(), buffer.arrayOffset() + buffer.position(), + binary.length()); + } else { + byte[] bytes = new byte[binary.length()]; + buffer.get(bytes); + c.putByteArray(rowId + i, bytes); + } + } + } + + @Override + public void skipBinary(int total) { + for(int i =0; i < total; i++) { Review comment: Fixed ########## File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedDeltaBinaryPackedReader.java ########## @@ -0,0 +1,318 @@ +/* + * 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.spark.sql.execution.datasources.parquet; + +import java.io.IOException; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.util.Arrays; + +import org.apache.parquet.Preconditions; +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.column.values.bitpacking.BytePackerForLong; +import org.apache.parquet.column.values.bitpacking.Packer; +import org.apache.parquet.io.ParquetDecodingException; +import org.apache.spark.sql.catalyst.util.RebaseDateTime; +import org.apache.spark.sql.execution.datasources.DataSourceUtils; +import org.apache.spark.sql.execution.vectorized.WritableColumnVector; + Review comment: Removed -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
