feat: add framework for File Format Options

Cargo.toml

@@ -49,7 +49,7 @@ object_store = { version = "0.12.1" }
 parquet = { version = "56.0.0" }
 
 # datafusion
-datafusion = "50.0.0"
+datafusion = { version = "50.0.0" , features = ["parquet_encryption", "default"]}
 datafusion-ffi = "50.0.0"
 datafusion-proto = "50.0.0"
 
@@ -70,7 +70,7 @@ tempfile = { version = "3" }
 uuid = { version = "1" }
 
 # runtime / async
-async-trait = { version = "0.1" }
+async-trait = { version = "0.1.89" }
 futures = { version = "0.3" }
 tokio = { version = "1" }
 

crates/core/Cargo.toml

@@ -31,7 +31,7 @@ arrow-ord = { workspace = true }
 arrow-row = { workspace = true }
 arrow-schema = { workspace = true, features = ["serde"] }
 arrow-select = { workspace = true }
-parquet = { workspace = true, features = ["async", "object_store"] }
+parquet = { workspace = true, features = ["async", "object_store", "encryption"] }
 object_store = { workspace = true }
 
 # datafusion
@@ -94,6 +94,8 @@ datatest-stable = "0.3"
 deltalake-test = { path = "../test" }
 dotenvy = "0"
 fs_extra = "1.2.0"
+parquet-key-management = { version = "0.4.0", features = ["_test_utils", "datafusion"] }
+paste = "1"
 pretty_assertions = "1.2.1"
 pretty_env_logger = "0.5.0"
 rstest = { version = "0.26.1" }
@@ -145,6 +147,10 @@ required-features = ["datafusion"]
 name = "command_vacuum"
 required-features = ["datafusion"]
 
+[[test]]
+name = "commands_with_encryption"
+required-features = ["datafusion"]
+
 [[test]]
 name = "commit_info_format"
 required-features = ["datafusion"]

crates/core/src/delta_datafusion/find_files.rs

