Add LZ4 compressor support. (#324)

* feat: add lz4 compressor support.

Signed-off-by: Wyatt Jia <i@eventloop.live>

* fix lz4 compressor buffer error.

Signed-off-by: Wyatt Jia <i@eventloop.live>

* format code by goimports.

* Add equal lz4 decompress test case and format code by goimports.

Signed-off-by: Wyatt Jia <i@eventloop.live>

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/natefinch/lumberjack v2.0.0+incompatible
 	github.com/parnurzeal/gorequest v0.2.16
+	github.com/pierrec/lz4/v4 v4.1.17
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.12.2
 	github.com/prometheus/common v0.32.1

go.sum

@@ -669,6 +669,8 @@ github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pierrec/lz4 v2.5.2+incompatible h1:WCjObylUIOlKy/+7Abdn34TLIkXiA4UWUMhxq9m9ZXI=
 github.com/pierrec/lz4 v2.5.2+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
+github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
+github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63 h1:+FZIDR/D97YOPik4N4lPDaUcLDF/EQPogxtlHB2ZZRM=
 github.com/pingcap/errors v0.11.5-0.20210425183316-da1aaba5fb63/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=

pkg/compressor/lz4_compress.go

@@ -16,3 +16,42 @@
  */
 
 package compressor
+
+import (
+	"fmt"
+
+	"github.com/pierrec/lz4/v4"
+)
+
+type Lz4 struct {
+}
+
+func (l *Lz4) Compress(data []byte) ([]byte, error) {
+
+	buffer := make([]byte, lz4.CompressBlockBound(len(data)))
+
+	var compressor lz4.Compressor
+
+	n, err := compressor.CompressBlock(data, buffer)
+	if err != nil {
+		return nil, err
+	}
+	if n >= len(data) {
+		return nil, fmt.Errorf("`%s` is not compressible", string(data))
+	}
+
+	return buffer[:n], nil
+}
+
+func (l *Lz4) Decompress(in []byte) ([]byte, error) {
+	out := make([]byte, 100*len(in))
+	n, err := lz4.UncompressBlock(in, out)
+	if err != nil {
+		return nil, err
+	}
+	return out[:n], nil
+}
+
+func (l *Lz4) GetCompressorType() CompressorType {
+	return CompressorLz4
+}

pkg/compressor/lz4_compress_test.go

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package compressor
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLz4Compress(t *testing.T) {
+	sample := strings.Repeat("hello world", 100)
+
+	lz4 := Lz4{}
+
+	compressResult, err := lz4.Compress([]byte(sample))
+	assert.NoError(t, err)
+	t.Logf("Compressed result: %v", string(compressResult))
+
+	decompressResult, err := lz4.Decompress(compressResult)
+	assert.NoError(t, err)
+	assert.Equal(t, sample, string(decompressResult))
+	t.Logf("Decompressed result: %v", string(decompressResult))
+}

pkg/datasource/sql/datasource/datasource_manager.go

@@ -119,7 +119,7 @@ func (dm *BasicSourceManager) BranchRegister(ctx context.Context, clientId strin
 	return 0, nil
 }
 
-//  Branch report
+// Branch report
 func (dm *BasicSourceManager) BranchReport(ctx context.Context, req message.BranchReportRequest) error {
 	return nil
 }
@@ -138,7 +138,7 @@ func (dm *BasicSourceManager) RegisterResource(resource rm.Resource) error {
 	return nil
 }
 
-//  Unregister a   model.Resource from the   model.Resource Manager
+// Unregister a   model.Resource from the   model.Resource Manager
 func (dm *BasicSourceManager) UnregisterResource(resource rm.Resource) error {
 	return errors.New("unsupport unregister resource")
 }

pkg/datasource/sql/types/executor.go

@@ -20,6 +20,7 @@ package types
 import "github.com/arana-db/parser/ast"
 
 // ExecutorType
+//
 //go:generate stringer -type=ExecutorType
 type ExecutorType int32