chore: dependency updates (#19)

Author: roeap

Cargo.toml

@@ -13,16 +13,12 @@ rust-version = "1.85"
 version = "0.0.1"
 
 [workspace.dependencies]
-delta_kernel = { version = "0.14.0", features = [
-  "default-engine",
-  "arrow",
+delta_kernel = { version = "0.15.1", features = [
+  "default-engine-rustls",
+  "arrow-55",
   "internal-api",
 ] }
-# unitycatalog-client = { path = "../unitycatalog-rs/crates/client", features = [
-#   "sharing",
-# ] }
+# unitycatalog-client = { path = "../unitycatalog-rs/crates/client" }
 # unitycatalog-common = { path = "../unitycatalog-rs/crates/common", default-features = false }
-unitycatalog-client = { git = "https://github.com/unitycatalog-incubator/unitycatalog-rs", rev = "68f0a4b25178f61bdbd4b8e4a1819bdc8f48a24c", features = [
-  "sharing",
-] }
-unitycatalog-common = { git = "https://github.com/unitycatalog-incubator/unitycatalog-rs", rev = "68f0a4b25178f61bdbd4b8e4a1819bdc8f48a24c", default-features = false }
+unitycatalog-client = { git = "https://github.com/unitycatalog-incubator/unitycatalog-rs", rev = "02e948cd1ac3bcebc751fed52b29e90aa0152f36" }
+unitycatalog-common = { git = "https://github.com/unitycatalog-incubator/unitycatalog-rs", rev = "02e948cd1ac3bcebc751fed52b29e90aa0152f36", default-features = false }

crates/acceptance/src/data.rs

@@ -6,14 +6,14 @@ use delta_kernel::arrow::compute::{
 };
 use delta_kernel::arrow::datatypes::{DataType, Schema};
 
-use delta_kernel::object_store::{ObjectStore, local::LocalFileSystem};
 use delta_kernel::parquet::arrow::async_reader::{
     ParquetObjectReader, ParquetRecordBatchStreamBuilder,
 };
 use delta_kernel::snapshot::Snapshot;
 use delta_kernel::{DeltaResult, Engine, Error, engine::arrow_data::ArrowEngineData};
 use futures::{StreamExt, stream::TryStreamExt};
 use itertools::Itertools;
+use object_store::{ObjectStore, local::LocalFileSystem};
 
 use crate::{TestCaseInfo, TestResult};
 

crates/acceptance/src/meta.rs

@@ -3,8 +3,8 @@ use std::fs::File;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use delta_kernel::object_store::{self, ObjectStore, local::LocalFileSystem};
 use futures::stream::TryStreamExt;
+use object_store::{self, ObjectStore, local::LocalFileSystem};
 use serde::{Deserialize, Serialize};
 use url::Url;
 

crates/datafusion/Cargo.toml

@@ -24,6 +24,7 @@ chrono = "0.4.40"
 dashmap = "6.0.1"
 futures = "0.3"
 itertools = "0.14"
+object_store = "0.12.3"
 parking_lot = "0.12"
 serde = "1.0"
 serde_json = "1.0"

crates/datafusion/src/engine/evaluation.rs

@@ -52,7 +52,7 @@ impl DataFusionEvaluationHandler {
     fn evaluator(
         &self,
         schema: SchemaRef,
-        expression: Expression,
+        expression: Arc<Expression>,
         output_type: DataType,
     ) -> Arc<DataFusionExpressionEvaluator> {
         let maybe_schema: Result<ArrowSchema, _> = schema.as_ref().try_into_arrow();
@@ -79,7 +79,7 @@ impl EvaluationHandler for DataFusionEvaluationHandler {
     fn new_expression_evaluator(
         &self,
         schema: SchemaRef,
-        expression: Expression,
+        expression: Arc<Expression>,
         output_type: DataType,
     ) -> Arc<dyn ExpressionEvaluator> {
         self.evaluator(schema, expression, output_type)
@@ -88,11 +88,11 @@ impl EvaluationHandler for DataFusionEvaluationHandler {
     fn new_predicate_evaluator(
         &self,
         schema: SchemaRef,
-        predicate: Predicate,
+        predicate: Arc<Predicate>,
     ) -> Arc<dyn PredicateEvaluator> {
         self.evaluator(
             schema,
-            Expression::Predicate(Box::new(predicate)),
+            Expression::Predicate(Box::new(predicate.as_ref().clone())).into(),
             DataType::BOOLEAN,
         )
     }
@@ -268,7 +268,7 @@ mod tests {
 
         let evaluator = handler.new_expression_evaluator(
             input_schema,
-            expr,
+            expr.into(),
             DataType::Primitive(PrimitiveType::Integer),
         );
 
@@ -305,7 +305,7 @@ mod tests {
             StructField::new("y", DataType::Primitive(PrimitiveType::String), true),
         ])));
 
-        let evaluator = handler.new_expression_evaluator(input_schema, expr, output_type);
+        let evaluator = handler.new_expression_evaluator(input_schema, expr.into(), output_type);
         let result = evaluator.evaluate(&data).unwrap();
         let engine_data = ArrowEngineData::try_from_engine_data(result).unwrap();
 
@@ -332,7 +332,7 @@ mod tests {
 
         let evaluator = handler.new_expression_evaluator(
             input_schema,
-            expr,
+            expr.into(),
             DataType::Primitive(PrimitiveType::Integer),
         );
 
@@ -362,7 +362,7 @@ mod tests {
 
         let evaluator = handler.new_expression_evaluator(
             input_schema,
-            expr,
+            expr.into(),
             DataType::Primitive(PrimitiveType::Integer),
         );
 

crates/datafusion/src/engine/expressions/to_datafusion.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use datafusion::common::scalar::ScalarStructBuilder;
 use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue, not_impl_err};
 use datafusion::functions::core::expr_ext::FieldAccessor;
@@ -28,6 +30,7 @@ pub(crate) fn to_datafusion_expr(expr: &Expression, output_type: &DataType) -> D
         Expression::Binary(expr) => binary_to_df(expr, output_type),
         Expression::Opaque(_) => not_impl_err!("Opaque expressions are not yet supported"),
         Expression::Unknown(_) => not_impl_err!("Unknown expressions are not yet supported"),
+        Expression::Transform(_) => not_impl_err!("Transform expressions are not yet supported"),
     }
 }
 
@@ -151,7 +154,7 @@ fn junction_to_df(junction: &JunctionPredicate, output_type: &DataType) -> DFRes
     }
 }
 
-fn struct_to_df(fields: &[Expression], output_type: &DataType) -> DFResult<Expr> {
+fn struct_to_df(fields: &[Arc<Expression>], output_type: &DataType) -> DFResult<Expr> {
     let DataType::Struct(struct_type) = output_type else {
         return Err(DataFusionError::Execution(
             "expected struct output type".into(),
@@ -518,8 +521,8 @@ mod tests {
     #[test]
     fn test_struct_expression() {
         let expr = Expression::Struct(vec![
-            Expression::Column(ColumnName::new(["a"])),
-            Expression::Column(ColumnName::new(["b"])),
+            Expression::Column(ColumnName::new(["a"])).into(),
+            Expression::Column(ColumnName::new(["b"])).into(),
         ]);
         let result = to_datafusion_expr(
             &expr,
@@ -797,12 +800,13 @@ mod tests {
 
         // Test struct expressions with nested fields
         let expr = Expression::Struct(vec![
-            Expression::Column(ColumnName::new(["a"])),
+            Expression::Column(ColumnName::new(["a"])).into(),
             Expression::Binary(BinaryExpression {
                 left: Box::new(Expression::Column(ColumnName::new(["b"]))),
                 op: BinaryExpressionOp::Plus,
                 right: Box::new(Expression::Column(ColumnName::new(["c"]))),
-            }),
+            })
+            .into(),
         ]);
         let result = to_datafusion_expr(
             &expr,

crates/datafusion/src/engine/file_format.rs

@@ -12,19 +12,19 @@ use datafusion::physical_plan::execute_stream;
 use datafusion::physical_plan::union::UnionExec;
 use datafusion::physical_plan::{ExecutionPlan, PhysicalExpr};
 use datafusion_session::{Session, SessionStore};
+use delta_kernel::PredicateRef;
 use delta_kernel::engine::arrow_conversion::TryIntoArrow as _;
 use delta_kernel::engine::arrow_data::ArrowEngineData;
 use delta_kernel::engine::default::executor::TaskExecutor;
 use delta_kernel::engine::{parse_json as arrow_parse_json, to_json_bytes};
-use delta_kernel::object_store::{PutMode, path::Path};
 use delta_kernel::schema::SchemaRef;
 use delta_kernel::{
     DeltaResult, EngineData, Error as DeltaError, Expression, ExpressionRef,
     FileDataReadResultIterator, FileMeta, JsonHandler, ParquetHandler,
 };
-use delta_kernel::{PredicateRef, object_store};
 use futures::TryStreamExt;
 use futures::stream::{self, BoxStream, StreamExt};
+use object_store::{PutMode, path::Path};
 use parking_lot::RwLock;
 use tracing::warn;
 use url::Url;

crates/datafusion/src/engine/storage.rs

@@ -92,10 +92,8 @@ mod tests {
     use std::ops::Range;
 
     use datafusion::prelude::SessionContext;
-    use delta_kernel::{
-        Engine,
-        object_store::{ObjectStore, local::LocalFileSystem, memory::InMemory, path::Path},
-    };
+    use delta_kernel::Engine;
+    use object_store::{ObjectStore, local::LocalFileSystem, memory::InMemory, path::Path};
     use rstest::*;
 
     use super::*;

crates/datafusion/src/session.rs

@@ -8,8 +8,8 @@ use datafusion_session::{Session, SessionStore};
 use delta_kernel::engine::default::executor::tokio::{
     TokioBackgroundExecutor, TokioMultiThreadExecutor,
 };
-use delta_kernel::object_store::ObjectStore;
 use delta_kernel::{Engine, Snapshot, Version};
+use object_store::ObjectStore;
 use parking_lot::RwLock;
 use tokio::runtime::{Handle, RuntimeFlavor};
 use unitycatalog_client::UnityCatalogClient;

crates/datafusion/src/sql/statements/catalogs.rs

@@ -22,8 +22,11 @@ impl CreateCatalogStatement {
         let name = self.name.to_string();
         let catalog_info = if let Some(location) = self.managed_location.as_ref() {
             // create a calog with explicit managed location
-            client
-                .create_catalog(&name, Some(location), self.comment.as_ref(), None)
+            let request = client
+                .create_catalog(&name)
+                .with_storage_root(location.to_string())
+                .with_comment(self.comment.clone());
+            request
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?
         } else if let Some(share) = self.using_share.as_ref() {
@@ -36,20 +39,20 @@ impl CreateCatalogStatement {
             }
             let provider_name = share.0[0].to_string();
             let share_name = share.0[1].to_string();
-            client
-                .create_sharing_catalog(
-                    &name,
-                    provider_name,
-                    share_name,
-                    self.comment.as_ref(),
-                    None,
-                )
+            let request = client
+                .create_catalog(&name)
+                .with_provider_name(provider_name)
+                .with_share_name(share_name)
+                .with_comment(self.comment.clone());
+            request
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?
         } else {
             // create a catalog with default settings
-            client
-                .create_catalog(&name, None::<String>, self.comment.as_ref(), None)
+            let request = client
+                .create_catalog(&name)
+                .with_comment(self.comment.clone());
+            request
                 .await
                 .map_err(|e| DataFusionError::External(Box::new(e)))?
         };
@@ -69,7 +72,8 @@ impl DropCatalogStatement {
         let name = self.name.to_string();
         client
             .catalog(&name)
-            .delete(self.cascade)
+            .delete()
+            .with_force(self.cascade)
             .await
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
         drop_response_to_batch(name, "Catalog", "success")