fix: enable parquet pushdown for DeltaScan via TableProvider impl for DeltaTable

crates/core/src/delta_datafusion/mod.rs

@@ -390,14 +390,13 @@ impl DeltaScanConfigBuilder {
 
  /// Build a DeltaScanConfig and ensure no column name conflicts occur during downstream processing
  pub fn build(&self, snapshot: &DeltaTableState) -> DeltaResult<DeltaScanConfig> {
- let input_schema = snapshot.input_schema()?;
- let mut file_column_name = None;
- let mut column_names: HashSet<&String> = HashSet::new();
- for field in input_schema.fields.iter() {
- column_names.insert(field.name());
- }
+ let file_column_name = if self.include_file_column {
+  let input_schema = snapshot.input_schema()?;
+  let mut column_names: HashSet<&String> = HashSet::new();
+  for field in input_schema.fields.iter() {
+  column_names.insert(field.name());
+  }
 
- if self.include_file_column {
  match &self.file_column_name {
  Some(name) => {
  if column_names.contains(name) {
@@ -407,7 +406,7 @@ impl DeltaScanConfigBuilder {
  )));
  }
 
- file_column_name = Some(name.to_owned())
+ Some(name.to_owned())
  }
  None => {
  let prefix = PATH_COLUMN;
@@ -419,10 +418,12 @@ impl DeltaScanConfigBuilder {
  name = format!("{}_{}", prefix, idx);
  }
 
- file_column_name = Some(name);
+ Some(name)
  }
  }
- }
+ } else {
+ None
+ };
 
  Ok(DeltaScanConfig {
  file_column_name,
@@ -452,7 +453,7 @@ pub(crate) struct DeltaScanBuilder<'a> {
  projection: Option<&'a Vec<usize>>,
  limit: Option<usize>,
  files: Option<&'a [Add]>,
- config: DeltaScanConfig,
+ config: Option<DeltaScanConfig>,
  schema: Option<SchemaRef>,
 }
 
@@ -470,7 +471,7 @@ impl<'a> DeltaScanBuilder<'a> {
  files: None,
  projection: None,
  limit: None,
- config: DeltaScanConfig::default(),
+ config: None,
  schema: None,
  }
  }
@@ -496,7 +497,7 @@ impl<'a> DeltaScanBuilder<'a> {
  }
 
  pub fn with_scan_config(mut self, config: DeltaScanConfig) -> Self {
- self.config = config;
+ self.config = Some(config);
  self
  }
 
@@ -507,7 +508,11 @@ impl<'a> DeltaScanBuilder<'a> {
  }
 
  pub async fn build(self) -> DeltaResult<DeltaScan> {
- let config = self.config;
+ let config = match self.config {
+ Some(config) => config,
+ None => DeltaScanConfigBuilder::new().build(self.snapshot)?,
+ };
+
  let schema = match self.schema {
  Some(schema) => schema,
  None => {
@@ -1719,7 +1724,10 @@ mod tests {
  use arrow::datatypes::{DataType, Field, Schema};
  use chrono::{TimeZone, Utc};
  use datafusion::assert_batches_sorted_eq;
+ use datafusion::datasource::physical_plan::ParquetExec;
  use datafusion::physical_plan::empty::EmptyExec;
+ use datafusion::physical_plan::{visit_execution_plan, ExecutionPlanVisitor};
+ use datafusion_expr::lit;
  use datafusion_proto::physical_plan::AsExecutionPlan;
  use datafusion_proto::protobuf;
  use object_store::path::Path;
@@ -2156,4 +2164,83 @@ mod tests {
 
  df.collect().await.unwrap();
  }
+
+ #[tokio::test]
+ async fn test_delta_scan_builder_no_scan_config() {
+ use crate::datafusion::prelude::SessionContext;
+ let arr: Arc<dyn Array> = Arc::new(arrow::array::StringArray::from(vec!["s"]));
+ let batch = RecordBatch::try_from_iter_with_nullable(vec![("a", arr, false)]).unwrap();
+ let table = crate::DeltaOps::new_in_memory()
+ .write(vec![batch])
+ .with_save_mode(crate::protocol::SaveMode::Append)
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ let scan =
+ DeltaScanBuilder::new(table.snapshot().unwrap(), table.log_store(), &ctx.state())
+ .with_filter(Some(col("a").eq(lit("s"))))
+ .build()
+ .await
+ .unwrap();
+
+ let mut visitor = ParquetPredicateVisitor::default();
+ visit_execution_plan(&scan, &mut visitor).unwrap();
+
+ assert_eq!(visitor.predicate.unwrap().to_string(), "a@0 = s");
+ assert_eq!(
+ visitor.pruning_predicate.unwrap().orig_expr().to_string(),
+ "a@0 = s"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_delta_scan_builder_scan_config_disable_pushdown() {
+ use crate::datafusion::prelude::SessionContext;
+ let arr: Arc<dyn Array> = Arc::new(arrow::array::StringArray::from(vec!["s"]));
+ let batch = RecordBatch::try_from_iter_with_nullable(vec![("a", arr, false)]).unwrap();
+ let table = crate::DeltaOps::new_in_memory()
+ .write(vec![batch])
+ .with_save_mode(crate::protocol::SaveMode::Append)
+ .await
+ .unwrap();
+
+ let ctx = SessionContext::new();
+ let snapshot = table.snapshot().unwrap();
+ let scan = DeltaScanBuilder::new(snapshot, table.log_store(), &ctx.state())
+ .with_filter(Some(col("a").eq(lit("s"))))
+ .with_scan_config(
+ DeltaScanConfigBuilder::new()
+ .with_parquet_pushdown(false)
+ .build(snapshot)
+ .unwrap(),
+ )
+ .build()
+ .await
+ .unwrap();
+
+ let mut visitor = ParquetPredicateVisitor::default();
+ visit_execution_plan(&scan, &mut visitor).unwrap();
+
+ assert!(visitor.predicate.is_none());
+ assert!(visitor.pruning_predicate.is_none());
+ }
+
+ #[derive(Default)]
+ struct ParquetPredicateVisitor {
+ predicate: Option<Arc<dyn PhysicalExpr>>,
+ pruning_predicate: Option<Arc<PruningPredicate>>,
+ }
+
+ impl ExecutionPlanVisitor for ParquetPredicateVisitor {
+ type Error = DataFusionError;
+
+ fn pre_visit(&mut self, plan: &dyn ExecutionPlan) -> Result<bool, Self::Error> {
+ if let Some(parquet_exec) = plan.as_any().downcast_ref::<ParquetExec>() {
+ self.predicate = parquet_exec.predicate().cloned();
+ self.pruning_predicate = parquet_exec.pruning_predicate().cloned();
+ }
+ Ok(true)
+ }
+ }
 }