feat(batch): bump opendal for batch spill out (#17550)

risingwavelabs · Jul 3, 2024 · 1c3bdc3 · 1c3bdc3
1 parent 1ac9583
commit 1c3bdc3
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 83 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/src/batch/Cargo.toml b/src/batch/Cargo.toml
@@ -32,7 +32,7 @@ hytra = "0.1.2"
 iceberg = { workspace = true }
 itertools = { workspace = true }
 memcomparable = "0.2"
-opendal = "0.45.1"
+opendal = "0.47"
 parking_lot = { workspace = true }
 paste = "1"
 prometheus = { version = "0.13", features = ["process"] }

diff --git a/src/batch/src/spill/spill_op.rs b/src/batch/src/spill/spill_op.rs
@@ -100,16 +100,16 @@ impl SpillOp {
         Ok(self
             .op
             .writer_with(name)
-            .buffer(DEFAULT_IO_BUFFER_SIZE)
             .concurrent(DEFAULT_IO_CONCURRENT_TASK)
+            .chunk(DEFAULT_IO_BUFFER_SIZE)
             .await?)
     }
 
     pub async fn reader_with(&self, name: &str) -> Result<opendal::Reader> {
         Ok(self
             .op
             .reader_with(name)
-            .buffer(DEFAULT_IO_BUFFER_SIZE)
+            .chunk(DEFAULT_IO_BUFFER_SIZE)
             .await?)
     }
 
@@ -123,7 +123,8 @@ impl SpillOp {
     /// [proto_bytes]
     /// ```
     #[try_stream(boxed, ok = DataChunk, error = BatchError)]
-    pub async fn read_stream(mut reader: opendal::Reader, spill_metrics: Arc<BatchSpillMetrics>) {
+    pub async fn read_stream(reader: opendal::Reader, spill_metrics: Arc<BatchSpillMetrics>) {
+        let mut reader = reader.into_futures_async_read(..).await?;
         let mut buf = [0u8; 4];
         loop {
             if let Err(err) = reader.read_exact(&mut buf).await {