Parquet: Support parquet modular encryption

[ggershinsky]

Implements #1413

[rdblue]

I'm approving this to run unit tests. Thanks for working on this, @ggershinsky!

[rdblue]

@ggershinsky, the doc you linked to is quite long and it isn't clear what part of it explains what you're doing here. Can you provide a quick summary of how to plug into Parquet encryption and what this does? I'm assuming that it provides Parquet's equivalent of an EncryptionManager that gets the file AAD and necessary key material from Iceberg's key_metadata field. Is that correct?

[ggershinsky]

I'm approving this to run unit tests. Thanks for working on this, @ggershinsky!

Thanks @rdblue!

[ggershinsky]

Can you provide a quick summary of how to plug into Parquet encryption and what this does?

Certainly. There are two encryption interfaces in parquet-mr-1.12.0 : low-level (direct impl of the spec; max flexibility; no key management) and high-level (a layer on top of low-level, with a lib-local key management tools driven by Hadoop properties). In Iceberg, we'll use directly the low-level Parquet encryption API - because the key management will be done by Iceberg, in a similar fashion for all formats; and because Iceberg has a centralized manifest capability, which makes key management more efficient than running lib-local nodes in each worker process.
Since Iceberg also taps directly into Parquet low-level API (general, no encryption), this PR enables to link it to the encryption feature, and translates general column encryption configuration (TBD) into Parquet encryption configuration.

it provides Parquet's equivalent of an EncryptionManager that gets the file AAD and necessary key material from Iceberg's key_metadata field.

Well, here we have the gap that I've described in the other PR. To encrypt data(/delete) files, we need the AES keys - the DEKs, "data encryption keys" which are used to actually encrypt the data and metadata modules (there must be a unique DEK per file/column). But the key_material is a binary field in the manifest entry for a data(/delete) file that keeps a "wrapped" version of these DEKs (encrypted with master keys, MEKs, in user's KMS system). It doesn't and shouldn't keep raw DEKs. Therefore, sending key_material to Parquet file writers (or any other writers) doesn't help. Per my proposal in the last sync, we can reverse this process - generate random DEKs at (Parquet) writers, use them for encryption, and send them in the DataFile/DeleteFile/ContentFile objects back to the manifest writer. This also seems to fit the current Iceberg model, where manifest entries are written after collecting ContentFile objects from file writers. At this point, the manifest writer process will contact the KMS to wrap these DEKs, and package them into the key_metadata field in the manifest file (for the readers).

In a future version, we might want to generate the DEKs (or get them from KMS) in the manifest writer process, and then distribute them to data/delete file writers, with a unique DEK per file (or a set of unique DEKs per file, for column encryption). This seems to be more complicated, and less fitting the current Iceberg flow; my suggestion would be to start with the reverse approach described above.

[flyrain]

Hi @ggershinsky , can you rebase this PR to have #2441?

[ggershinsky]

Hi @flyrain, will do, thanks for the notice.

[ggershinsky]

