feat: delta engine injection support

Author: dvlascenco

crates/core/src/delta_datafusion/engine/mod.rs

@@ -15,6 +15,10 @@ mod expressions;
 mod file_formats;
 mod storage;
 
+/// Wraps a custom [`Engine`] for injection into a DataFusion session config extension.
+#[derive(Clone)]
+pub struct InjectedEngine(pub Arc<dyn Engine>);
+
 /// A Datafusion based Kernel Engine
 #[derive(Clone)]
 pub struct DataFusionEngine {

crates/core/src/delta_datafusion/mod.rs

@@ -72,6 +72,7 @@ pub use cdf::scan::DeltaCdfTableProvider;
 pub(crate) use data_validation::{
     DataValidationExec, constraints_to_exprs, generated_columns_to_exprs, validation_predicates,
 };
+pub use engine::{DataFusionEngine, InjectedEngine};
 pub(crate) use find_files::*;
 pub use table_provider::{
     DeltaScan, DeltaScanConfig, DeltaScanConfigBuilder, DeltaTableProvider, TableProviderBuilder,

crates/core/src/delta_datafusion/table_provider/next/mod.rs

@@ -40,13 +40,14 @@ use datafusion::{
     physical_plan::ExecutionPlan,
 };
 use datafusion::catalog::{ScanArgs, ScanResult};
+use delta_kernel::Engine;
 use delta_kernel::table_configuration::TableConfiguration;
 use serde::{Deserialize, Serialize};
 
 pub use self::scan::{DeltaScanExec, DeltaNextPhysicalCodec};
 pub(crate) use self::scan::KernelScanPlan;
 use crate::delta_datafusion::DeltaScanConfig;
-use crate::delta_datafusion::engine::DataFusionEngine;
+use crate::delta_datafusion::engine::{DataFusionEngine, InjectedEngine};
 use crate::delta_datafusion::table_provider::TableProviderBuilder;
 use crate::kernel::{EagerSnapshot, Snapshot};
 
@@ -175,7 +176,11 @@ impl TableProvider for DeltaScan {
         filters: &[Expr],
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        let engine = DataFusionEngine::new_from_session(session);
+        let engine: Arc<dyn Engine> = session
+            .config()
+            .get_extension::<InjectedEngine>()
+            .map(|e| e.0.clone())
+            .unwrap_or_else(|| DataFusionEngine::new_from_session(session));
 
         // Filter out file_id column from projection if present
         let file_id_idx = self
@@ -218,7 +223,11 @@ impl TableProvider for DeltaScan {
     }
 
     async fn scan_with_args<'a>(&self, state: &dyn Session, args: ScanArgs<'a>) -> Result<ScanResult> {
-        let engine = DataFusionEngine::new_from_session(state);
+        let engine: Arc<dyn Engine> = state
+            .config()
+            .get_extension::<InjectedEngine>()
+            .map(|e| e.0.clone())
+            .unwrap_or_else(|| DataFusionEngine::new_from_session(state));
 
         // Filter out file_id column from projection if present
         let file_id_idx = self