apache / apache/arrow-rs-object-store

Multipart Upload Concurrency Issues

Open
#276 16 comments 0 reactions 0 assignees View on GitHub
question
Dominant language
Rust
Stars
322
Forks
212
Avg merge
5d 2h
Merged PRs (30d)
10

Description

**Describe the bug**

So recently we started seeing two issues:
- multiple shutdowns of the same writer will cause an issue, related to [this](https://github.com/apache/arrow-rs/blob/50e9e4927ed4d763c76227c588aa324aa173032e/object_store/src/buffered.rs#L220
- partition size issue, related to [this](https://github.com/apache/arrow-rs/blob/50e9e4927ed4d763c76227c588aa324aa173032e/object_store/src/upload.rs#L120)

# Multiple Shutdown

First of all, I want to note that multiple shutdown calls to the same writer are the issue by itself, but I think we can make the situation better with minimum effort.

Here is the code:
```rust
BufWriterState::Write(x) => {
let upload = x.take().ok_or_else(|| {
std::io::Error::new(
ErrorKind::InvalidInput,
"Cannot shutdown a writer that has already been shut down",
)
})?;
self.state = BufWriterState::Flush(
async move {
upload.finish().await?;
Ok(())
}
.boxed(),
)
}
```

I think we can change it to something more friendly like this:
```rust
BufWriterState::Write(x) => {
if let Some(upload) = x.take() {
self.state = BufWriterState::Flush(
async move { upload.finish().await.map(|_| ()) }.boxed(),
)
} else {
return Poll::Ready(Ok(()));
}
}
```

This way on a second shutdown call we just immediately return `Ok(())`

# Upload part size issue

Something leftover during the shutdown, complete before the previous upload, in this case, we're getting:
```
Your proposed upload is smaller than the minimum allowed size
```

To mitigate this issue we probably should wait for all previous part uploads to complete and then upload the final part which may be smaller than the minimum size of the last one.

Here is the original code I propose to change:

```rust
pub async fn finish(mut self) -> Result {
if !self.buffer.is_empty() {
let part = std::mem::take(&mut self.buffer);
self.put_part(part.into())
}

self.wait_for_capacity(0).await?;

match self.upload.complete().await {
Err(e) => {
self.tasks.shutdown().await;
self.upload.abort().await?;
Err(e)
}
Ok(result) => Ok(result),
}
}
```

by injecting `self.wait_for_capacity(0).await?;` before actually putting the last chunk we can mitigate this issue.

```rust
pub async fn finish(mut self) -> Result {
if !self.buffer.is_empty() {
self.wait_for_capacity(0).await?; // here

let part = std::mem::take(&mut self.buffer);
self.put_part(part.into())
}

self.wait_for_capacity(0).await?;

match self.upload.complete().await {
Err(e) => {
self.tasks.shutdown().await;
self.upload.abort().await?;
Err(e)
}
Ok(result) => Ok(result),
}
}
```

This way we wait for all ongoing uploads before submitting the last part

Contributor guide

Open the contributing guide

Research direction

Start in object_store/src/buffered.rs at the BufWriterState shutdown handling and in object_store/src/upload.rs at MultipartUpload::finish. Trace repeated shutdown calls and the order of outstanding part uploads relative to the final part. Done means repeated shutdown is harmless and a final undersized part is submitted only after earlier uploads complete, without the minimum-size error.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
backend
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.