[HSTACK] Add support for deep projections

crates/core/src/delta_datafusion/mod.rs

@@ -461,6 +461,7 @@ pub(crate) struct DeltaScanBuilder<'a> {
     filter: Option<Expr>,
     session: &'a dyn Session,
     projection: Option<&'a Vec<usize>>,
+    projection_deep: Option<&'a HashMap<usize, Vec<String>>>, 
     limit: Option<usize>,
     files: Option<&'a [Add]>,
     config: Option<DeltaScanConfig>,
@@ -478,6 +479,7 @@ impl<'a> DeltaScanBuilder<'a> {
             filter: None,
             session,
             projection: None,
+            projection_deep: None,
             limit: None,
             files: None,
             config: None,
@@ -499,6 +501,12 @@ impl<'a> DeltaScanBuilder<'a> {
         self
     }
 
+    pub fn with_projection_deep(mut self, projection_deep: Option<&'a HashMap<usize, Vec<String>>>) -> Self {
+        self.projection_deep = projection_deep;
+        self
+    }
+
+
     pub fn with_limit(mut self, limit: Option<usize>) -> Self {
         self.limit = limit;
         self
@@ -655,6 +663,7 @@ impl<'a> DeltaScanBuilder<'a> {
             },
             statistics: stats,
             projection: self.projection.cloned(),
+            projection_deep: self.projection_deep.cloned(),
             limit: self.limit,
             table_partition_cols,
             output_ordering: vec![],
@@ -732,6 +741,30 @@ impl TableProvider for DeltaTable {
         Ok(Arc::new(scan))
     }
 
+    async fn scan_deep(
+        &self,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        projection_deep: Option<&HashMap<usize, Vec<String>>>,
+        filters: &[Expr],
+        limit: Option<usize>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        let session_state = state.as_any().downcast_ref::<SessionState>().unwrap();
+
+        register_store(self.log_store(), session_state.runtime_env().clone());
+        let filter_expr = conjunction(filters.iter().cloned());
+
+        let scan = DeltaScanBuilder::new(self.snapshot()?, self.log_store(), session_state)
+            .with_projection(projection)
+            .with_projection_deep(projection_deep)
+            .with_limit(limit)
+            .with_filter(filter_expr)
+            .build()
+            .await?;
+
+        Ok(Arc::new(scan))
+    }
+
     fn supports_filters_pushdown(
         &self,
         filter: &[&Expr],
@@ -824,6 +857,30 @@ impl TableProvider for DeltaTableProvider {
         Ok(Arc::new(scan.build().await?))
     }
 
+    async fn scan_deep(
+        &self,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        projection_deep: Option<&HashMap<usize, Vec<String>>>,
+        filters: &[Expr],
+        limit: Option<usize>,
+    ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+        let session_state = state.as_any().downcast_ref::<SessionState>().unwrap();
+        register_store(self.log_store.clone(), session_state.runtime_env().clone());
+        let filter_expr = conjunction(filters.iter().cloned());
+
+        let scan = DeltaScanBuilder::new(&self.snapshot, self.log_store.clone(), session_state)
+            .with_projection(projection)
+            .with_projection_deep(projection_deep)
+            .with_limit(limit)
+            .with_filter(filter_expr)
+            .with_scan_config(self.config.clone())
+            .build()
+            .await?;
+
+        Ok(Arc::new(scan))
+    }
+
     fn supports_filters_pushdown(
         &self,
         filter: &[&Expr],

crates/core/src/operations/load_cdf.rs

@@ -364,6 +364,7 @@ impl CdfLoadBuilder {
                     file_groups: cdc_file_groups.into_values().collect(),
                     statistics: Statistics::new_unknown(&cdc_file_schema),
                     projection: None,
+                    projection_deep: None,
                     limit: None,
                     table_partition_cols: cdc_partition_cols,
                     output_ordering: vec![],
@@ -381,6 +382,7 @@ impl CdfLoadBuilder {
                     file_groups: add_file_groups.into_values().collect(),
                     statistics: Statistics::new_unknown(&add_remove_file_schema.clone()),
                     projection: None,
+                    projection_deep: None,
                     limit: None,
                     table_partition_cols: add_remove_partition_cols.clone(),
                     output_ordering: vec![],
@@ -398,6 +400,7 @@ impl CdfLoadBuilder {
                     file_groups: remove_file_groups.into_values().collect(),
                     statistics: Statistics::new_unknown(&add_remove_file_schema),
                     projection: None,
+                    projection_deep: None,
                     limit: None,
                     table_partition_cols: add_remove_partition_cols,
                     output_ordering: vec![],