Pick up delta-rs DF upgrade and fix a couple of tests

Author: gruuya

Cargo.lock

@@ -96,7 +96,7 @@ datafusion-functions-nested = { workspace = true }
 
 datafusion-remote-tables = { path = "./datafusion_remote_tables", optional = true }
 
-deltalake = { git = "https://github.com/ion-elgreco/delta-rs", rev = "e363ee6a41f2c0b12a748b8fcbd45faf6f29dc93", features = ["datafusion"] }
+deltalake = { git = "https://github.com/splitgraph/delta-rs", branch = "fix-decimal-stat-overflow", features = ["datafusion"] }
 
 futures = "0.3"
 hex = ">=0.4.0"

Cargo.toml

@@ -37,6 +37,7 @@ use datafusion::{
     physical_plan::{ExecutionPlan, SendableRecordBatchStream},
     sql::TableReference,
 };
+use datafusion_common::config::TableParquetOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{Column as ColumnExpr, ResolvedTableReference, SchemaReference};
 use datafusion_expr::logical_plan::{
@@ -941,7 +942,12 @@ impl SeafowlContext {
         let table_path = ListingTableUrl::parse(file_path)?;
         let file_format: Arc<dyn FileFormat> = match file_type {
             "csv" => Arc::new(CsvFormat::default().with_has_header(has_header)),
-            "parquet" => Arc::new(ParquetFormat::default()),
+            "parquet" => {
+                // TODO: We can remove this once delta-rs supports Utf8View
+                let mut options = TableParquetOptions::default();
+                options.global.schema_force_view_types = false;
+                Arc::new(ParquetFormat::default().with_options(options))
+            }
             _ => {
                 return Err(Error::Plan(format!(
                     "File type {file_type:?} not supported!"

src/context/physical.rs

@@ -985,7 +985,7 @@ mod tests {
             "              ProjectionExec: expr=[c1@0 as c1, c2@1 as c2, true as __lower_rel]",
             "                DeltaScan",
             "                  RepartitionExec: partitioning=RoundRobinBatch",
-            "                    ParquetExec: file_groups={0 groups: []}, projection=[c1, c2]",
+            "                    ParquetExec: file_groups={1 group: [[]]}, projection=[c1, c2]",
             "          CoalesceBatchesExec: target_batch_size=8192",
             "            RepartitionExec: partitioning=Hash",
             "              UnnestExec",