fixup duckdb/python/java

Signed-off-by: Andrew Duffy <[email protected]>

Author: a10y

vortex-duckdb/src/copy.rs

@@ -126,17 +126,21 @@ impl CopyFunction for VortexCopyFunction {
         let array_stream = ArrayStreamAdapter::new(bind_data.dtype.clone(), rx.into_stream());
 
         let handle = SESSION.handle();
-        let writer = handle.spawn(async move {
+        let write_task = handle.spawn(async move {
             let mut file = async_fs::File::create(file_path).await?;
-            SESSION.write_options().write(&mut file, array_stream).await
+            SESSION
+                .write_options()
+                .write(&mut file, array_stream)
+                .await
+                .map(|(summary, _)| summary)
         });
 
         let worker_pool = RUNTIME.new_pool();
         worker_pool.set_workers_to_available_parallelism();
 
         Ok(GlobalState {
             worker_pool,
-            write_task: Mutex::new(Some(writer)),
+            write_task: Mutex::new(Some(write_task)),
             sink: Some(sink),
         })
     }

vortex-jni/src/writer.rs

@@ -176,8 +176,8 @@ pub extern "system" fn Java_dev_vortex_jni_NativeWriterMethods_create(
 
         let (store, _scheme) = make_object_store(&url, &properties)?;
         let write_handle = SESSION.handle().spawn(async move {
-            let mut write = ObjectStoreWriter::new(store, &path).await?;
-            let summary = SESSION.write_options().write(&mut write, w).await?;
+            let write = ObjectStoreWriter::new(store, &path).await?;
+            let (summary, mut write) = SESSION.write_options().write(write, w).await?;
             write.shutdown().await?;
             Ok(summary)
         });

vortex-python/src/io.rs

@@ -171,10 +171,10 @@ pub fn read_url<'py>(
 pub fn write(py: Python, iter: PyIntoArrayIterator, path: &str) -> PyResult<()> {
     py.detach(|| {
         TOKIO_RUNTIME.block_on(async move {
-            let mut file = File::create(path).await?;
+            let file = File::create(path).await?;
             SESSION
                 .write_options()
-                .write(&mut file, iter.into_inner().into_array_stream())
+                .write(file, iter.into_inner().into_array_stream())
                 .await
         })
     })?;
@@ -285,7 +285,7 @@ impl PyVortexWriteOptions {
     pub fn write_path(&self, py: Python, iter: PyIntoArrayIterator, path: &str) -> PyResult<()> {
         py.detach(|| {
             TOKIO_RUNTIME.block_on(async move {
-                let mut file = File::create(path).await?;
+                let file = File::create(path).await?;
 
                 let mut strategy = WriteStrategyBuilder::new();
                 if let Some(compressor) = self.compressor.as_ref() {
@@ -295,7 +295,7 @@ impl PyVortexWriteOptions {
                 SESSION
                     .write_options()
                     .with_strategy(strategy.build())
-                    .write(&mut file, iter.into_inner().into_array_stream())
+                    .write(file, iter.into_inner().into_array_stream())
                     .await
             })
         })?;