Using WriteMultipart::put results in 0 bytes being written

[orf]

WriteMultipart::put seems to be trivially broken. The method depends on the size of the buffer (via the content_length() method):

arrow-rs/object_store/src/upload.rs

Lines 183 to 194 in b25c441

	pub fn put(&mut self, mut bytes: Bytes) {
	while !bytes.is_empty() {
	let remaining = self.chunk_size - self.buffer.content_length();
	if bytes.len() < remaining {
	self.buffer.push(bytes);
	return;
	}
	self.buffer.push(bytes.split_to(remaining));
	let buffer = std::mem::take(&mut self.buffer);
	self.put_part(buffer.into())
	}
	}

However this is not updated by push():

arrow-rs/object_store/src/payload.rs

Lines 250 to 256 in b25c441

	pub fn push(&mut self, bytes: Bytes) {
	if !self.in_progress.is_empty() {
	let completed = std::mem::take(&mut self.in_progress);
	self.completed.push(completed.into())
	}
	self.completed.push(bytes)
	}

Repeated calls to .put() will not modify buffer.len (it will always be 0), so no part will ever be uploaded to S3.

Reproduction: foobar will have 0 bytes, even if you write more than 25mb of data.

let store = AmazonS3Builder::new().with_bucket_name("foo").build()?;
let upload = store.put_multipart("foobar").await?;
let uploader = WriteMultipart::new_with_chunk_size(upload, 1024 * 1024 * 25);
uploader.put(Bytes::from_static(b"yo"));
uploader.put(Bytes::from_static(b"yo2"));
uploader.put(Bytes::from_static(b"yo3"));
uploader.finish().await.unwrap();

[tustvold]

label_issue.py automatically added labels {'object-store'} from #5744

[orf]

Thanks for the super quick fix @tustvold <3