repository·1.x·Indexed 21 days ago
https://github.com/datadog/zstdA Go wrapper for the Zstd compression library providing one-shot compression/decompression and a streaming API compatible with io.Reader and io.Writer. It includes a BulkProcessor for high-throughput dictionary-based compression, support for parallel compression via SetNbWorkers, and the ability to build against an external libzstd (version 1.4.0 or higher) using the external_libzstd build tag.
By default, this package uses vendored zstd source code. To build against an external static or shared libzstd library instead, use the external_libzstd build tag. This requires libzstd version 1.4.0 or higher and uses pkg-config to resolve build and linking parameters.
go build -tags external_libzstdThe BulkProcessor is designed for scenarios where you need to compress or decompress many small messages or blocks using the same pre-trained dictionary.
Key Benefits:
BulkProcessor instance can be shared by multiple goroutines concurrently because its usage is read-only.BulkProcessor object.Workflow:
BulkProcessor using NewBulkProcessor with your dictionary and desired compression level.Compress or Decompress using the processor instance.dst to minimize allocations.// Example initialization
bulkProc, err := zstd.NewBulkProcessor(dictionaryBytes, 3)
if err != nil {
log.Fatal(err)
}
// Example compression
compressed, err := bulkProc.Compress(nil, originalData)To perform compression or decompression, you should create a Ctx using NewCtx().
Best Practices:
Ctx instance per thread to allow for parallel execution.ctx := zstd.NewCtx()
// Use ctx for multiple operations
compressed, err := ctx.Compress(nil, data)You can run the included benchmarks against your own data by setting the PAYLOAD environment variable to the path of your payload file and using the standard Go benchmark command.
export PAYLOAD=/path/to/your/data
go test -bench .The simple API is designed for one-shot compression and decompression of byte arrays, mirroring the lz4 API. You can provide a pre-allocated destination buffer to dst to minimize allocations. If dst is nil, the functions will allocate the necessary space.
// Compress compresses src into dst. If dst is nil, it allocates the worst case size.
// If dst is too small, it will be reallocated and returned.
Compress(dst, src []byte) ([]byte, error)
// CompressLevel is the same as Compress but allows specifying a compression level.
CompressLevel(dst, src []byte, level int) ([]byte, error)
// Decompress decompresses src into dst. If dst is nil, it allocates 4*src as default.
// If dst is too small, it retries up to 3 times by doubling the size before
// switching to the slower stream API.
Decompress(dst, src []byte) ([]byte, error)The streaming API is designed as a drop-in replacement for zlib. It uses io.Writer for compression and io.Reader for decompression.
Important: You MUST call Close() on writers to ensure the last bytes of the stream are written and C objects are freed. For readers, you must call Close() to free the underlying C objects.
// --- Compression ---
// NewWriter creates a writer for the provided io.Writer.
// NewWriterLevel allows specifying a compression level.
// NewWriterLevelDict allows using a precomputed dictionary (dict must not be modified during use).
// MUST call Close() to flush and free C objects.
NewWriter(w io.Writer) *Writer
NewWriterLevel(w io.Writer, level int) *Writer
NewWriterLevelDict(w io.Writer, level int, dict []byte) *Writer
// Write compresses input data to the underlying writer.
// Flush writes unwritten data to the underlying writer.
// Close flushes the buffer and frees C objects.
(w *Writer) Write(p []byte) (int, error)
(w *Writer) Flush() error
(w *Writer) Close() error
// --- Decompression ---
// NewReader returns an io.ReadCloser for decompressing data from the underlying reader.
// NewReaderDict allows using a precomputed dictionary (dict must not be modified until Close is called).
// MUST call Close() to free C objects.
NewReader(r io.Reader) io.ReadCloser
NewReaderDict(r io.Reader, dict []byte) io.ReadCloserThe zstd.Writer is an io.WriteCloser that compresses input data using the Zstandard algorithm. You can initialize it with default settings, a specific compression level, or a compression dictionary.
Key Methods:
NewWriter(w io.Writer): Creates a writer with DefaultCompression.NewWriterLevel(w io.Writer, level int): Creates a writer with a specific level (e.g., BestSpeed to BestCompression).NewWriterLevelDict(w io.Writer, level int, dict []byte): Creates a writer using a provided dictionary for improved compression ratios.Write(p []byte): Compresses and writes data to the underlying writer.Flush(): Flushes any unwritten compressed data to the underlying writer.Close(): Flushes remaining data, finishes the Zstd stream, and frees C resources. Always call Close() to ensure all data is written and memory is released.SetNbWorkers(n int): Enables parallel compression using n threads. Note that if n > 1, Write() calls become asynchronous and may buffer data in memory.// Basic usage with default compression
writer := zstd.NewWriter(file)
defer writer.Close()
// Writing data
n, err := writer.Write([]byte("hello world"))
// Using a specific compression level
writer := zstd.NewWriterLevel(file, zstd.BestCompression)To speed up compression, you can use multiple threads by calling SetNbWorkers(n int) on a zstd.Writer.
Important Considerations:
n > 1, Write() calls become asynchronous. Data is buffered internally until processed.Flush() periodically.libzstd was not compiled with parallel support, this method will return ErrNoParallelSupport.writer := zstd.NewWriter(file)
err := writer.SetNbWorkers(4) // Use 4 threads
if err != nil {
if errors.Is(err, zstd.ErrNoParallelSupport) {
// Fallback or handle lack of parallel support
}
}The zstd package uses an ErrorCode type (an alias for int) to represent errors returned by the underlying C library. When a function returns a negative integer, you can convert it to an ErrorCode to retrieve the human-readable error string provided by zstd.
To check for specific error conditions, use the IsDstSizeTooSmallError helper function, which identifies if an error is specifically due to the destination buffer being too small for decompression.
// Example of checking for a specific error type
err := zstd.SomeFunction()
if err != nil {
if zstd.IsDstSizeTooSmallError(err) {
// Handle buffer size issue
}
fmt.Println("Zstd error:", err.Error())
}Decompresses src into dst using the dictionary associated with the BulkProcessor.
Parameters:
dst []byte: The destination buffer.cap(dst) is large enough to hold the decompressed data, it will be reused.dst is nil or too small, a new buffer will be allocated. Note that unlike Compress, if the buffer is too small, it cannot be recovered via streaming fallbacks.src []byte: The compressed source data.Returns:
[]byte: The decompressed data slice.error: Returns ErrEmptySlice if src is empty, or other errors if decompression fails.// Decompressing into a new buffer
decompressed, err := processor.Decompress(nil, compressedData)There are two primary ways to decompress data:
Decompress(dst, src): The high-level API. It attempts to use the provided dst buffer if it's large enough. If the buffer is too small or the payload size is unknown, it may allocate a new buffer or fall back to a streaming reader to ensure the full payload is retrieved.DecompressInto(dst, src): A low-level API that requires dst to be pre-allocated with sufficient capacity to hold the entire decompressed payload. It returns an error if dst is too small.Note: For frames that do not advertise their size, Decompress may partially overwrite the dst buffer even if a new slice is returned. Do not rely on the contents of dst after such a call.
// High-level decompression (handles allocation/streaming automatically)
decompressed, err := zstd.Decompress(nil, src)
// Low-level decompression (requires pre-allocated dst)
dest := make([]byte, expectedSize)
n, err := zstd.DecompressInto(dest, src)Creates a new BulkProcessor by digesting a provided dictionary. This prepares both compression (CDict) and decompression (DDict) states.
Parameters:
dictionary []byte: The pre-trained Zstd dictionary.compressionLevel int: The Zstd compression level to use for compression operations.Returns:
*BulkProcessor: A pointer to the initialized processor.error: Returns ErrEmptyDictionary if the dictionary is empty, or ErrBadDictionary if the dictionary cannot be loaded.processor, err := zstd.NewBulkProcessor(myDict, 3)