fix: release the GIL during sync iteration

python/pytests/execution_test.py

@@ -34,14 +34,27 @@ async def test_iter_pandas(golden, source_int64) -> None:
         next(batches)
 
 
-async def test_iter_rows(golden, source_int64) -> None:
+async def test_iter_rows(source_int64) -> None:
     results: Iterator[dict] = source_int64.run_iter("row", row_limit=2)
     assert next(results)["m"] == 5
     assert next(results)["m"] == 24
     with pytest.raises(StopIteration):
         next(results)
 
 
+@kd.udf("add_one<N: number>(x: N) -> N")
+def add_one(x: pd.Series) -> pd.Series:
+    """Use Pandas to one"""
+    print("In UDF")
+    return x + 1
+
+
+async def test_iter_udf(source_int64) -> None:
+    results = source_int64.col("m").pipe(add_one).run_iter("row", row_limit=2)
+    assert next(results)["result"] == 6
+    assert next(results)["result"] == 25
+
+
 async def test_iter_pandas_async(golden, source_int64) -> None:
     batches: AsyncIterator[pd.DataFrame] = source_int64.run_iter(
         row_limit=4, max_batch_size=2

python/src/execution.rs

@@ -73,7 +73,11 @@ impl Execution {
 
     fn next_pyarrow(&mut self, py: Python<'_>) -> Result<Option<PyObject>> {
         let mut execution = self.execution()?;
-        let batch = execution.as_mut().unwrap().next_blocking()?;
+
+        // Explicitly allow threads to take the gil during execution, so
+        // the udf evaluator can acquire it to execute python.
+        let batch = py.allow_threads(move || execution.as_mut().unwrap().next_blocking())?;
+
         let result = match batch {
             Some(batch) => Some(batch.to_pyarrow(py)?),
             None => None,