seekable

Documentation

Overview

Package adds an ability create ZSTD files in seekable format and randomly access them using uncompressed offsets.

Example

package main

import (
	"fmt"
	"io"
	"log"
	"os"

	"github.com/klauspost/compress/zstd"

	seekable "github.com/SaveTheRbtz/zstd-seekable-format-go/pkg"
)

func main() {
	f, err := os.CreateTemp("", "example")
	if err != nil {
		log.Fatal(err)
	}
	defer os.Remove(f.Name())

	enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedFastest))
	if err != nil {
		log.Fatal(err)
	}
	defer enc.Close()

	w, err := seekable.NewWriter(f, enc)
	if err != nil {
		log.Fatal(err)
	}

	// Write data in chunks.
	for _, b := range [][]byte{[]byte("Hello"), []byte(" "), []byte("World!")} {
		_, err = w.Write(b)
		if err != nil {
			log.Fatal(err)
		}
	}

	// Close and flush seek table.
	err = w.Close()
	if err != nil {
		log.Fatal(err)
	}

	dec, err := zstd.NewReader(nil)
	if err != nil {
		log.Fatal(err)
	}
	defer dec.Close()

	r, err := seekable.NewReader(f, dec)
	if err != nil {
		log.Fatal(err)
	}
	defer r.Close()

	ello := make([]byte, 4)
	// ReaderAt
	_, err = r.ReadAt(ello, 1)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Offset: 1 from the start: %s\n", string(ello))

	world := make([]byte, 5)
	// Seeker
	_, err = r.Seek(-6, io.SeekEnd)
	if err != nil {
		log.Fatal(err)
	}
	// Reader
	_, err = r.Read(world)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Offset: -6 from the end: %s\n", string(world))

	_, _ = f.Seek(0, io.SeekStart)

	// Standard ZSTD Reader.
	dec, err = zstd.NewReader(f)
	if err != nil {
		log.Fatal(err)
	}
	defer dec.Close()

	all, err := io.ReadAll(dec)
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("Whole string: %s\n", string(all))

}

Output:

Offset: 1 from the start: ello
Offset: -6 from the end: World
Whole string: Hello World!

Index

• func WithREnvironment(e env.REnvironment) rOption
• func WithRLogger(l *zap.Logger) rOption
• func WithWEnvironment(e env.WEnvironment) wOption
• func WithWLogger(l *zap.Logger) wOption
• type ConcurrentWriter
  • func NewWriter(w io.Writer, encoder ZSTDEncoder, opts ...wOption) (ConcurrentWriter, error)
• type Decoder
  • func NewDecoder(seekTable []byte, decoder ZSTDDecoder, opts ...rOption) (Decoder, error)
• type Encoder
  • func NewEncoder(encoder ZSTDEncoder, opts ...wOption) (Encoder, error)
• type FrameSource
• type Reader
  • func NewReader(rs io.ReadSeeker, decoder ZSTDDecoder, opts ...rOption) (Reader, error)
• type WriteManyOption
  • func WithConcurrency(concurrency int) WriteManyOption
  • func WithWriteCallback(cb func(size uint32)) WriteManyOption
• type Writer
• type ZSTDDecoder
• type ZSTDEncoder

Examples

• Package

Functions

func WithREnvironment

func WithREnvironment(e env.REnvironment) rOption

func WithRLogger

func WithRLogger(l *zap.Logger) rOption

func WithWEnvironment

func WithWEnvironment(e env.WEnvironment) wOption

func WithWLogger

func WithWLogger(l *zap.Logger) wOption

Types

type ConcurrentWriter

type ConcurrentWriter interface {
	Writer

	// WriteMany writes many frames concurrently
	WriteMany(ctx context.Context, frameSource FrameSource, options ...WriteManyOption) error
}

ConcurrentWriter allows writing many frames concurrently

func NewWriter

func NewWriter(w io.Writer, encoder ZSTDEncoder, opts ...wOption) (ConcurrentWriter, error)

