Update deps, fix some warnings

Cargo.toml

@@ -32,23 +32,23 @@ repository = "https://github.com/sjrusso8/spark-connect-rs"
 rust-version = "1.81"
 
 [workspace.dependencies]
-tonic = { version ="0.11", default-features = false }
+tonic = { version = "0.14", default-features = false, features = ["default", "transport", "codegen"] }
+tonic-prost = "0.14"
 
-tokio = { version = "1.44", default-features = false, features = ["macros"] }
+tokio = { version = "1.44", default-features = false, features = ["macros", "sync"] }
 tower = { version = "0.5" }
 
 futures-util = { version = "0.3" }
 thiserror = { version = "2.0" }
 
-http-body = { version = "0.4.6" }
 
-arrow = { version = "55", features = ["prettyprint"] }
-arrow-ipc = { version = "55" }
+arrow = { version = "^56", features = ["prettyprint"] }
+arrow-ipc = { version = "^56" }
 
 serde_json = { version = "1" }
 
-prost = { version = "0.12" }
-prost-types = { version = "0.12" }
+prost = { version = "^0.14" }
+prost-types = { version = "^0.14" }
 
 rand = { version = "0.9" }
 uuid = { version = "1.16", features = ["v4"] }
@@ -57,6 +57,6 @@ regex = { version = "1" }
 
 chrono = { version = "0.4" }
 
-datafusion = { version = "47.0", default-features = false }
-polars = { version = "0.43", default-features = false }
-polars-arrow = { version = "0.43", default-features = false, features = ["arrow_rs"]  }
+datafusion = { version = "50", default-features = false }
+polars = { version = "0.51", default-features = false }
+polars-arrow = { version = "0.51", default-features = false, features = ["arrow-format"]  }

crates/connect/Cargo.toml

@@ -36,15 +36,14 @@ include = [
 
 [dependencies]
 tonic = { workspace = true, default-features = false, optional = true }
+tonic-prost = { workspace = true }
 
 tower = { workspace = true }
 tokio = { workspace = true, optional = true }
 
 futures-util = { workspace = true }
 thiserror = { workspace = true }
 
-http-body = { workspace = true }
-
 arrow = { workspace = true }
 arrow-ipc = { workspace = true }
 
@@ -55,7 +54,7 @@ prost-types = { workspace = true }
 
 rand = { workspace = true }
 uuid = { workspace = true }
-url =  { workspace = true }
+url = { workspace = true }
 regex = { workspace = true }
 
 chrono = { workspace = true }
@@ -70,29 +69,27 @@ futures = "0.3"
 tokio = { workspace = true, features = ["rt-multi-thread"] }
 
 [build-dependencies]
-tonic-build = "0.11"
+tonic-prost-build = "0.14"
 
 [lib]
 doctest = false
 
 [features]
 default = [
-  "tokio",
-  "tonic/codegen",
-  "tonic/prost",
-  "tonic/transport",
+    "tokio",
+    "tonic"
 ]
 
 tls = [
-  "tonic/tls",
-  "tonic/tls-roots"
+    "tonic/tls-aws-lc"
 ]
 
 datafusion = [
-  "dep:datafusion"
+    "dep:datafusion"
 ]
 
 polars = [
-  "dep:polars",
-  "dep:polars-arrow"
+    "dep:polars",
+    "dep:polars-arrow",
+    "arrow/ffi"
 ]

crates/connect/build.rs

@@ -27,12 +27,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         file_paths.push(entry.to_str().unwrap().to_string());
     }
 
-    tonic_build::configure()
+    tonic_prost_build::configure()
         .protoc_arg("--experimental_allow_proto3_optional")
         .build_server(false)
         .build_client(true)
         .build_transport(true)
-        .compile(file_paths.as_ref(), &["./protobuf/spark-3.5/"])?;
+        .compile_protos(file_paths.as_ref(), &["./protobuf/spark-3.5/".to_string()])?;
 
     Ok(())
 }

crates/connect/src/client/middleware.rs

@@ -23,7 +23,6 @@ use std::str::FromStr;
 use std::task::{Context, Poll};
 
 use futures_util::future::BoxFuture;
-use http_body::combinators::UnsyncBoxBody;
 
 use tonic::codegen::http::Request;
 use tonic::codegen::http::{HeaderName, HeaderValue};
@@ -66,13 +65,9 @@ impl<S> HeadersMiddleware<S> {
 
 // TODO! as of now Request is not clone. So the retry logic does not work.
 // https://github.com/tower-rs/tower/pull/790
-impl<S> Service<Request<UnsyncBoxBody<prost::bytes::Bytes, tonic::Status>>> for HeadersMiddleware<S>
+impl<S> Service<Request<tonic::body::Body>> for HeadersMiddleware<S>
 where
-    S: Service<Request<UnsyncBoxBody<prost::bytes::Bytes, tonic::Status>>>
-        + Clone
-        + Send
-        + Sync
-        + 'static,
+    S: Service<Request<tonic::body::Body>> + Clone + Send + Sync + 'static,
     S::Future: Send + 'static,
     S::Response: Send + Debug + 'static,
     S::Error: Debug,
@@ -85,10 +80,7 @@ where
         self.inner.poll_ready(cx).map_err(Into::into)
     }
 
