Refactor sql methods in PySessionContext to use immutable self reference

Enhance wait_for_future function to handle current runtime context

Author: kosiew

src/context.rs

@@ -429,15 +429,15 @@ impl PySessionContext {
     }
 
     /// Returns a PyDataFrame whose plan corresponds to the SQL statement.
-    pub fn sql(&mut self, query: &str, py: Python) -> PyDataFusionResult<PyDataFrame> {
+    pub fn sql(&self, query: &str, py: Python) -> PyDataFusionResult<PyDataFrame> {
         let result = self.ctx.sql(query);
         let df = wait_for_future(py, result)??;
         Ok(PyDataFrame::new(df))
     }
 
     #[pyo3(signature = (query, options=None))]
     pub fn sql_with_options(
-        &mut self,
+        &self,
         query: &str,
         options: Option<PySQLOptions>,
         py: Python,

src/utils.rs

@@ -82,7 +82,7 @@ where
     const INTERVAL_CHECK_SIGNALS: Duration = Duration::from_millis(1_000);
 
     py.allow_threads(|| {
-        runtime.block_on(async {
+        let wait_future = || async {
             tokio::pin!(fut);
             loop {
                 tokio::select! {
@@ -92,7 +92,13 @@ where
                     }
                 }
             }
-        })
+        };
+
+        if tokio::runtime::Handle::try_current().is_ok() {
+            tokio::task::block_in_place(|| runtime.block_on(wait_future()))
+        } else {
+            runtime.block_on(wait_future())
+        }
     })
 }