@@ -21,6 +21,7 @@ use crate::delta_datafusion::{
 use crate::errors::{DeltaResult, DeltaTableError};
 use crate::kernel::{Add, EagerSnapshot};
 use crate::logstore::LogStoreRef;
+use crate::table::file_format_options::{to_table_parquet_options_from_ffo, FileFormatRef};
 
 #[derive(Debug, Hash, Eq, PartialEq)]
 /// Representing the result of the [find_files] function.
@@ -45,6 +46,7 @@ pub(crate) async fn find_files(
     snapshot: &EagerSnapshot,
     log_store: LogStoreRef,
     session: &dyn Session,
+    file_format_options: Option<&FileFormatRef>,
     predicate: Option<Expr>,
 ) -> DeltaResult<FindFiles> {
     let current_metadata = snapshot.metadata();
@@ -71,8 +73,14 @@ pub(crate) async fn find_files(
                 Span::current().record("candidate_count", result.candidates.len());
                 Ok(result)
             } else {
-                let candidates =
-                    find_files_scan(snapshot, log_store, session, predicate.to_owned()).await?;
+                let candidates = find_files_scan(
+                    snapshot,
+                    log_store,
+                    session,
+                    file_format_options,
+                    predicate.to_owned(),
+                )
+                .await?;
 
                 let result = FindFiles {
                     candidates,
@@ -221,6 +229,7 @@ async fn find_files_scan(
     snapshot: &EagerSnapshot,
     log_store: LogStoreRef,
     session: &dyn Session,
+    file_format_options: Option<&FileFormatRef>,
     expression: Expr,
 ) -> DeltaResult<Vec<Add>> {
     let candidate_map: HashMap<String, Add> = snapshot
@@ -250,10 +259,13 @@ async fn find_files_scan(
     // Add path column
     used_columns.push(logical_schema.index_of(scan_config.file_column_name.as_ref().unwrap())?);
 
+    let table_parquet_options = to_table_parquet_options_from_ffo(file_format_options);
+
     let scan = DeltaScanBuilder::new(snapshot, log_store, session)
         .with_filter(Some(expression.clone()))
         .with_projection(Some(&used_columns))
         .with_scan_config(scan_config)
+        .with_parquet_options(table_parquet_options)
         .build()
         .await?;
     let scan = Arc::new(scan);

crates/core/src/delta_datafusion/table_provider.rs

@@ -59,6 +59,7 @@ use crate::kernel::{Action, Add, EagerSnapshot, Remove};
 use crate::operations::write::writer::{DeltaWriter, WriterConfig};
 use crate::operations::write::WriterStatsConfig;
 use crate::protocol::{DeltaOperation, SaveMode};
+use crate::table::file_format_options::{to_table_parquet_options_from_ffo, FileFormatRef};
 use crate::{ensure_table_uri, DeltaTable};
 use crate::{logstore::LogStoreRef, DeltaResult, DeltaTableError};
 
@@ -373,6 +374,7 @@ pub(crate) struct DeltaScanBuilder<'a> {
     limit: Option<usize>,
     files: Option<&'a [Add]>,
     config: Option<DeltaScanConfig>,
+    parquet_options: Option<TableParquetOptions>,
 }
 
 impl<'a> DeltaScanBuilder<'a> {
@@ -390,9 +392,15 @@ impl<'a> DeltaScanBuilder<'a> {
             limit: None,
             files: None,
             config: None,
+            parquet_options: None,
         }
     }
 
+    pub fn with_parquet_options(mut self, parquet_options: Option<TableParquetOptions>) -> Self {
+        self.parquet_options = parquet_options;
+        self
+    }
+
     pub fn with_filter(mut self, filter: Option<Expr>) -> Self {
         self.filter = filter;
         self
@@ -643,13 +651,30 @@ impl<'a> DeltaScanBuilder<'a> {
 
         let stats = stats.unwrap_or(Statistics::new_unknown(&schema));
 
-        let parquet_options = TableParquetOptions {
-            global: self.session.config().options().execution.parquet.clone(),
-            ..Default::default()
-        };
+        let parquet_options = self
+            .parquet_options
+            .unwrap_or_else(|| self.session.table_options().parquet.clone());
+
+        // We have to set the encryption factory on the ParquetSource based on the Parquet options,
+        // as this is usually handled by the ParquetFormat type in DataFusion,
+        // which is not used in delta-rs.
+        let encryption_factory = parquet_options
+            .crypto
+            .factory_id
+            .as_ref()
+            .map(|factory_id| {
+                self.session
+                    .runtime_env()
+                    .parquet_encryption_factory(factory_id)
+            })
+            .transpose()?;
 
         let mut file_source = ParquetSource::new(parquet_options);
 
+        if let Some(encryption_factory) = encryption_factory {
+            file_source = file_source.with_encryption_factory(encryption_factory);
+        }
+
         // Sometimes (i.e Merge) we want to prune files that don't make the
         // filter and read the entire contents for files that do match the
         // filter
@@ -730,9 +755,17 @@ impl TableProvider for DeltaTable {
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         register_store(self.log_store(), session.runtime_env().as_ref());
+        if let Some(format_options) = &self.config.file_format_options {
+            format_options.update_session(session)?;
+        }
         let filter_expr = conjunction(filters.iter().cloned());
 
         let scan = DeltaScanBuilder::new(self.snapshot()?.snapshot(), self.log_store(), session)
+            .with_parquet_options(
+                crate::table::file_format_options::to_table_parquet_options_from_ffo(
+                    self.config.file_format_options.as_ref(),
+                ),
+            )
             .with_projection(projection)
             .with_limit(limit)
             .with_filter(filter_expr)
@@ -763,6 +796,7 @@ pub struct DeltaTableProvider {
     config: DeltaScanConfig,
     schema: Arc<Schema>,
     files: Option<Vec<Add>>,
+    file_format_options: Option<FileFormatRef>,
 }
 
 impl DeltaTableProvider {
@@ -778,9 +812,15 @@ impl DeltaTableProvider {
             log_store,
             config,
             files: None,
+            file_format_options: None,
         })
     }
 
+    pub fn with_file_format_options(mut self, file_format_options: Option<FileFormatRef>) -> Self {
+        self.file_format_options = file_format_options;
+        self
+    }
+
     /// Define which files to consider while building a scan, for advanced usecases
     pub fn with_files(mut self, files: Vec<Add>) -> DeltaTableProvider {
         self.files = Some(files);
@@ -818,9 +858,17 @@ impl TableProvider for DeltaTableProvider {
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         register_store(self.log_store.clone(), session.runtime_env().as_ref());
+        if let Some(format_options) = &self.file_format_options {
+            format_options.update_session(session)?;
+        }
+
         let filter_expr = conjunction(filters.iter().cloned());
 
+        let table_parquet_options =
+            to_table_parquet_options_from_ffo(self.file_format_options.as_ref());
+
         let mut scan = DeltaScanBuilder::new(&self.snapshot, self.log_store.clone(), session)
+            .with_parquet_options(table_parquet_options)
             .with_projection(projection)
             .with_limit(limit)
             .with_filter(filter_expr)

crates/core/src/operations/add_column.rs

@@ -1,10 +1,9 @@
 //! Add a new column to a table
 
-use std::sync::Arc;
-
 use delta_kernel::schema::StructType;
 use futures::future::BoxFuture;
 use itertools::Itertools;
+use std::sync::Arc;
 
 use super::{CustomExecuteHandler, Operation};
 use crate::kernel::schema::merge_delta_struct;

crates/core/src/operations/create.rs

@@ -1,12 +1,11 @@
 //! Command for creating a new delta table
 // https://github.com/delta-io/delta/blob/master/core/src/main/scala/org/apache/spark/sql/delta/commands/CreateDeltaTableCommand.scala
 
-use std::collections::HashMap;
-use std::sync::Arc;
-
 use delta_kernel::schema::MetadataValue;
 use futures::future::BoxFuture;
 use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::Arc;
 use tracing::log::*;
 use uuid::Uuid;
 
@@ -21,7 +20,7 @@ use crate::logstore::LogStoreRef;
 use crate::protocol::{DeltaOperation, SaveMode};
 use crate::table::builder::ensure_table_uri;
 use crate::table::config::TableProperty;
-use crate::{DeltaTable, DeltaTableBuilder};
+use crate::{DeltaTable, DeltaTableBuilder, DeltaTableConfig};
 
 #[derive(thiserror::Error, Debug)]
 enum CreateError {
@@ -61,6 +60,7 @@ pub struct CreateBuilder {
     storage_options: Option<HashMap<String, String>>,
     actions: Vec<Action>,
     log_store: Option<LogStoreRef>,
+    table_config: DeltaTableConfig,
     configuration: HashMap<String, Option<String>>,
     /// Additional information to add to the commit
     commit_properties: CommitProperties,
@@ -98,6 +98,7 @@ impl CreateBuilder {
             storage_options: None,
             actions: Default::default(),
             log_store: None,
+            table_config: DeltaTableConfig::default(),
             configuration: Default::default(),
             commit_properties: CommitProperties::default(),
             raise_if_key_not_exists: true,
@@ -238,6 +239,12 @@ impl CreateBuilder {
         self
     }
 
+    /// Set configuration options for the table
+    pub fn with_table_config(mut self, table_config: DeltaTableConfig) -> Self {
+        self.table_config = table_config;
+        self
+    }
+
     /// Set a custom execute handler, for pre and post execution
     pub fn with_custom_execute_handler(mut self, handler: Arc<dyn CustomExecuteHandler>) -> Self {
         self.custom_execute_handler = Some(handler);
@@ -262,14 +269,15 @@ impl CreateBuilder {
         let (storage_url, table) = if let Some(log_store) = self.log_store {
             (
                 ensure_table_uri(log_store.root_uri())?.as_str().to_string(),
-                DeltaTable::new(log_store, Default::default()),
+                DeltaTable::new(log_store, self.table_config.clone()),
             )
         } else {
             let storage_url =
                 ensure_table_uri(self.location.clone().ok_or(CreateError::MissingLocation)?)?;
             (
                 storage_url.as_str().to_string(),
                 DeltaTableBuilder::from_uri(storage_url)?
+                    .with_table_config(self.table_config.clone())
                     .with_storage_options(self.storage_options.clone().unwrap_or_default())
                     .build()?,
             )