How to compress streamed items in batches of (roughly) the same (compressed) size?

[KrzysFR]

How can we batch items according to the resulting compressed (batch) size, and not the number of items, or the input buffer size, especially when the compression ratio can not be predicted, and vary wildly?

Use case

We have a list of small items of various size (a few hundred bytes to a few KB) that we want to compress in batches, by filling as much items as possible per batch, so that the compressed size of each batch would always end up around 30 KB, plus or minus a few KB (there is no hard limit).

Each batch will be stored individually, and must contain whole items: the decompressor will read batches individually, and must end up with an array of complete items. This means we cannot simply compress the items has a giant stream, and chunk the compressed output stream into blocks of exactly 30KB.

We don't need each batch to be exactly the same size. We have a margin of a plus/minus few KB. Also, we don't care about ordering. Each batch can be read or deleted in any order, once produced.

A finally, the items have very different compression ratio, which makes it almost impossible to estimate how much input size would compress down to 30KB. The only real way is to compress them, and see what we get. This means that some batch may require only a few items to output 30KB of compressed text, while others may require a few hundred.

Example

An example of that would be compressing JSON or XML fragments, streamed from a socket, while in the background outputting blocks of ~30 KB or compressed data every know and then, and without having to try compressing 1 fragment, then 2 fragment, then 3 fragments, ... until finally we have enough lines to fill the target compressed batch size.

Each batch produced would then be sent to some worker thread somewhere, decompressed, and passed through a filter that would see complete JSON documents (and not truncated chunks of JSON at the start or end of each batch).

Solution that does not work

We already tried estimating how much input size would compress down to ~30KB (let's say about 100KB of "text"), but in practice this does not work because of too much variation in the data we consume. Compression ratio vary widely, and we sometimes end up with batches from anywhere between 3KB to 90KB, which is way outside the desired result.

Solution that works but is slow

The current algorithm (which is too slow)

• Start with an empty buffer
• Read next item from stream
  • Append item to the compression buffer.
  • Compress the buffer, and measure the compressed size:
    • If >= 30KB, complete the block, output it, and empty the buffer
    • If not, continue on with the next item
• If no more items, but buffer is not empty, compress it and output it.

Ideal solution

What we would prefer:

• Start a new "batch" compression context
• Read next item from stream
  • "Append" the item to the compression context, and get current compressed size of the partial block
    • If the partial compression block length >= 30KB, complete the block, get the result compressed text, output it, and "reset" the compression context (to start a new block)
    • If not, continue with the next item
• If no more items, but a compression block is still pending, complete the block and output it.

Even though the compressed block would have been created by "appending" N fragments to it, we want the compressed result to be identical to compressing the concatenation of the N fragments in a single "zstd_compress" call.

Questions

I think that zstd_compress_continue could be used for this but I'm not sure about the following points:

• Does zstd_compress_continue output a full block when called? Ie: Would calling zstd_compress_continue on some input data produce the same output has calling it 4 times with a quarter of the same input data?
• Let's say the current compressed result is at 28 KB (almost there), and I output the final item that would make it reach 30KB. What would happen in this item would be uncompressible and large enough, to overflow the output compression buffer? How would I be able to recover from the error, and output the compressed block as it was BEFORE the call?
• What would it take to have a compression method that takes in an array of input buffers, and a target destination buffer+size, and have the method return the compressed size, as well as the number of input buffers that were consumed (or fail if even the first buffer would overflow the destination buffer).

[Cyan4973]

we want the compressed result to be identical to compressing the concatenation of the N fragments in a single "zstd_compress" call

That's the key point in your objective definition.

I initially expected you wanted N frames from N fragments just stored back to back up to maxBufferSize. That would have been easier, but also introduce some other limitations.

The scenario you are looking for is not supported yet.

But let's see how it could be approached.

ZSTD_compressContinue() is a good place to start, but there is a big difference between calling N times ZSTD_compressContinue() and calling a single time ZSTD_compress() with the N fragments : each invocation of ZSTD_compressContinue() generates at least one block (more if input > 128 KB). N blocks means more headers, hence less compression ratio, than a single block.

This is because ZSTD_compressContinue() is a synchronous interface, with no intermediate buffer.

Such capability is the domain of ZBUFF interface. Here, adding input doesn't immediately translate into creating a new block. You'll have to either reach an accumulated source size > one block (typically 128 KB), or send a flush() command, or send an end() command.

Not producing a block is good for compression ratio, but is bad for size estimation.

So here is an imagined workflow which could match your objective, although it's not completely possible yet (see details below) :

• Create a ZBUFF context and init it.
• For each fragment
  • Save the ZBUFF context, using ZBUFF_copyCCtx() (which doesn't exist !)
  • Add the fragment using ZBUFF_compressContinue()
  • Evaluate final size by using ZBUFF_compressEnd()
  • Restore ZBUFF context
    • If size is fine : Add the fragment again, continue
    • if size is too large : don't add the fragment, end with ZBUFF_compressEnd()

The downside of this method :

• 2 compressions per fragment
• copy / restoration of context for each fragment

So my guess is that it will be quite slower. How much precisely is unclear.
In particular, it's not even clear if it will be better than your own "works but too slow" approach.

Better than that will be more difficult though.
It would require specific developments.
Basic spirit would be the same : add, try, decide,
but at least intermediate results could be preserved, avoiding re-compressing them.
There is also a chance that compression size could be estimated, rather than discovered at the end of compression operation.

So it would likely be faster (and use less memory).
But it will require specific developments...

[KrzysFR]

Thanks for the detailed response. I was afraid that it was not so simple :)

If we add more constraints to it, would be it help simplify this?

1. Instead of deciding if adding the new fragment would overflow, and have to rollback, it would be perfectly acceptable to add the new fragment, and then decide if we stop now, or continue on with the next fragment (i.e.: no "try add and maybe rollback", simply "add and then maybe stop after that").

We would then just need to make sure that every time we add a new fragment, we accept that it could not compress at all, and make sure that the remaining buffer size is large enough to make it fit (ie: ensure that there is always at least zstd_compress_bound(next_fragment->size) bytes left in the dest buffer)

or 2) could there be a way to get a rough estimate of how much ZBUFF_compressEnd() would produce, just enough to help decide if we should stop after that fragment, but without actually doing the work and mutate the context? Again, we don't need perfect accuracy, and already have large enough security margins.

