Skip to content

GH-3946: Encrypt the levels of DataPageV2 together with the values #3947

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
raviston wants to merge 1 commit into from
+678 −65
Closed

GH-3946: Encrypt the levels of DataPageV2 together with the values #3947

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 88 additions & 12 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,15 @@
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;
import org.apache.parquet.column.page.DictionaryPage;
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;
Expand Down Expand Up @@ -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()) {
Expand All @@ -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);
Expand All @@ -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);
Expand All @@ -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());
Expand All @@ -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());
Expand Down Expand Up @@ -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;
}
}
}
28 changes: 15 additions & 13 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -317,27 +317,30 @@ 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);
}
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(
Expand Down Expand Up @@ -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);
}

Expand Down
Loading