Repository: risingwavelabs/risingwave
File: src/batch/src/spill/spill_op.rs (Rust)

Code before:
const DEFAULT_IO_CONCURRENT_TASK: usize = 8;

#[derive(Clone)]
pub enum SpillBackend {
    Disk,
    /// Only for testing purpose
    Memory,
}

/// `SpillOp` is used to manage the spill directory of the spilling executor and it will drop the directory with a RAII style.
pub struct SpillOp {
    pub op: Operator,
}

impl SpillOp {
    fn batch_spill_root() -> PathBuf {
        batch_spill_base_dir().join(RW_MANAGED_SPILL_DIR)
    }

    pub fn create(path: impl AsRef<Path>, spill_backend: SpillBackend) -> Result<SpillOp> {
        let path = path.as_ref();
        assert!(path.is_relative());

        let root = Self::batch_spill_root().join(path);

        let op = match spill_backend {
            SpillBackend::Disk => {
                let builder = Fs::default().root(&root.to_string_lossy());
                Operator::new(builder)?
                    .layer(RetryLayer::default())
                    .finish()
            }
            SpillBackend::Memory => {
                let builder = Memory::default().root(&root.to_string_lossy());
                Operator::new(builder)?
                    .layer(RetryLayer::default())
                    .finish()
            }
        };
        Ok(SpillOp { op })
    }

    pub async fn clean_spill_directory() -> opendal::Result<()> {
        static LOCK: LazyLock<Mutex<usize>> = LazyLock::new(|| Mutex::new(0));
        let _guard = LOCK.lock().await;

        let root = Self::batch_spill_root();

        let builder = Fs::default().root(&root.to_string_lossy());

        let op: Operator = Operator::new(builder)?

Review comment (performance):
`SpillOp::create` uses `assert!(path.is_relative())` for input validation. Since this is a `pub` API, a bad `path` would panic the process instead of returning a structured `BatchError`. Consider returning an error and also rejecting paths containing `..` components to prevent escaping the managed spill directory.

Code after:
const DEFAULT_IO_CONCURRENT_TASK: usize = 8;

#[derive(Clone)]
pub enum SpillBackend {
    Disk,
    /// Only for testing purpose
    Memory,
}

/// `SpillOp` is used to manage the spill directory of the spilling executor and it will drop the directory with a RAII style.
pub struct SpillOp {
    pub op: Operator,
}

impl SpillOp {
    fn batch_spill_root() -> PathBuf {
        batch_spill_base_dir().join(RW_MANAGED_SPILL_DIR)
    }

    pub fn create(path: impl AsRef<Path>, spill_backend: SpillBackend) -> Result<SpillOp> {
        let path = path.as_ref();
        if !path.is_relative() {
            bail!("Spill path must be relative, but got {:?}", path);
        }

        let root = Self::batch_spill_root().join(path);

        let op = match spill_backend {
            SpillBackend::Disk => {
                let builder = Fs::default().root(&root.to_string_lossy());
                Operator::new(builder)?
                    .layer(RetryLayer::default())
                    .finish()
            }
            SpillBackend::Memory => {
                let builder = Memory::default().root(&root.to_string_lossy());
                Operator::new(builder)?
                    .layer(RetryLayer::default())
                    .finish()
            }
        };
        Ok(SpillOp { op })
    }

    pub async fn clean_spill_directory() -> opendal::Result<()> {
        static LOCK: LazyLock<Mutex<usize>> = LazyLock::new(|| Mutex::new(0));
        let _guard = LOCK.lock().await;

        let root = Self::batch_spill_root();

        let builder = Fs::default().root(&root.to_string_lossy());