lightning: Compress Reader/Writer supports reading/writing Snappy/Zstd type compressed files

br/pkg/storage/compress.go

@@ -81,7 +81,8 @@ func (w *withCompression) ReadFile(ctx context.Context, name string) ([]byte, er
 }
 
 type compressReader struct {
-	io.ReadCloser
+	io.Reader
+	io.Closer
 }
 
 // nolint:interfacer
@@ -94,14 +95,20 @@ func newInterceptReader(fileReader ExternalFileReader, compressType CompressType
 		return nil, errors.Trace(err)
 	}
 	return &compressReader{
-		ReadCloser: r,
+		Reader: r,
+		Closer: fileReader,
 	}, nil
 }
 
 func (*compressReader) Seek(_ int64, _ int) (int64, error) {
 	return int64(0), errors.Annotatef(berrors.ErrStorageInvalidConfig, "compressReader doesn't support Seek now")
 }
 
+func (c *compressReader) Close() error {
+	err := c.Closer.Close()
+	return err
+}
+
 type flushStorageWriter struct {
 	writer  io.Writer
 	flusher flusher

br/pkg/storage/writer.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"io"
 
+	"github.com/golang/snappy"
 	"github.com/pingcap/errors"
 )
 
@@ -17,6 +18,10 @@ const (
 	NoCompression CompressType = iota
 	// Gzip will compress given bytes in gzip format.
 	Gzip
+	// LZO will compress given bytes in lzo format.
+	LZO
+	// Snappy will compress given bytes in snappy format.
+	Snappy
 )
 
 type flusher interface {
@@ -50,15 +55,19 @@ func newCompressWriter(compressType CompressType, w io.Writer) simpleCompressWri
 	switch compressType {
 	case Gzip:
 		return gzip.NewWriter(w)
+	case Snappy:
+		return snappy.NewBufferedWriter(w)
 	default:
 		return nil
 	}
 }
 
-func newCompressReader(compressType CompressType, r io.Reader) (io.ReadCloser, error) {
+func newCompressReader(compressType CompressType, r io.Reader) (io.Reader, error) {
 	switch compressType {
 	case Gzip:
 		return gzip.NewReader(r)
+	case Snappy:
+		return snappy.NewReader(r), nil
 	default:
 		return nil, nil
 	}

br/pkg/storage/writer_test.go

@@ -102,8 +102,15 @@ func TestCompressReaderWriter(t *testing.T) {
 		ctx := context.Background()
 		storage, err := Create(ctx, backend, true)
 		require.NoError(t, err)
-		storage = WithCompression(storage, Gzip)
-		fileName := strings.ReplaceAll(test.name, " ", "-") + ".txt.gz"
+		storage = WithCompression(storage, test.compressType)
+		var suffix string
+		if test.compressType == Gzip {
+			suffix = ".txt.gz"
+		}
+		if test.compressType == Snappy {
+			suffix = ".txt.snappy"
+		}
+		fileName := strings.ReplaceAll(test.name, " ", "-") + suffix
 		writer, err := storage.Create(ctx, fileName)
 		require.NoError(t, err)
 		for _, str := range test.content {
@@ -124,7 +131,6 @@ func TestCompressReaderWriter(t *testing.T) {
 		_, err = bf.ReadFrom(r)
 		require.NoError(t, err)
 		require.Equal(t, strings.Join(test.content, ""), bf.String())
-		require.Nil(t, r.Close())
 
 		// test withCompression Open
 		r, err = storage.Open(ctx, fileName)
@@ -135,7 +141,8 @@ func TestCompressReaderWriter(t *testing.T) {
 
 		require.Nil(t, file.Close())
 	}
-	compressTypeArr := []CompressType{Gzip}
+	compressTypeArr := []CompressType{Gzip, Snappy}
+
 	tests := []testcase{
 		{
 			name: "long text medium chunks",