feat: Add a configuration to make parquet encryption optional (#16649)

* Initial commit to form PR for datafusion encryption support

* Add tests for encryption configuration

* Apply cargo fmt

* Add a roundtrip encryption test to the parquet tests.

* cargo fmt

* Update test to add decryption parameter to called functions.

* Try to get DataFrame.write_parquet to work with encryption. Doesn't quite, column encryption is broken.

* Update datafusion/datasource-parquet/src/opener.rs

Co-authored-by: Adam Reeve <[email protected]>

* Update datafusion/datasource-parquet/src/source.rs

Co-authored-by: Adam Reeve <[email protected]>

* Fix write test in parquet.rs

* Simplify encryption test. Remove unused imports.

* Run cargo fmt.

* Further streamline roundtrip test.

* Change From methods for FileEncryptionProperties and FileDecryptionProperties to use references.

* Change encryption config to directly hold column keys using custom config fields.

* Fix generated field names in visit for encryptor and decryptor to use "." instead of "::"

* 1. Disable parallel writes with enccryption.
2. Fixed unused header warning in config.rs.
3. Fix test case in encryption.rs to call conversion to ConfigFileDecryption properties correctly.

* cargo fmt

* Update datafusion/common/src/file_options/parquet_writer.rs

Co-authored-by: Copilot <[email protected]>

* fix variables shown in information schema test.

* Backout bad suggestion from copilot

* Remove unused serde reference
Add an example to read and write encrypted parquet files.

* cargo fmt

* change file_format.rs to use global encryption options in struct.

* Turn off page_index for encrypted example. Get encrypted example working with filter.

* Tidy up example output.

* Add missing license. Run taplo format

* Update configs.md by running dev/update_config_docs.sh

* Cargo fmt + clippy changes.

* Add filter test for encrypted files.

* Cargo clippy changes.

* Fix link in README.md

* Add issue tag for parallel writes.

* Move file encryption and decryption properties out of global options

* Use config_namespace_with_hashmap for column encryption/decryption props

* Remove outdated docs on crypto settings.

Signed-off-by: Corwin Joy <[email protected]>

* 1. Add docs for using encryption configuration.
2. Add example SQL for using encryption from CLI.
3. Fix removed variables in test for configuration information.
4. Clippy and cargo fmt.

Signed-off-by: Corwin Joy <[email protected]>

* Update code to add missing ParquetOpener parameter due to merge from main

Signed-off-by: Corwin Joy <[email protected]>

* Add CLI documentation for Parquet options and provide an encryption example

Signed-off-by: Corwin Joy <[email protected]>

* Use ConfigFileDecryptionProperties in ParquetReadOptions

Signed-off-by: Adam Reeve <[email protected]>

* Implement default for ConfigFileEncryptionProperties

Signed-off-by: Corwin Joy <[email protected]>

* Add sqllogictest for parquet with encryption

Signed-off-by: Corwin Joy <[email protected]>

* Apply prettier changes from CI

Signed-off-by: Corwin Joy <[email protected]>

* Begin adding config for encryption.

* Fix merge errors.

* Fix config dependency in datafusion/core/Cargo.toml

* ci reformat

* Fix missing Cargo dependency.

* Update Cargo.toml

Co-authored-by: Copilot <[email protected]>

* Revert bad suggestion from copilot

Signed-off-by: Corwin Joy <[email protected]>

* taplo format Cargo.toml

Signed-off-by: Corwin Joy <[email protected]>

* 1. Add CI test for parquet_encryption. 2. Document new configuration

Signed-off-by: Corwin Joy <[email protected]>

* Refactor cfg parquet_encryption feature into a few functions defined in encryption.rs

Signed-off-by: Corwin Joy <[email protected]>

* Remove #allow(unusued) that is no longer needed

Signed-off-by: Corwin Joy <[email protected]>

* cargo fmt

Signed-off-by: Corwin Joy <[email protected]>

* Fix error when cfg parquet_encryption is disabled.

Signed-off-by: Corwin Joy <[email protected]>

* cargo fmt

Signed-off-by: Corwin Joy <[email protected]>

* Remove unused mut.

Signed-off-by: Corwin Joy <[email protected]>

---------

Signed-off-by: Corwin Joy <[email protected]>
Signed-off-by: Adam Reeve <[email protected]>
Co-authored-by: Adam Reeve <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: Adam Reeve <[email protected]>

Author: 3 people

.github/workflows/rust.yml

@@ -218,6 +218,8 @@ jobs:
         run: cargo check --profile ci --no-default-features -p datafusion --features=string_expressions
       - name: Check datafusion (unicode_expressions)
         run: cargo check --profile ci --no-default-features -p datafusion --features=unicode_expressions
+      - name: Check parquet encryption (parquet_encryption)
+        run: cargo check --profile ci --no-default-features -p datafusion --features=parquet_encryption
 
   # Check datafusion-functions crate features
   #

Cargo.lock

@@ -150,6 +150,7 @@ env_logger = "0.11"
 futures = "0.3"
 half = { version = "2.6.0", default-features = false }
 hashbrown = { version = "0.14.5", features = ["raw"] }
+hex = { version = "0.4.3" }
 indexmap = "2.10.0"
 itertools = "0.14"
 log = "^0.4"

Cargo.toml

@@ -120,6 +120,7 @@ Default features:
 - `datetime_expressions`: date and time functions such as `to_timestamp`
 - `encoding_expressions`: `encode` and `decode` functions
 - `parquet`: support for reading the [Apache Parquet] format
+- `parquet_encryption`: support for using [Parquet Modular Encryption]
 - `regex_expressions`: regular expression functions, such as `regexp_match`
 - `unicode_expressions`: Include unicode aware functions such as `character_length`
 - `unparser`: enables support to reverse LogicalPlans back into SQL
@@ -134,6 +135,7 @@ Optional features:
 
 [apache avro]: https://avro.apache.org/
 [apache parquet]: https://parquet.apache.org/
+[parquet modular encryption]: https://parquet.apache.org/docs/file-format/data-pages/encryption/
 
 ## DataFusion API Evolution and Deprecation Guidelines
 

README.md

@@ -40,9 +40,15 @@ name = "datafusion_common"
 [features]
 avro = ["apache-avro"]
 backtrace = []
+parquet_encryption = [
+    "parquet",
+    "parquet/encryption",
+    "dep:hex",
+]
 pyarrow = ["pyo3", "arrow/pyarrow", "parquet"]
 force_hash_collisions = []
 recursive_protection = ["dep:recursive"]
+parquet = ["dep:parquet"]
 
 [dependencies]
 ahash = { workspace = true }
@@ -58,7 +64,7 @@ base64 = "0.22.1"
 chrono = { workspace = true }
 half = { workspace = true }
 hashbrown = { workspace = true }
-hex = "0.4.3"
+hex = { workspace = true, optional = true }
 indexmap = { workspace = true }
 libc = "0.2.174"
 log = { workspace = true }

datafusion/common/Cargo.toml

@@ -19,6 +19,8 @@
 
 use arrow_ipc::CompressionType;
 
+#[cfg(feature = "parquet_encryption")]
+use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
 use crate::error::_config_err;
 use crate::parsers::CompressionTypeVariant;
 use crate::utils::get_available_parallelism;
@@ -29,12 +31,8 @@ use std::error::Error;
 use std::fmt::{self, Display};
 use std::str::FromStr;
 
-#[cfg(feature = "parquet")]
+#[cfg(feature = "parquet_encryption")]
 use hex;
-#[cfg(feature = "parquet")]
-use parquet::encryption::decrypt::FileDecryptionProperties;
-#[cfg(feature = "parquet")]
-use parquet::encryption::encrypt::FileEncryptionProperties;
 
 /// A macro that wraps a configuration struct and automatically derives
 /// [`Default`] and [`ConfigField`] for it, allowing it to be used
@@ -2148,7 +2146,7 @@ impl ConfigField for ConfigFileEncryptionProperties {
     }
 }
 