NewWriter wraps the passed io.Writer and Encoder into and indexed ZSTD stream. Resulting stream then can be randomly accessed through the Reader and Decoder interfaces.

type Decoder

type Decoder interface {
	// GetIndexByDecompOffset returns FrameOffsetEntry for an offset in the decompressed stream.
	// Will return nil if offset is greater or equal than Size().
	GetIndexByDecompOffset(off uint64) *env.FrameOffsetEntry

	// GetIndexByID returns FrameOffsetEntry for a given frame id.
	// Will return nil if offset is greater or equal than NumFrames() or less than 0.
	GetIndexByID(id int64) *env.FrameOffsetEntry

	// Size returns the size of the uncompressed stream.
	Size() int64

	// NumFrames returns number of frames in the compressed stream.
	NumFrames() int64

	// Close closes the decoder feeing up any resources.
	Close() error
}

Decoder is a byte-oriented API that is useful for cases where wrapping io.ReadSeeker is not desirable.

func NewDecoder

func NewDecoder(seekTable []byte, decoder ZSTDDecoder, opts ...rOption) (Decoder, error)

NewDecoder creates a byte-oriented Decode interface from a given seektable index. This index can either be produced by either Writer's WriteSeekTable or Encoder's EndStream. Decoder can be used concurrently.

type Encoder

type Encoder interface {
	// Encode returns compressed data and appends a frame to in-memory seek table.
	Encode(src []byte) ([]byte, error)

	// EndStream returns in-memory seek table as a ZSTD's skippable frame.
	EndStream() ([]byte, error)
}

Encoder is a byte-oriented API that is useful where wrapping io.Writer is not desirable.

func NewEncoder

func NewEncoder(encoder ZSTDEncoder, opts ...wOption) (Encoder, error)

type FrameSource

type FrameSource func() ([]byte, error)

FrameSource returns one frame of data at a time. When there are no more frames, returns nil.

type Reader

type Reader interface {
	// Seek implements io.Seeker interface to randomly access data.
	// This method is NOT goroutine-safe and CAN NOT be called
	// concurrently since it modifies the underlying offset.
	Seek(offset int64, whence int) (int64, error)

	// Read implements io.Reader interface to sequentially access data.
	// This method is NOT goroutine-safe and CAN NOT be called
	// concurrently since it modifies the underlying offset.
	Read(p []byte) (n int, err error)

	// ReadAt implements io.ReaderAt interface to randomly access data.
	// This method is goroutine-safe and can be called concurrently ONLY if
	// the underlying reader supports io.ReaderAt interface.
	ReadAt(p []byte, off int64) (n int, err error)

	// Close implements io.Closer interface free up any resources.
	Close() error
}

func NewReader

func NewReader(rs io.ReadSeeker, decoder ZSTDDecoder, opts ...rOption) (Reader, error)

NewReader returns ZSTD stream reader that can be randomly accessed using uncompressed data offset. Ideally, passed io.ReadSeeker should implement io.ReaderAt interface.

type WriteManyOption

type WriteManyOption func(options *writeManyOptions) error

func WithConcurrency

func WithConcurrency(concurrency int) WriteManyOption

func WithWriteCallback

func WithWriteCallback(cb func(size uint32)) WriteManyOption

type Writer

type Writer interface {
	// Write writes a chunk of data as a separate frame into the datastream.
	//
	// Note that Write does not do any coalescing nor splitting of data,
	// so each write will map to a separate ZSTD Frame.
	Write(src []byte) (int, error)

	// Close implement io.Closer interface.  It writes the seek table footer
	// and releases occupied memory.
	//
	// Caller is still responsible to Close the underlying writer.
	Close() (err error)
}

type ZSTDDecoder

type ZSTDDecoder interface {
	DecodeAll(input, dst []byte) ([]byte, error)
}

ZSTDDecoder is the decompressor. Tested with github.com/klauspost/compress/zstd.

type ZSTDEncoder

type ZSTDEncoder interface {
	EncodeAll(src, dst []byte) []byte
}

ZSTDEncoder is the compressor. Tested with github.com/klauspost/compress/zstd.