**Describe the situation**

The author implemented it but enabled only for DiskS3.
CC @excitoon 
void consume(Chunk chunk) override
    {
        writer->write(getHeader().cloneWithColumns(chunk.detachColumns()));
    }
    void onFinish() override
    {
        try
        {
            writer->finalize();
            writer->flush();
            write_buf->finalize();
        }
        catch (...)
        {
            
            writer.reset();
            throw;
        }
    }
private:
    Block sample_block;
    std::optional<FormatSettings> format_settings;
    std::unique_ptr<WriteBuffer> write_buf;
    OutputFormatPtr writer;
};
class PartitionedStorageS3Sink : public PartitionedSink
{
public:
    PartitionedStorageS3Sink(