feat(python, rust): arrow large/view types passthrough, rust default engine

.github/workflows/python_build.yml

@@ -31,7 +31,7 @@ jobs:
         run: make check-rust
 
   test-minimal:
-    name: Python Build (Python 3.8 PyArrow 8.0.0)
+    name: Python Build (Python 3.8 PyArrow 16.0.0)
     runs-on: ubuntu-latest
     env:
       RUSTFLAGS: "-C debuginfo=line-tables-only"
@@ -51,7 +51,7 @@ jobs:
           source venv/bin/activate
           make setup
           # Install minimum PyArrow version
-          pip install -e .[pandas,devel] pyarrow==8.0.0
+          pip install -e .[pandas,devel] pyarrow==16.0.0
         env:
           RUSTFLAGS: "-C debuginfo=line-tables-only"
 
@@ -60,10 +60,6 @@ jobs:
           source venv/bin/activate
           make unit-test
 
-    # - name: Run Integration tests
-    #   run: |
-    #     py.test --cov tests -m integration
-
   test:
     name: Python Build (Python 3.10 PyArrow latest)
     runs-on: ubuntu-latest

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

@@ -137,7 +137,7 @@ async fn scan_table_by_files(
         .with_file_column(true)
         .build(&snapshot)?;
 
-    let logical_schema = df_logical_schema(&snapshot, &scan_config)?;
+    let logical_schema = df_logical_schema(&snapshot, &scan_config.file_column_name, None)?;
 
     // Identify which columns we need to project
     let mut used_columns = expression

crates/core/src/delta_datafusion/mod.rs

@@ -305,9 +305,13 @@ pub(crate) fn register_store(store: LogStoreRef, env: Arc<RuntimeEnv>) {
 /// at the physical level
 pub(crate) fn df_logical_schema(
     snapshot: &DeltaTableState,
-    scan_config: &DeltaScanConfig,
+    file_column_name: &Option<String>,
+    schema: Option<ArrowSchemaRef>,
 ) -> DeltaResult<SchemaRef> {
-    let input_schema = snapshot.arrow_schema()?;
+    let input_schema = match schema {
+        Some(schema) => schema,
+        None => snapshot.input_schema()?,
+    };
     let table_partition_cols = &snapshot.metadata().partition_columns;
 
     let mut fields: Vec<Arc<Field>> = input_schema
@@ -326,7 +330,7 @@ pub(crate) fn df_logical_schema(
         ));
     }
 
-    if let Some(file_column_name) = &scan_config.file_column_name {
+    if let Some(file_column_name) = file_column_name {
         fields.push(Arc::new(Field::new(file_column_name, DataType::Utf8, true)));
     }
 
@@ -528,7 +532,11 @@ impl<'a> DeltaScanBuilder<'a> {
             None => self.snapshot.arrow_schema(),
         }?;
 
-        let logical_schema = df_logical_schema(self.snapshot, &config)?;
+        let logical_schema = df_logical_schema(
+            self.snapshot,
+            &config.file_column_name,
+            Some(schema.clone()),
+        )?;
 
         let logical_schema = if let Some(used_columns) = self.projection {
             let mut fields = vec![];
@@ -733,7 +741,7 @@ impl TableProvider for DeltaTable {
         filter: &[&Expr],
     ) -> DataFusionResult<Vec<TableProviderFilterPushDown>> {
         Ok(filter
-            .into_iter()
+            .iter()
             .map(|_| TableProviderFilterPushDown::Inexact)
             .collect())
     }
@@ -760,7 +768,7 @@ impl DeltaTableProvider {
         config: DeltaScanConfig,
     ) -> DeltaResult<Self> {
         Ok(DeltaTableProvider {
-            schema: df_logical_schema(&snapshot, &config)?,
+            schema: df_logical_schema(&snapshot, &config.file_column_name, config.schema.clone())?,
             snapshot,
             log_store,
             config,
@@ -1524,7 +1532,7 @@ pub(crate) async fn find_files_scan<'a>(
     }
     .build(snapshot)?;
 
-    let logical_schema = df_logical_schema(snapshot, &scan_config)?;
+    let logical_schema = df_logical_schema(snapshot, &scan_config.file_column_name, None)?;
 
     // Identify which columns we need to project
     let mut used_columns = expression

crates/core/src/operations/add_column.rs

@@ -4,11 +4,10 @@ use delta_kernel::schema::StructType;
 use futures::future::BoxFuture;
 use itertools::Itertools;
 
-use super::cast::merge_struct;
 use super::transaction::{CommitBuilder, CommitProperties, PROTOCOL};
-
 use crate::kernel::StructField;
 use crate::logstore::LogStoreRef;
+use crate::operations::cast::merge_schema::merge_delta_struct;
 use crate::protocol::DeltaOperation;
 use crate::table::state::DeltaTableState;
 use crate::{DeltaResult, DeltaTable, DeltaTableError};
@@ -67,7 +66,7 @@ impl std::future::IntoFuture for AddColumnBuilder {
 
             let fields_right = &StructType::new(fields.clone());
             let table_schema = this.snapshot.schema();
-            let new_table_schema = merge_struct(table_schema, fields_right)?;
+            let new_table_schema = merge_delta_struct(table_schema, fields_right)?;
 
             // TODO(ion): Think of a way how we can simply this checking through the API or centralize some checks.
             let contains_timestampntz = PROTOCOL.contains_timestampntz(fields.iter());