[jira] [Created] (AVRO-3509) Allow stateful compression codecs in avro schemas

["Aaron Zinger (Jira)" <ji...@apache.org>]

Aaron Zinger created AVRO-3509:
----------------------------------

             Summary: Allow stateful compression codecs in avro schemas
                 Key: AVRO-3509
                 URL: https://issues.apache.org/jira/browse/AVRO-3509
             Project: Apache Avro
          Issue Type: Improvement
            Reporter: Aaron Zinger


Avro schemas specify a compression codec using a named string like "deflate". However, some workflows aren't able to use compression effectively because they're applying deflate separately to each individual record, when most compressibility comes from similarities between records.

For example, say I have a table containing the text of the book Moby Dick. Each row contains a chapter number, a view counter, and the text of a chapter. Whenever a row is updated, either to fix a typo in the text or to increment the view counter, I compress the row, encode it using an avro schema and codec stored separately (e.g. using a confluent schema registry), and send the compressed message to an external webhook, which fetches and caches the indicated metadata and uses it to decompress and decode the message. At first, the inefficiency is not too bad. The entire text of Moby Dick compresses to 41% of its original size, but compressing each chapter individually leads to 148 different records of total size 46% of the original. But now when I make a small text change to chapter 5, we re-emit the entire compressed row, for a 4320-byte message. If instead I were compressing the full text of Moby Dick plus a slightly-modified copy of chapter 5, the compressed file would only be 8 bytes longer! An ideal stateful streaming compression algorithm only needs to communicate which chapter is being modified (about 1 byte of entropy), and the diff. So over time, stateless compression stays at a steady 59% efficiency, while stateful compression would tend towards close to 100% efficiency.

We could enable stateful compression by allowing codecs of the form, say, "deflate+customdictionary={bytes}", where bytes is a dynamically generated Huffman coding of common strings, similar to the start of a compressed block in a normal deflate format. To decompress, you'd just fetch and cache the codec along with the schema and prepend it to each compressed message. Updating the codec would mean bumping the metadata version number, just like updating the schema, and would use the same flow. 

On the compression side, there are a couple of different implementation options. In my project that's motivating this, the same distributed workers that process rows and send messages are responsible for generating and registering schemas. So a worker would start out compressing using a vanilla codec, but build up a Huffman tree in memory as it sends each message. When it's both sent some threshold number of messages and compression using the custom tree is some threshold more efficient than using the most-recently-registered codec, it registers a new codec using the custom tree, then repeats this process, compressing in parallel using both the static registered codec and a stateful one it continues to update (maybe only for a random sample of records to save resources) until both thresholds are crossed again.

In other projects, the codec might be generated separately out-of-band by a centralized process that has access to all records. Compression workers would subscribe to or poll the schema registry and use the most up-to-date codec they have.

The result would be compression that is transparent to the consumer as long as they have access to a library that can decompress using these custom codecs, but that tends toward maximum efficiency over time, such that a 9122-byte message typically gets compressed to around 10 bytes.



--
This message was sent by Atlassian Jira
(v8.20.7#820007)