This is an automated email from the ASF dual-hosted git repository.
wForget pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new c0b525194 [AURON #2352] Lock spills before new spiller (#2350)
c0b525194 is described below
commit c0b525194fa04f4e51b483965a6c026e079f19dc
Author: Zhen Wang <[email protected]>
AuthorDate: Mon Jul 20 16:48:14 2026 +0800
[AURON #2352] Lock spills before new spiller (#2350)
# Which issue does this PR close?
Closes #2352
# Rationale for this change
`try_new_spill` always uses `spills.len` as the `spill_id`, so we should
always lock `spills` before `trying_new_spill`.
# What changes are included in this PR?
# Are there any user-facing changes?
# How was this patch tested?
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git
a/native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs
b/native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs
index b76ed6314..c32c91e0a 100644
--- a/native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs
+++ b/native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs
@@ -98,6 +98,7 @@ impl MemConsumer for SortShuffleRepartitioner {
async fn spill(&self) -> Result<()> {
let data = self.data.lock().await.drain();
let spill_metrics = self.exec_ctx.spill_metrics().clone();
+ let mut spills = self.spills.lock().await;
let spill = tokio::task::spawn_blocking(move || {
let mut spill = try_new_spill(&spill_metrics)?;
let offsets = data.write(spill.get_buf_writer())?;
@@ -105,8 +106,8 @@ impl MemConsumer for SortShuffleRepartitioner {
})
.await
.expect("tokio spawn_blocking error")?;
-
- self.spills.lock().await.push(spill);
+ spills.push(spill);
+ drop(spills);
self.update_mem_used(0).await?;
Ok(())
}
