bug(deadlock): change to async rwlock

Cargo.toml

@@ -25,8 +25,6 @@ arrow-ipc = { version = "51" }
 
 serde_json = { version = "1" }
 
-parking_lot = { version = "0.12", features = ["send_guard"]}
-
 prost = { version = "0.12" }
 prost-types = { version = "0.12" }
 

core/Cargo.toml

@@ -28,8 +28,6 @@ arrow-ipc = { workspace = true }
 
 serde_json = { workspace = true }
 
-parking_lot = { workspace = true }
-
 prost = { workspace = true }
 prost-types = { workspace = true }
 
@@ -46,6 +44,9 @@ datafusion = { workspace = true, optional = true }
 polars = { workspace = true, optional = true }
 polars-arrow = { workspace = true, optional = true }
 
+[dev-dependencies]
+futures = "0.3"
+
 [build-dependencies]
 tonic-build = "0.11.0"
 

core/src/client/mod.rs

@@ -16,15 +16,15 @@ use arrow::error::ArrowError;
 use arrow::record_batch::RecordBatch;
 use arrow_ipc::reader::StreamReader;
 
-use parking_lot::RwLock;
-
 use tonic::codegen::{Body, Bytes, StdError};
 use tonic::metadata::{
     Ascii, AsciiMetadataValue, KeyAndValueRef, MetadataKey, MetadataMap, MetadataValue,
 };
 use tonic::service::Interceptor;
 use tonic::Status;
 
+use tokio::sync::RwLock;
+
 use url::Url;
 
 use uuid::Uuid;
@@ -410,15 +410,13 @@ where
         }
     }
 
-    #[allow(clippy::await_holding_lock)]
     async fn execute_and_fetch(
         &mut self,
         req: spark::ExecutePlanRequest,
     ) -> Result<(), SparkError> {
-        let mut client = self.stub.write();
+        let mut client = self.stub.write().await;
 
         let mut resp = client.execute_plan(req).await?.into_inner();
-
         drop(client);
 
         // clear out any prior responses
@@ -436,7 +434,6 @@ where
         Ok(())
     }
 
-    #[allow(clippy::await_holding_lock)]
     pub async fn analyze(
         &mut self,
         analyze: spark::analyze_plan_request::Analyze,
@@ -445,19 +442,16 @@ where
 
         req.analyze = Some(analyze);
 
-        let mut client = self.stub.write();
-
         // clear out any prior responses
         self.analyzer = AnalyzeHandler::new();
 
+        let mut client = self.stub.write().await;
         let resp = client.analyze_plan(req).await?.into_inner();
-
         drop(client);
 
         self.handle_analyze(resp)
     }
 
-    #[allow(clippy::await_holding_lock)]
     pub async fn interrupt_request(
         &mut self,
         interrupt_type: spark::interrupt_request::InterruptType,
@@ -494,10 +488,9 @@ where
             }
         };
 
-        let mut client = self.stub.write();
+        let mut client = self.stub.write().await;
 
         let resp = client.interrupt(req).await?.into_inner();
-        drop(client);
 
         Ok(resp)
     }
@@ -538,7 +531,6 @@ where
         self.tags = vec![];
     }
 
-    #[allow(clippy::await_holding_lock)]
     pub async fn config_request(
         &mut self,
         operation: spark::config_request::Operation,
@@ -550,10 +542,9 @@ where
             operation: Some(operation),
         };
 
-        let mut client = self.stub.write();
+        let mut client = self.stub.write().await;
 
         let resp = client.config(operation).await?.into_inner();
-        drop(client);
 
         Ok(resp)
     }

core/src/dataframe.rs

@@ -2202,4 +2202,22 @@ mod tests {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_df_explain_concurrent() -> Result<(), SparkError> {
+        let spark = setup().await;
+        let spark_clone = spark.clone();
+
+        let data = mock_data();
+
+        let df = spark.createDataFrame(&data)?;
+        let df_clone = spark_clone.createDataFrame(&data)?;
+
+        let (res, res_clone) = futures::join!(df.explain(None), df_clone.explain(None));
+        let (val, val_clone) = (res?, res_clone?);
+
+        assert!(val.contains("== Physical Plan =="));
+        assert!(val_clone.contains("== Physical Plan =="));
+        Ok(())
+    }
 }

core/src/session.rs

@@ -16,7 +16,7 @@ use spark::spark_connect_service_client::SparkConnectServiceClient;
 
 use arrow::record_batch::RecordBatch;
 
-use parking_lot::RwLock;
+use tokio::sync::RwLock;
 
 #[cfg(not(feature = "wasm"))]
 use tonic::transport::{Channel, Endpoint};