Add flush after writing file and add some docs to file writepath (#2922)

1. Make `VortexWriteOptions::write` flush the internal writer before
returning it (especially important for tokio and `ObjectStore` writes,
as pointed out by @a10y in #2921).
2. Remove the unused `TokioAdapter` wrapper struct.
3. Adds some docs

Author: AdamGS

vortex-datafusion/src/persistent/sink.rs

@@ -106,7 +106,6 @@ impl FileSink for VortexSink {
                 .write(writer, stream_adapter)
                 .await?;
 
-            writer.flush().await?;
             writer.shutdown().await?;
         }
 

vortex-file/src/writer.rs

@@ -13,6 +13,10 @@ use crate::segments::writer::BufferedSegmentWriter;
 use crate::strategy::VortexLayoutStrategy;
 use crate::{EOF_SIZE, MAGIC_BYTES, MAX_FOOTER_SIZE, VERSION};
 
+/// Configure a new writer, which can eventually be used to write an [`ArrayStream`] into a sink that implements [`VortexWrite`].
+///
+/// By default, the [`LayoutStrategy`] will be the [`VortexLayoutStrategy`], which includes re-chunking and will also
+/// uncompress all data back to its canonical form before compressing it using the [`BtrBlocksCompressor`](vortex_btrblocks::BtrBlocksCompressor).
 pub struct VortexWriteOptions {
     strategy: Box<dyn LayoutStrategy>,
     exclude_dtype: bool,
@@ -147,6 +151,8 @@ impl VortexWriteOptions {
         eof[4..8].copy_from_slice(&MAGIC_BYTES);
         write.write_all(eof).await?;
 
+        write.flush().await?;
+
         Ok(write.into_inner())
     }
 

vortex-io/src/tokio.rs

@@ -5,35 +5,18 @@ use std::os::unix::fs::FileExt;
 use std::path::Path;
 use std::sync::Arc;
 
-use tokio::io::{AsyncWrite, AsyncWriteExt};
+use tokio::io::AsyncWriteExt;
 use tokio::task::spawn_blocking;
 use vortex_buffer::{Alignment, ByteBuffer, ByteBufferMut};
 use vortex_error::VortexExpect;
 
 use crate::{IoBuf, PerformanceHint, VortexReadAt, VortexWrite};
 
-pub struct TokioAdapter<IO>(pub IO);
-
-impl<W: AsyncWrite + Unpin> VortexWrite for TokioAdapter<W> {
-    async fn write_all<B: IoBuf>(&mut self, buffer: B) -> io::Result<B> {
-        self.0.write_all(buffer.as_slice()).await?;
-        Ok(buffer)
-    }
-
-    async fn flush(&mut self) -> io::Result<()> {
-        self.0.flush().await
-    }
-
-    async fn shutdown(&mut self) -> io::Result<()> {
-        self.0.shutdown().await
-    }
-}
-
 /// A cheaply cloneable, readonly file that executes operations
 /// on a tokio blocking threadpool.
 ///
-/// We use this because the builtin tokio `File` type is not `Clone` and
-/// also does actually implement a `read_exact_at` operation.
+/// We use this because tokio's [`File`](tokio::fs::File) type is not `Clone` and
+/// also does not implement a `read_exact_at` operation.
 #[derive(Debug, Clone)]
 pub struct TokioFile(Arc<File>);