sink_parquet_cloud doesnt work when updating from 0.40 -> 0.41

[a-agmon]

Checks

• I have checked that this issue has not already been reported.
• I have confirmed this bug exists on the latest version of Polars.

Reproducible example

let parquet_write_opt = ParquetWriteOptions {
        compression: ParquetCompression::Uncompressed,
        ..Default::default()
    };

    LazyFrame::scan_parquet(source, cloud_args.clone())?
        .with_streaming(true)
        .with_predicate_pushdown(true)
        .with_projection_pushdown(true)
        .select([col("A"), col("B"), col("C")])
        .group_by([col("B"), col("C")])
        .agg([col("A").count().alias("apps")])
     
        .sink_parquet_cloud(target, cloud_args.cloud_options.clone(), parquet_write_opt);

Log output

POLARS PREFETCH_SIZE: 20
RUN STREAMING PIPELINE
[parquet -> generic-group_by -> parquet_sink]
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
STREAMING CHUNK SIZE: 16666 rows
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
POLARS ROW_GROUP PREFETCH_SIZE: 128
finish streaming aggregation with local in-memory table

Issue description

On version 0.40 this code was used to write a small aggregation to cloud storage (S3), after updating to 0.41.* the file is not written and no error is being thrown

Expected behavior

On version File will be written to S3 or error thrown

Installed versions

polars = { version = "0.41.2", features = [
"lazy",
"aws",
"parquet",
"streaming",
"performant",
"cloud_write",
] }

[ritchie46]

This functionality was community provided. At the moment it is a bit low priority for us. We want to redesign this properly in the new engine. I would accept a fix.

[philss]

It seems that with the new object_store v0.10.0 (landed in Polars after PR #16920), the trait that puts the parts is not performing an internal buffer anymore. So each call to writer.put_part makes a new network request, and I believe this is leading to errors.

This is not explicitly documented in the ObjectStore's change log, but is described in the "removed docs" of this commit: apache/arrow-rs@96c4c0b

As a solution, we may need to use the WriteMultipart that does buffer the data to a certain capacity before making the request, or we can use the new BufWriter that automatically choose between doing a single "put" or a "put multipart" depending on the size of the payload, and does the buffer as the name suggests.

Do you have any preference? I can try to submit a PR with a fix for this.