fix: Implement file writing directly for VortexSink instead of DataFusion demuxer

vortex-datafusion/Cargo.toml

@@ -36,6 +36,7 @@ object_store = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread", "fs"] }
 tokio-stream = { workspace = true }
 tracing = { workspace = true, features = ["std", "attributes"] }
+uuid = { workspace = true, features = ["v4"] }
 vortex = { workspace = true, features = ["object_store", "tokio", "files"] }
 vortex-utils = { workspace = true, features = ["dashmap"] }
 

vortex-datafusion/src/persistent/format.rs

@@ -96,6 +96,11 @@ config_namespace! {
         /// Values smaller than `MAX_POSTSCRIPT_SIZE + EOF_SIZE` will be clamped to that minimum
         /// during footer parsing.
         pub footer_initial_read_size_bytes: usize, default = DEFAULT_FOOTER_INITIAL_READ_SIZE_BYTES
+        /// Target file size in megabytes for written Vortex files.
+        ///
+        /// The writer will attempt to split output into multiple files, each approximately
+        /// this size. Defaults to 16 MB. A value of 0 is treated as the default (16 MB).
+        pub target_file_size_mb: usize, default = 16
     }
 }
 
@@ -417,8 +422,20 @@ impl FileFormat for VortexFormat {
             return not_impl_err!("Overwrites are not implemented yet for Vortex");
         }
 
+        let target_file_size_mb = if self.opts.target_file_size_mb > 0 {
+            self.opts.target_file_size_mb
+        } else {
+            16 // Default to 16 MB when set to 0
+        };
+        let target_file_size = target_file_size_mb as u64 * 1024 * 1024;
+
         let schema = conf.output_schema().clone();
-        let sink = Arc::new(VortexSink::new(conf, schema, self.session.clone()));
+        let sink = Arc::new(VortexSink::new(
+            conf,
+            schema,
+            self.session.clone(),
+            target_file_size,
+        ));
 
         Ok(Arc::new(DataSinkExec::new(input, sink, order_requirements)) as _)
     }