Revert "Refactor ArrowArrayStream handling in PyDataFrame to improve memory management and encapsulation"

This reverts commit b5b9759.

Author: kosiew

src/dataframe.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use std::collections::HashMap;
-use std::ffi::CString;
+use std::ffi::{c_void, CString};
 use std::sync::Arc;
 
 use arrow::array::{new_null_array, RecordBatch, RecordBatchReader};
@@ -58,8 +58,6 @@ use crate::{
     expr::{sort_expr::PySortExpr, PyExpr},
 };
 
-const ARROW_STREAM_NAME: &str = "arrow_array_stream";
-
 // https://github.com/apache/datafusion-python/pull/1016#discussion_r1983239116
 // - we have not decided on the table_provider approach yet
 // this is an interim implementation
@@ -954,10 +952,26 @@ impl PyDataFrame {
         };
         let reader: Box<dyn RecordBatchReader + Send> = Box::new(reader);
 
-        let stream = FFI_ArrowArrayStream::new(reader);
-        let name = CString::new(ARROW_STREAM_NAME).unwrap();
-        let capsule = PyCapsule::new(py, stream, Some(name)).map_err(PyDataFusionError::from)?;
-        Ok(capsule)
+        let stream = Box::new(FFI_ArrowArrayStream::new(reader));
+        let stream_ptr = Box::into_raw(stream);
+        assert!(
+            !stream_ptr.is_null(),
+            "ArrowArrayStream pointer should never be null"
+        );
+        let stream_capsule_name = CString::new("arrow_array_stream").unwrap();
+        unsafe {
+            PyCapsule::new_bound_with_destructor(
+                py,
+                stream_ptr,
+                Some(stream_capsule_name),
+                |ptr: *mut FFI_ArrowArrayStream, _| {
+                    if !ptr.is_null() {
+                        unsafe { Box::from_raw(ptr) };
+                    }
+                },
+            )
+        }
+        .map_err(PyDataFusionError::from)
     }
 
     fn execute_stream(&self, py: Python) -> PyDataFusionResult<PyRecordBatchStream> {