bug: lock spills before new spiller in sort_repartitioner

[wForget]

Describe the bug

try_new_spill always uses spills.len as the spill_id, so we should always lock spills before trying_new_spill.

auron/native-engine/datafusion-ext-plans/src/shuffle/sort_repartitioner.rs

Lines 98 to 112 in ca3bece

	async fn spill(&self) -> Result<()> {
	let data = self.data.lock().await.drain();
	let spill_metrics = self.exec_ctx.spill_metrics().clone();
	let spill = tokio::task::spawn_blocking(move || {
	let mut spill = try_new_spill(&spill_metrics)?;
	let offsets = data.write(spill.get_buf_writer())?;
	Ok::<_, DataFusionError>(Offsetted::new(offsets, spill))
	})
	.await
	.expect("tokio spawn_blocking error")?;
	self.spills.lock().await.push(spill);
	self.update_mem_used(0).await?;
	Ok(())
	}

To Reproduce

Expected behavior

Screenshots

Additional context