fix: Prevent deadlock in sink_csv

crates/polars-pipe/src/executors/sinks/file_sink.rs

@@ -1,5 +1,6 @@
 use std::any::Any;
 use std::path::Path;
+use std::sync::Mutex;
 use std::thread::JoinHandle;
 
 use crossbeam_channel::{bounded, Receiver, Sender};
@@ -54,17 +55,6 @@ impl<W: std::io::Write> SinkWriter for polars_io::ipc::BatchedWriter<W> {
     }
 }
 
-#[cfg(feature = "csv")]
-impl SinkWriter for polars_io::csv::BatchedWriter<std::fs::File> {
-    fn _write_batch(&mut self, df: &DataFrame) -> PolarsResult<()> {
-        self.write_batch(df)
-    }
-
-    fn _finish(&mut self) -> PolarsResult<()> {
-        Ok(())
-    }
-}
-
 #[cfg(feature = "json")]
 impl SinkWriter for BatchedWriter<std::fs::File> {
     fn _write_batch(&mut self, df: &DataFrame) -> PolarsResult<()> {
@@ -229,11 +219,14 @@ impl IpcCloudSink {
 }
 
 #[cfg(feature = "csv")]
-pub struct CsvSink {}
+#[derive(Clone)]
+pub struct CsvSink {
+    writer: Arc<Mutex<polars_io::csv::BatchedWriter<std::fs::File>>>,
+}
 #[cfg(feature = "csv")]
 impl CsvSink {
     #[allow(clippy::new_ret_no_self)]
-    pub fn new(path: &Path, options: CsvWriterOptions, schema: &Schema) -> PolarsResult<FilesSink> {
+    pub fn new(path: &Path, options: CsvWriterOptions, schema: &Schema) -> PolarsResult<Self> {
         let file = std::fs::File::create(path)?;
         let writer = CsvWriter::new(file)
             .include_bom(options.include_bom)
@@ -250,23 +243,39 @@ impl CsvSink {
             .with_quote_style(options.serialize_options.quote_style)
             .batched(schema)?;
 
-        let writer = Box::new(writer) as Box<dyn SinkWriter + Send + Sync>;
+        Ok(Self {
+            writer: Arc::new(Mutex::new(writer)),
+        })
+    }
+}
 
-        let morsels_per_sink = morsels_per_sink();
-        let backpressure = morsels_per_sink * 2;
-        let (sender, receiver) = bounded(backpressure);
+// Csv has a sync implementation because it writes in parallel. The file sink would deadlock.
+#[cfg(feature = "csv")]
+impl Sink for CsvSink {
+    fn sink(&mut self, _: &PExecutionContext, chunk: DataChunk) -> PolarsResult<SinkResult> {
+        let mut writer = self.writer.lock().unwrap();
+        writer.write_batch(&chunk.data)?;
+        Ok(SinkResult::CanHaveMoreInput)
+    }
 
-        let io_thread_handle = Arc::new(Some(init_writer_thread(
-            receiver,
-            writer,
-            options.maintain_order,
-            morsels_per_sink,
-        )));
+    fn combine(&mut self, _other: &mut dyn Sink) {
+        // already synchronized
+    }
 
-        Ok(FilesSink {
-            sender,
-            io_thread_handle,
-        })
+    fn split(&self, _thread_no: usize) -> Box<dyn Sink> {
+        Box::new(self.clone())
+    }
+
+    fn finalize(&mut self, _context: &PExecutionContext) -> PolarsResult<FinalizedSink> {
+        Ok(FinalizedSink::Finished(Default::default()))
+    }
+
+    fn as_any(&mut self) -> &mut dyn Any {
+        self
+    }
+
+    fn fmt(&self) -> &str {
+        "csv_sink"
     }
 }
 
@@ -408,6 +417,6 @@ impl Sink for FilesSink {
         self
     }
     fn fmt(&self) -> &str {
-        "parquet_sink"
+        "file_sink"
     }
 }