Repository: risingwavelabs/risingwave
File: src/connector/src/sink/file_sink/opendal_sink.rs (Rust)

Code before:
}

/// Public interface exposed to `BatchingLogSinker`, used to write chunk and commit files.
impl OpenDalSinkWriter {
    /// This method writes a chunk.
    pub async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
        if self.sink_writer.is_none() {
            assert_eq!(self.current_bached_row_num, 0);
            self.create_sink_writer().await?;
        };
        self.append_only(chunk).await?;
        Ok(())
    }

    /// This method close current writer, finish writing a file and returns whether the commit is successful.
    pub async fn commit(&mut self) -> Result<bool> {
        if let Some(sink_writer) = self.sink_writer.take() {
            match sink_writer {
                FileWriterEnum::ParquetFileWriter(w) => {
                    if w.bytes_written() > 0 {
                        let metadata = w.close().await?;
                        tracing::info!(
                            "writer {} finish write file, metadata: {:?}",
                            self.unique_writer_id,
                            metadata
                        );
                    }
                }
                FileWriterEnum::FileWriter(mut w) => {
                    w.close().await?;
                }
            };
            self.current_bached_row_num = 0;
            return Ok(true);
        }
        Ok(false)
    }

    // Try commit if the batching condition is met.
    pub async fn try_commit(&mut self) -> Result<bool> {
        if self.can_commit() {
            return self.commit().await;
        }
        Ok(false)
    }
}

/// Private methods related to batching.
impl OpenDalSinkWriter {
    /// Method for judging whether batch condition is met.
    fn can_commit(&self) -> bool {

Review comment (bug):
I think it is better for debugging to include `executor_id` and the `created_time` as part of the path.

Also, is this a breaking change? Do we explicitly mention the sink file name format somewhere to our user?

Code after:
    ParquetFileWriter(AsyncArrowWriter<Compat<FuturesAsyncWriter>>),
    FileWriter(OpendalWriter),
}

/// Public interface exposed to `BatchingLogSinker`, used to write chunk and commit files.
impl OpenDalSinkWriter {
    /// This method writes a chunk.
    pub async fn write_batch(&mut self, chunk: StreamChunk) -> Result<()> {
        if self.sink_writer.is_none() {
            assert_eq!(self.current_bached_row_num, 0);
            self.create_sink_writer().await?;
        };
        self.append_only(chunk).await?;
        Ok(())
    }

    /// This method close current writer, finish writing a file and returns whether the commit is successful.
    pub async fn commit(&mut self) -> Result<bool> {
        if let Some(sink_writer) = self.sink_writer.take() {
            match sink_writer {
                FileWriterEnum::ParquetFileWriter(w) => {
                    if w.bytes_written() > 0 {
                        let metadata = w.close().await?;
                        tracing::info!(
                            "writer {} (executor_id: {}, created_time: {}) finish write file, metadata: {:?}",
                            self.unique_writer_id,
                            self.executor_id,
                            self.created_time
                                .duration_since(UNIX_EPOCH)
                                .expect("Time went backwards")
                                .as_secs(),
                            metadata
                        );
                    }
                }
                FileWriterEnum::FileWriter(mut w) => {
                    w.close().await?;
                }
            };
            self.current_bached_row_num = 0;
            return Ok(true);
        }
        Ok(false)
    }

    // Try commit if the batching condition is met.
    pub async fn try_commit(&mut self) -> Result<bool> {
        if self.can_commit() {
            return self.commit().await;
        }
        Ok(false)