Support streaming put payloads

[senzzzi]

Which issue does this PR close?

Closes #281.

Rationale for this change

PutPayload currently only supports collections of Bytes. As a result, uploading a file through put or put_opts requires loading the entire file into memory or using a multipart upload.

This change allows callers to stream a payload with a known content length through a single PUT request. File-backed payloads are a primary use case.

What changes are included in this PR?

• Extends PutPayload to support replayable streaming bodies with a known content length.
• Adds PutPayload::from_stream for constructing streaming payloads.
• Adds PutPayload::from_file, which reads files in 16 KiB chunks by default.
• Adds PutPayload::from_file_with_chunk_size for configuring the file chunk size.
• Updates HTTP request bodies to poll payload streams asynchronously.
• Recreates the stream for every request attempt so retryable uploads can resend the complete payload.
• Updates S3 signing and checksum calculation to consume streaming payloads.
• Updates the local filesystem and in-memory backends to accept streaming payloads.
• Adds regression tests covering:
  • the default 16 KiB file chunk size;
  • replaying cloned streaming request bodies; and
  • retrying an upload using a file-backed streaming payload.

Validation performed:

• cargo test
• cargo test --all-features
• cargo clippy --all-targets --all-features -- -D warnings
• compilation with only the fs feature
• compilation without default features

Are there any user-facing changes?

Yes. Callers can now upload a file without first loading the complete file into memory:

let payload = PutPayload::from_file(file_path).await?;
store.put(&location, payload).await?;

Custom streaming payloads can be created with PutPayload::from_stream. The stream factory must produce a new stream from the beginning for every invocation, and the
supplied content length must exactly match the number of bytes yielded.

Existing byte-backed PutPayload construction and iteration behavior is unchanged. Synchronous APIs such as iter, AsRef<[Bytes]>, and conversion into Bytes are not supported for streaming payloads; callers should use PutPayload::stream or PutPayload::bytes instead.

API compatibility consideration

PutPayload remains a public struct backed by a private enum so that existing byte-backed construction and usage remain source compatible:

enum PutPayloadInner {
    Bytes(Arc<[Bytes]>),
    Streaming {
        factory: StreamFactory,
        content_length: usize,
    },
}

There is an API-design limitation around the existing synchronous accessors. Methods and trait implementations such as PutPayload::iter, AsRef<[Bytes]>, IntoIterator, and conversion into Bytes cannot synchronously represent an asynchronous, fallible stream.

In this implementation, their existing behavior is unchanged for byte-backed payloads, but calling them with a streaming payload panics. Streaming-aware implementations should instead use PutPayload::stream or PutPayload::bytes.

Changing the existing synchronous APIs to return Option or Result would avoid the panic but would be a breaking public API change. Returning an empty iterator for streaming payloads was rejected because it could silently result in incomplete uploads.

Feedback from maintainers on the preferred long-term API shape would be appreciated. Possible alternatives include:

• accepting the documented limitation for synchronous APIs;
• changing these APIs in a future breaking release; or
• introducing streaming uploads through a separate payload type or method.

[alamb]

You can also use Mulit-part upload to achieve something similar (though most object stores would still require you to buffer the parts in 5mb chunks)

We haven't released these docs yet, btu here is an example of how to do it

[alamb]

There is more rationale for non streaming APIs here:

• https://github.com/apache/arrow-rs/issues/5458
• Replace AsyncWrite with Upload trait and rename MultiPartStore to MultipartStore (#5458) arrow-rs#5500

TLDR is that streaming writes make a bunch of other operations (like retry , error handling, etc)

[alamb]

I see you have talked about multi-part uploads here: #281 (comment)

[senzzzi]

You can also use Mulit-part upload to achieve something similar (though most object stores would still require you to buffer the parts in 5mb chunks)

We haven't released these docs yet, btu here is an example of how to do it

It's not the same because of costs. Uploading a 64Mb file in a single put request is still quite fast (depending on your needs) and you are charged for 1 request. If you upload 64 different 1Mb parts, you are charged by the initial request, 64 parts individually and 1 final requests. So it's 66 times more expensive. You can check the link in this comment: #281 (comment)

[tustvold]

I think the idea is a good one, but as Andrew says we need to think about how retries, etc... would work

[tustvold]

This would likely be rather limiting in practice as it would break retries in client implementations

[alamb]

There is a bunch more discussion on #281 now too