-    fn call(
-        &mut self,
-        mut request: Request<UnsyncBoxBody<prost::bytes::Bytes, tonic::Status>>,
-    ) -> Self::Future {
+    fn call(&mut self, mut request: Request<tonic::body::Body>) -> Self::Future {
         let clone = self.inner.clone();
         let mut inner = std::mem::replace(&mut self.inner, clone);
 

crates/connect/src/client/mod.rs

@@ -97,7 +97,7 @@ pub struct SparkConnectClient<T> {
 
 impl<T> SparkConnectClient<T>
 where
-    T: tonic::client::GrpcService<tonic::body::BoxBody>,
+    T: tonic::client::GrpcService<tonic::body::Body>,
     T::Error: Into<StdError>,
     T::ResponseBody: Body<Data = Bytes> + Send + 'static,
     <T::ResponseBody as Body>::Error: Into<StdError> + Send,

crates/connect/src/dataframe.rs

@@ -1201,15 +1201,37 @@ impl DataFrame {
     /// Converts a [DataFrame] into a [polars::frame::DataFrame]
     #[cfg(any(feature = "default", feature = "polars"))]
     pub async fn to_polars(self) -> Result<polars::frame::DataFrame, SparkError> {
+        use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
         let batch = self.collect().await?;
         let schema = batch.schema();
 
         let mut columns = Vec::with_capacity(batch.num_columns());
         for (i, column) in batch.columns().iter().enumerate() {
-            let arrow = Box::<dyn polars_arrow::array::Array>::from(&**column);
+            let array_data = column.to_data();
+            let (array, ffi_schema) = arrow::ffi::to_ffi(&array_data)?;
+
+            // SAFETY: This follows the Arrow C FFI interface. The transmute converts types that are
+            // across crates, but are the same layout.
+            let field = unsafe {
+                polars_arrow::ffi::import_field_from_c(std::mem::transmute::<
+                    &FFI_ArrowSchema,
+                    &polars_arrow::ffi::ArrowSchema,
+                >(&ffi_schema))
+            }?;
+
+            // SAFETY: This follows the Arrow C FFI interface. The transmute converts types that are
+            // across crates, but are the same layout.
+            let data = unsafe {
+                polars_arrow::ffi::import_array_from_c(
+                    std::mem::transmute::<FFI_ArrowArray, polars_arrow::ffi::ArrowArray>(array),
+                    field.dtype().clone(),
+                )
+            }?;
+
+            // Create Polars series from arrow column
             columns.push(polars::series::Series::from_arrow(
                 schema.fields().get(i).unwrap().name().into(),
-                arrow,
+                data,
             )?);
         }
 
@@ -2542,37 +2564,6 @@ mod tests {
         Ok(())
     }
 
-    #[tokio::test]
-    #[cfg(feature = "polars")]
-    async fn test_df_to_polars() -> Result<(), SparkError> {
-        let spark = setup().await;
-
-        let data = mock_data();
-
-        let schema = data.schema();
-
-        // transform arrow into polars_arrow
-        // same code as used in the function
-        let mut columns = Vec::with_capacity(data.num_columns());
-        for (i, column) in data.columns().iter().enumerate() {
-            let arrow = Box::<dyn polars_arrow::array::Array>::from(&**column);
-            columns.push(polars::series::Series::from_arrow(
-                schema.fields().get(i).unwrap().name().into(),
-                arrow,
-            )?);
-        }
-
-        let df_expected = polars::frame::DataFrame::from_iter(columns);
-
-        let df = spark.create_dataframe(&data)?;
-
-        let df_output = df.to_polars().await?;
-
-        assert_eq!(df_expected, df_output);
-
-        Ok(())
-    }
-
     #[tokio::test]
     async fn test_df_explain_concurrent() -> Result<(), SparkError> {
         let spark = setup().await;

crates/connect/src/functions/mod.rs

@@ -643,37 +643,18 @@ pub fn window(
     start_time: Option<&str>,
 ) -> Column {
     let window_duration = lit(window_duration);
-
-    if slide_duration.is_some() & start_time.is_some() {
-        invoke_func(
-            "window",
-            vec![
-                time_column.into(),
-                window_duration,
-                lit(slide_duration.unwrap()),
-                lit(start_time.unwrap()),
-            ],
-        )
-    } else if slide_duration.is_some() & start_time.is_none() {
-        invoke_func(
-            "window",
-            vec![
-                time_column.into(),
-                window_duration,
-                lit(slide_duration.unwrap()),
-            ],
-        )
-    } else if slide_duration.is_none() & start_time.is_some() {
-        invoke_func(
+    match (slide_duration, start_time) {
+        (Some(sd), Some(st)) => invoke_func(
             "window",
-            vec![
-                time_column.into(),
-                window_duration,
-                lit(start_time.unwrap()),
-            ],
-        )
-    } else {
-        invoke_func("window", vec![time_column.into(), window_duration])
+            vec![time_column.into(), window_duration, lit(sd), lit(st)],
+        ),
+        (Some(sd), None) => {
+            invoke_func("window", vec![time_column.into(), window_duration, lit(sd)])
+        }
+        (None, Some(st)) => {
+            invoke_func("window", vec![time_column.into(), window_duration, lit(st)])
+        }
+        (None, None) => invoke_func("window", vec![time_column.into(), window_duration]),
     }
 }