Documentation
¶
Index ¶
- Constants
- Variables
- func AppendDecoded(dst, src []byte) ([]byte, error)
- func AppendEncoded(dst, src []byte, level int) ([]byte, error)
- func BuildSidecar(dst io.Writer, src io.Reader, opts ...SidecarOption) error
- func Decode(dst, src []byte) ([]byte, error)
- func DecodedLen(src []byte) (int, error)
- func Encode(dst, src []byte, level int) ([]byte, error)
- func ExtractSidecar(sidecarDst, newStreamDst io.Writer, src io.Reader, opts ...SidecarOption) error
- func IndexStream(r io.Reader) ([]byte, error)
- func IsMinLZ(src []byte) (ok bool, size int, err error)
- func MaxEncodedLen(srcLen int) int
- func RemoveIndexHeaders(b []byte) []byte
- func RestoreIndexHeaders(in []byte) []byte
- func TryEncode(dst, src []byte, level int) []byte
- type BlockSearchOption
- type BlockSearcher
- type CompressedSearchOption
- type CompressedSearchStats
- type ErrCantSeek
- type Index
- type OffsetPair
- type ReadSeeker
- type Reader
- func (r *Reader) DecodeConcurrent(w io.Writer, concurrent int) (written int64, err error)
- func (r *Reader) GetBufferCapacity() int
- func (r *Reader) Read(p []byte) (int, error)
- func (r *Reader) ReadByte() (byte, error)
- func (r *Reader) ReadSeeker(index []byte) (rrs *ReadSeeker, err error)
- func (r *Reader) Reset(reader io.Reader)
- func (r *Reader) Skip(n int64) error
- func (r *Reader) UserChunkCB(id uint8, fn func(r io.Reader) error) error
- func (r *Reader) WriteTo(w io.Writer) (n int64, err error)
- type ReaderOption
- type SearchResult
- type SearchStats
- type SearchTableConfig
- func (c SearchTableConfig) String() string
- func (c SearchTableConfig) WithBytePrefix(prefixes ...byte) SearchTableConfig
- func (c SearchTableConfig) WithCompression(opts ...CompressedSearchOption) SearchTableConfig
- func (c SearchTableConfig) WithExtras(n int) SearchTableConfig
- func (c SearchTableConfig) WithLongPrefix(prefix []byte) SearchTableConfig
- func (c SearchTableConfig) WithMaskPrefix(mask [32]byte) SearchTableConfig
- func (c SearchTableConfig) WithMatchLen(n int) SearchTableConfig
- func (c SearchTableConfig) WithMaxPopulation(pct int) SearchTableConfig
- func (c SearchTableConfig) WithMaxReducedPopulation(pct int) SearchTableConfig
- func (c SearchTableConfig) WithoutCompression() SearchTableConfig
- type SidecarOption
- type SidecarSearcher
- type WindowStat
- type Writer
- func (w *Writer) AddUserChunk(id uint8, data []byte) (err error)
- func (w *Writer) AsyncFlush() error
- func (w *Writer) Close() error
- func (w *Writer) CloseIndex() ([]byte, error)
- func (w *Writer) EncodeBuffer(buf []byte) (err error)
- func (w *Writer) Flush() error
- func (w *Writer) ReadFrom(r io.Reader) (n int64, err error)
- func (w *Writer) Reset(writer io.Writer)
- func (w *Writer) SetSidecar(sidecar io.Writer) error
- func (w *Writer) Write(p []byte) (nRet int, errRet error)
- func (w *Writer) Written() (input, output int64)
- type WriterOption
- func WriterAddIndex(b bool) WriterOption
- func WriterBlockSize(n int) WriterOption
- func WriterConcurrency(n int) WriterOption
- func WriterCreateIndex(b bool) WriterOption
- func WriterCustomEncoder(fn func(dst, src []byte) int) WriterOption
- func WriterFlushOnWrite() WriterOption
- func WriterLevel(n int) WriterOption
- func WriterPadding(n int) WriterOption
- func WriterPaddingSrc(reader io.Reader) WriterOption
- func WriterSearchTable(cfg SearchTableConfig) WriterOption
- func WriterSidecar(sidecar io.Writer) WriterOption
- func WriterUncompressed() WriterOption
Examples ¶
Constants ¶
const ( // LevelSuperFast is the fastest compression level. // This will take significant shortcuts and usually provide much worse compression. // Use only if LevelFastest is confirmed to be too slow. LevelSuperFast = -1 // LevelUncompressed will bypass compression. LevelUncompressed = 0 // LevelFastest is the fastest compression level. LevelFastest = 1 // LevelBalanced is the balanced compression level. // This is targeted to be approximately half the speed of LevelFastest. LevelBalanced = 2 // LevelSmallest will attempt the best possible compression. // There is no speed target for this level. LevelSmallest = 3 )
const ( IndexHeader = "s2idx\x00" IndexTrailer = "\x00xdi2s" )
const ( // MaxBlockSize is the maximum value where MaxEncodedLen will return a valid block size. MaxBlockSize = 8 << 20 // MinUserSkippableChunk is the lowest user defined skippable chunk ID. // All chunks IDs within this range will be ignored if not handled. MinUserSkippableChunk = 0x80 // MaxUserSkippableChunk is the last user defined skippable chunk ID. MaxUserSkippableChunk = 0xbf // MinUserNonSkippableChunk is the lowest user defined non-skippable chunk ID. // All chunks IDs within this range will cause an error if not handled. MinUserNonSkippableChunk = 0xc0 // MaxUserNonSkippableChunk is the last user defined non-skippable chunk ID. MaxUserNonSkippableChunk = 0xfd // ChunkTypePadding is a padding chunk. ChunkTypePadding = 0xfe // ChunkTypeStreamIdentifier is the Snappy/S2/MinLZ stream id chunk. ChunkTypeStreamIdentifier = 0xff // MaxUserChunkSize is the maximum possible size of a single chunk. MaxUserChunkSize = 1<<24 - 1 // 16777215 )
Variables ¶
var ( // ErrCorrupt reports that the input is invalid. ErrCorrupt = errors.New("minlz: corrupt input") // ErrCRC reports that the input failed CRC validation (streams only) ErrCRC = errors.New("minlz: corrupt input, crc mismatch") // ErrTooLarge reports that the uncompressed length is too large. ErrTooLarge = errors.New("minlz: decoded block is too large") // ErrUnsupported reports that the input isn't supported. ErrUnsupported = errors.New("minlz: unsupported input") // ErrInvalidLevel is returned when an invalid compression level is requested. ErrInvalidLevel = errors.New("minlz: invalid compression level") )
var ErrSearchForward = errors.New("minlz: forward to next block")
ErrSearchForward can be returned from the search callback to request forward context. The searcher will decode the next block, shift Blocks forward, and re-call the callback with the same match but more context.
var ErrSearchTablesUnusable = errors.New("minlz: search tables cannot be used for this pattern")
ErrSearchTablesUnusable is returned by BlockSearcher.Search when BailOnMissing is set and search tables cannot accelerate the query.
var ErrSidecarInvalid = errors.New("minlz: invalid sidecar stream")
ErrSidecarInvalid is returned when a sidecar stream is malformed.
Functions ¶
func AppendDecoded ¶
AppendDecoded will append the decoded version of src to dst. If the decoded content cannot fit within dst, it will cause an allocation. This decoder has automatic fallback to Snappy/S2. To reject fallback check with IsMinLZ. The dst and src must not overlap. It is valid to pass a nil dst.
func AppendEncoded ¶
AppendEncoded will append the encoded version of src to dst. If dst has MaxEncodedLen(len(src)) capacity left it will be done without allocation. See Encode for more information.
func BuildSidecar ¶ added in v1.2.0
BuildSidecar reads a compressed MinLZ stream from src, generates fresh per-block search tables according to the provided SidecarSearchTable configs, and writes the sidecar to dst. src is read sequentially and is not modified. The 0x47 Remote Block References in the sidecar point at the absolute offsets of data blocks within src.
At least one SidecarSearchTable option must be supplied. Multiple configs produce multiple 0x44 info chunks and per-block 0x45/0x46 chunks; the SidecarSearcher tries each config independently and ANDs the skip results.
Compression-only blocks in src (chunk types 0x02 / 0x03) are decoded to build the tables; uncompressed blocks (0x01) are also indexed. Other chunks (existing 0x45/0x46 in src, padding, index, user) are skipped.
Example ¶
package main
import (
"bytes"
"fmt"
"github.com/minio/minlz"
)
func main() {
// 1) Compress without any inline search tables.
var main bytes.Buffer
w := minlz.NewWriter(&main,
minlz.WriterBlockSize(4096),
minlz.WriterConcurrency(1),
)
data := make([]byte, 8192)
copy(data[:4096], bytes.Repeat([]byte("aaaa"), 1024))
copy(data[4096:], bytes.Repeat([]byte("bbbb"), 1024))
copy(data[4096+200:], []byte("NEEDLE_PATTERN"))
w.Write(data)
w.Close()
// 2) Build a sidecar against that pre-existing main stream. The source
// is read once and not modified.
var side bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(4)
if err := minlz.BuildSidecar(&side, bytes.NewReader(main.Bytes()),
minlz.SidecarSearchTable(cfg),
); err != nil {
fmt.Println("build:", err)
return
}
// 3) Search via the sidecar against the original main.
ss := minlz.NewSidecarSearcher(
bytes.NewReader(main.Bytes()),
bytes.NewReader(side.Bytes()),
minlz.BlockSearchCollectStats(),
)
ss.Search([]byte("NEEDLE_PATTERN"), func(r minlz.SearchResult) error {
fmt.Printf("found at stream offset %d\n", r.StreamOffset)
return nil
})
stats := ss.Stats()
fmt.Printf("blocks: %d total, %d skipped\n", stats.BlocksTotal, stats.BlocksSkipped)
}
Output: found at stream offset 4296 blocks: 2 total, 1 skipped
func Decode ¶
Decode returns the decoded form of src. The returned slice may be a sub- slice of dst if dst was large enough to hold the entire decoded block. Otherwise, a newly allocated slice will be returned.
This decoder has automatic fallback to Snappy/S2. To reject fallback check with IsMinLZ.
The dst and src must not overlap. It is valid to pass a nil dst.
func DecodedLen ¶
DecodedLen returns the length of the decoded block. This length will never be exceeded when decoding a block.
func Encode ¶
Encode returns the encoded form of src. The returned slice may be a sub- slice of dst if dst was large enough to hold the entire encoded block. Otherwise, a newly allocated slice will be returned.
The dst and src must not overlap. It is valid to pass a nil dst.
The blocks will require the same amount of memory to decode as encoding, and does not make for concurrent decoding. Also note that blocks do not contain CRC information, so corruption may be undetected.
If you need to encode larger amounts of data, consider using the streaming interface which gives all of these features.
func ExtractSidecar ¶ added in v1.2.0
ExtractSidecar reads a MinLZ stream from src that already contains search tables (0x44/0x45/0x46) and writes a sidecar to sidecarDst. The tables are preserved verbatim (no rebuilding).
If newStreamDst is non-nil, src is also re-emitted to it WITHOUT the search-table chunks; sidecar offsets reference the new layout. The seek index (chunk 0x40) in src is dropped and a fresh one is appended to newStreamDst after the EOF chunk (mirroring Writer.closeIndex).
If newStreamDst is nil, sidecar offsets reference src's original layout and src must be the stream used for subsequent searches.
Example ¶
package main
import (
"bytes"
"fmt"
"io"
"github.com/minio/minlz"
)
func main() {
// 1) Compress WITH inline search tables.
var orig bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(4)
w := minlz.NewWriter(&orig,
minlz.WriterBlockSize(4096),
minlz.WriterConcurrency(1),
minlz.WriterSearchTable(cfg),
)
data := make([]byte, 8192)
copy(data[:4096], bytes.Repeat([]byte("aaaa"), 1024))
copy(data[4096:], bytes.Repeat([]byte("bbbb"), 1024))
copy(data[4096+200:], []byte("NEEDLE_PATTERN"))
w.Write(data)
w.Close()
// 2) Pull the existing search-index chunks out into a sidecar and also
// write a stripped main containing only the data chunks. The seek
// index in the stripped stream is rebuilt automatically.
var side, stripped bytes.Buffer
if err := minlz.ExtractSidecar(&side, &stripped, bytes.NewReader(orig.Bytes())); err != nil {
fmt.Println("extract:", err)
return
}
// The stripped main is smaller and still round-trips to the same data.
fmt.Printf("stripped smaller than orig: %v\n", stripped.Len() < orig.Len())
got, _ := io.ReadAll(minlz.NewReader(bytes.NewReader(stripped.Bytes())))
fmt.Printf("round-trip: %v\n", bytes.Equal(got, data))
// Searching via sidecar against the stripped main returns the same match.
ss := minlz.NewSidecarSearcher(bytes.NewReader(stripped.Bytes()), bytes.NewReader(side.Bytes()))
ss.Search([]byte("NEEDLE_PATTERN"), func(r minlz.SearchResult) error {
fmt.Printf("found at stream offset %d\n", r.StreamOffset)
return nil
})
}
Output: stripped smaller than orig: true round-trip: true found at stream offset 4296
func IndexStream ¶
IndexStream will return an index for a stream. The stream structure will be checked, but data within blocks is not verified. The returned index can either be appended to the end of the stream or stored separately.
Example ¶
ExampleIndexStream shows an example of indexing a stream and indexing it after it has been written. The index can either be appended.
package main
import (
"bytes"
"fmt"
"io"
"math/rand"
"os"
"github.com/minio/minlz"
)
func main() {
fatalErr := func(err error) {
if err != nil {
fmt.Println("ERR:", err)
}
}
// Create a test stream without index
streamName := ""
tmp := make([]byte, 5<<20)
{
rng := rand.New(rand.NewSource(0xbeefcafe))
rng.Read(tmp)
// Make it compressible...
for i, v := range tmp {
tmp[i] = '0' + v&3
}
// Compress it...
output, err := os.CreateTemp("", "IndexStream")
streamName = output.Name()
fatalErr(err)
// We use smaller blocks just for the example...
enc := minlz.NewWriter(output, minlz.WriterBlockSize(64<<10))
err = enc.EncodeBuffer(tmp)
fatalErr(err)
// Close and get index...
err = enc.Close()
fatalErr(err)
err = output.Close()
fatalErr(err)
}
// Open our compressed stream without an index...
stream, err := os.Open(streamName)
fatalErr(err)
defer stream.Close()
indexInput := io.Reader(stream)
var indexOutput io.Writer
var indexedName string
// Should index be combined with stream by appending?
// This could also be done by appending to an os.File
// If not it will be written to a separate file.
const combineOutput = false
// Function to easier use defer.
func() {
if combineOutput {
output, err := os.CreateTemp("", "IndexStream-Combined")
fatalErr(err)
defer func() {
fatalErr(output.Close())
if false {
fi, err := os.Stat(output.Name())
fatalErr(err)
fmt.Println("Combined:", fi.Size(), "bytes")
} else {
fmt.Println("Index saved")
}
}()
// Everything read from stream will also be written to output.
indexedName = output.Name()
indexInput = io.TeeReader(stream, output)
indexOutput = output
} else {
output, err := os.CreateTemp("", "IndexStream-Index")
fatalErr(err)
defer func() {
fatalErr(output.Close())
fi, err := os.Stat(output.Name())
fatalErr(err)
if false {
fmt.Println("Index:", fi.Size(), "bytes")
} else {
fmt.Println("Index saved")
}
}()
indexedName = output.Name()
indexOutput = output
}
// Index the input
idx, err := minlz.IndexStream(indexInput)
fatalErr(err)
// Write the index
_, err = indexOutput.Write(idx)
fatalErr(err)
}()
if combineOutput {
// Read from combined stream only.
stream, err := os.Open(indexedName)
fatalErr(err)
defer stream.Close()
// Create a reader with the input.
// We assert that the stream is an io.ReadSeeker.
r := minlz.NewReader(io.ReadSeeker(stream))
// Request a ReadSeeker with random access.
// This will load the index from the stream.
rs, err := r.ReadSeeker(nil)
fatalErr(err)
_, err = rs.Seek(-10, io.SeekEnd)
fatalErr(err)
b, err := io.ReadAll(rs)
fatalErr(err)
if want := tmp[len(tmp)-10:]; !bytes.Equal(b, want) {
fatalErr(fmt.Errorf("wanted %v, got %v", want, b))
}
fmt.Println("last 10 bytes read")
_, err = rs.Seek(10, io.SeekStart)
fatalErr(err)
_, err = io.ReadFull(rs, b)
fatalErr(err)
if want := tmp[10:20]; !bytes.Equal(b, want) {
fatalErr(fmt.Errorf("wanted %v, got %v", want, b))
}
fmt.Println("10 bytes at offset 10 read")
} else {
// Read from separate stream and index.
stream, err := os.Open(streamName)
fatalErr(err)
defer stream.Close()
// Create a reader with the input.
// We assert that the stream is an io.ReadSeeker.
r := minlz.NewReader(io.ReadSeeker(stream))
// Read the separate index.
index, err := os.ReadFile(indexedName)
fatalErr(err)
// Request a ReadSeeker with random access.
// The provided index will be used.
rs, err := r.ReadSeeker(index)
fatalErr(err)
_, err = rs.Seek(-10, io.SeekEnd)
fatalErr(err)
b, err := io.ReadAll(rs)
fatalErr(err)
if want := tmp[len(tmp)-10:]; !bytes.Equal(b, want) {
fatalErr(fmt.Errorf("wanted %v, got %v", want, b))
}
fmt.Println("last 10 bytes read")
_, err = rs.Seek(10, io.SeekStart)
fatalErr(err)
_, err = io.ReadFull(rs, b)
fatalErr(err)
if want := tmp[10:20]; !bytes.Equal(b, want) {
fatalErr(fmt.Errorf("wanted %v, got %v", want, b))
}
fmt.Println("10 bytes at offset 10 read")
}
}
Output: Index saved last 10 bytes read 10 bytes at offset 10 read
func IsMinLZ ¶
IsMinLZ returns whether the block is a minlz block and returns the size of the decompressed block.
func MaxEncodedLen ¶
MaxEncodedLen returns the maximum length of a snappy block, given its uncompressed length.
It will return a negative value if srcLen is too large to encode.
func RemoveIndexHeaders ¶
RemoveIndexHeaders will trim all headers and trailers from a given index. This is expected to save 20 bytes. These can be restored using RestoreIndexHeaders. This removes a layer of security, but is the most compact representation. Returns nil if headers contains errors. The returned slice references the provided slice.
func RestoreIndexHeaders ¶
RestoreIndexHeaders will index restore headers removed by RemoveIndexHeaders. No error checking is performed on the input. If a 0 length slice is sent, it is returned without modification.
Types ¶
type BlockSearchOption ¶ added in v1.2.0
type BlockSearchOption func(*BlockSearcher) error
BlockSearchOption configures a BlockSearcher.
func BlockSearchBailOnMissing ¶ added in v1.2.0
func BlockSearchBailOnMissing() BlockSearchOption
BlockSearchBailOnMissing makes Search return ErrSearchTablesUnusable if search tables are not available or not compatible with the pattern.
func BlockSearchCollectStats ¶ added in v1.2.0
func BlockSearchCollectStats() BlockSearchOption
BlockSearchCollectStats enables collection of search statistics. When enabled, Stats() returns detailed metrics after Search completes.
func BlockSearchIgnoreCRC ¶ added in v1.2.0
func BlockSearchIgnoreCRC() BlockSearchOption
BlockSearchIgnoreCRC skips CRC validation during search.
func BlockSearchInfoCallback ¶ added in v1.2.0
func BlockSearchInfoCallback(fn func(SearchTableConfig)) BlockSearchOption
BlockSearchInfoCallback installs fn to be invoked when a Search Table Info chunk (0x44) is parsed from the stream. The callback receives the parsed SearchTableConfig and is intended for logging / introspection. The callback must not retain references to mutable state inside the config.
func BlockSearchMaxBlockSize ¶ added in v1.2.0
func BlockSearchMaxBlockSize(n int) BlockSearchOption
BlockSearchMaxBlockSize limits the maximum block size the searcher will decode.
type BlockSearcher ¶ added in v1.2.0
type BlockSearcher struct {
// contains filtered or unexported fields
}
BlockSearcher reads a MinLZ stream and searches for patterns using per-block search tables to skip blocks that definitely don't contain the pattern.
Example ¶
package main
import (
"bytes"
"fmt"
"github.com/minio/minlz"
)
func main() {
// Create test data: two blocks, needle only in the second.
data := make([]byte, 8192)
copy(data[:4096], bytes.Repeat([]byte("aaaa"), 1024))
copy(data[4096:], bytes.Repeat([]byte("bbbb"), 1024))
copy(data[4096+100:], []byte("NEEDLE_PATTERN"))
var buf bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(4)
w := minlz.NewWriter(&buf,
minlz.WriterSearchTable(cfg),
minlz.WriterBlockSize(4096),
minlz.WriterConcurrency(1),
)
w.Write(data)
w.Close()
// Search for the needle.
searcher := minlz.NewBlockSearcher(bytes.NewReader(buf.Bytes()), minlz.BlockSearchCollectStats())
err := searcher.Search([]byte("NEEDLE_PATTERN"), func(r minlz.SearchResult) error {
fmt.Printf("found at stream offset %d\n", r.StreamOffset)
return nil
})
if err != nil {
fmt.Println("error:", err)
}
stats := searcher.Stats()
fmt.Printf("blocks: %d total, %d skipped\n", stats.BlocksTotal, stats.BlocksSkipped)
}
Output: found at stream offset 4196 blocks: 2 total, 1 skipped
Example (Bail) ¶
package main
import (
"bytes"
"fmt"
"github.com/minio/minlz"
)
func main() {
// Compress WITHOUT search tables.
var buf bytes.Buffer
w := minlz.NewWriter(&buf, minlz.WriterBlockSize(4096), minlz.WriterConcurrency(1))
w.Write(bytes.Repeat([]byte("test data "), 1000))
w.Close()
searcher := minlz.NewBlockSearcher(
bytes.NewReader(buf.Bytes()),
minlz.BlockSearchBailOnMissing(),
)
err := searcher.Search([]byte("pattern"), func(r minlz.SearchResult) error {
return nil
})
fmt.Println(err)
}
Output: minlz: search tables cannot be used for this pattern
func NewBlockSearcher ¶ added in v1.2.0
func NewBlockSearcher(r io.Reader, opts ...BlockSearchOption) *BlockSearcher
NewBlockSearcher creates a BlockSearcher for the given reader.
func (*BlockSearcher) Search ¶ added in v1.2.0
func (s *BlockSearcher) Search(pattern []byte, fn func(r SearchResult) error) error
Search iterates blocks in the stream, calling fn for each pattern match.
The callback receives a SearchResult with the match offset and surrounding block data. Return nil to continue, ErrSearchForward to request the next block for forward context (re-calls fn with shifted Blocks), or any other error to abort.
func (*BlockSearcher) Stats ¶ added in v1.2.0
func (s *BlockSearcher) Stats() SearchStats
Stats returns search statistics accumulated during the last Search call.
type CompressedSearchOption ¶ added in v1.2.0
type CompressedSearchOption func(*compressedOpts)
CompressedSearchOption configures the compressed search-table encoder.
func CompressedSearchForce ¶ added in v1.2.0
func CompressedSearchForce() CompressedSearchOption
CompressedSearchForce forces the encoder to emit the compressed chunk type (0x46) even when it is larger than the uncompressed (0x45) alternative. Intended for benchmarking only.
func CompressedSearchSkipPct ¶ added in v1.2.0
func CompressedSearchSkipPct(pct float64) CompressedSearchOption
CompressedSearchSkipPct sets the popcount-band half-width (in percent) below which compression is skipped. The default is 5.0.
func CompressedSearchStatsHook ¶ added in v1.2.0
func CompressedSearchStatsHook(fn func(CompressedSearchStats)) CompressedSearchOption
CompressedSearchStatsHook installs a callback receiving stats for every search-table bitmap processed by the encoder. The callback must not retain references into the supplied struct.
type CompressedSearchStats ¶ added in v1.2.0
type CompressedSearchStats struct {
BitmapBytes int
Reductions uint8 // reductions applied to the bitmap before encoding
SubBlockSize int // 1<<h0_bs, or 0 if SkippedBand or below minimum size
SubBlocks int
SetBits int
TotalBits int
SkippedBand bool // popcount band rejected compression
SkippedSize bool // bitmap below cstMinBitmapForCompression
Chunk0x45Size int // alternative uncompressed-form size
Chunk0x46Size int // produced compressed-form size; 0 if not emitted
Emitted0x46 bool
Tables int // distinct tables embedded
// Per-disposition sub-block counts (sum to SubBlocks when emitted).
BlocksOwnTable int
BlocksGlobalTable int
BlocksRaw int
BlocksRLE int
BlocksSparse int
// Per-disposition on-wire payload bytes (excludes disposition byte;
// includes the uvarint length for table-using blocks).
BytesOwnPayload int
BytesGlobalPayload int
BytesRawPayload int
BytesRLEPayload int
BytesSparsePayload int
// Bytes consumed by serialized table headers (sums over the tables that
// were actually embedded in the chunk).
BytesOwnTables int
BytesGlobalTable int
}
CompressedSearchStats reports per-bitmap stats produced by the compressed search-table encoder. It is delivered via CompressedSearchStatsHook.
type ErrCantSeek ¶
type ErrCantSeek struct {
Reason string
}
ErrCantSeek is returned if the stream cannot be seeked.
type Index ¶
type Index struct {
// Total Uncompressed size.
TotalUncompressed int64
// Total Compressed size if known. Will be -1 if unknown.
TotalCompressed int64
// Offset pairs are pairs of Compressed -> Uncompressed positions.
// Offsets are stream offsets from first stream byte.
// It will be safe to start decompressing from any of these offsets.
// The slice is sorted by offset.
Offsets []OffsetPair
// contains filtered or unexported fields
}
Index represents an S2/Snappy/MinLZ index.
func (*Index) Find ¶
Find the offset at or before the wanted (uncompressed) offset. If offset is 0 or positive it is the offset from the beginning of the file. If the uncompressed size is known, the offset must be within the file. If an offset outside the file is requested io.ErrUnexpectedEOF is returned. If the offset is negative, it is interpreted as the distance from the end of the file, where -1 represents the last byte. If offset from the end of the file is requested, but size is unknown, ErrUnsupported will be returned.
func (*Index) Load ¶
Load a binary index. A zero value Index can be used or a previous one can be reused.
Example ¶
package main
import (
"bytes"
"fmt"
"io"
"math/rand"
"sync"
"github.com/minio/minlz"
)
func main() {
fatalErr := func(err error) {
if err != nil {
panic(err)
}
}
// Create a test corpus
tmp := make([]byte, 5<<20)
rng := rand.New(rand.NewSource(0xbeefcafe))
rng.Read(tmp)
// Make it compressible...
for i, v := range tmp {
tmp[i] = '0' + v&3
}
// Compress it...
var buf bytes.Buffer
// We use smaller blocks just for the example...
enc := minlz.NewWriter(&buf, minlz.WriterBlockSize(100<<10))
err := enc.EncodeBuffer(tmp)
fatalErr(err)
// Close and get index...
idxBytes, err := enc.CloseIndex()
fatalErr(err)
// This is our compressed stream...
compressed := buf.Bytes()
var once sync.Once
for wantOffset := int64(0); wantOffset < int64(len(tmp)); wantOffset += 555555 {
// Let's assume we want to read from uncompressed offset 'i'
// and we cannot seek in input, but we have the index.
want := tmp[wantOffset:]
// Load the index.
var index minlz.Index
_, err = index.Load(idxBytes)
fatalErr(err)
// Find offset in file:
compressedOffset, uncompressedOffset, err := index.Find(wantOffset)
fatalErr(err)
// Offset the input to the compressed offset.
// Notice how we do not provide any bytes before the offset.
input := io.Reader(bytes.NewBuffer(compressed[compressedOffset:]))
if _, ok := input.(io.Seeker); !ok {
// Notice how the input cannot be seeked...
once.Do(func() {
fmt.Println("Input does not support seeking...")
})
} else {
panic("did you implement seeking on bytes.Buffer?")
}
// When creating the decoder we must specify that it should not
// expect a stream identifier at the beginning og the frame.
dec := minlz.NewReader(input, minlz.ReaderIgnoreStreamIdentifier())
// We now have a reader, but it will start outputting at uncompressedOffset,
// and not the actual offset we want, so skip forward to that.
toSkip := wantOffset - uncompressedOffset
err = dec.Skip(toSkip)
fatalErr(err)
// Read the rest of the stream...
got, err := io.ReadAll(dec)
fatalErr(err)
if bytes.Equal(got, want) {
fmt.Println("Successfully skipped forward to", wantOffset)
} else {
fmt.Println("Failed to skip forward to", wantOffset)
}
}
}
Output: Input does not support seeking... Successfully skipped forward to 0 Successfully skipped forward to 555555 Successfully skipped forward to 1111110 Successfully skipped forward to 1666665 Successfully skipped forward to 2222220 Successfully skipped forward to 2777775 Successfully skipped forward to 3333330 Successfully skipped forward to 3888885 Successfully skipped forward to 4444440 Successfully skipped forward to 4999995
func (*Index) LoadStream ¶
func (i *Index) LoadStream(rs io.ReadSeeker) error
LoadStream will load an index from the end of the supplied stream. ErrUnsupported will be returned if the signature cannot be found. ErrCorrupt will be returned if unexpected values are found. io.ErrUnexpectedEOF is returned if there are too few bytes. IO errors are returned as-is.
type OffsetPair ¶
type ReadSeeker ¶
type ReadSeeker struct {
*Reader
// contains filtered or unexported fields
}
ReadSeeker provides random or forward seeking in compressed content. See Reader.ReadSeeker
func (*ReadSeeker) ReadAt ¶
func (r *ReadSeeker) ReadAt(p []byte, offset int64) (int, error)
ReadAt reads len(p) bytes into p starting at offset off in the underlying input source. It returns the number of bytes read (0 <= n <= len(p)) and any error encountered.
When ReadAt returns n < len(p), it returns a non-nil error explaining why more bytes were not returned. In this respect, ReadAt is stricter than Read.
Even if ReadAt returns n < len(p), it may use all of p as scratch space during the call. If some data is available but not len(p) bytes, ReadAt blocks until either all the data is available or an error occurs. In this respect ReadAt is different from Read.
If the n = len(p) bytes returned by ReadAt are at the end of the input source, ReadAt may return either err == EOF or err == nil.
If ReadAt is reading from an input source with a seek offset, ReadAt should not affect nor be affected by the underlying seek offset.
Clients of ReadAt can execute parallel ReadAt calls on the same input source. This is however not recommended.
type Reader ¶
type Reader struct {
// contains filtered or unexported fields
}
Reader is an io.Reader that can read Snappy-compressed bytes.
func NewReader ¶
func NewReader(r io.Reader, opts ...ReaderOption) *Reader
NewReader returns a new Reader that decompresses from r, using the framing format described at https://github.com/google/snappy/blob/master/framing_format.txt with S2 changes.
func (*Reader) DecodeConcurrent ¶
DecodeConcurrent will decode the full stream to w. This function should not be combined with reading, seeking or other operations. Up to 'concurrent' goroutines will be used. If <= 0, min(runtime.NumCPU, runtime.GOMAXPROCS, 8) will be used. On success the number of bytes decompressed nil and is returned. This is mainly intended for bigger streams, since it will cause more allocations.
func (*Reader) GetBufferCapacity ¶
GetBufferCapacity returns the capacity of the internal buffer. This might be useful to know when reusing the same reader in combination with the lazy buffer option.
func (*Reader) ReadSeeker ¶
func (r *Reader) ReadSeeker(index []byte) (rrs *ReadSeeker, err error)
ReadSeeker will return an io.ReadSeeker and io.ReaderAt compatible version of the reader. The original input must support the io.Seeker interface. A custom index can be specified which will be used if supplied. When using a custom index, it will not be read from the input stream. The ReadAt position will affect regular reads and the current position of Seek. So using Read after ReadAt will continue from where the ReadAt stopped. No functions should be used concurrently. The returned ReadSeeker contains a shallow reference to the existing Reader, meaning changes performed to one is reflected in the other.
func (*Reader) Reset ¶
Reset discards any buffered data, resets all state, and switches the Snappy reader to read from r. This permits reusing a Reader rather than allocating a new one.
func (*Reader) Skip ¶
Skip will skip n bytes forward in the decompressed output. For larger skips this consumes less CPU and is faster than reading output and discarding it. CRC is not checked on skipped blocks. io.ErrUnexpectedEOF is returned if the stream ends before all bytes have been skipped. If a decoding error is encountered subsequent calls to Read will also fail.
func (*Reader) UserChunkCB ¶
UserChunkCB will register a callback for chunks with the specified ID. ID must be a reserved user chunks ID, 0x80-0xfd (inclusive). For each chunk with the ID, the callback is called with the content. Any returned non-nil error will abort decompression. Only one callback per ID is supported, latest sent will be used. Sending a nil function will disable previous callbacks. You can peek the stream, triggering the callback, by doing a Read with a 0 byte buffer.
type ReaderOption ¶
ReaderOption is an option for creating a decoder.
func ReaderFallback ¶
func ReaderFallback(b bool) ReaderOption
ReaderFallback will enable/disable S2/Snappy fallback.
func ReaderIgnoreCRC ¶
func ReaderIgnoreCRC() ReaderOption
ReaderIgnoreCRC will make the reader skip CRC calculation and checks.
func ReaderIgnoreStreamIdentifier ¶
func ReaderIgnoreStreamIdentifier() ReaderOption
ReaderIgnoreStreamIdentifier will make the reader skip the expected stream identifier at the beginning of the stream. This can be used when serving a stream that has been forwarded to a specific point. Validation of EOF length is also disabled.
func ReaderMaxBlockSize ¶
func ReaderMaxBlockSize(blockSize int) ReaderOption
ReaderMaxBlockSize allows controlling allocations if the stream has been compressed with a smaller WriterBlockSize, or with the default 1MB. Blocks must be this size or smaller to decompress, otherwise the decoder will return ErrUnsupported.
For streams compressed with Snappy this can safely be set to 64KB (64 << 10).
Default is the maximum limit of 8MB.
func ReaderUserChunkCB ¶
func ReaderUserChunkCB(id uint8, fn func(r io.Reader) error) ReaderOption
ReaderUserChunkCB will register a callback for chunks with the specified ID. ID must be a Reserved skippable chunks ID, 0x40-0xfd (inclusive). For each chunk with the ID, the callback is called with the content. Any returned non-nil error will abort decompression. Only one callback per ID is supported, latest sent will be used. Sending a nil function will disable previous callbacks. You can peek the stream, triggering the callback, by doing a Read with a 0 byte buffer.
type SearchResult ¶ added in v1.2.0
type SearchResult struct {
// Blocks contains the block data surrounding the match.
// Blocks[0] is the previous block (nil if previous was skipped or lazy).
// Blocks[1] is the current block. The match may start in Blocks[0] (boundary match).
// Both slices are invalid after the callback returns.
Blocks [2][]byte
// Offset is the position of the match relative to PrevBlock()+Blocks[1].
// When PrevBlock() returns nil, Offset is relative to just Blocks[1].
Offset int
// StreamOffset is the absolute uncompressed stream offset of the match.
StreamOffset int64
// BlockStart is the stream offset of PrevBlock() data (or Blocks[1] if no prev).
// Invariant: Offset == int(StreamOffset - BlockStart)
BlockStart int64
// PrevBlockLen is the decompressed size of the previous block.
// This equals len(PrevBlock()) but avoids forcing a lazy decode.
PrevBlockLen int
// contains filtered or unexported fields
}
SearchResult is passed to the search callback for each pattern match.
func (SearchResult) PrevBlock ¶ added in v1.2.0
func (r SearchResult) PrevBlock() []byte
PrevBlock returns the previous block's decompressed data. Returns Blocks[0] if non-nil, otherwise lazily decompresses the previous block if available. Returns nil if no previous block exists.
type SearchStats ¶ added in v1.2.0
type SearchStats struct {
BlocksTotal int // Total data blocks encountered
BlocksSkipped int // Blocks skipped via search table (definitely no match)
BlocksSearched int // Blocks decoded and passed to callback
CompBytesSkipped int64 // Compressed bytes skipped
UncompBytesSearched int64 // Uncompressed bytes decoded and searched
TablesPresent int // Number of 0x45 + 0x46 search table chunks seen
TablesBytes int64 // Total wire bytes of search table chunks (0x45 + 0x46)
TablesMissing int // Data blocks without a preceding search table
TablesUnusable int // Tables present but incompatible with query
TableBitsSum int // Sum of effective table bits (for avg: TableBitsSum/TablesPresent)
TableReductionsSum int // Sum of reductions applied (for avg: TableReductionsSum/TablesPresent)
BlocksDeferred int // Blocks that entered the deferred path
BlocksDeferredSkipped int // Deferred blocks ultimately skipped (subset of BlocksSkipped)
BlocksFalsePositive int // Blocks decoded due to table match but containing no actual matches
BlocksBoundaryScanned int // Blocks the table proved free of the pattern but decoded anyway, to rule out a match straddling in from the previous block (subset of BlocksSearched). A high count means boundary checks are forcing scans that the tables alone would have skipped.
TablePopMin float64 // Min population % across tables seen
TablePopMax float64 // Max population % across tables seen
TablePopSum float64 // Sum of population % (for computing average)
UncompressedSize int64 // Total uncompressed bytes in stream
// Compressed search-table (chunk type 0x46) breakdown. These count a
// subset of TablesPresent / TablesBytes.
TablesCompressed int // Number of 0x46 chunks seen
TablesCompressedBytes int64 // Total wire bytes of 0x46 chunks (subset of TablesBytes)
TableBitmapBytes int64 // Total uncompressed bitmap bytes across 0x46 chunks
// Compressed sub-block breakdown for the 0x46 chunks above.
CompressedBlocksTotal int // Sum of compressed sub-blocks across all 0x46 chunks
CompressedBlocksRaw int // Sub-blocks with disposition = raw
CompressedBlocksRLE int // Sub-blocks with disposition = RLE
CompressedBlocksSparse int // Sub-blocks with disposition = sparse bit table
CompressedTablesSum int // Sum of distinct tables emitted across all 0x46 chunks
// Wire-byte breakdown of 0x46 sub-block payloads (excludes the disposition
// byte itself; for tabled blocks includes the uvarint length prefix and
// compressed data, but NOT the table header — that's tracked separately).
CompressedBytesTabled int64
CompressedBytesRaw int64
CompressedBytesRLE int64
CompressedBytesSparse int64
CompressedBytesTableHeaders int64 // bytes consumed by serialized tables
// Windows holds per-pattern-window table-presence counts, populated when
// stats collection is enabled. Each entry is one matchLen-window the
// searcher tests against every block's table; Present/Absent count the
// tables whose bitmap had that window's hash set/clear. Rendered by
// FprintExtended — useful for seeing how selective the chosen matchLen and
// prefix are for a given pattern (the first prefix window is the skip gate).
Windows []WindowStat
}
SearchStats contains statistics from a BlockSearcher.Search call.
func (SearchStats) Fprint ¶ added in v1.2.0
func (st SearchStats) Fprint(w io.Writer)
Fprint writes a human-readable summary of the search stats to w.
Example ¶
package main
import (
"bytes"
"fmt"
"strings"
"github.com/minio/minlz"
)
func main() {
// Create and search a stream to show stats output.
data := make([]byte, 16384)
copy(data[:8192], bytes.Repeat([]byte("aaaa"), 2048))
copy(data[8192:], bytes.Repeat([]byte("bbbb"), 2048))
copy(data[8192+100:], []byte("FINDME"))
var buf bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(4)
w := minlz.NewWriter(&buf,
minlz.WriterSearchTable(cfg),
minlz.WriterBlockSize(8192),
minlz.WriterConcurrency(1),
)
w.Write(data)
w.Close()
searcher := minlz.NewBlockSearcher(bytes.NewReader(buf.Bytes()), minlz.BlockSearchCollectStats())
searcher.Search([]byte("FINDME"), func(r minlz.SearchResult) error { return nil })
var out strings.Builder
searcher.Stats().Fprint(&out)
s := strings.Replace(out.String(), fmt.Sprintf("skipped: %d compressed", searcher.Stats().CompBytesSkipped), "skipped: <N> compressed", 1)
fmt.Print(s)
}
Output: Blocks total: 2, skipped: 1 (50.0%), deferred: 0 (0.0%, 0 skipped) Blocks searched: 1 (50.0%), false positive: 0 (0.0%) Bytes skipped: <N> compressed, searched: 8192 uncompressed Tables: 2 present, 0 missing, 0 unusable Table bits/byte: 0.0225, log2: 8.0, avg reductions: 5.0 Table total: 46 bytes, avg 23 bytes/table, 0.28% of 16384 uncompressed Table population: avg 2.7%, min 1.6%, max 3.9% Table types: 0 uncompressed (0x45), 2 compressed (0x46) Compressed tables: 2 (100.0% of total), 46 wire bytes, 64 uncompressed bitmap bytes (71.88% ratio) Sub-blocks: 2 total (0 tabled, 0 raw, 0 RLE, 2 sparse); 0 tables emitted (share=0.00 tables/tabled-block) Payload bytes: tabled=0 raw=0 RLE=0 sparse=16; table-header bytes=0
func (SearchStats) FprintExtended ¶ added in v1.2.0
func (st SearchStats) FprintExtended(w io.Writer)
FprintExtended writes the standard Fprint summary followed by a per-window breakdown of how often each of the pattern's matchLen-windows was present in the per-block tables. The first prefix window is the skip "gate"; the raw fallback (prefix < 0) anchors a match whose prefix byte sits in the previous block. Requires stats collection to have populated Windows.
type SearchTableConfig ¶ added in v1.2.0
type SearchTableConfig struct {
// contains filtered or unexported fields
}
SearchTableConfig configures search table generation for a Writer. Use NewSearchTableConfig to create, then chain With* methods to customize.
func NewSearchTableConfig ¶ added in v1.2.0
func NewSearchTableConfig() SearchTableConfig
NewSearchTableConfig creates a search table config. Defaults: matchLen=6, no prefix (type 1), auto table size, 70% max population, 25% max reduced population (auto-tightened to 10% when a prefix is configured), compression on.
func (SearchTableConfig) String ¶ added in v1.2.0
func (c SearchTableConfig) String() string
String returns a one-line, human-readable summary of the search table configuration. Useful for logging the contents of an 0x44 chunk.
func (SearchTableConfig) WithBytePrefix ¶ added in v1.2.0
func (c SearchTableConfig) WithBytePrefix(prefixes ...byte) SearchTableConfig
WithBytePrefix sets prefix byte values. With 1-8 unique values, table type 2 is used. With more than 8, it automatically switches to a bitmask (type 3).
func (SearchTableConfig) WithCompression ¶ added in v1.2.0
func (c SearchTableConfig) WithCompression(opts ...CompressedSearchOption) SearchTableConfig
WithCompression enables huff0 compression of per-block search tables and optionally tunes it. Compression is on by default — call this only to pass non-default options.
Search indexes that can only be marginally compressed are stored uncompressed.
With no options, defaults are: 10.0% popcount band, no stats hook, non-forced (emit compressed only when smaller).
func (SearchTableConfig) WithExtras ¶ added in v1.2.0
func (c SearchTableConfig) WithExtras(n int) SearchTableConfig
WithExtras sets the number of extra hashes emitted after each long-prefix occurrence. With extras=E, table type 4 writes E+1 consecutive overlapping matchLen-byte windows per indexed position; the searcher checks the same E+1 windows per pattern occurrence. matchLen+extras must not exceed 16.
Extras only applies to type 4 (long prefix). Setting extras > 0 on any other table type produces a validation error.
func (SearchTableConfig) WithLongPrefix ¶ added in v1.2.0
func (c SearchTableConfig) WithLongPrefix(prefix []byte) SearchTableConfig
WithLongPrefix sets a long prefix (1-256 bytes, table type 4).
func (SearchTableConfig) WithMaskPrefix ¶ added in v1.2.0
func (c SearchTableConfig) WithMaskPrefix(mask [32]byte) SearchTableConfig
WithMaskPrefix sets a 256-bit bitmask of prefix bytes (table type 3).
func (SearchTableConfig) WithMatchLen ¶ added in v1.2.0
func (c SearchTableConfig) WithMatchLen(n int) SearchTableConfig
WithMatchLen sets the match length (1-8). Shorter values use less of the search pattern but are more likely to collide. Default is 6.
func (SearchTableConfig) WithMaxPopulation ¶ added in v1.2.0
func (c SearchTableConfig) WithMaxPopulation(pct int) SearchTableConfig
WithMaxPopulation sets the max population percentage (0-100). Tables with more bits set are skipped entirely.
func (SearchTableConfig) WithMaxReducedPopulation ¶ added in v1.2.0
func (c SearchTableConfig) WithMaxReducedPopulation(pct int) SearchTableConfig
WithMaxReducedPopulation sets the max population percentage (0-100) for the reduced table. Reductions stop before exceeding this threshold.
Default is 25%, automatically tightened to 10% when a prefix is configured (byte/mask/long). Calling this method disables that auto-tightening — the supplied value is used regardless of prefix.
func (SearchTableConfig) WithoutCompression ¶ added in v1.2.0
func (c SearchTableConfig) WithoutCompression() SearchTableConfig
WithoutCompression disables the per-block search-table compression that is on by default. The stream will emit only 0x45 chunks; readers that don't implement 0x46 can read it.
type SidecarOption ¶ added in v1.2.0
type SidecarOption func(*sidecarOpts) error
SidecarOption configures sidecar generation.
func SidecarIgnoreCRC ¶ added in v1.2.0
func SidecarIgnoreCRC() SidecarOption
SidecarIgnoreCRC disables CRC validation when reading the source stream.
func SidecarSearchTable ¶ added in v1.2.0
func SidecarSearchTable(cfg SearchTableConfig) SidecarOption
SidecarSearchTable adds a search-table config. May be called multiple times to embed multiple configs in one sidecar. Each config produces its own 0x44 info chunk and per-block 0x45/0x46 chunks. The searcher tries every usable config and ANDs the skip decisions.
type SidecarSearcher ¶ added in v1.2.0
type SidecarSearcher struct {
// contains filtered or unexported fields
}
SidecarSearcher searches a main MinLZ data stream (via io.ReaderAt) using search indexes from a sidecar stream (via io.Reader). Adjacent must-read blocks are coalesced into a single ReadAt call to minimise I/O.
Example ¶
package main
import (
"bytes"
"fmt"
"github.com/minio/minlz"
)
func main() {
// Build a small stream + sidecar in one go via WriterSidecar.
var main, side bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(4)
w := minlz.NewWriter(&main,
minlz.WriterBlockSize(4096),
minlz.WriterConcurrency(1),
minlz.WriterSearchTable(cfg),
minlz.WriterSidecar(&side),
)
data := make([]byte, 8192)
copy(data[:4096], bytes.Repeat([]byte("aaaa"), 1024))
copy(data[4096:], bytes.Repeat([]byte("bbbb"), 1024))
copy(data[4096+200:], []byte("NEEDLE_PATTERN"))
w.Write(data)
w.Close()
// The main reader must support io.ReaderAt — a *bytes.Reader or *os.File
// works. Each NewSidecarSearcher reads its own sidecar from start to end.
ss := minlz.NewSidecarSearcher(
bytes.NewReader(main.Bytes()),
bytes.NewReader(side.Bytes()),
minlz.BlockSearchCollectStats(),
)
ss.Search([]byte("NEEDLE_PATTERN"), func(r minlz.SearchResult) error {
fmt.Printf("found at stream offset %d\n", r.StreamOffset)
return nil
})
stats := ss.Stats()
fmt.Printf("blocks: %d total, %d skipped\n", stats.BlocksTotal, stats.BlocksSkipped)
}
Output: found at stream offset 4296 blocks: 2 total, 1 skipped
func NewSidecarSearcher ¶ added in v1.2.0
func NewSidecarSearcher(main io.ReaderAt, sidecar io.Reader, opts ...BlockSearchOption) *SidecarSearcher
NewSidecarSearcher creates a searcher reading the sidecar sequentially from sidecar and the main data stream via main. main must support concurrent ReadAt calls (per io.ReaderAt's contract).
func (*SidecarSearcher) Search ¶ added in v1.2.0
func (s *SidecarSearcher) Search(pattern []byte, fn func(SearchResult) error) error
Search iterates blocks referenced by the sidecar, decoding only those whose search tables do not prove the pattern absent, and calls fn for each pattern occurrence.
func (*SidecarSearcher) Stats ¶ added in v1.2.0
func (s *SidecarSearcher) Stats() SearchStats
Stats returns search statistics accumulated during the last Search call.
type WindowStat ¶ added in v1.2.0
type WindowStat struct {
Pos int // start index of the window within the pattern
Prefix int // pattern index of the preceding prefix byte; -1 = raw / no-prefix anchor
PrefixByte byte // the prefix byte (0 when Prefix < 0)
Bytes []byte // the matchLen bytes that get hashed
Present int // tables with the bit set
Absent int // tables with the bit clear
}
WindowStat reports, across every per-block search table, how often one pattern window's hash bit was set ("present", block might contain it) or clear ("absent", block definitely lacks it).
type Writer ¶
type Writer struct {
// contains filtered or unexported fields
}
Writer is an io.Writer that can write Snappy-compressed bytes.
func NewWriter ¶
func NewWriter(w io.Writer, opts ...WriterOption) *Writer
NewWriter returns a new Writer that compresses a MinLZ stream to w.
Users must call Close to guarantee all data has been forwarded to the underlying io.Writer and that resources are released.
func (*Writer) AddUserChunk ¶
AddUserChunk will add a (non)skippable chunk to the stream. The ID must be in the range 0x80 -> 0xfe - inclusive. The length of the block must be <= MaxUserChunkSize bytes.
Example ¶
var buf bytes.Buffer
w := NewWriter(&buf)
// Add a skippable chunk
if err := w.AddUserChunk(MinUserSkippableChunk, []byte("Chunk Custom Data")); err != nil {
log.Fatalf("w.AddUserChunk: %v", err)
}
//
if _, err := w.Write([]byte("some data")); err != nil {
log.Fatalf("w.Write: %v", err)
}
w.Close()
// Read back what we wrote.
r := NewReader(&buf)
err := r.UserChunkCB(MinUserSkippableChunk, func(sr io.Reader) error {
var err error
b, err := io.ReadAll(sr)
fmt.Println("Callback:", string(b), err)
return err
})
if err != nil {
log.Fatal(err)
}
// Read stream data
b, err := io.ReadAll(r)
if err != nil {
log.Fatal(err)
}
fmt.Println("Stream data:", string(b))
Output: Callback: Chunk Custom Data <nil> Stream data: some data
func (*Writer) AsyncFlush ¶
AsyncFlush writes any buffered bytes to a block and starts compressing it. It does not wait for the output has been written as Flush() does.
func (*Writer) Close ¶
Close calls Flush and then closes the Writer. This is required to mark the end of the stream. Calling Close multiple times is ok, but calling CloseIndex after this will make it not return the index.
func (*Writer) CloseIndex ¶
CloseIndex calls Close and returns an index on first call. This is not required if you are only adding index to a stream.
func (*Writer) EncodeBuffer ¶
EncodeBuffer will add a buffer to the stream. This is the fastest way to encode a stream, but the input buffer cannot be written to by the caller until Flush or Close has been called when concurrency != 1.
If you cannot control that, use the regular Write function.
Note that input is not buffered. This means that each write will result in discrete blocks being created. For buffered writes, use the regular Write function.
func (*Writer) Flush ¶
Flush flushes the Writer to its underlying io.Writer. This does not apply padding.
When search tables are enabled, the block flushed here is written without a search table (its following bytes aren't available yet to index boundary windows); it remains fully searchable but is always scanned, not skipped.
func (*Writer) ReadFrom ¶
ReadFrom implements the io.ReaderFrom interface. Using this is typically more efficient since it avoids a memory copy. ReadFrom reads data from r until EOF or error. The return value n is the number of bytes read. Any error except io.EOF encountered during the read is also returned.
func (*Writer) Reset ¶
Reset discards the writer's state and switches the Snappy writer to write to w. This permits reusing a Writer rather than allocating a new one.
func (*Writer) SetSidecar ¶ added in v1.2.0
SetSidecar configures or replaces the sidecar destination for subsequent blocks. Pass nil to disable sidecar mode. Must be called before any data is written for the current stream (i.e., immediately after NewWriter or after Reset, before the first Write/EncodeBuffer/ReadFrom). The Writer must have been constructed with WriterSearchTable.
This is primarily intended for tools that reuse a single Writer across many files and want to direct each file's index to a separate sidecar.
type WriterOption ¶
WriterOption is an option for creating a encoder.
func WriterAddIndex ¶
func WriterAddIndex(b bool) WriterOption
WriterAddIndex will append an index to the end of a stream when it is closed.
func WriterBlockSize ¶
func WriterBlockSize(n int) WriterOption
WriterBlockSize allows to override the default block size. Blocks will be this size or smaller. Minimum size is 4KB and the maximum size is 8MB.
Bigger blocks may give bigger throughput on systems with many cores, and will increase compression slightly, but it will limit the possible concurrency for smaller payloads for both encoding and decoding. Default block size is 2MB.
func WriterConcurrency ¶
func WriterConcurrency(n int) WriterOption
WriterConcurrency will set the concurrency, meaning the maximum number of decoders to run concurrently. The value supplied must be at least 1. By default this will be set to GOMAXPROCS.
func WriterCreateIndex ¶
func WriterCreateIndex(b bool) WriterOption
WriterCreateIndex allows to disable the default index creation. This can be used when no index will be needed - for example on network streams.
func WriterCustomEncoder ¶
func WriterCustomEncoder(fn func(dst, src []byte) int) WriterOption
WriterCustomEncoder allows to override the encoder for blocks on the stream. The function must compress 'src' into 'dst' and return the bytes used in dst as an integer. Block size (initial varint) should not be added by the encoder. Returning value 0 indicates the block could not be compressed. Returning a negative value indicates that compression should be attempted. The function should expect to be called concurrently.
func WriterFlushOnWrite ¶
func WriterFlushOnWrite() WriterOption
WriterFlushOnWrite will compress blocks on each call to the Write function.
This is quite inefficient as blocks size will depend on the write size.
Use WriterConcurrency(1) to also make sure that output is flushed. When Write calls return, otherwise they will be written when compression is done.
func WriterPadding ¶
func WriterPadding(n int) WriterOption
WriterPadding will add padding to all output, so the size will be a multiple of n. This can be used to obfuscate the exact output size or make blocks of a certain size. The contents will be a skippable frame, so it will be invisible by the decoder. n must be > 0 and <= 8MB. The padded area will be filled with data from crypto/rand.Reader. The padding will be applied whenever Close is called on the writer.
func WriterPaddingSrc ¶
func WriterPaddingSrc(reader io.Reader) WriterOption
WriterPaddingSrc will get random data for padding from the supplied source. By default, crypto/rand is used.
func WriterSearchTable ¶ added in v1.2.0
func WriterSearchTable(cfg SearchTableConfig) WriterOption
WriterSearchTable enables per-block search table generation. The config controls match length, prefix filtering, and heuristics. Use NewSearchTableConfig to create the config.
Example ¶
package main
import (
"bytes"
"fmt"
"io"
"github.com/minio/minlz"
)
func main() {
// Compress data with search tables enabled.
var buf bytes.Buffer
cfg := minlz.NewSearchTableConfig() // matchLen defaults to 6
w := minlz.NewWriter(&buf, minlz.WriterSearchTable(cfg))
w.Write([]byte("Hello World! This is searchable compressed data."))
w.Close()
// The stream is still readable by a normal Reader.
decoded, _ := io.ReadAll(minlz.NewReader(bytes.NewReader(buf.Bytes())))
fmt.Println("decoded:", string(decoded))
}
Output: decoded: Hello World! This is searchable compressed data.
Example (WithPrefix) ¶
package main
import (
"bytes"
"fmt"
"github.com/minio/minlz"
)
func main() {
// Prefix bytes dramatically reduce table size for structured data.
// Only positions after '"' and ':' are indexed, so the table is much sparser.
blockSize := 4096
data := make([]byte, blockSize*3)
// Fill with compressible JSON-like lines.
line := []byte(`{"name":"Alice","age":30,"id":"xyz"}` + "\n")
for i := 0; i < len(data); i += len(line) {
copy(data[i:], line)
}
// Place a unique value only in block 1.
copy(data[blockSize+100:], []byte(`{"name":"FindMe","id":"unique-9876"}`))
var buf bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithBytePrefix('"', ':')
w := minlz.NewWriter(&buf,
minlz.WriterSearchTable(cfg),
minlz.WriterBlockSize(blockSize),
minlz.WriterConcurrency(1),
)
w.Write(data)
w.Close()
// Search for a pattern that contains prefix bytes internally.
// The searcher finds '"' at position 2 and ':' at position 3 inside the pattern,
// and uses those to check the table — no need to start with a prefix byte.
searcher := minlz.NewBlockSearcher(bytes.NewReader(buf.Bytes()), minlz.BlockSearchCollectStats())
searcher.Search([]byte(`"unique-9876"`), func(r minlz.SearchResult) error {
fmt.Println("found at stream offset", r.StreamOffset)
return nil
})
stats := searcher.Stats()
fmt.Printf("blocks: %d total, %d skipped\n", stats.BlocksTotal, stats.BlocksSkipped)
}
Output: found at stream offset 4218 blocks: 3 total, 1 skipped
func WriterSidecar ¶ added in v1.2.0
func WriterSidecar(sidecar io.Writer) WriterOption
WriterSidecar redirects search-index chunks (0x44/0x45/0x46) to a separate sidecar writer rather than embedding them in the main stream. For each data block written to the main stream a 0x47 Remote Block Reference is appended to the sidecar that points at the block's offset in the main stream.
WriterSearchTable must also be supplied. Both the main stream and the sidecar are valid MinLZ streams on their own; the sidecar can be searched using NewSidecarSearcher with the main stream as the data source.
Example ¶
package main
import (
"bytes"
"fmt"
"io"
"github.com/minio/minlz"
)
func main() {
// Compress to one writer while diverting search-index chunks to a
// second writer. The main output contains only the compressed data; the
// sidecar contains 0x44/0x45/0x46 + 0x47 remote block references.
var main, side bytes.Buffer
cfg := minlz.NewSearchTableConfig().WithMatchLen(6)
w := minlz.NewWriter(&main,
minlz.WriterBlockSize(4096),
minlz.WriterConcurrency(1),
minlz.WriterSearchTable(cfg),
minlz.WriterSidecar(&side),
)
w.Write(bytes.Repeat([]byte("compressed log line, log line, log line\n"), 200))
w.Close()
// The main stream decompresses normally.
decoded, _ := io.ReadAll(minlz.NewReader(bytes.NewReader(main.Bytes())))
fmt.Printf("decoded: %d bytes\n", len(decoded))
// The sidecar is also a valid MinLZ stream that produces zero data bytes
// (all chunks in it are skippable).
sideDecoded, _ := io.ReadAll(minlz.NewReader(bytes.NewReader(side.Bytes())))
fmt.Printf("sidecar data bytes: %d\n", len(sideDecoded))
fmt.Printf("both streams non-empty: %v\n", main.Len() > 0 && side.Len() > 0)
}
Output: decoded: 8000 bytes sidecar data bytes: 0 both streams non-empty: true
func WriterUncompressed ¶
func WriterUncompressed() WriterOption
WriterUncompressed will bypass compression. The stream will be written as uncompressed blocks only. If concurrency is > 1 CRC calculation and output will be done async.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
_generate
module
|
|
|
cmd
|
|
|
internal/filepathx
Package filepathx adds double-star globbing support to the Glob function from the core path/filepath package.
|
Package filepathx adds double-star globbing support to the Glob function from the core path/filepath package. |
|
internal/readahead
Package readahead will do asynchronous read-ahead from an input io.Reader and make the data available as an io.Reader.
|
Package readahead will do asynchronous read-ahead from an input io.Reader and make the data available as an io.Reader. |
|
mz
command
|
|
|
internal
|
|








