pepijnve commented on code in PR #23522:
URL: https://github.com/apache/datafusion/pull/23522#discussion_r3588284155
##########
datafusion/physical-plan/src/spill/spill_pool.rs:
##########
@@ -194,52 +194,48 @@ impl SpillPoolWriter {
// Re-acquire lock and push to shared queue
shared = self.shared.lock();
shared.files.push_back(Arc::clone(&file_shared));
- shared.current_write_file = Some(file_shared);
shared.wake(); // Wake readers waiting for new files
- }
+ file_shared
+ };
- let current_write_file = shared.current_write_file.take();
// Release shared lock before file I/O (fine-grained locking)
// This allows readers to access the queue while we do disk I/O
drop(shared);
// Write batch to current file - lock only the specific file
- if let Some(current_file) = current_write_file {
- // Now lock just this file for I/O (separate from shared lock)
- let mut file_shared = current_file.lock();
-
- // Append the batch
- if let Some(ref mut writer) = file_shared.writer {
- writer.append_batch(batch)?;
- // make sure we flush the writer for readers
- writer.flush()?;
- file_shared.batches_written += 1;
- file_shared.estimated_size += batch_size;
- }
+ let mut file_shared = current_write_file.lock();
+
+ // Append the batch
+ if let Some(ref mut writer) = file_shared.writer {
+ writer.append_batch(batch)?;
+ // make sure we flush the writer for readers
+ writer.flush()?;
+ file_shared.batches_written += 1;
+ file_shared.estimated_size += batch_size;
+ }
- // Wake reader waiting on this specific file
- file_shared.wake();
+ // Wake reader waiting on this specific file
+ file_shared.wake();
- // Check if we need to rotate
- let needs_rotation = file_shared.estimated_size >
self.max_file_size_bytes;
+ // Check if we need to rotate
+ let needs_rotation = file_shared.estimated_size >
self.max_file_size_bytes;
- if needs_rotation {
- // Finish the IPC writer
- if let Some(mut writer) = file_shared.writer.take() {
- writer.finish()?;
- }
- // Mark as finished so readers know not to wait for more data
- file_shared.writer_finished = true;
- // Wake reader waiting on this file (it's now finished)
- file_shared.wake();
- // Don't put back current_write_file - let it rotate
- } else {
- // Release file lock
- drop(file_shared);
- // Put back the current file for further writing
- let mut shared = self.shared.lock();
- shared.current_write_file = Some(current_file);
+ if needs_rotation {
+ // Finish the IPC writer
+ if let Some(mut writer) = file_shared.writer.take() {
+ writer.finish()?;
}
+ // Mark as finished so readers know not to wait for more data
+ file_shared.writer_finished = true;
+ // Wake reader waiting on this file (it's now finished)
+ file_shared.wake();
+ // Don't put back current_write_file - let it rotate
Review Comment:
Singular is actually correct here, but I'll change the wording a bit to
clarify what's being done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
