**Describe the situation**

The author implemented it but enabled only for DiskS3.
CC @excitoon 
{
public:
    StorageS3Sink(
        const String & format,
        const Block & sample_block_,
        ContextPtr context,
        std::optional<FormatSettings> format_settings_,
        const CompressionMethod compression_method,
        const std::shared_ptr<Aws::S3::S3Client> & client,
        const String & bucket,
        const String & key,
        size_t min_upload_part_size,
        size_t upload_part_size_multiply_factor,
        size_t upload_part_size_multiply_parts_count_threshold,
        size_t max_single_part_upload_size)
        : SinkToStorage(sample_block_)
        , sample_block(sample_block_)
        , format_settings(format_settings_)
    {
        write_buf = wrapWriteBufferWithCompressionMethod(
            std::make_unique<WriteBufferFromS3>(
                client, bucket, key, min_upload_part_size,
                upload_part_size_multiply_factor, upload_part_size_multiply_parts_count_threshold,
                max_single_part_upload_size), compression_method, 3);
        writer = FormatFactory::instance().getOutputFormatParallelIfPossible(format, *write_buf, sample_block, context, {}, format_settings);
    }
    String getName() const override { return "StorageS3Sink"; }