Per the comments above, there will be changes in this pull request (and in #2638, #2640). Converting the three to drafts.

[liujinhui1994]

@ggershinsky hello，Is there any latest development of this feature, we are looking forward to this feature

[ggershinsky]

hi @liujinhui1994 , yes, this feature is under active development. Per the community discussions, the encryption PRs will be updated in Q1'22; probably I'll have the commits ready later this month or in Feb.

[shangxinli]

@ggershinsky Let me know when it is ready for review. I would love to review this change.

[ggershinsky]

Hi @shangxinli ; frankly, the only dependency left is the parquet-mr update. This PR depends on the "uniform encryption" feature, which is already merged in the parquet master, but not released yet. We've started this discussion in the last parquet community sync, it's on track. Once the parquet-mr master is cut (as 1.12.3 or 1.13.0 :), this Iceberg PR will be able to pass CI, ready for a review.

[shangxinli]

OK. I will have a look once I have time.

…

On Tue, Feb 8, 2022 at 10:35 AM ggershinsky ***@***.***> wrote: Hi @shangxinli <https://github.com/shangxinli> ; frankly, the only dependency left is the parquet-mr update. This PR depends on the "uniform encryption" feature, which is already merged in the parquet master, but not released yet. We've started this discussion in the last parquet community sync, it's on track. Once the parquet-mr master is cut (as 1.12.3 or 1.13.0 :), this Iceberg PR will be able to pass CI, ready for a review. — Reply to this email directly, view it on GitHub <#2639 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AHPXKMIIJC57SQDO2WA22UTU2FPALANCNFSM45SBJ3JQ> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>. You are receiving this because you were mentioned.Message ID: ***@***.***>

-- Xinli Shang

[zhongyujiang]

Nit: a switch would look simpler than an if block.

[zhongyujiang]

File key must not be null when we push down encryption, right? If so, can we move this check to NativeFileCryptoParameters#build ? So we won't need to check if the file key is null every time when converting NativeFileCryptoParameters to Parquet / ORC config.

[ggershinsky]

I believe ORC works with column keys, not files keys. So this is a parquet-specific check.

[zhongyujiang]

Got it, thanks for explanation.

[pvary]

nit: empty line between blocks

[rdblue]

The "footer key" is used for the entire file?

[ggershinsky]

Correct, we're leveraging the EncryptionWithFooterKey mode ( https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift#L770 ), that encrypts all columns in a file with the "footer key".

[rdblue]

Style: it would be better to create the new read options on a separate line.

[ggershinsky]

I tried a couple of ways to break the line(s), but the spotless check hasn't accepted them, the spotless apply always returns the code to this form.

[rdblue]

I'm not talking about reformatting the line breaks, I mean to create a variable:

ParquetReadOptionsreadOptions = ParquetReadOptions.builder().withDecryption(decryptionProperties).build();
try (ParquetFileReaderfileReader = newReader(file, readOptions)) {
...
}

[rdblue]

@ggershinsky, this is looking good, but there are a few minor updates needed and it requires more testing to exercise more of the Iceberg specific code. You can check out code coverage to see when you're hitting all the areas you've changed.

Also, for the change to remove reflection, here's a diff since I explored that locally:

[blue@work iceberg]$ git diff
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java
index af2fb0e80a..ee3a8f50be 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetWriter.java@@ -28,20 +28,16 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.Metrics;
import org.apache.iceberg.MetricsConfig;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.common.DynConstructors;-import org.apache.iceberg.common.DynMethods;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.parquet.bytes.ByteBufferAllocator;
import org.apache.parquet.column.ColumnWriteStore;
import org.apache.parquet.column.ParquetProperties;
-import org.apache.parquet.column.page.PageWriteStore;-import org.apache.parquet.column.values.bloomfilter.BloomFilterWriteStore;
import org.apache.parquet.crypto.FileEncryptionProperties;
import org.apache.parquet.crypto.InternalFileEncryptor;
import org.apache.parquet.hadoop.CodecFactory;
+import org.apache.parquet.hadoop.ColumnChunkPageWriteStore;
import org.apache.parquet.hadoop.ParquetFileWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import org.apache.parquet.schema.MessageType;
@@ -50,25 +46,6 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
private static final Metrics EMPTY_METRICS = new Metrics(0L, null, null, null, null);
- private static final DynConstructors.Ctor<PageWriteStore> pageStoreCtorParquet =- DynConstructors.builder(PageWriteStore.class)- .hiddenImpl(- "org.apache.parquet.hadoop.ColumnChunkPageWriteStore",- CodecFactory.BytesCompressor.class,- MessageType.class,- ByteBufferAllocator.class,- int.class,- boolean.class,- InternalFileEncryptor.class,- int.class)- .build();-- private static final DynMethods.UnboundMethod flushToWriter =- DynMethods.builder("flushToFileWriter")- .hiddenImpl(- "org.apache.parquet.hadoop.ColumnChunkPageWriteStore", ParquetFileWriter.class)- .build();-
private final long targetRowGroupSize;
private final Map<String, String> metadata;
private final ParquetProperties props;
@@ -81,7 +58,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
private final OutputFile output;
private final Configuration conf;
- private DynMethods.BoundMethod flushPageStoreToWriter;+ private ColumnChunkPageWriteStore pageStore = null;
private ColumnWriteStore writeStore;
private long recordCount = 0;
private long nextCheckRecordCount = 10;
@@ -232,7 +209,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
ensureWriterInitialized();
writer.startBlock(recordCount);
writeStore.flush();
- flushPageStoreToWriter.invoke(writer);+ pageStore.flushToFileWriter(writer);
writer.endBlock();
if (!finished) {
writeStore.close();
@@ -253,8 +230,8 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
props.getMaxRowCountForPageSizeCheck());
this.recordCount = 0;
- PageWriteStore pageStore =- pageStoreCtorParquet.newInstance(+ this.pageStore =+ new ColumnChunkPageWriteStore(
compressor,
parquetSchema,
props.getAllocator(),
@@ -264,9 +241,7 @@ class ParquetWriter<T> implements FileAppender<T>, Closeable {
rowGroupOrdinal);
this.rowGroupOrdinal++;
- this.flushPageStoreToWriter = flushToWriter.bind(pageStore);- this.writeStore =- props.newColumnWriteStore(parquetSchema, pageStore, (BloomFilterWriteStore) pageStore);+ this.writeStore = props.newColumnWriteStore(parquetSchema, pageStore, pageStore);
model.setColumnStore(writeStore);
}

[rdblue]

@ggershinsky this looks close. Just style and naming issues left.

[rdblue]

@ggershinsky, looks like this is out of date. Can you rebase?

[ggershinsky]

@rdblue The PR is rebased

[rdblue]

Thanks, @ggershinsky!

[siddiquebagwan]

Hi @ggershinsky
I am looking to use parquet column level encryption in Apache IceBerg.
I have set below properties in spark.config but still encryption is not working

Here is my config code:

spark = SparkSession.builder
.appName("ParquetColumnEncryption")
.config("spark.hadoop.parquet.encryption.key.list", "key1:AAECAwQFBgcICQoLDA0ODw==,key2:AAECAwQFBgcICQoLDA0ODw==")
.config("spark.hadoop.parquet.encryption.kms.client.class", "org.apache.parquet.crypto.keytools.mocks.InMemoryKMS")
.config("spark.hadoop.parquet.crypto.factory.class", "org.apache.parquet.crypto.keytools.PropertiesDrivenCryptoFactory")
.config("spark.jars", "/home/foo/spark-env/jars/parquet-hadoop-1.13.1-tests.jar")
.config("spark.hadoop.parquet.encryption.plaintext.footer", "true")
.getOrCreate()

[ggershinsky]

Hi @siddiquebagwan . In Iceberg, parquet encryption is not activated by configuration. The mechanism is different, but it is not fully merged yet. For the progress, please track #5544