feat: Parquet modular encryption

Cargo.toml

@@ -155,6 +155,7 @@ parquet = { version = "55.1.0", default-features = false, features = [
     "arrow",
     "async",
     "object_store",
+    "encryption",
 ] }
 pbjson = { version = "0.7.0" }
 pbjson-types = "0.7"

benchmarks/src/bin/dfbench.rs

@@ -60,11 +60,11 @@ pub async fn main() -> Result<()> {
         Options::Cancellation(opt) => opt.run().await,
         Options::Clickbench(opt) => opt.run().await,
         Options::H2o(opt) => opt.run().await,
-        Options::Imdb(opt) => opt.run().await,
+        Options::Imdb(opt) => Box::pin(opt.run()).await,
         Options::ParquetFilter(opt) => opt.run().await,
         Options::Sort(opt) => opt.run().await,
         Options::SortTpch(opt) => opt.run().await,
-        Options::Tpch(opt) => opt.run().await,
+        Options::Tpch(opt) => Box::pin(opt.run()).await,
         Options::TpchConvert(opt) => opt.run().await,
     }
 }

benchmarks/src/bin/imdb.rs

@@ -53,7 +53,7 @@ pub async fn main() -> Result<()> {
     env_logger::init();
     match ImdbOpt::from_args() {
         ImdbOpt::Benchmark(BenchmarkSubCommandOpt::DataFusionBenchmark(opt)) => {
-            opt.run().await
+            Box::pin(opt.run()).await
         }
         ImdbOpt::Convert(opt) => opt.run().await,
     }

benchmarks/src/bin/tpch.rs

@@ -58,7 +58,7 @@ async fn main() -> Result<()> {
     env_logger::init();
     match TpchOpt::from_args() {
         TpchOpt::Benchmark(BenchmarkSubCommandOpt::DataFusionBenchmark(opt)) => {
-            opt.run().await
+            Box::pin(opt.run()).await
         }
         TpchOpt::Convert(opt) => opt.run().await,
     }

datafusion-cli/tests/sql/encrypted_parquet.sql

@@ -0,0 +1,75 @@
+/*
+    Test parquet encryption and decryption in DataFusion SQL.
+    See datafusion/common/src/config.rs for equivalent rust code
+*/
+
+-- Keys are hex encoded, you can generate these via encode, e.g.
+select encode('0123456789012345', 'hex');
+/*
+Expected output:
++----------------------------------------------+
+| encode(Utf8("0123456789012345"),Utf8("hex")) |
++----------------------------------------------+
+| 30313233343536373839303132333435             |
++----------------------------------------------+
+*/
+
+CREATE EXTERNAL TABLE encrypted_parquet_table
+(
+double_field double,
+float_field float
+)
+STORED AS PARQUET LOCATION 'pq/' OPTIONS (
+    'format.crypto.file_encryption.encrypt_footer' 'true',
+    'format.crypto.file_encryption.footer_key_as_hex' '30313233343536373839303132333435',  -- b"0123456789012345"
+    'format.crypto.file_encryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450" 
+    'format.crypto.file_encryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451" 
+          -- Same for decryption 
+    'format.crypto.file_decryption.footer_key_as_hex' '30313233343536373839303132333435', -- b"0123456789012345" 
+    'format.crypto.file_decryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450" 
+    'format.crypto.file_decryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451"
+);
+
+CREATE TABLE temp_table (
+    double_field double,
+    float_field float
+);
+
+INSERT INTO temp_table VALUES(-1.0, -1.0);
+INSERT INTO temp_table VALUES(1.0, 2.0);
+INSERT INTO temp_table VALUES(3.0, 4.0);
+INSERT INTO temp_table VALUES(5.0, 6.0);
+
+INSERT INTO TABLE encrypted_parquet_table(double_field, float_field) SELECT * FROM temp_table;
+
+SELECT * FROM encrypted_parquet_table
+WHERE double_field > 0.0 AND float_field > 0.0;
+
+/*
+Expected output:
++--------------+-------------+
+| double_field | float_field |
++--------------+-------------+
+| 1.0          | 2.0         |
+| 5.0          | 6.0         |
+| 3.0          | 4.0         |
++--------------+-------------+
+*/
+
+CREATE EXTERNAL TABLE parquet_table
+(
+double_field double,
+float_field float
+)
+STORED AS PARQUET LOCATION 'pq/';
+
+SELECT * FROM parquet_table;
+/*
+Expected output:
+Parquet error: Parquet error: Parquet file has an encrypted footer but decryption properties were not provided
+*/
+
+
+
+
+

datafusion-examples/README.md

@@ -65,6 +65,7 @@ cargo run --example dataframe
 - [`flight_sql_server.rs`](examples/flight/flight_sql_server.rs): Run DataFusion as a standalone process and execute SQL queries from JDBC clients
 - [`function_factory.rs`](examples/function_factory.rs): Register `CREATE FUNCTION` handler to implement SQL macros
 - [`optimizer_rule.rs`](examples/optimizer_rule.rs): Use a custom OptimizerRule to replace certain predicates
+- [`parquet_encrypted.rs`](examples/parquet_encrypted.rs): Read and write encrypted Parquet files using DataFusion
 - [`parquet_index.rs`](examples/parquet_index.rs): Create an secondary index over several parquet files and use it to speed up queries
 - [`parquet_exec_visitor.rs`](examples/parquet_exec_visitor.rs): Extract statistics by visiting an ExecutionPlan after execution
 - [`parse_sql_expr.rs`](examples/parse_sql_expr.rs): Parse SQL text into DataFusion `Expr`.

datafusion-examples/examples/parquet_encrypted.rs

@@ -0,0 +1,118 @@
+// 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.
+
+use datafusion::common::DataFusionError;
+use datafusion::config::TableParquetOptions;
+use datafusion::dataframe::{DataFrame, DataFrameWriteOptions};
+use datafusion::logical_expr::{col, lit};
+use datafusion::parquet::encryption::decrypt::FileDecryptionProperties;
+use datafusion::parquet::encryption::encrypt::FileEncryptionProperties;
+use datafusion::prelude::{ParquetReadOptions, SessionContext};
+use tempfile::TempDir;
+
+#[tokio::main]
+async fn main() -> datafusion::common::Result<()> {
+    // The SessionContext is the main high level API for interacting with DataFusion
+    let ctx = SessionContext::new();
+
+    // Find the local path of "alltypes_plain.parquet"
+    let testdata = datafusion::test_util::parquet_test_data();
+    let filename = &format!("{testdata}/alltypes_plain.parquet");
+
+    // Read the sample parquet file
+    let parquet_df = ctx
+        .read_parquet(filename, ParquetReadOptions::default())
+        .await?;
+
+    // Show information from the dataframe
+    println!(
+        "==============================================================================="
+    );
+    println!("Original Parquet DataFrame:");
+    query_dataframe(&parquet_df).await?;
+
+    // Setup encryption and decryption properties
+    let (encrypt, decrypt) = setup_encryption(&parquet_df)?;
+
+    // Create a temporary file location for the encrypted parquet file
+    let tmp_dir = TempDir::new()?;
+    let tempfile = tmp_dir.path().join("alltypes_plain-encrypted.parquet");
+    let tempfile_str = tempfile.into_os_string().into_string().unwrap();
+
+    // Write encrypted parquet
+    let mut options = TableParquetOptions::default();
+    options.crypto.file_encryption = Some((&encrypt).into());
+    parquet_df
+        .write_parquet(
+            tempfile_str.as_str(),
+            DataFrameWriteOptions::new().with_single_file_output(true),
+            Some(options),
+        )
+        .await?;
+
+    // Read encrypted parquet
+    let ctx: SessionContext = SessionContext::new();
+    let read_options = ParquetReadOptions::default().file_decryption_properties(decrypt);
+
+    let encrypted_parquet_df = ctx.read_parquet(tempfile_str, read_options).await?;
+
+    // Show information from the dataframe
+    println!("\n\n===============================================================================");
+    println!("Encrypted Parquet DataFrame:");
+    query_dataframe(&encrypted_parquet_df).await?;
+
+    Ok(())
+}
+
+// Show information from the dataframe
+async fn query_dataframe(df: &DataFrame) -> Result<(), DataFusionError> {
+    // show its schema using 'describe'
+    println!("Schema:");
+    df.clone().describe().await?.show().await?;
+
+    // Select three columns and filter the results
+    // so that only rows where id > 1 are returned
+    println!("\nSelected rows and columns:");
+    df.clone()
+        .select_columns(&["id", "bool_col", "timestamp_col"])?
+        .filter(col("id").gt(lit(5)))?
+        .show()
+        .await?;
+
+    Ok(())
+}
+
+// Setup encryption and decryption properties
+fn setup_encryption(
+    parquet_df: &DataFrame,
+) -> Result<(FileEncryptionProperties, FileDecryptionProperties), DataFusionError> {
+    let schema = parquet_df.schema();
+    let footer_key = b"0123456789012345".to_vec(); // 128bit/16
+    let column_key = b"1234567890123450".to_vec(); // 128bit/16
+
+    let mut encrypt = FileEncryptionProperties::builder(footer_key.clone());
+    let mut decrypt = FileDecryptionProperties::builder(footer_key.clone());
+
+    for field in schema.fields().iter() {
+        encrypt = encrypt.with_column_key(field.name().as_str(), column_key.clone());
+        decrypt = decrypt.with_column_key(field.name().as_str(), column_key.clone());
+    }
+
+    let encrypt = encrypt.build()?;
+    let decrypt = decrypt.build()?;
+    Ok((encrypt, decrypt))
+}

datafusion/common/Cargo.toml

@@ -57,6 +57,7 @@ arrow-ipc = { workspace = true }
 base64 = "0.22.1"
 half = { workspace = true }
 hashbrown = { workspace = true }
+hex = "0.4.3"
 indexmap = { workspace = true }
 libc = "0.2.172"
 log = { workspace = true }