Repository: matrixorigin/matrixone
Pull request #3012 (2022-06-15 06:11:18)

support decimal aggregation in plan2
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #3011 

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/go.mod b/go.mod
index 4568fc5e7..da4041b36 100644
--- a/go.mod
+++ b/go.mod
@@ -21,7 +21,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/google/btree v1.0.1
 	github.com/google/gofuzz v1.2.0
-	github.com/lni/dragonboat/v4 v4.0.0-20220604123758-e40bf3f57b59
+	github.com/lni/dragonboat/v4 v4.0.0-20220612172217-6ae1d0b2e535
 	github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4
 	github.com/matrixorigin/matrixcube v0.3.1-0.20220606032431-c944d801f1e5
 	github.com/matrixorigin/simdcsv v0.0.0-20210926114300-591bf748a770
diff --git a/go.sum b/go.sum
index f4a75cda9..b3f27f718 100644
--- a/go.sum
+++ b/go.sum
@@ -472,8 +472,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/labstack/echo/v4 v4.1.11/go.mod h1:i541M3Fj6f76NZtHSj7TXnyM8n2gaodfvfxNnFqi74g=
 github.com/labstack/echo/v4 v4.5.0/go.mod h1:czIriw4a0C1dFun+ObrXp7ok03xON0N1awStJ6ArI7Y=
 github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
-github.com/lni/dragonboat/v4 v4.0.0-20220604123758-e40bf3f57b59 h1:QQnr5/JOG9upSf0WXP6LVg18vBlbycJbY5egnpyWcKQ=
-github.com/lni/dragonboat/v4 v4.0.0-20220604123758-e40bf3f57b59/go.mod h1:5dd7hX8qUJ+72UVKdZZIBw5UbhZTaGY5PIt/TTdIeSY=
+github.com/lni/dragonboat/v4 v4.0.0-20220612172217-6ae1d0b2e535 h1:tjaMFsEucNN3tOkzT6FHxEWJQZW0skBmK7CXR25bVHo=
+github.com/lni/dragonboat/v4 v4.0.0-20220612172217-6ae1d0b2e535/go.mod h1:5dd7hX8qUJ+72UVKdZZIBw5UbhZTaGY5PIt/TTdIeSY=
 github.com/lni/goutils v1.3.0/go.mod h1:PUPtBAnZlRPUKWUXCsYkIRWubJbtNHpTAee0sczhlf4=
 github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4 h1:6gzI38ZJmbzZ7oZebXz6jII0uVK+MZ3+ds+7mIt1ioI=
 github.com/lni/goutils v1.3.1-0.20220604063047-388d67b4dbc4/go.mod h1:LIHvF0fflR+zyXUQFQOiHPpKANf3UIr7DFIv5CBPOoU=
diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go
index b4999d139..90e1a52f8 100644
--- a/pkg/frontend/mysql_cmd_executor.go
+++ b/pkg/frontend/mysql_cmd_executor.go
@@ -498,6 +498,7 @@ func getDataFromPipeline(obj interface{}, bat *batch.Batch) error {
 				if !nulls.Any(vec.Nsp) { //all data in this column are not null
 					vs := vec.Col.([]types.Decimal128)
 					row[i] = vs[rowIndex].Decimal128ToString(scale)
+					fmt.Println(row[i])
 				} else {
 					if nulls.Contains(vec.Nsp, uint64(rowIndex)) {
 						row[i] = nil
diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go
index dbcd10412..18a6da867 100644
--- a/pkg/logservice/client.go
+++ b/pkg/logservice/client.go
@@ -27,6 +27,14 @@ var (
 	ErrIncompatibleClient = moerr.NewError(moerr.INVALID_INPUT, "incompatible client")
 )
 
+// IsTempError returns a boolean value indicating whether the specified error is a temp
+// error that worth to be retried, e.g. timeouts, temp network issues, operation can be
+// completed as Raft leader is being elected. Non-temp error means the error is caused
+// by program logics rather than some external factors.
+func IsTempError(err error) bool {
+	return isTempError(err)
+}
+
 type LogServiceClientConfig struct {
 	ReadOnly  bool
 	ShardID   uint64
diff --git a/pkg/logservice/errors.go b/pkg/logservice/errors.go
index 2bdddaeb8..1d7b3f6d3 100644
--- a/pkg/logservice/errors.go
+++ b/pkg/logservice/errors.go
@@ -15,6 +15,8 @@
 package logservice
 
 import (
+	"net"
+
 	"github.com/cockroachdb/errors"
 	"github.com/lni/dragonboat/v4"
 
@@ -45,10 +47,10 @@ func getErrorToCodeMapping() []errorToCode {
 		{dragonboat.ErrShardNotReady, pb.ErrorCode_ShardNotReady, true},
 		{dragonboat.ErrSystemBusy, pb.ErrorCode_ShardNotReady, false},
 		{dragonboat.ErrClosed, pb.ErrorCode_SystemClosed, true},
+		{dragonboat.ErrInvalidRange, pb.ErrorCode_OutOfRange, true},
 
 		{ErrInvalidTruncateIndex, pb.ErrorCode_IndexAlreadyTruncated, true},
 		{ErrNotLeaseHolder, pb.ErrorCode_NotLeaseHolder, true},
-		{ErrOutOfRange, pb.ErrorCode_OutOfRange, true},
 	}
 }
 
@@ -83,3 +85,13 @@ func toError(resp pb.Response) error {
 	// will never reach here
 	return nil
 }
+
+func isTempError(err error) bool {
+	if dragonboat.IsTempError(err) {
+		return true
+	}
+	if _, ok := err.(net.Error); ok {
+		return true
+	}
+	return false
+}
diff --git a/pkg/logservice/service.go b/pkg/logservice/service.go
index 900df86c6..e09367b8a 100644
--- a/pkg/logservice/service.go
+++ b/pkg/logservice/service.go
@@ -141,7 +141,7 @@ func (s *Service) serve(conn net.Conn) {
 				}
 				return
 			}
-			if errors.Is(err, ErrBadMessage) {
+			if errors.Is(err, errBadMessage) {
 				return
 			}
 			operr, ok := err.(net.Error)
diff --git a/pkg/logservice/store.go b/pkg/logservice/store.go
index 4699e5a84..50b2670b9 100644
--- a/pkg/logservice/store.go
+++ b/pkg/logservice/store.go
@@ -35,8 +35,9 @@ import (
 var (
 	ErrInvalidTruncateIndex = moerr.NewError(moerr.INVALID_INPUT, "invalid input")
 	ErrNotLeaseHolder       = moerr.NewError(moerr.INVALID_STATE, "not lease holder")
-	ErrOutOfRange           = moerr.NewError(moerr.INVALID_INPUT, "query out of range")
 	ErrInvalidShardID       = moerr.NewError(moerr.INVALID_INPUT, "invalid shard ID")
+
+	ErrOutOfRange = dragonboat.ErrInvalidRange
 )
 
 type logStoreMeta struct {
diff --git a/pkg/logservice/transport.go b/pkg/logservice/transport.go
index 44ed870af..bc719c467 100644
--- a/pkg/logservice/transport.go
+++ b/pkg/logservice/transport.go
@@ -26,8 +26,7 @@ import (
 )
 
 var (
-	ErrInvalidBuffer   = moerr.NewError(moerr.INVALID_STATE, "invalid buffer size")
-	ErrBadMessage      = moerr.NewError(moerr.INVALID_INPUT, "invalid message")
+	errBadMessage      = moerr.NewError(moerr.INVALID_INPUT, "invalid message")
 	errPoisonReceived  = moerr.NewError(moerr.INVALID_STATE, "poison received")
 	ErrDeadlineNotSet  = moerr.NewError(moerr.INVALID_INPUT, "deadline not set")
 	ErrInvalidDeadline = moerr.NewError(moerr.INVALID_INPUT, "invalid deadline")
@@ -273,7 +272,7 @@ func readMagicNumber(conn net.Conn, buf []byte) error {
 		return errPoisonReceived
 	}
 	if !bytes.Equal(buf, magicNumber[:]) {
-		return ErrBadMessage
+		return errBadMessage
 	}
 	return nil
 }
diff --git a/pkg/sql/colexec2/aggregate/aggregate.go b/pkg/sql/colexec2/aggregate/aggregate.go
index ff4acdcfa..01abc5875 100755
--- a/pkg/sql/colexec2/aggregate/aggregate.go
+++ b/pkg/sql/colexec2/aggregate/aggregate.go
@@ -36,21 +36,26 @@ import (
 )
 
 var sumReturnTypes = map[types.T]types.T{
-	types.T_int8:    types.T_int64,
-	types.T_int16:   types.T_int64,
-	types.T_int32:   types.T_int64,
-	types.T_int64:   types.T_int64,
-	types.T_uint8:   types.T_uint64,
-	types.T_uint16:  types.T_uint64,
-	types.T_uint32:  types.T_uint64,
-	types.T_uint64:  types.T_uint64,
-	types.T_float32: types.T_float64,
-	types.T_float64: types.T_float64,
+	types.T_int8:       types.T_int64,
+	types.T_int16:      types.T_int64,
+	types.T_int32:      types.T_int64,
+	types.T_int64:      types.T_int64,
+	types.T_uint8:      types.T_uint64,
+	types.T_uint16:     types.T_uint64,
+	types.T_uint32:     types.T_uint64,
+	types.T_uint64:     types.T_uint64,
+	types.T_float32:    types.T_float64,
+	types.T_float64:    types.T_float64,
+	types.T_decimal64:  types.T_decimal64,
+	types.T_decimal128: types.T_decimal128,
 }
 
 func ReturnType(op int, typ types.T) types.T {
 	switch op {
 	case Avg:
+		if typ == types.T_decimal64 || typ == types.T_decimal128 {
+			return types.T_decimal128
+		}
 		return types.T_float64
 	case Max:
 		return typ
@@ -146,6 +151,10 @@ func NewSum(typ types.Type) (ring.Ring, error) {
 		return sum.NewInt(typ), nil
 	case types.T_uint8, types.T_uint16, types.T_uint32, types.T_uint64:
 		return sum.NewUint(typ), nil
+	case types.T_decimal64:
+		return sum.NewDecimal64(typ), nil
+	case types.T_decimal128:
+		return sum.NewDecimal128(typ), nil
 	}
 	return nil, fmt.Errorf("'%v' not support Sum", typ)
 }
@@ -180,6 +189,10 @@ func NewMax(typ types.Type) (ring.Ring, error) {
 		return max.NewDate(typ), nil
 	case types.T_datetime:
 		return max.NewDatetime(typ), nil
+	case types.T_decimal64:
+		return max.NewDecimal64(typ), nil
+	case types.T_decimal128:
+		return max.NewDecimal128(typ), nil
 	}
 	return nil, fmt.Errorf("'%v' not support Max", typ)
 }
@@ -214,6 +227,10 @@ func NewMin(typ types.Type) (ring.Ring, error) {
 		return min.NewDate(typ), nil
 	case types.T_datetime:
 		return min.NewDatetime(typ), nil
+	case types.T_decimal64:
+		return min.NewDecimal64(typ), nil
+	case types.T_decimal128:
+		return min.NewDecimal128(typ), nil
 	}
 	return nil, fmt.Errorf("'%v' not support Min", typ)
 }


Test patch:
diff --git a/pkg/logservice/errors_test.go b/pkg/logservice/errors_test.go
index 8ed7ecd12..76867b330 100644
--- a/pkg/logservice/errors_test.go
+++ b/pkg/logservice/errors_test.go
@@ -18,6 +18,7 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/lni/dragonboat/v4"
 	"github.com/stretchr/testify/assert"
 
 	pb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
@@ -44,3 +45,36 @@ func TestUnknownErrorIsHandled(t *testing.T) {
 	assert.Equal(t, pb.ErrorCode_OtherSystemError, code)
 	assert.Equal(t, err.Error(), str)
 }
+
+func TestIsTempError(t *testing.T) {
+	tests := []struct {
+		err  error
+		temp bool
+	}{
+		{dragonboat.ErrInvalidOperation, false},
+		{dragonboat.ErrInvalidAddress, false},
+		{dragonboat.ErrInvalidSession, false},
+		{dragonboat.ErrTimeoutTooSmall, false},
+		{dragonboat.ErrPayloadTooBig, false},
+		{dragonboat.ErrSystemBusy, true},
+		{dragonboat.ErrShardClosed, true},
+		{dragonboat.ErrShardNotInitialized, true},
+		{dragonboat.ErrTimeout, true},
+		{dragonboat.ErrClosed, true},
+		{dragonboat.ErrCanceled, false},
+		{dragonboat.ErrRejected, false},
+		{dragonboat.ErrShardNotReady, true},
+		{dragonboat.ErrInvalidTarget, false},
+		{dragonboat.ErrInvalidRange, false},
+
+		{ErrDeadlineNotSet, false},
+		{ErrInvalidDeadline, false},
+		{ErrInvalidTruncateIndex, false},
+		{ErrNotLeaseHolder, false},
+		{ErrOutOfRange, false},
+		{ErrInvalidShardID, false},
+	}
+	for _, tt := range tests {
+		assert.Equal(t, tt.temp, isTempError(tt.err))
+	}
+}
diff --git a/pkg/logservice/transport_test.go b/pkg/logservice/transport_test.go
index d51c8be7a..aa8103df2 100644
--- a/pkg/logservice/transport_test.go
+++ b/pkg/logservice/transport_test.go
@@ -234,7 +234,7 @@ func TestReadMagicNumber(t *testing.T) {
 	conn := newTestConn()
 	_, err := conn.Write([]byte{0, 0})
 	assert.NoError(t, err)
-	assert.Equal(t, ErrBadMessage, readMagicNumber(conn, nil))
+	assert.Equal(t, errBadMessage, readMagicNumber(conn, nil))
 
 	conn = newTestConn()
 	_, err = conn.Write(magicNumber[:])