-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement compression for incoming/outgoing records and for sstables (#…
…349)
- Loading branch information
Showing
40 changed files
with
916 additions
and
300 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package compress | ||
|
|
||
| import "testing" | ||
|
|
||
| func BenchmarkCompressGzip(b *testing.B) { | ||
| benchmarkCompress(b, CompressionTypeGzip) | ||
| } | ||
|
|
||
| func BenchmarkCompressSnappy(b *testing.B) { | ||
| benchmarkCompress(b, CompressionTypeSnappy) | ||
| } | ||
|
|
||
| func BenchmarkCompressLz4(b *testing.B) { | ||
| benchmarkCompress(b, CompressionTypeLz4) | ||
| } | ||
|
|
||
| func BenchmarkCompressZstd(b *testing.B) { | ||
| benchmarkCompress(b, CompressionTypeZstd) | ||
| } | ||
|
|
||
| func benchmarkCompress(b *testing.B, compressionType CompressionType) { | ||
| buff := randomBytes(100000) | ||
| for i := 0; i < b.N; i++ { | ||
| compressed, err := Compress(compressionType, nil, buff) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| decompressed, err := Decompress(compressionType, compressed) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| if len(decompressed) != len(buff) { | ||
| panic("wrong compressed size") | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| package compress | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "compress/gzip" | ||
| "github.com/DataDog/zstd" | ||
| "github.com/klauspost/compress/s2" | ||
| "github.com/pierrec/lz4/v4" | ||
| "github.com/pkg/errors" | ||
| "io" | ||
| ) | ||
|
|
||
| type CompressionType byte | ||
|
|
||
| const ( | ||
| CompressionTypeNone CompressionType = 0 | ||
| CompressionTypeGzip CompressionType = 1 | ||
| CompressionTypeSnappy CompressionType = 2 | ||
| CompressionTypeLz4 CompressionType = 3 | ||
| CompressionTypeZstd CompressionType = 4 | ||
| CompressionTypeUnknown CompressionType = 255 | ||
| ) | ||
|
|
||
| func FromString(str string) CompressionType { | ||
| switch str { | ||
| case "none": | ||
| return CompressionTypeNone | ||
| case "gzip": | ||
| return CompressionTypeGzip | ||
| case "snappy": | ||
| return CompressionTypeSnappy | ||
| case "lz4": | ||
| return CompressionTypeLz4 | ||
| case "zstd": | ||
| return CompressionTypeZstd | ||
| default: | ||
| return CompressionTypeUnknown | ||
| } | ||
| } | ||
|
|
||
| func (t CompressionType) String() string { | ||
| switch t { | ||
| case CompressionTypeNone: | ||
| return "none" | ||
| case CompressionTypeGzip: | ||
| return "gzip" | ||
| case CompressionTypeSnappy: | ||
| return "snappy" | ||
| case CompressionTypeLz4: | ||
| return "lz4" | ||
| case CompressionTypeZstd: | ||
| return "zstd" | ||
| case CompressionTypeUnknown: | ||
| return "unknown" | ||
| default: | ||
| panic("unknown compression type") | ||
| } | ||
| } | ||
|
|
||
| func Compress(compressionType CompressionType, buff []byte, data []byte) ([]byte, error) { | ||
| var w io.WriteCloser | ||
| var buf *bytes.Buffer | ||
| switch compressionType { | ||
| case CompressionTypeGzip: | ||
| buf = bytes.NewBuffer(buff) | ||
| w = gzip.NewWriter(buf) | ||
| if _, err := w.Write(data); err != nil { | ||
| return nil, err | ||
| } | ||
| case CompressionTypeSnappy: | ||
| compressed := s2.EncodeSnappy(nil, data) | ||
| if buff == nil { | ||
| return compressed, nil | ||
| } | ||
| return append(buff, compressed...), nil | ||
| case CompressionTypeLz4: | ||
| buf = bytes.NewBuffer(buff) | ||
| w = lz4.NewWriter(buf) | ||
| if _, err := w.Write(data); err != nil { | ||
| return nil, err | ||
| } | ||
| case CompressionTypeZstd: | ||
| compressed, err := zstd.Compress(nil, data) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if buff == nil { | ||
| return compressed, nil | ||
| } | ||
| return append(buff, compressed...), nil | ||
| default: | ||
| return nil, errors.Errorf("unexpected compression type: %d", compressionType) | ||
| } | ||
| if err := w.Close(); err != nil { | ||
| return nil, err | ||
| } | ||
| return buf.Bytes(), nil | ||
| } | ||
|
|
||
| func Decompress(compressionType CompressionType, data []byte) ([]byte, error) { | ||
| var r io.Reader | ||
| switch compressionType { | ||
| case CompressionTypeGzip: | ||
| var err error | ||
| r, err = gzip.NewReader(bytes.NewReader(data)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| case CompressionTypeSnappy: | ||
| return s2.Decode(nil, data) | ||
| case CompressionTypeLz4: | ||
| r = lz4.NewReader(bytes.NewReader(data)) | ||
| case CompressionTypeZstd: | ||
| return zstd.Decompress(nil, data) | ||
| default: | ||
| return nil, errors.Errorf("unexpected compression type: %d", compressionType) | ||
| } | ||
| buf := new(bytes.Buffer) | ||
| _, err := io.Copy(buf, r) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return buf.Bytes(), nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,53 @@ | ||
| package compress | ||
|
|
||
| import ( | ||
| "crypto/rand" | ||
| "github.com/stretchr/testify/require" | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestCompressionGzip(t *testing.T) { | ||
| testCompressor(t, CompressionTypeGzip) | ||
| } | ||
|
|
||
| func TestCompressionSnappy(t *testing.T) { | ||
| testCompressor(t, CompressionTypeSnappy) | ||
| } | ||
|
|
||
| func TestCompressionLz4(t *testing.T) { | ||
| testCompressor(t, CompressionTypeLz4) | ||
| } | ||
|
|
||
| func TestCompressionZstd(t *testing.T) { | ||
| testCompressor(t, CompressionTypeZstd) | ||
| } | ||
|
|
||
| func testCompressor(t *testing.T, compressionType CompressionType) { | ||
| testCompressorWithInitialBytes(t, compressionType, 0) | ||
| testCompressorWithInitialBytes(t, compressionType, 100) | ||
| } | ||
|
|
||
| func testCompressorWithInitialBytes(t *testing.T, compressionType CompressionType, numInitialBytes int) { | ||
| data := randomBytes(10000) | ||
| var initialBytes []byte | ||
| if numInitialBytes > 0 { | ||
| initialBytes = randomBytes(numInitialBytes) | ||
| } | ||
| compressed, err := Compress(compressionType, initialBytes, data) | ||
| require.NoError(t, err) | ||
| if numInitialBytes > 0 { | ||
| require.Equal(t, initialBytes, compressed[:len(initialBytes)]) | ||
| } | ||
| decompressed, err := Decompress(compressionType, compressed[len(initialBytes):]) | ||
| require.NoError(t, err) | ||
| require.Equal(t, data, decompressed) | ||
| } | ||
|
|
||
| func randomBytes(n int) []byte { | ||
| b := make([]byte, n) | ||
| _, err := rand.Read(b) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| return b | ||
| } |
Oops, something went wrong.