-#[cfg(feature = "parquet")]
+#[cfg(feature = "parquet_encryption")]
 impl From<ConfigFileEncryptionProperties> for FileEncryptionProperties {
     fn from(val: ConfigFileEncryptionProperties) -> Self {
         let mut fep = FileEncryptionProperties::builder(
@@ -2194,7 +2192,7 @@ impl From<ConfigFileEncryptionProperties> for FileEncryptionProperties {
     }
 }
 
-#[cfg(feature = "parquet")]
+#[cfg(feature = "parquet_encryption")]
 impl From<&FileEncryptionProperties> for ConfigFileEncryptionProperties {
     fn from(f: &FileEncryptionProperties) -> Self {
         let (column_names_vec, column_keys_vec, column_metas_vec) = f.column_keys();
@@ -2308,7 +2306,7 @@ impl ConfigField for ConfigFileDecryptionProperties {
     }
 }
 
-#[cfg(feature = "parquet")]
+#[cfg(feature = "parquet_encryption")]
 impl From<ConfigFileDecryptionProperties> for FileDecryptionProperties {
     fn from(val: ConfigFileDecryptionProperties) -> Self {
         let mut column_names: Vec<&str> = Vec::new();
@@ -2342,7 +2340,7 @@ impl From<ConfigFileDecryptionProperties> for FileDecryptionProperties {
     }
 }
 
-#[cfg(feature = "parquet")]
+#[cfg(feature = "parquet_encryption")]
 impl From<&FileDecryptionProperties> for ConfigFileDecryptionProperties {
     fn from(f: &FileDecryptionProperties) -> Self {
         let (column_names_vec, column_keys_vec) = f.column_keys();
@@ -2688,7 +2686,7 @@ mod tests {
         );
     }
 
-    #[cfg(feature = "parquet")]
+    #[cfg(feature = "parquet_encryption")]
     #[test]
     fn parquet_table_encryption() {
         use crate::config::{

datafusion/common/src/config.rs

@@ -0,0 +1,76 @@
+// 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.
+
+// Support optional features for encryption in Parquet files.
+//! This module provides types and functions related to encryption in Parquet files.
+
+#[cfg(feature = "parquet_encryption")]
+pub use parquet::encryption::decrypt::FileDecryptionProperties;
+#[cfg(feature = "parquet_encryption")]
+pub use parquet::encryption::encrypt::FileEncryptionProperties;
+
+#[cfg(not(feature = "parquet_encryption"))]
+pub struct FileDecryptionProperties;
+#[cfg(not(feature = "parquet_encryption"))]
+pub struct FileEncryptionProperties;
+
+#[cfg(feature = "parquet")]
+use crate::config::ParquetEncryptionOptions;
+pub use crate::config::{ConfigFileDecryptionProperties, ConfigFileEncryptionProperties};
+#[cfg(feature = "parquet")]
+use parquet::file::properties::WriterPropertiesBuilder;
+
+#[cfg(feature = "parquet")]
+pub fn add_crypto_to_writer_properties(
+    #[allow(unused)] crypto: &ParquetEncryptionOptions,
+    #[allow(unused_mut)] mut builder: WriterPropertiesBuilder,
+) -> WriterPropertiesBuilder {
+    #[cfg(feature = "parquet_encryption")]
+    if let Some(file_encryption_properties) = &crypto.file_encryption {
+        builder = builder
+            .with_file_encryption_properties(file_encryption_properties.clone().into());
+    }
+    builder
+}
+
+#[cfg(feature = "parquet_encryption")]
+pub fn map_encryption_to_config_encryption(
+    encryption: Option<&FileEncryptionProperties>,
+) -> Option<ConfigFileEncryptionProperties> {
+    encryption.map(|fe| fe.into())
+}
+
+#[cfg(not(feature = "parquet_encryption"))]
+pub fn map_encryption_to_config_encryption(
+    _encryption: Option<&FileEncryptionProperties>,
+) -> Option<ConfigFileEncryptionProperties> {
+    None
+}
+
+#[cfg(feature = "parquet_encryption")]
+pub fn map_config_decryption_to_decryption(
+    decryption: Option<&ConfigFileDecryptionProperties>,
+) -> Option<FileDecryptionProperties> {
+    decryption.map(|fd| fd.clone().into())
+}
+
+#[cfg(not(feature = "parquet_encryption"))]
+pub fn map_config_decryption_to_decryption(
+    _decryption: Option<&ConfigFileDecryptionProperties>,
+) -> Option<FileDecryptionProperties> {
+    None
+}

datafusion/common/src/encryption.rs

@@ -27,6 +27,7 @@ use crate::{
 
 use arrow::datatypes::Schema;
 // TODO: handle once deprecated
+use crate::encryption::add_crypto_to_writer_properties;
 #[allow(deprecated)]
 use parquet::{
     arrow::ARROW_SCHEMA_META_KEY,
@@ -100,11 +101,7 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
 
         let mut builder = global.into_writer_properties_builder()?;
 
-        if let Some(file_encryption_properties) = &crypto.file_encryption {
-            builder = builder.with_file_encryption_properties(
-                file_encryption_properties.clone().into(),
-            );
-        }
+        builder = add_crypto_to_writer_properties(crypto, builder);
 
         // check that the arrow schema is present in the kv_metadata, if configured to do so
         if !global.skip_arrow_metadata
@@ -456,12 +453,10 @@ mod tests {
     };
     use std::collections::HashMap;
 
-    use crate::config::{
-        ConfigFileEncryptionProperties, ParquetColumnOptions, ParquetEncryptionOptions,
-        ParquetOptions,
-    };
-
     use super::*;
+    use crate::config::{ParquetColumnOptions, ParquetEncryptionOptions, ParquetOptions};
+    #[cfg(feature = "parquet_encryption")]
+    use crate::encryption::map_encryption_to_config_encryption;
 
     const COL_NAME: &str = "configured";
 
@@ -590,8 +585,10 @@ mod tests {
             HashMap::from([(COL_NAME.into(), configured_col_props)])
         };
 
-        let fep: Option<ConfigFileEncryptionProperties> =
-            props.file_encryption_properties().map(|fe| fe.into());
+        #[cfg(feature = "parquet_encryption")]
+        let fep = map_encryption_to_config_encryption(props.file_encryption_properties());
+        #[cfg(not(feature = "parquet_encryption"))]
+        let fep = None;
 
         #[allow(deprecated)] // max_statistics_size
         TableParquetOptions {

datafusion/common/src/file_options/parquet_writer.rs

@@ -41,6 +41,7 @@ pub mod config;
 pub mod cse;
 pub mod diagnostic;
 pub mod display;
+pub mod encryption;
 pub mod error;
 pub mod file_options;
 pub mod format;

datafusion/common/src/lib.rs

@@ -61,13 +61,21 @@ default = [
     "unicode_expressions",
     "compression",
     "parquet",
+    "parquet_encryption",
     "recursive_protection",
 ]
 encoding_expressions = ["datafusion-functions/encoding_expressions"]
 # Used for testing ONLY: causes all values to hash to the same value (test for collisions)
 force_hash_collisions = ["datafusion-physical-plan/force_hash_collisions", "datafusion-common/force_hash_collisions"]
 math_expressions = ["datafusion-functions/math_expressions"]
 parquet = ["datafusion-common/parquet", "dep:parquet", "datafusion-datasource-parquet"]
+parquet_encryption = [
+    "dep:parquet",
+    "parquet/encryption",
+    "datafusion-common/parquet_encryption",
+    "datafusion-datasource-parquet/parquet_encryption",
+    "dep:hex",
+]
 pyarrow = ["datafusion-common/pyarrow", "parquet"]
 regex_expressions = [
     "datafusion-functions/regex_expressions",
@@ -127,6 +135,7 @@ datafusion-session = { workspace = true }
 datafusion-sql = { workspace = true }
 flate2 = { version = "1.1.2", optional = true }
 futures = { workspace = true }
+hex = { workspace = true, optional = true }
 itertools = { workspace = true }
 log = { workspace = true }
 object_store = { workspace = true }