Shutdown Tokio file (#4621)

Signed-off-by: Nicholas Gates <nick@nickgates.com>

Author: gatesn

vortex-cxx/src/write.rs

@@ -13,6 +13,7 @@ use vortex::dtype::DType;
 use vortex::dtype::arrow::FromArrowType;
 use vortex::error::{VortexError, VortexExpect};
 use vortex::file::VortexWriteOptions as WriteOptions;
+use vortex::io::VortexWrite;
 use vortex::io::runtime::tokio::TokioRuntime;
 use vortex::iter::{ArrayIteratorAdapter, ArrayIteratorExt};
 use vortex::stream::ArrayStream;
@@ -67,6 +68,7 @@ pub(crate) unsafe fn write_array_stream(
     RUNTIME.block_on(async {
         let mut file = tokio::fs::File::create(path).await?;
         options.inner.write(&mut file, vortex_stream).await?;
+        file.shutdown().await?;
         Ok(())
     })
 }

vortex-tui/src/convert.rs

@@ -8,6 +8,7 @@ use futures::StreamExt;
 use indicatif::ProgressBar;
 use parquet::arrow::ParquetRecordBatchStreamBuilder;
 use tokio::fs::File;
+use tokio::io::AsyncWriteExt;
 use vortex::ArrayRef;
 use vortex::arrow::FromArrowArray;
 use vortex::compressor::CompactCompressor;
@@ -80,13 +81,12 @@ pub async fn exec_convert(flags: Flags) -> anyhow::Result<()> {
         Strategy::Compact => strategy.with_compressor(CompactCompressor::default()),
     };
 
+    let mut file = File::create(output_path).await?;
     VortexWriteOptions::default()
         .with_strategy(strategy.build())
-        .write(
-            &mut File::create(output_path).await?,
-            ArrayStreamAdapter::new(dtype, vortex_stream),
-        )
+        .write(&mut file, ArrayStreamAdapter::new(dtype, vortex_stream))
         .await?;
+    file.shutdown().await?;
 
     Ok(())
 }