or 3) instead of adding items one by one until it fills up, let them fill up to 128KB, compress them into blocks, and hold on these blocks until we have about ~30KB +/- size of a block. If the compressed blocks are small enough then it can be doable to group them in such a way, but at the cost of adding some latency to the system (ie: instead of outputting blocks once they are full, we would output multiple blocks in bursts)

Method 3 seems doable with the current API, if we put enough constraints on the input data. Only downside is that we need more in-memory buffering, and we would need special casing when a 128KB block compress really bad (we'd need to split the input data in 2 or 4 pieces, and retry compression).

Finally, would the idea of passing a list of buffers to zstd, instead of a single buffer, plus a compressed size constraint (with 'will stop once it goes over' semantic, not 'absolute maximum size'), and sufficiently large destination buffer for the worst case, would it help simplify the problem ?

If we say that there could be no literal match across two input buffers boundary, would this simplify the internal logic? (i.e.: allowing the buffers to be disjoint in memory, at the cost of a little bit of reduced compression ratio).

[Cyan4973]

OK, let's try a less precise method, but with much better speed perspectives.

I would suggest a 2-passes approach.

Let's assume you have a hard-limit at 32 KB, and a soft objective at 30 KB.

• Compress up to 32 KB of input.
• Get the compression ratio from this first try
• Use the compression ratio to reach the 30 KB objective

Example :

• Suppose first stage produced a 50% compression ratio.
  So 32 KB get compressed to 16 KB.
  Now there is a 30-16 = 14 KB budget remaining.
  14 KB / 50% => add 28 KB
  • Now compress 32+28=60 KB worth of input data
• Less than max == 32 KB ? => you are done
• More than 32 KB ? Try again, reduce input a bit.
  Use new compression ratio to determine by how much.

I would expect the process to work in 2 passes in most cases.
Of course, there are also corner cases to take care of (if it doesn't compress in first pass for example), but it looks manageable.
Its main downside is that it's not "precise", that is maybe in above example we could fit 61 KB of input, but in the interest of speed and simplicity, we don't even try, and get away with 60 KB.

The entire process can be done externally right now, just using ZSTD_compress().
It could nonetheless be encapsulated into an advanced interface, which would do the same thing, just be more convenient to call.
Something like :

size_t ZSTD_compress_destSize(ZSTD_CCtx* cctx, 
void* dstBuffer, size_t dstCapacity, 
const void* srcBuffer, const size_t* segmentSizes, unsigned* nbSegmentsPtr);

( Similar to ZDICT_trainFromBuffer() ) .
The result would be both a compressed size and a nb of segments used to reach that.

At a later stage, the inner process could be improved (both on speed and packing efficiency). But the advanced interface would remain the same, so improvement would be transparent.

[KrzysFR]

Yes, this looks like something that could be done. We are currently also exploring other ideas that would maybe improve the compression ratio by reordering data, but at the cost of allocating more memory (10's of MiB versus a few 64 or 128 KiB buffers).

To be more explicit, our goal is to compress a stream of small items into compressed blobs, that would then be stored into a B-Tree with a page size of 64 KiB (maybe even 128 KiB). A consumer that would need to read one item (or a range of items), would need to find the corresponding blob, decompress the payload, and extract what it needs, throwing the rest away (the reason why items cannot be truncated).

If you are familiar with B-Trees, you may remember that a page size is almost always a power of 2 (ex: 65,536 bytes), but there is some overhead per entry (page header, key bytes, various lengths and offsets, ...), which means that we cannot blindly chunk the compressed output stream into nice round buffers with power of two sizes (ex: 32,768).

=> If we store a value of exactly 32,768 bytes into a 64 KB B-tree leaf page, there will only remain ~32,500 bytes free space in the page. The next value of 32,768 bytes will not fit, hence creating a b-tree with about 51% fill ratio, and doubling the size on disk, as well as halving the read and write speed of the database.

Our goal is to generate values that are slightly smaller, so that we can fit at least 2 values per page (more ideally), and keep the wasted free space to a minimum. The idea being that we have a target of ~30,000 bytes, and tolerate up to 2,500 bytes extra. And if there are a very few blocks that overshoot the limit and waste half a page, this should be acceptable (if rare enough).

We may even need to lower our threshold to something like 15,000 bytes, so that we can fit at least 4 blobs per 64 KiB leaf page, depending on what the final benchmarks will produce, though probably lowering the compression ratio.

[FrancescAlted]

Not completely sure why you want to store the compressed data straight into the B-Tree, but have you thought on storing the different compressed blobs in a linear structure and then add another level of indirection to so that what it is stored in the B-Tree is actually where the blob starts and its length? This would avoid the wasted space (at the cost of just a --pretty quick-- additional indirection). My 2 cents.

[KrzysFR]

The existing blobs can be removed randomly, which mean having to garbage collect the flat files. This looks too much like an LSM to me, and I don't think it would be a good fit for what we need.

Also, the B-Tree supports overflow pages for large values, where the cells in the leaf pages contain reference pointers to the remote value. This looks a lot like storing a reference to another file, except that I can let the existing B-Tree implementation deal with allocation and most importantly keep serializable transactions (not sure how it would work if I needed to fsync multiple files in the same transaction...).

Using the overflow pages, I would simply need to set the threshold to be fill an entire page, but not more. Except now, all my leaf pages are in fact the same as branch pages (they contain pointers to other pages), and my overflow pages are the actual leaf pages (with only one value, but still). Back to square one, except now the tree is one level taller.

Another approach, which could work also, would be to allow even higher latency, and compress very large batches, producing compressed blobs that are a lot larger than the page size. That way, the wasted free space (on average, half the size of a page) would become a negligible percentage. The downside is having to deal with huge memory buffers, and potentially more CPU wasted when decompressing.

All of this makes me think that the ideal solution would be to be able to produce small compressed blobs that play nicely with the constraints of b-trees (or any store that uses fixed-size pages, like most file systems).

[FrancescAlted]

Yeah, many different possibilities. For what is worth, what I am doing for Blosc2 is to have 2 different representations for data on-disk and data in-memory. On disk, you have all the chunks sequentially stored (so with no holes in media), whereas in memory, there is a structure that holds pointers to the different chunks that are malloced/freed independently (see https://speakerdeck.com/francescalted/blosc-and-bcolz-efficient-compressed-data-containers?slide=16). This allows for efficient storage of compressed chunks in bot disk or memory, but the inconvenient is that when you need to modify of remove a chunk you have to load in memory the data that is on disk and then re-serialize it again. Anyway, different use cases require different approaches.

[Cyan4973]

Is there any action left on this item ?

[KrzysFR]

We have decided to go the other way around and create larger blobs that span multiple pages so that the free space waste (half a page in average) is now a small percentage of the total (less than 10% waste with at least 5 pages long). I changed the format of the file to be able to fit power-of-2 sized blobs without extra overhead, which also helped a bit (ie: worst case of 128KB uncompressed block would fit perfectly).

The one remaining place I'd like to use this, is to be able to optimize the batch of transactions written to a log file that uses FILE_FLAG_WRITE_THROUGH mode (ie: writes must be aligned and of size multiple of the page size).

Given an incoming async stream of transaction data to write (each going from changing a single value to bulk inserting several MB of data), be able to compress them on the fly, and when the next compressed transaction would overflow a page, stop the batch there, and fsync everything, before starting over a new batch from that transaction onward (until the next overflow).

So, I still think that in the long run it would be very useful to have some way to achieve that when dealing with very unstable data. Having a way to present a new chunk of data to the current block, and get an estimate of the new compressed size, without actually having to write to a buffer or mutate the current state (even at the cost of some memory allocations/copy), could help solve this.

In most scenarios, I would guess that the extra CPU cost would be way smaller than the I/O cost of overshooting by one page (and especially the cost of fsync!), so it would still be a net gain.