diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java index a21fec9ab8..fdc79370d2 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java @@ -33,6 +33,7 @@ import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Encoding; import org.apache.parquet.column.page.DataPage; import org.apache.parquet.column.page.DataPageV1; import org.apache.parquet.column.page.DataPageV2; @@ -40,6 +41,7 @@ import org.apache.parquet.column.page.DictionaryPageReadStore; import org.apache.parquet.column.page.PageReadStore; import org.apache.parquet.column.page.PageReader; +import org.apache.parquet.column.statistics.Statistics; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputDecompressor; import org.apache.parquet.crypto.AesCipher; import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; @@ -221,6 +223,22 @@ public DataPage visit(DataPageV2 dataPageV2) { if (!dataPageV2.isCompressed() && offsetIndex == null && null == blockDecryptor) { return dataPageV2; } + // In the standard encrypted layout, the levels are encrypted together with the values, and are + // split from the decrypted page body. Otherwise, they are already split (and in plaintext). + final boolean levelsInPageBody = dataPageV2 instanceof EncryptedDataPageV2; + final int rlLength; + final int dlLength; + BytesInput repetitionLevels = dataPageV2.getRepetitionLevels(); + BytesInput definitionLevels = dataPageV2.getDefinitionLevels(); + if (levelsInPageBody) { + rlLength = ((EncryptedDataPageV2) dataPageV2).getRepetitionLevelsByteLength(); + dlLength = ((EncryptedDataPageV2) dataPageV2).getDefinitionLevelsByteLength(); + } else { + rlLength = Math.toIntExact(repetitionLevels.size()); + dlLength = Math.toIntExact(definitionLevels.size()); + } + final int uncompressedSize = + Math.toIntExact(dataPageV2.getUncompressedSize() - dlLength - rlLength); BytesInput pageBytes = dataPageV2.getData(); try { if (options.getAllocator().isDirect() && options.useOffHeapDecryptBuffer()) { @@ -231,11 +249,14 @@ public DataPage visit(DataPageV2 dataPageV2) { if (blockDecryptor != null) { byteBuffer = blockDecryptor.decrypt(byteBuffer, dataPageAAD); } + if (levelsInPageBody) { + repetitionLevels = BytesInput.from(slice(byteBuffer, 0, rlLength)); + definitionLevels = BytesInput.from(slice(byteBuffer, rlLength, dlLength)); + byteBuffer = slice( + byteBuffer, rlLength + dlLength, byteBuffer.remaining() - rlLength - dlLength); + } long compressedSize = byteBuffer.limit(); if (dataPageV2.isCompressed()) { - int uncompressedSize = Math.toIntExact(dataPageV2.getUncompressedSize() - - dataPageV2.getDefinitionLevels().size() - - dataPageV2.getRepetitionLevels().size()); ByteBuffer decompressedBuffer = options.getAllocator().allocate(uncompressedSize); releaser.releaseLater(decompressedBuffer); @@ -250,13 +271,17 @@ public DataPage visit(DataPageV2 dataPageV2) { } } else { if (null != blockDecryptor) { - pageBytes = - BytesInput.from(blockDecryptor.decrypt(pageBytes.toByteArray(), dataPageAAD)); + byte[] decrypted = blockDecryptor.decrypt(pageBytes.toByteArray(), dataPageAAD); + if (levelsInPageBody) { + repetitionLevels = BytesInput.from(decrypted, 0, rlLength); + definitionLevels = BytesInput.from(decrypted, rlLength, dlLength); + pageBytes = BytesInput.from( + decrypted, rlLength + dlLength, decrypted.length - rlLength - dlLength); + } else { + pageBytes = BytesInput.from(decrypted); + } } if (dataPageV2.isCompressed()) { - int uncompressedSize = Math.toIntExact(dataPageV2.getUncompressedSize() - - dataPageV2.getDefinitionLevels().size() - - dataPageV2.getRepetitionLevels().size()); long start = System.nanoTime(); pageBytes = BytesInput.copy(decompressor.decompress(pageBytes, uncompressedSize)); setDecompressMetrics(pageBytes, start); @@ -272,8 +297,8 @@ public DataPage visit(DataPageV2 dataPageV2) { dataPageV2.getRowCount(), dataPageV2.getNullCount(), dataPageV2.getValueCount(), - dataPageV2.getRepetitionLevels(), - dataPageV2.getDefinitionLevels(), + repetitionLevels, + definitionLevels, dataPageV2.getDataEncoding(), pageBytes, dataPageV2.getStatistics()); @@ -283,8 +308,8 @@ public DataPage visit(DataPageV2 dataPageV2) { dataPageV2.getNullCount(), dataPageV2.getValueCount(), offsetIndex.getFirstRowIndex(currentPageIndex), - dataPageV2.getRepetitionLevels(), - dataPageV2.getDefinitionLevels(), + repetitionLevels, + definitionLevels, dataPageV2.getDataEncoding(), pageBytes, dataPageV2.getStatistics()); @@ -423,4 +448,55 @@ public void close() { toClose.add(releaser); AutoCloseables.uncheckedClose(toClose); } + + private static ByteBuffer slice(ByteBuffer buffer, int offset, int length) { + ByteBuffer slice = buffer.duplicate(); + slice.position(buffer.position() + offset); + slice.limit(buffer.position() + offset + length); + return slice.slice(); + } + + /** + * An encrypted v2 data page, whose repetition and definition levels are encrypted together with the values, as a + * single page module. {@link #getData()} returns the encrypted page body; the levels are split from the page body + * after decryption, using the level byte lengths from the page header. + */ + static final class EncryptedDataPageV2 extends DataPageV2 { + private final int repetitionLevelsByteLength; + private final int definitionLevelsByteLength; + + EncryptedDataPageV2( + int rowCount, + int nullCount, + int valueCount, + int repetitionLevelsByteLength, + int definitionLevelsByteLength, + Encoding dataEncoding, + BytesInput encryptedPageBody, + int uncompressedSize, + Statistics statistics, + boolean isCompressed) { + super( + rowCount, + nullCount, + valueCount, + BytesInput.empty(), + BytesInput.empty(), + dataEncoding, + encryptedPageBody, + uncompressedSize, + statistics, + isCompressed); + this.repetitionLevelsByteLength = repetitionLevelsByteLength; + this.definitionLevelsByteLength = definitionLevelsByteLength; + } + + int getRepetitionLevelsByteLength() { + return repetitionLevelsByteLength; + } + + int getDefinitionLevelsByteLength() { + return definitionLevelsByteLength; + } + } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java index 9f5bc48f8d..fb91e7d483 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java @@ -317,12 +317,19 @@ public void writePageV2( compressedData = compressor.compress(data); compressed = true; } + // The page body, as written after the page header: the levels followed by the (optionally + // compressed) values. In encrypted columns, the whole body - levels included - is encrypted + // as a single page module, as required by the Parquet Modular Encryption specification. + BytesInput[] pageBodyParts = {repetitionLevels, definitionLevels, compressedData}; if (null != pageBlockEncryptor) { AesCipher.quickUpdatePageAAD(dataPageAAD, pageOrdinal); - compressedData = BytesInput.from(pageBlockEncryptor.encrypt(compressedData.toByteArray(), dataPageAAD)); + pageBodyParts = new BytesInput[] { + BytesInput.from(pageBlockEncryptor.encrypt( + BytesInput.concat(pageBodyParts).toByteArray(), dataPageAAD)) + }; } - int compressedSize = - toIntWithCheck(compressedData.size() + repetitionLevels.size() + definitionLevels.size()); + BytesInput pageBody = BytesInput.concat(pageBodyParts); + int compressedSize = toIntWithCheck(pageBody.size()); tempOutputStream.reset(); if (null != headerBlockEncryptor) { AesCipher.quickUpdatePageAAD(dataPageHeaderAAD, pageOrdinal); @@ -330,14 +337,10 @@ public void writePageV2( if (pageWriteChecksumEnabled) { crc.reset(); try (ByteBufferReleaser pageReleaser = new ByteBufferReleaser(allocator)) { - if (repetitionLevels.size() > 0) { - crc.update(repetitionLevels.toByteBuffer(pageReleaser)); - } - if (definitionLevels.size() > 0) { - crc.update(definitionLevels.toByteBuffer(pageReleaser)); - } - if (compressedData.size() > 0) { - crc.update(compressedData.toByteBuffer(pageReleaser)); + for (BytesInput part : pageBodyParts) { + if (part.size() > 0) { + crc.update(part.toByteBuffer(pageReleaser)); + } } } parquetMetadataConverter.writeDataPageV2Header( @@ -382,8 +385,7 @@ public void writePageV2( // by concatenating before collecting instead of collecting twice, // we only allocate one buffer to copy into instead of multiple. - buf.collect(BytesInput.concat( - BytesInput.from(tempOutputStream), repetitionLevels, definitionLevels, compressedData)); + buf.collect(BytesInput.concat(BytesInput.from(tempOutputStream), pageBody)); dataEncodings.add(dataEncoding); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 5a611b41d1..44265bd36b 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -34,6 +34,7 @@ import java.io.ByteArrayInputStream; import java.io.Closeable; +import java.io.EOFException; import java.io.IOException; import java.io.InputStream; import java.io.InterruptedIOException; @@ -65,9 +66,15 @@ import org.apache.parquet.HadoopReadOptions; import org.apache.parquet.ParquetReadOptions; import org.apache.parquet.Preconditions; +import org.apache.parquet.SemanticVersion; +import org.apache.parquet.Strings; +import org.apache.parquet.VersionParser; +import org.apache.parquet.VersionParser.ParsedVersion; +import org.apache.parquet.VersionParser.VersionParseException; import org.apache.parquet.bytes.ByteBufferInputStream; import org.apache.parquet.bytes.ByteBufferReleaser; import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.column.page.DataPage; import org.apache.parquet.column.page.DataPageV1; @@ -2215,34 +2222,60 @@ public ColumnChunkPageReader readAllPages( break; case DATA_PAGE_V2: DataPageHeaderV2 dataHeaderV2 = pageHeader.getData_page_header_v2(); - int dataSize = compressedPageSize - - dataHeaderV2.getRepetition_levels_byte_length() - - dataHeaderV2.getDefinition_levels_byte_length(); - final BytesInput repetitionLevels = - this.readAsBytesInput(dataHeaderV2.getRepetition_levels_byte_length()); - final BytesInput definitionLevels = - this.readAsBytesInput(dataHeaderV2.getDefinition_levels_byte_length()); - final BytesInput values = this.readAsBytesInput(dataSize); - if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { - verifyCrc( - pageHeader.getCrc(), - "could not verify page integrity, CRC checksum verification failed", + int rlLength = dataHeaderV2.getRepetition_levels_byte_length(); + int dlLength = dataHeaderV2.getDefinition_levels_byte_length(); + DataPageV2 dataPageV2; + if (null != pageBlockDecryptor + && rlLength + dlLength > 0 + && !isLegacyEncryptedDataPageV2(rlLength + dlLength, compressedPageSize)) { + // The levels are encrypted together with the values, as a single page module. They + // are split after decryption, in ColumnChunkPageReader. + final BytesInput encryptedPageBody = this.readAsBytesInput(compressedPageSize); + if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { + verifyCrc( + pageHeader.getCrc(), + "could not verify page integrity, CRC checksum verification failed", + encryptedPageBody); + } + dataPageV2 = new ColumnChunkPageReadStore.EncryptedDataPageV2( + dataHeaderV2.getNum_rows(), + dataHeaderV2.getNum_nulls(), + dataHeaderV2.getNum_values(), + rlLength, + dlLength, + converter.getEncoding(dataHeaderV2.getEncoding()), + encryptedPageBody, + uncompressedPageSize, + converter.fromParquetStatistics( + getFileMetaData().getCreatedBy(), dataHeaderV2.getStatistics(), type), + dataHeaderV2.isIs_compressed()); + } else { + // Unencrypted page, or page without levels, or legacy encrypted page with plaintext levels + int dataSize = compressedPageSize - rlLength - dlLength; + final BytesInput repetitionLevels = this.readAsBytesInput(rlLength); + final BytesInput definitionLevels = this.readAsBytesInput(dlLength); + final BytesInput values = this.readAsBytesInput(dataSize); + if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { + verifyCrc( + pageHeader.getCrc(), + "could not verify page integrity, CRC checksum verification failed", + repetitionLevels, + definitionLevels, + values); + } + dataPageV2 = new DataPageV2( + dataHeaderV2.getNum_rows(), + dataHeaderV2.getNum_nulls(), + dataHeaderV2.getNum_values(), repetitionLevels, definitionLevels, - values); + converter.getEncoding(dataHeaderV2.getEncoding()), + values, + uncompressedPageSize, + converter.fromParquetStatistics( + getFileMetaData().getCreatedBy(), dataHeaderV2.getStatistics(), type), + dataHeaderV2.isIs_compressed()); } - DataPageV2 dataPageV2 = new DataPageV2( - dataHeaderV2.getNum_rows(), - dataHeaderV2.getNum_nulls(), - dataHeaderV2.getNum_values(), - repetitionLevels, - definitionLevels, - converter.getEncoding(dataHeaderV2.getEncoding()), - values, - uncompressedPageSize, - converter.fromParquetStatistics( - getFileMetaData().getCreatedBy(), dataHeaderV2.getStatistics(), type), - dataHeaderV2.isIs_compressed()); // Copy crc to new page, used for testing if (pageHeader.isSetCrc()) { dataPageV2.setCrc(pageHeader.getCrc()); @@ -2280,6 +2313,61 @@ public ColumnChunkPageReader readAllPages( options); } + /** + * Detects the layout of an encrypted v2 data page with non-empty levels, without consuming the stream. + *

+ * Per the Parquet Modular Encryption specification, the levels and the values of a v2 data page are encrypted + * together, as a single page module that starts with a 4-byte little-endian length. Older parquet-java + * versions wrote the levels in plaintext, followed by a page module that encrypts the values only. + * + * @param levelsLength the total byte length of the repetition and definition levels + * @param compressedPageSize the page size, as written to the file + * @return true if the page uses the legacy layout, with plaintext levels + */ + private boolean isLegacyEncryptedDataPageV2(int levelsLength, int compressedPageSize) throws IOException { + if (compressedPageSize < levelsLength + 4) { + // Too small for a page module after the levels: can only be the standard layout + return false; + } + long moduleLengthAtStart; + long moduleLengthAfterLevels; + stream.mark(levelsLength + 4); + try { + if (levelsLength >= 4) { + moduleLengthAtStart = BytesUtils.readIntLittleEndian(stream) & 0xFFFFFFFFL; + stream.skipFully(levelsLength - 4); + moduleLengthAfterLevels = BytesUtils.readIntLittleEndian(stream) & 0xFFFFFFFFL; + } else { + // The two candidate length fields overlap + byte[] prefix = new byte[levelsLength + 4]; + for (int i = 0; i < prefix.length; i++) { + int b = stream.read(); + if (b < 0) { + throw new EOFException( + "Unexpected end of encrypted data page v2 in column " + descriptor.col); + } + prefix[i] = (byte) b; + } + moduleLengthAtStart = BytesUtils.readIntLittleEndian(prefix, 0) & 0xFFFFFFFFL; + moduleLengthAfterLevels = BytesUtils.readIntLittleEndian(prefix, levelsLength) & 0xFFFFFFFFL; + } + } finally { + stream.reset(); + } + boolean standard = moduleLengthAtStart + 4 == compressedPageSize; + boolean legacy = levelsLength + moduleLengthAfterLevels + 4 == compressedPageSize; + if (standard && legacy) { + // Ambiguous (very unlikely): rely on the writer version + return isWrittenWithPlaintextEncryptedV2Levels(getFileMetaData().getCreatedBy()); + } + if (!standard && !legacy) { + throw new ParquetDecodingException("Encrypted data page v2 in column " + descriptor.col + + " is corrupted: the page size " + compressedPageSize + + " does not match the size of its encrypted page module"); + } + return legacy; + } + private boolean hasMorePages(long valuesCountReadSoFar, int dataPageCountReadSoFar) { return offsetIndex == null ? valuesCountReadSoFar < descriptor.metadata.getValueCount() @@ -2304,6 +2392,35 @@ public BytesInput readAsBytesInput(int size) throws IOException { } } + /** + * First parquet-java version that encrypts the levels of v2 data pages together with the values. + */ + private static final SemanticVersion ENCRYPTED_V2_LEVELS_FIXED_VERSION = new SemanticVersion(1, 19, 0); + + /** + * @param createdBy the created_by field of the file + * @return true if the file was written by a parquet-java version that left the levels of encrypted v2 data + * pages in plaintext + */ + static boolean isWrittenWithPlaintextEncryptedV2Levels(String createdBy) { + if (Strings.isNullOrEmpty(createdBy)) { + return false; + } + try { + ParsedVersion version = VersionParser.parse(createdBy); + if (!"parquet-mr".equals(version.application) || !version.hasSemanticVersion()) { + return false; + } + SemanticVersion semver = version.getSemanticVersion(); + // Ignore pre-release and build info: snapshots of the fixed version already use the standard layout + return new SemanticVersion(semver.major, semver.minor, semver.patch) + .compareTo(ENCRYPTED_V2_LEVELS_FIXED_VERSION) + < 0; + } catch (RuntimeException | VersionParseException e) { + return false; + } + } + /** * deals with a now fixed bug where compressedLength was missing a few bytes. */ diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index a453d4e756..c707548759 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -1385,13 +1385,81 @@ public void writeDataPageV2( byte[] pageHeaderAAD, SizeStatistics sizeStatistics) throws IOException { + writeDataPageV2( + rowCount, + nullCount, + valueCount, + repetitionLevels, + definitionLevels, + dataEncoding, + bytes, + compressed, + uncompressedDataSize, + statistics, + metadataBlockEncryptor, + pageHeaderAAD, + sizeStatistics, + null, + null); + } + + /** + * Writes a single v2 data page. + *

+ * If a {@code pageBlockEncryptor} is provided, {@code bytes} must not be encrypted: the repetition levels, + * definition levels and {@code bytes} are encrypted together, as a single page module, as required by the + * Parquet Modular Encryption specification. If {@code pageBlockEncryptor} is {@code null}, the page body is + * written as is. + * + * @param rowCount count of rows + * @param nullCount count of nulls + * @param valueCount count of values + * @param repetitionLevels repetition level bytes (plaintext) + * @param definitionLevels definition level bytes (plaintext) + * @param dataEncoding encoding for data + * @param bytes data bytes (optionally compressed; not encrypted if {@code pageBlockEncryptor} is provided) + * @param compressed whether the data bytes is compressed + * @param uncompressedDataSize the size of uncompressed data + * @param statistics the statistics of the page + * @param metadataBlockEncryptor encryptor for the page header + * @param pageHeaderAAD pageHeader AAD + * @param sizeStatistics size statistics for the page + * @param pageBlockEncryptor encryptor for the page body, or {@code null} + * @param pageAAD page body AAD, used if {@code pageBlockEncryptor} is provided + * @throws IOException if any I/O error occurs during writing the file + */ + public void writeDataPageV2( + int rowCount, + int nullCount, + int valueCount, + BytesInput repetitionLevels, + BytesInput definitionLevels, + Encoding dataEncoding, + BytesInput bytes, + boolean compressed, + int uncompressedDataSize, + Statistics statistics, + BlockCipher.Encryptor metadataBlockEncryptor, + byte[] pageHeaderAAD, + SizeStatistics sizeStatistics, + BlockCipher.Encryptor pageBlockEncryptor, + byte[] pageAAD) + throws IOException { withAbortOnFailure(() -> { state = state.write(); int rlByteLength = toIntWithCheck(repetitionLevels.size(), "page repetition levels"); int dlByteLength = toIntWithCheck(definitionLevels.size(), "page definition levels"); - int compressedSize = - toIntWithCheck(bytes.size() + repetitionLevels.size() + definitionLevels.size(), "page"); + BytesInput[] pageBodyParts = {repetitionLevels, definitionLevels, bytes}; + if (pageBlockEncryptor != null) { + pageBodyParts = new BytesInput[] { + BytesInput.from(pageBlockEncryptor.encrypt( + BytesInput.concat(pageBodyParts).toByteArray(), pageAAD)) + }; + } + BytesInput pageBody = BytesInput.concat(pageBodyParts); + + int compressedSize = toIntWithCheck(pageBody.size(), "page"); int uncompressedSize = toIntWithCheck(uncompressedDataSize + repetitionLevels.size() + definitionLevels.size(), "page"); @@ -1403,14 +1471,10 @@ public void writeDataPageV2( if (pageWriteChecksumEnabled) { crc.reset(); - if (repetitionLevels.size() > 0) { - crcUpdate(repetitionLevels); - } - if (definitionLevels.size() > 0) { - crcUpdate(definitionLevels); - } - if (bytes.size() > 0) { - crcUpdate(bytes); + for (BytesInput part : pageBodyParts) { + if (part.size() > 0) { + crcUpdate(part); + } } metadataConverter.writeDataPageV2Header( uncompressedSize, @@ -1451,7 +1515,7 @@ public void writeDataPageV2( currentEncodings.add(dataEncoding); encodingStatsBuilder.addDataEncoding(dataEncoding); - BytesInput.concat(repetitionLevels, definitionLevels, bytes).writeAllTo(out); + pageBody.writeAllTo(out); offsetIndexBuilder.add( toIntWithCheck(out.getPos() - beforeHeader, "page"), diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java index fce8866850..1787cab4b3 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java @@ -738,6 +738,8 @@ private void processChunk( BytesInput dlLevels = readBlockAllocate(dlLength, reader); int payLoadLength = pageHeader.getCompressed_page_size() - rlLength - dlLength; int rawDataLength = pageHeader.getUncompressed_page_size() - rlLength - dlLength; + // The values are not encrypted here: in encrypted columns, ParquetFileWriter encrypts + // the levels and the values together, as a single page module. pageLoad = processPageLoad( reader, headerV2.is_compressed, @@ -745,9 +747,9 @@ private void processChunk( decompressor, payLoadLength, rawDataLength, - encryptColumn, - dataEncryptor, - dataPageAAD); + false, + null, + null); statistics = convertStatistics( originalCreatedBy, normalizeNameInType(chunk.getPrimitiveType()), @@ -777,7 +779,10 @@ private void processChunk( rawDataLength, statistics, metaEncryptor, - dataPageHeaderAAD); + dataPageHeaderAAD, + null, + encryptColumn ? dataEncryptor : null, + dataPageAAD); pageOrdinal++; break; default: diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestEncryptedDataPageV2Layout.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestEncryptedDataPageV2Layout.java new file mode 100644 index 0000000000..67b06996ec --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestEncryptedDataPageV2Layout.java @@ -0,0 +1,349 @@ +/* + * 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.parquet.hadoop; + +import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.bytes.DirectByteBufferAllocator; +import org.apache.parquet.bytes.HeapByteBufferAllocator; +import org.apache.parquet.bytes.TrackingByteBufferAllocator; +import org.apache.parquet.column.ColumnDescriptor; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.statistics.Statistics; +import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder; +import org.apache.parquet.crypto.AesCipher; +import org.apache.parquet.crypto.ColumnEncryptionProperties; +import org.apache.parquet.crypto.DecryptionKeyRetrieverMock; +import org.apache.parquet.crypto.FileDecryptionProperties; +import org.apache.parquet.crypto.FileEncryptionProperties; +import org.apache.parquet.crypto.InternalColumnEncryptionSetup; +import org.apache.parquet.crypto.InternalFileDecryptor; +import org.apache.parquet.crypto.InternalFileEncryptor; +import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.format.BlockCipher; +import org.apache.parquet.format.PageHeader; +import org.apache.parquet.format.PageType; +import org.apache.parquet.format.Util; +import org.apache.parquet.hadoop.ParquetFileWriter.Mode; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ColumnPath; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Verifies that the repetition and definition levels of encrypted v2 data pages are encrypted together with the + * values, as a single page module (as written by Arrow C++ and arrow-rs), and that the legacy layout written by older + * parquet-java versions, with plaintext levels, is still readable. + */ +public class TestEncryptedDataPageV2Layout { + + private static final byte[] FOOTER_KEY = "0123456789012345".getBytes(); + private static final byte[] COLUMN_KEY = "1234567890123450".getBytes(); + private static final String FOOTER_KEY_ID = "kf"; + private static final String COLUMN_KEY_ID = "kc"; + private static final String COLUMN = "a"; + + private static final MessageType SCHEMA = + Types.buildMessage().repeated(INT32).named(COLUMN).named("msg"); + + /** Rows: [1, 2], [], [3] */ + private static final int[][] ROWS = {{1, 2}, {}, {3}}; + + private static final int[] REPETITION_LEVELS = {0, 1, 0, 0}; + private static final int[] DEFINITION_LEVELS = {1, 1, 0, 1}; + private static final int[] VALUES = {1, 2, 3}; + + @TempDir + private java.nio.file.Path tempDir; + + private enum Layout { + STANDARD, + LEGACY + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testWriterEncryptsLevelsWithValues(boolean offHeapDecryption) throws IOException { + Path path = new Path(tempDir.resolve("standard-writer.parquet").toUri()); + Configuration conf = new Configuration(); + GroupWriteSupport.setSchema(SCHEMA, conf); + SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA); + try (ParquetWriter writer = ExampleParquetWriter.builder(path) + .withConf(conf) + .withWriterVersion(WriterVersion.PARQUET_2_0) + .withDictionaryEncoding(false) + .withCompressionCodec(CompressionCodecName.SNAPPY) + .withPageWriteChecksumEnabled(true) + .withEncryption(encryptionProperties()) + .build()) { + for (int[] row : ROWS) { + Group group = factory.newGroup(); + for (int value : row) { + group.append(COLUMN, value); + } + writer.write(group); + } + } + + assertThat(readPageLayout(path)).isEqualTo(Layout.STANDARD); + assertRows(path, offHeapDecryption); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testFileWriterEncryptsLevelsWithValues(boolean offHeapDecryption) throws IOException { + Path path = new Path(tempDir.resolve("standard-file-writer.parquet").toUri()); + writePage(path, Layout.STANDARD); + + assertThat(readPageLayout(path)).isEqualTo(Layout.STANDARD); + assertRows(path, offHeapDecryption); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testLegacyLayoutWithPlaintextLevelsIsReadable(boolean offHeapDecryption) throws IOException { + Path path = new Path(tempDir.resolve("legacy.parquet").toUri()); + writePage(path, Layout.LEGACY); + + assertThat(readPageLayout(path)).isEqualTo(Layout.LEGACY); + assertRows(path, offHeapDecryption); + } + + @Test + public void testWriterVersionsWithPlaintextEncryptedV2Levels() { + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels( + "parquet-mr version 1.12.0 (build db75a6815f2ba1d1ee89d1a90aeb296f1f3a8f20)")) + .isTrue(); + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels( + "parquet-mr version 1.16.0 (build 0123456789)")) + .isTrue(); + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels( + "parquet-mr version 1.19.0-SNAPSHOT (build 0123456789)")) + .isFalse(); + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels( + "parquet-mr version 1.19.0 (build 0123456789)")) + .isFalse(); + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels("parquet-cpp-arrow version 17.0.0")) + .isFalse(); + assertThat(ParquetFileReader.isWrittenWithPlaintextEncryptedV2Levels(null)) + .isFalse(); + } + + /** + * Writes a single v2 data page with {@link ParquetFileWriter}: either with the levels encrypted together with the + * values, or in the legacy layout (plaintext levels followed by encrypted values) written by older parquet-java + * versions. + */ + private void writePage(Path path, Layout layout) throws IOException { + Configuration conf = new Configuration(); + ParquetProperties props = ParquetProperties.builder() + .withWriterVersion(WriterVersion.PARQUET_2_0) + .build(); + ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(path, conf), + SCHEMA, + Mode.CREATE, + ParquetWriter.DEFAULT_BLOCK_SIZE, + ParquetWriter.MAX_PADDING_SIZE_DEFAULT, + encryptionProperties(), + props); + ColumnDescriptor descriptor = SCHEMA.getColumns().get(0); + + BytesInput repetitionLevels = encodeLevels(REPETITION_LEVELS); + BytesInput definitionLevels = encodeLevels(DEFINITION_LEVELS); + ByteBuffer values = ByteBuffer.allocate(VALUES.length * 4).order(ByteOrder.LITTLE_ENDIAN); + Statistics statistics = Statistics.createStats(descriptor.getPrimitiveType()); + for (int value : VALUES) { + values.putInt(value); + statistics.updateStats(value); + } + statistics.incrementNumNulls(1); + BytesInput plainValues = BytesInput.from(values.array()); + + writer.start(); + writer.startBlock(ROWS.length); + InternalFileEncryptor fileEncryptor = writer.getEncryptor(); + InternalColumnEncryptionSetup columnSetup = fileEncryptor.getColumnSetup(ColumnPath.get(COLUMN), true, 0); + BlockCipher.Encryptor dataEncryptor = columnSetup.getDataEncryptor(); + byte[] pageAAD = AesCipher.createModuleAAD(fileEncryptor.getFileAAD(), ModuleType.DataPage, 0, 0, 0); + byte[] pageHeaderAAD = + AesCipher.createModuleAAD(fileEncryptor.getFileAAD(), ModuleType.DataPageHeader, 0, 0, 0); + + writer.startColumn(descriptor, REPETITION_LEVELS.length, CompressionCodecName.UNCOMPRESSED); + if (layout == Layout.LEGACY) { + // Encrypt the values only, and write the levels in plaintext + BytesInput encryptedValues = BytesInput.from(dataEncryptor.encrypt(plainValues.toByteArray(), pageAAD)); + writer.writeDataPageV2( + ROWS.length, + 1, + REPETITION_LEVELS.length, + repetitionLevels, + definitionLevels, + Encoding.PLAIN, + encryptedValues, + false, + (int) plainValues.size(), + statistics, + columnSetup.getMetaDataEncryptor(), + pageHeaderAAD, + null); + } else { + writer.writeDataPageV2( + ROWS.length, + 1, + REPETITION_LEVELS.length, + repetitionLevels, + definitionLevels, + Encoding.PLAIN, + plainValues, + false, + (int) plainValues.size(), + statistics, + columnSetup.getMetaDataEncryptor(), + pageHeaderAAD, + null, + dataEncryptor, + pageAAD); + } + writer.endColumn(); + writer.endBlock(); + writer.end(Collections.emptyMap()); + } + + private static BytesInput encodeLevels(int[] levels) throws IOException { + try (RunLengthBitPackingHybridEncoder encoder = + new RunLengthBitPackingHybridEncoder(1, 64, 1024, new HeapByteBufferAllocator())) { + for (int level : levels) { + encoder.writeInt(level); + } + return BytesInput.copy(encoder.toBytes()); + } + } + + /** + * Reads the raw (encrypted) body of the first data page, and checks where its page module starts. + */ + private Layout readPageLayout(Path path) throws IOException { + Configuration conf = new Configuration(); + ParquetReadOptions options = ParquetReadOptions.builder() + .withDecryption(decryptionProperties()) + .build(); + try (ParquetFileReader reader = ParquetFileReader.open(HadoopInputFile.fromPath(path, conf), options)) { + ColumnChunkMetaData chunk = + reader.getFooter().getBlocks().get(0).getColumns().get(0); + InternalFileDecryptor fileDecryptor = + reader.getFooter().getFileMetaData().getFileDecryptor(); + byte[] pageHeaderAAD = + AesCipher.createModuleAAD(fileDecryptor.getFileAAD(), ModuleType.DataPageHeader, 0, 0, 0); + reader.f.seek(chunk.getStartingPos()); + PageHeader pageHeader = Util.readPageHeader( + reader.f, fileDecryptor.getColumnSetup(chunk.getPath()).getMetaDataDecryptor(), pageHeaderAAD); + assertThat(pageHeader.getType()).isEqualTo(PageType.DATA_PAGE_V2); + + int levelsLength = pageHeader.getData_page_header_v2().getRepetition_levels_byte_length() + + pageHeader.getData_page_header_v2().getDefinition_levels_byte_length(); + assertThat(levelsLength).isPositive(); + byte[] pageBody = new byte[pageHeader.getCompressed_page_size()]; + reader.f.readFully(pageBody); + + if (BytesUtils.readIntLittleEndian(pageBody, 0) + 4 == pageBody.length) { + return Layout.STANDARD; + } + assertThat(levelsLength + BytesUtils.readIntLittleEndian(pageBody, levelsLength) + 4) + .isEqualTo(pageBody.length); + return Layout.LEGACY; + } + } + + private void assertRows(Path path, boolean offHeapDecryption) throws IOException { + Configuration conf = new Configuration(); + conf.setBoolean(ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED, offHeapDecryption); + conf.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true); + List> actual = new ArrayList<>(); + try (TrackingByteBufferAllocator allocator = TrackingByteBufferAllocator.wrap( + offHeapDecryption ? new DirectByteBufferAllocator() : new HeapByteBufferAllocator()); + ParquetReader reader = ParquetReader.builder(new GroupReadSupport(), path) + .withConf(conf) + .withAllocator(allocator) + .withDecryption(decryptionProperties()) + .build()) { + for (Group group = reader.read(); group != null; group = reader.read()) { + List row = new ArrayList<>(); + for (int i = 0; i < group.getFieldRepetitionCount(COLUMN); i++) { + row.add(group.getInteger(COLUMN, i)); + } + actual.add(row); + } + } + List> expected = new ArrayList<>(); + for (int[] row : ROWS) { + List expectedRow = new ArrayList<>(); + for (int value : row) { + expectedRow.add(value); + } + expected.add(expectedRow); + } + assertThat(actual).isEqualTo(expected); + } + + private static FileEncryptionProperties encryptionProperties() { + ColumnEncryptionProperties columnProperties = ColumnEncryptionProperties.builder(COLUMN) + .withKey(COLUMN_KEY) + .withKeyID(COLUMN_KEY_ID) + .build(); + return FileEncryptionProperties.builder(FOOTER_KEY) + .withFooterKeyID(FOOTER_KEY_ID) + .withEncryptedColumns(Collections.singletonMap(columnProperties.getPath(), columnProperties)) + .build(); + } + + private static FileDecryptionProperties decryptionProperties() { + return FileDecryptionProperties.builder() + .withKeyRetriever(new DecryptionKeyRetrieverMock() + .putKey(FOOTER_KEY_ID, FOOTER_KEY) + .putKey(COLUMN_KEY_ID, COLUMN_KEY)) + .build(); + } +}