Repository: matrixorigin/matrixone
Pull request #2998 (2022-06-14 14:15:17)

Hakeeper: add expected replica count
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [x] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

issue #

**What this PR does / why we need it:**

Added the expected replica count used by HAKeeper

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/cmd/generate-config/system_vars_def.toml b/cmd/generate-config/system_vars_def.toml
index 7d110ad22..02eabc04a 100644
--- a/cmd/generate-config/system_vars_def.toml
+++ b/cmd/generate-config/system_vars_def.toml
@@ -394,8 +394,8 @@ scope = ["global"]
 access = ["file"]
 type = "string"
 domain-type = "set"
-values = ["aoe","tae"]
-comment = "default engine is 'aoe'."
+values = ["tae", "aoe"]
+comment = "default engine is 'tae'."
 update-mode = "dynamic"
 
 [[parameter]]
@@ -414,8 +414,8 @@ scope = ["global"]
 access = ["file"]
 type = "bool"
 domain-type = "set"
-values = []
-comment = "default is false. true : use plan2/compile2 false : use stuff in v0.4.0"
+values = ["true", "false"]
+comment = "default is true. true : use plan2/compile2 false : use stuff in v0.4.0"
 update-mode = "dynamic"
 
 [[parameter]]
diff --git a/docs/cn/MatrixOne/MatrixCampEvent.md b/docs/cn/MatrixOne/MatrixCampEvent.md
index f00755dfd..1502d79fa 100644
--- a/docs/cn/MatrixOne/MatrixCampEvent.md
+++ b/docs/cn/MatrixOne/MatrixCampEvent.md
@@ -18,7 +18,7 @@ MatrixOne社区一共准备了4个类别的56个任务，有不同的难度级
 在开始之前，先给MatrixOne项目Star, Fork, Watch一下吧。
 
 1. 选择很重要！加小助手微信“MatrixOrigin001”填写活动注册表<https://www.wjx.top/vm/Ys1rz1I.aspx> 选择你想要完成的函数任务，并加入MatrixOne社区群。
-2. 登陆Github，在你选择的函数issue留下你的comment，比如“I'd like to work on this issue”, 小助手会将相关issue分配给你。
+2. 登录Github，在你选择的函数issue留下你的comment，比如“I'd like to work on this issue”, 小助手会将相关issue分配给你。
 3. 开始只属于你的函数体验任务。如有任何问题，MatrixOne社区群里的技术大牛全程在线支持。
 
 > 请注意：1个开发者可以选择多个挑战任务，但是1个任务只能由1个开发者完成
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/container/types/date.go b/pkg/container/types/date.go
index d5b8a75f7..23ee4ebed 100644
--- a/pkg/container/types/date.go
+++ b/pkg/container/types/date.go
@@ -16,6 +16,7 @@ package types
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/matrixorigin/matrixone/pkg/errno"
@@ -201,6 +202,36 @@ func (d Date) Year() uint16 {
 	return year
 }
 
+func (d Date) YearMonth() uint32 {
+	year, month, _, _ := d.Calendar(true)
+	yearStr := fmt.Sprintf("%04d", year)
+	monthStr := fmt.Sprintf("%02d", month)
+	// fmt.Println(yearStr, monthStr, "--------")
+	result, _ := strconv.ParseUint(yearStr+monthStr, 10, 32)
+	// fmt.Println(result)
+	return uint32(result)
+}
+
+var monthToQuarter = map[uint8]uint32{
+	1:  1,
+	2:  1,
+	3:  1,
+	4:  2,
+	5:  2,
+	6:  2,
+	7:  3,
+	8:  3,
+	9:  3,
+	10: 4,
+	11: 4,
+	12: 4,
+}
+
+func (d Date) Quarter() uint32 {
+	_, month, _, _ := d.Calendar(true)
+	return monthToQuarter[month]
+}
+
 func (d Date) Calendar(full bool) (year int32, month, day uint8, yday uint16) {
 	// Account for 400 year cycles.
 	n := d / daysPer400Years
diff --git a/pkg/container/types/interval.go b/pkg/container/types/interval.go
index ebc61eb2e..177ee17df 100644
--- a/pkg/container/types/interval.go
+++ b/pkg/container/types/interval.go
@@ -127,18 +127,18 @@ func IntervalTypeOf(s string) (IntervalType, error) {
 //  2. 1-1 is parsed out as 1, 1 '-' is delim, so is '+', '.' etc.
 //	3. we will not support int32 overflow.
 //
-func parseInts(s string, isxxxMicrosecond bool, typeMaxLength int) ([]int32, error) {
-	ret := make([]int32, 0)
+func parseInts(s string, isxxxMicrosecond bool, typeMaxLength int) ([]int64, error) {
+	ret := make([]int64, 0)
 	numLength := 0
 	cur := -1
 	for _, c := range s {
 		if c >= rune('0') && c <= rune('9') {
 			if cur < 0 {
 				cur = len(ret)
-				ret = append(ret, c-rune('0'))
+				ret = append(ret, int64(c-rune('0')))
 				numLength++
 			} else {
-				ret[cur] = 10*ret[cur] + c - rune('0')
+				ret[cur] = 10*ret[cur] + int64(c-rune('0'))
 				numLength++
 				if ret[cur] < 0 {
 					return nil, errors.New(errno.DataException, "Invalid string interval value")
@@ -153,13 +153,13 @@ func parseInts(s string, isxxxMicrosecond bool, typeMaxLength int) ([]int32, err
 	}
 	if isxxxMicrosecond {
 		if len(ret) == typeMaxLength {
-			ret[len(ret)-1] *= int32(math.Pow10(6 - numLength))
+			ret[len(ret)-1] *= int64(math.Pow10(6 - numLength))
 		}
 	}
 	return ret, nil
 }
 
-func conv(a []int32, mul []int64, rt IntervalType) (int64, IntervalType, error) {
+func conv(a []int64, mul []int64, rt IntervalType) (int64, IntervalType, error) {
 	if len(a) > len(mul) {
 		return 0, IntervalTypeInvalid, errors.New(errno.DataException, "Invalid interval format")
 	}
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/hakeeper/state.go b/pkg/hakeeper/state.go
index f7bcccb20..8ae14f95b 100644
--- a/pkg/hakeeper/state.go
+++ b/pkg/hakeeper/state.go
@@ -26,8 +26,8 @@ const (
 	NoLeader uint64 = 0
 )
 
-// DNShardInfo contins information on a list of shards.
-type DNShardInfo struct {
+// DNStoreInfo contins information on a list of shards.
+type DNStoreInfo struct {
 	Tick   uint64
 	Shards []logservice.DNShardInfo
 }
@@ -35,14 +35,14 @@ type DNShardInfo struct {
 // DNState contains all DN details known to the HAKeeper.
 type DNState struct {
 	// Stores is keyed by DN store UUID, it contains details found on each DN
-	// store.
-	Stores map[string]DNShardInfo
+	// store. Each DNStoreInfo reflects what was last reported by each DN store.
+	Stores map[string]DNStoreInfo
 }
 
 // NewDNState creates a new DNState.
 func NewDNState() DNState {
 	return DNState{
-		Stores: make(map[string]DNShardInfo),
+		Stores: make(map[string]DNStoreInfo),
 	}
 }
 
@@ -50,17 +50,17 @@ func NewDNState() DNState {
 // current tick of the HAKeeper which can be used as the timestamp of the
 // heartbeat.
 func (s *DNState) Update(hb logservice.DNStoreHeartbeat, tick uint64) {
-	shardInfo, ok := s.Stores[hb.UUID]
+	storeInfo, ok := s.Stores[hb.UUID]
 	if !ok {
-		shardInfo = DNShardInfo{}
+		storeInfo = DNStoreInfo{}
 	}
-	shardInfo.Tick = tick
-	shardInfo.Shards = hb.Shards
-	s.Stores[hb.UUID] = shardInfo
+	storeInfo.Tick = tick
+	storeInfo.Shards = hb.Shards
+	s.Stores[hb.UUID] = storeInfo
 }
 
-// LogShardInfo contains information of all replicas found on a Log store.
-type LogShardInfo struct {
+// LogStoreInfo contains information of all replicas found on a Log store.
+type LogStoreInfo struct {
 	Tick           uint64
 	RaftAddress    string
 	ServiceAddress string
@@ -75,15 +75,15 @@ type LogState struct {
 	// shard.
 	Shards map[uint64]logservice.LogShardInfo
 	// Stores is keyed by log store UUID, it contains details found on each store.
-	// Each LogShardInfo here reflects what was last reported by each Log store.
-	Stores map[string]LogShardInfo
+	// Each LogStoreInfo here reflects what was last reported by each Log store.
+	Stores map[string]LogStoreInfo
 }
 
 // NewLogState creates a new LogState.
 func NewLogState() LogState {
 	return LogState{
 		Shards: make(map[uint64]logservice.LogShardInfo),
-		Stores: make(map[string]LogShardInfo),
+		Stores: make(map[string]LogStoreInfo),
 	}
 }
 
@@ -95,16 +95,16 @@ func (s *LogState) Update(hb logservice.LogStoreHeartbeat, tick uint64) {
 }
 
 func (s *LogState) updateStores(hb logservice.LogStoreHeartbeat, tick uint64) {
-	shardInfo, ok := s.Stores[hb.UUID]
+	storeInfo, ok := s.Stores[hb.UUID]
 	if !ok {
-		shardInfo = LogShardInfo{}
+		storeInfo = LogStoreInfo{}
 	}
-	shardInfo.Tick = tick
-	shardInfo.RaftAddress = hb.RaftAddress
-	shardInfo.ServiceAddress = hb.ServiceAddress
-	shardInfo.GossipAddress = hb.GossipAddress
-	shardInfo.Shards = hb.Shards
-	s.Stores[hb.UUID] = shardInfo
+	storeInfo.Tick = tick
+	storeInfo.RaftAddress = hb.RaftAddress
+	storeInfo.ServiceAddress = hb.ServiceAddress
+	storeInfo.GossipAddress = hb.GossipAddress
+	storeInfo.Shards = hb.Shards
+	s.Stores[hb.UUID] = storeInfo
 }
 
 func (s *LogState) updateShards(hb logservice.LogStoreHeartbeat) {
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/pb/metadata/metadata.pb.go b/pkg/pb/metadata/metadata.pb.go
index b38e98107..cca8a473e 100644
--- a/pkg/pb/metadata/metadata.pb.go
+++ b/pkg/pb/metadata/metadata.pb.go
@@ -23,39 +23,35 @@ var _ = math.Inf
 // proto package needs to be updated.
 const _ = proto.GoGoProtoPackageIsVersion3 // please upgrade the proto package
 
-// DN dn shard metadata
-type DN struct {
-	// ID the id of the DN. The ID is not modified after the DN is created.
-	ID uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
-	// ReplicaID only one replica for one DN. The ReplicaID is specified at
-	// the time the DN is created. DN restart ReplicaID will not change, when
-	// a DN is migrated to another node, the ReplicaID will be reset.
-	ReplicaID uint64 `protobuf:"varint,2,opt,name=replicaID,proto3" json:"replicaID,omitempty"`
+// DNShardRecord is DN shard metadata describing what is a DN shard. It
+// is internally used by HAKeeper to maintain how many DNs available in
+// the system.
+type DNShardRecord struct {
+	// ShardID the id of the DN shard.
+	ShardID uint64 `protobuf:"varint,1,opt,name=ShardID,proto3" json:"ShardID,omitempty"`
 	// LogShardID a DN corresponds to a unique Shard of LogService.
-	LogShardID uint64 `protobuf:"varint,3,opt,name=logShardID,proto3" json:"logShardID,omitempty"`
+	LogShardID uint64 `protobuf:"varint,2,opt,name=LogShardID,proto3" json:"LogShardID,omitempty"`
 	// Start A DN manages data within a range [Start, end)
-	Start []byte `protobuf:"bytes,4,opt,name=start,proto3" json:"start,omitempty"`
+	Start []byte `protobuf:"bytes,3,opt,name=Start,proto3" json:"Start,omitempty"`
 	// End A DN manages data within a range [Start, end)
-	End []byte `protobuf:"bytes,5,opt,name=end,proto3" json:"end,omitempty"`
-	// Address DN external service address.
-	Address              string   `protobuf:"bytes,6,opt,name=address,proto3" json:"address,omitempty"`
+	End                  []byte   `protobuf:"bytes,4,opt,name=End,proto3" json:"End,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
 }
 
-func (m *DN) Reset()         { *m = DN{} }
-func (m *DN) String() string { return proto.CompactTextString(m) }
-func (*DN) ProtoMessage()    {}
-func (*DN) Descriptor() ([]byte, []int) {
+func (m *DNShardRecord) Reset()         { *m = DNShardRecord{} }
+func (m *DNShardRecord) String() string { return proto.CompactTextString(m) }
+func (*DNShardRecord) ProtoMessage()    {}
+func (*DNShardRecord) Descriptor() ([]byte, []int) {
 	return fileDescriptor_56d9f74966f40d04, []int{0}
 }
-func (m *DN) XXX_Unmarshal(b []byte) error {
+func (m *DNShardRecord) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
 }
-func (m *DN) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+func (m *DNShardRecord) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 	if deterministic {
-		return xxx_messageInfo_DN.Marshal(b, m, deterministic)
+		return xxx_messageInfo_DNShardRecord.Marshal(b, m, deterministic)
 	} else {
 		b = b[:cap(b)]
 		n, err := m.MarshalToSizedBuffer(b)
@@ -65,87 +61,210 @@ func (m *DN) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 		return b[:n], nil
 	}
 }
-func (m *DN) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_DN.Merge(m, src)
+func (m *DNShardRecord) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DNShardRecord.Merge(m, src)
 }
-func (m *DN) XXX_Size() int {
+func (m *DNShardRecord) XXX_Size() int {
 	return m.Size()
 }
-func (m *DN) XXX_DiscardUnknown() {
-	xxx_messageInfo_DN.DiscardUnknown(m)
+func (m *DNShardRecord) XXX_DiscardUnknown() {
+	xxx_messageInfo_DNShardRecord.DiscardUnknown(m)
 }
 
-var xxx_messageInfo_DN proto.InternalMessageInfo
+var xxx_messageInfo_DNShardRecord proto.InternalMessageInfo
 
-func (m *DN) GetID() uint64 {
+func (m *DNShardRecord) GetShardID() uint64 {
 	if m != nil {
-		return m.ID
+		return m.ShardID
 	}
 	return 0
 }
 
-func (m *DN) GetReplicaID() uint64 {
-	if m != nil {
-		return m.ReplicaID
-	}
-	return 0
-}
-
-func (m *DN) GetLogShardID() uint64 {
+func (m *DNShardRecord) GetLogShardID() uint64 {
 	if m != nil {
 		return m.LogShardID
 	}
 	return 0
 }
 
-func (m *DN) GetStart() []byte {
+func (m *DNShardRecord) GetStart() []byte {
 	if m != nil {
 		return m.Start
 	}
 	return nil
 }
 
-func (m *DN) GetEnd() []byte {
+func (m *DNShardRecord) GetEnd() []byte {
 	if m != nil {
 		return m.End
 	}
 	return nil
 }
 
-func (m *DN) GetAddress() string {
+// DNShard
+type DNShard struct {
+	// DNShard extends DNShardRecord
+	DNShardRecord `protobuf:"bytes,1,opt,name=DNShardRecord,proto3,embedded=DNShardRecord" json:"DNShardRecord"`
+	// ReplicaID only one replica for one DN. The ReplicaID is specified at
+	// the time the DN is created. DN restart ReplicaID will not change, when
+	// a DN is migrated to another node, the ReplicaID will be reset.
+	ReplicaID uint64 `protobuf:"varint,2,opt,name=ReplicaID,proto3" json:"ReplicaID,omitempty"`
+	// Address is DN's external service address.
+	Address              string   `protobuf:"bytes,3,opt,name=Address,proto3" json:"Address,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *DNShard) Reset()         { *m = DNShard{} }
+func (m *DNShard) String() string { return proto.CompactTextString(m) }
+func (*DNShard) ProtoMessage()    {}
+func (*DNShard) Descriptor() ([]byte, []int) {
+	return fileDescriptor_56d9f74966f40d04, []int{1}
+}
+func (m *DNShard) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *DNShard) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_DNShard.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *DNShard) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_DNShard.Merge(m, src)
+}
+func (m *DNShard) XXX_Size() int {
+	return m.Size()
+}
+func (m *DNShard) XXX_DiscardUnknown() {
+	xxx_messageInfo_DNShard.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_DNShard proto.InternalMessageInfo
+
+func (m *DNShard) GetReplicaID() uint64 {
+	if m != nil {
+		return m.ReplicaID
+	}
+	return 0
+}
+
+func (m *DNShard) GetAddress() string {
 	if m != nil {
 		return m.Address
 	}
 	return ""
 }
 
+// LogShardRecord is Log Shard Metadata describing what is a Log shard. It is
+// internally used by the HAKeeper to maintain how many Log shards are available
+// in the system.
+type LogShardRecord struct {
+	// ShardID is the id of the Log Shard.
+	ShardID uint64 `protobuf:"varint,1,opt,name=ShardID,proto3" json:"ShardID,omitempty"`
+	// NumberOfReplicas is the number of replicas in the shard.
+	NumberOfReplicas uint64 `protobuf:"varint,2,opt,name=NumberOfReplicas,proto3" json:"NumberOfReplicas,omitempty"`
+	// Name is the human readable name of the shard given by the DN.
+	Name                 string   `protobuf:"bytes,3,opt,name=Name,proto3" json:"Name,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *LogShardRecord) Reset()         { *m = LogShardRecord{} }
+func (m *LogShardRecord) String() string { return proto.CompactTextString(m) }
+func (*LogShardRecord) ProtoMessage()    {}
+func (*LogShardRecord) Descriptor() ([]byte, []int) {
+	return fileDescriptor_56d9f74966f40d04, []int{2}
+}
+func (m *LogShardRecord) XXX_Unmarshal(b []byte) error {
+	return m.Unmarshal(b)
+}
+func (m *LogShardRecord) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	if deterministic {
+		return xxx_messageInfo_LogShardRecord.Marshal(b, m, deterministic)
+	} else {
+		b = b[:cap(b)]
+		n, err := m.MarshalToSizedBuffer(b)
+		if err != nil {
+			return nil, err
+		}
+		return b[:n], nil
+	}
+}
+func (m *LogShardRecord) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_LogShardRecord.Merge(m, src)
+}
+func (m *LogShardRecord) XXX_Size() int {
+	return m.Size()
+}
+func (m *LogShardRecord) XXX_DiscardUnknown() {
+	xxx_messageInfo_LogShardRecord.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogShardRecord proto.InternalMessageInfo
+
+func (m *LogShardRecord) GetShardID() uint64 {
+	if m != nil {
+		return m.ShardID
+	}
+	return 0
+}
+
+func (m *LogShardRecord) GetNumberOfReplicas() uint64 {
+	if m != nil {
+		return m.NumberOfReplicas
+	}
+	return 0
+}
+
+func (m *LogShardRecord) GetName() string {
+	if m != nil {
+		return m.Name
+	}
+	return ""
+}
+
 func init() {
-	proto.RegisterType((*DN)(nil), "metadata.DN")
+	proto.RegisterType((*DNShardRecord)(nil), "metadata.DNShardRecord")
+	proto.RegisterType((*DNShard)(nil), "metadata.DNShard")
+	proto.RegisterType((*LogShardRecord)(nil), "metadata.LogShardRecord")
 }
 
 func init() { proto.RegisterFile("metadata.proto", fileDescriptor_56d9f74966f40d04) }
 
 var fileDescriptor_56d9f74966f40d04 = []byte{
-	// 245 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x4c, 0x90, 0x41, 0x4a, 0xc4, 0x30,
-	0x14, 0x86, 0x79, 0x99, 0x99, 0xea, 0x3c, 0x44, 0x24, 0x88, 0x04, 0x91, 0x5a, 0x5c, 0x75, 0xe3,
-	0x04, 0xf1, 0x00, 0xc2, 0xd0, 0x4d, 0x37, 0x2e, 0xea, 0xce, 0xdd, 0xeb, 0x24, 0x66, 0x82, 0xd3,
-	0xa6, 0xa4, 0x19, 0xf0, 0x30, 0x1e, 0xc8, 0xa5, 0x27, 0x10, 0xe9, 0x49, 0x64, 0x32, 0x14, 0xbb,
-	0xfb, 0xbf, 0xef, 0x4f, 0x16, 0xff, 0xc3, 0xf3, 0x46, 0x07, 0x52, 0x14, 0x68, 0xd5, 0x79, 0x17,
-	0x1c, 0x3f, 0x1d, 0xf9, 0xfa, 0xde, 0xd8, 0xb0, 0xdd, 0xd7, 0xab, 0x8d, 0x6b, 0xa4, 0x71, 0xc6,
-	0xc9, 0xf8, 0xa0, 0xde, 0xbf, 0x45, 0x8a, 0x10, 0xd3, 0xf1, 0xe3, 0xdd, 0x27, 0x20, 0x2b, 0x9e,
-	0xf9, 0x15, 0x32, 0xab, 0x04, 0x64, 0x90, 0xcf, 0xd7, 0xc9, 0xf0, 0x73, 0xcb, 0xca, 0xa2, 0x62,
-	0x56, 0xf1, 0x1b, 0x5c, 0x7a, 0xdd, 0xed, 0xec, 0x86, 0xca, 0x42, 0xb0, 0x43, 0x5d, 0xfd, 0x0b,
-	0x9e, 0x22, 0xee, 0x9c, 0x79, 0xd9, 0x92, 0x57, 0x65, 0x21, 0x66, 0xb1, 0x9e, 0x18, 0x7e, 0x89,
-	0x8b, 0x3e, 0x90, 0x0f, 0x62, 0x9e, 0x41, 0x7e, 0x56, 0x1d, 0x81, 0x5f, 0xe0, 0x4c, 0xb7, 0x4a,
-	0x2c, 0xa2, 0x3b, 0x44, 0x2e, 0xf0, 0x84, 0x94, 0xf2, 0xba, 0xef, 0x45, 0x92, 0x41, 0xbe, 0xac,
-	0x46, 0x5c, 0x3f, 0x7d, 0x0d, 0x29, 0x7c, 0x0f, 0x29, 0xfc, 0x0e, 0x29, 0xbc, 0x3e, 0x4c, 0xb6,
-	0x35, 0x14, 0xbc, 0xfd, 0x70, 0xde, 0x1a, 0xdb, 0x8e, 0xd0, 0x6a, 0xd9, 0xbd, 0x1b, 0xd9, 0xd5,
-	0x72, 0x3c, 0x47, 0x9d, 0xc4, 0x99, 0x8f, 0x7f, 0x01, 0x00, 0x00, 0xff, 0xff, 0xf4, 0xf5, 0x5c,
-	0xb9, 0x31, 0x01, 0x00, 0x00,
-}
-
-func (m *DN) Marshal() (dAtA []byte, err error) {
+	// 306 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x91, 0xd1, 0x4a, 0xc3, 0x30,
+	0x14, 0x86, 0x8d, 0x9b, 0x6e, 0x3b, 0xea, 0x18, 0x41, 0xb0, 0x88, 0x74, 0x63, 0x57, 0x43, 0xb0,
+	0x45, 0x7d, 0x00, 0x71, 0x4c, 0x44, 0x90, 0x0a, 0xd9, 0x9d, 0x77, 0x69, 0x93, 0x65, 0x55, 0xdb,
+	0xd4, 0x34, 0x05, 0x9f, 0xc1, 0x27, 0xdb, 0xe5, 0x9e, 0x60, 0x48, 0x9f, 0x44, 0x96, 0x35, 0x73,
+	0xce, 0x0b, 0xef, 0xce, 0xf7, 0x9f, 0x36, 0xff, 0xf9, 0xcf, 0x81, 0x76, 0xc2, 0x35, 0x65, 0x54,
+	0x53, 0x2f, 0x53, 0x52, 0x4b, 0xdc, 0xb4, 0x7c, 0x7a, 0x21, 0x62, 0x3d, 0x2d, 0x42, 0x2f, 0x92,
+	0x89, 0x2f, 0xa4, 0x90, 0xbe, 0xf9, 0x20, 0x2c, 0x26, 0x86, 0x0c, 0x98, 0x6a, 0xf5, 0x63, 0xff,
+	0x1d, 0x8e, 0x46, 0xc1, 0x78, 0x4a, 0x15, 0x23, 0x3c, 0x92, 0x8a, 0x61, 0x07, 0x1a, 0x06, 0x1f,
+	0x46, 0x0e, 0xea, 0xa1, 0x41, 0x9d, 0x58, 0xc4, 0x2e, 0xc0, 0xa3, 0x14, 0xb6, 0xb9, 0x6b, 0x9a,
+	0x1b, 0x0a, 0x3e, 0x86, 0xbd, 0xb1, 0xa6, 0x4a, 0x3b, 0xb5, 0x1e, 0x1a, 0x1c, 0x92, 0x15, 0xe0,
+	0x0e, 0xd4, 0xee, 0x52, 0xe6, 0xd4, 0x8d, 0xb6, 0x2c, 0xfb, 0x9f, 0x08, 0x1a, 0x95, 0x27, 0xbe,
+	0xdf, 0xb2, 0x37, 0x9e, 0x07, 0x57, 0x27, 0xde, 0x3a, 0xdf, 0xaf, 0xf6, 0xb0, 0x39, 0x5b, 0x74,
+	0x77, 0xe6, 0x8b, 0x2e, 0x22, 0x5b, 0x63, 0x9f, 0x41, 0x8b, 0xf0, 0xec, 0x2d, 0x8e, 0xe8, 0x7a,
+	0xb6, 0x1f, 0x61, 0x19, 0xea, 0x96, 0x31, 0xc5, 0xf3, 0xdc, 0x0c, 0xd7, 0x22, 0x16, 0xfb, 0x2f,
+	0xd0, 0xb6, 0x11, 0xfe, 0x5d, 0xc0, 0x39, 0x74, 0x82, 0x22, 0x09, 0xb9, 0x7a, 0x9a, 0x54, 0x4f,
+	0xe7, 0x95, 0xd5, 0x1f, 0x1d, 0x63, 0xa8, 0x07, 0x34, 0xe1, 0x95, 0x9d, 0xa9, 0x87, 0x37, 0xb3,
+	0xd2, 0x45, 0xf3, 0xd2, 0x45, 0x5f, 0xa5, 0x8b, 0x9e, 0x2f, 0x37, 0x0e, 0x95, 0x50, 0xad, 0xe2,
+	0x0f, 0xa9, 0x62, 0x11, 0xa7, 0x16, 0x52, 0xee, 0x67, 0xaf, 0xc2, 0xcf, 0x42, 0xdf, 0xee, 0x22,
+	0xdc, 0x37, 0x37, 0xbb, 0xfe, 0x0e, 0x00, 0x00, 0xff, 0xff, 0x44, 0xd1, 0x9e, 0x99, 0xfe, 0x01,
+	0x00, 0x00,
+}
+
+func (m *DNShardRecord) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
 	n, err := m.MarshalToSizedBuffer(dAtA[:size])
@@ -155,12 +274,12 @@ func (m *DN) Marshal() (dAtA []byte, err error) {
 	return dAtA[:n], nil
 }
 
-func (m *DN) MarshalTo(dAtA []byte) (int, error) {
+func (m *DNShardRecord) MarshalTo(dAtA []byte) (int, error) {
 	size := m.Size()
 	return m.MarshalToSizedBuffer(dAtA[:size])
 }
 
-func (m *DN) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+func (m *DNShardRecord) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	i := len(dAtA)
 	_ = i
 	var l int
@@ -169,39 +288,120 @@ func (m *DN) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 		i -= len(m.XXX_unrecognized)
 		copy(dAtA[i:], m.XXX_unrecognized)
 	}
-	if len(m.Address) > 0 {
-		i -= len(m.Address)
-		copy(dAtA[i:], m.Address)
-		i = encodeVarintMetadata(dAtA, i, uint64(len(m.Address)))
-		i--
-		dAtA[i] = 0x32
-	}
 	if len(m.End) > 0 {
 		i -= len(m.End)
 		copy(dAtA[i:], m.End)
 		i = encodeVarintMetadata(dAtA, i, uint64(len(m.End)))
 		i--
-		dAtA[i] = 0x2a
+		dAtA[i] = 0x22
 	}
 	if len(m.Start) > 0 {
 		i -= len(m.Start)
 		copy(dAtA[i:], m.Start)
 		i = encodeVarintMetadata(dAtA, i, uint64(len(m.Start)))
 		i--
-		dAtA[i] = 0x22
+		dAtA[i] = 0x1a
 	}
 	if m.LogShardID != 0 {
 		i = encodeVarintMetadata(dAtA, i, uint64(m.LogShardID))
 		i--
-		dAtA[i] = 0x18
+		dAtA[i] = 0x10
+	}
+	if m.ShardID != 0 {
+		i = encodeVarintMetadata(dAtA, i, uint64(m.ShardID))
+		i--
+		dAtA[i] = 0x8
+	}
+	return len(dAtA) - i, nil
+}
+
+func (m *DNShard) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *DNShard) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *DNShard) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Address) > 0 {
+		i -= len(m.Address)
+		copy(dAtA[i:], m.Address)
+		i = encodeVarintMetadata(dAtA, i, uint64(len(m.Address)))
+		i--
+		dAtA[i] = 0x1a
 	}
 	if m.ReplicaID != 0 {
 		i = encodeVarintMetadata(dAtA, i, uint64(m.ReplicaID))
 		i--
 		dAtA[i] = 0x10
 	}
-	if m.ID != 0 {
-		i = encodeVarintMetadata(dAtA, i, uint64(m.ID))
+	{
+		size, err := m.DNShardRecord.MarshalToSizedBuffer(dAtA[:i])
+		if err != nil {
+			return 0, err
+		}
+		i -= size
+		i = encodeVarintMetadata(dAtA, i, uint64(size))
+	}
+	i--
+	dAtA[i] = 0xa
+	return len(dAtA) - i, nil
+}
+
+func (m *LogShardRecord) Marshal() (dAtA []byte, err error) {
+	size := m.Size()
+	dAtA = make([]byte, size)
+	n, err := m.MarshalToSizedBuffer(dAtA[:size])
+	if err != nil {
+		return nil, err
+	}
+	return dAtA[:n], nil
+}
+
+func (m *LogShardRecord) MarshalTo(dAtA []byte) (int, error) {
+	size := m.Size()
+	return m.MarshalToSizedBuffer(dAtA[:size])
+}
+
+func (m *LogShardRecord) MarshalToSizedBuffer(dAtA []byte) (int, error) {
+	i := len(dAtA)
+	_ = i
+	var l int
+	_ = l
+	if m.XXX_unrecognized != nil {
+		i -= len(m.XXX_unrecognized)
+		copy(dAtA[i:], m.XXX_unrecognized)
+	}
+	if len(m.Name) > 0 {
+		i -= len(m.Name)
+		copy(dAtA[i:], m.Name)
+		i = encodeVarintMetadata(dAtA, i, uint64(len(m.Name)))
+		i--
+		dAtA[i] = 0x1a
+	}
+	if m.NumberOfReplicas != 0 {
+		i = encodeVarintMetadata(dAtA, i, uint64(m.NumberOfReplicas))
+		i--
+		dAtA[i] = 0x10
+	}
+	if m.ShardID != 0 {
+		i = encodeVarintMetadata(dAtA, i, uint64(m.ShardID))
 		i--
 		dAtA[i] = 0x8
 	}
@@ -219,17 +419,14 @@ func encodeVarintMetadata(dAtA []byte, offset int, v uint64) int {
 	dAtA[offset] = uint8(v)
 	return base
 }
-func (m *DN) Size() (n int) {
+func (m *DNShardRecord) Size() (n int) {
 	if m == nil {
 		return 0
 	}
 	var l int
 	_ = l
-	if m.ID != 0 {
-		n += 1 + sovMetadata(uint64(m.ID))
-	}
-	if m.ReplicaID != 0 {
-		n += 1 + sovMetadata(uint64(m.ReplicaID))
+	if m.ShardID != 0 {
+		n += 1 + sovMetadata(uint64(m.ShardID))
 	}
 	if m.LogShardID != 0 {
 		n += 1 + sovMetadata(uint64(m.LogShardID))
@@ -242,6 +439,23 @@ func (m *DN) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovMetadata(uint64(l))
 	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
+func (m *DNShard) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	l = m.DNShardRecord.Size()
+	n += 1 + l + sovMetadata(uint64(l))
+	if m.ReplicaID != 0 {
+		n += 1 + sovMetadata(uint64(m.ReplicaID))
+	}
 	l = len(m.Address)
 	if l > 0 {
 		n += 1 + l + sovMetadata(uint64(l))
@@ -252,13 +466,35 @@ func (m *DN) Size() (n int) {
 	return n
 }
 
+func (m *LogShardRecord) Size() (n int) {
+	if m == nil {
+		return 0
+	}
+	var l int
+	_ = l
+	if m.ShardID != 0 {
+		n += 1 + sovMetadata(uint64(m.ShardID))
+	}
+	if m.NumberOfReplicas != 0 {
+		n += 1 + sovMetadata(uint64(m.NumberOfReplicas))
+	}
+	l = len(m.Name)
+	if l > 0 {
+		n += 1 + l + sovMetadata(uint64(l))
+	}
+	if m.XXX_unrecognized != nil {
+		n += len(m.XXX_unrecognized)
+	}
+	return n
+}
+
 func sovMetadata(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
 func sozMetadata(x uint64) (n int) {
 	return sovMetadata(uint64((x << 1) ^ uint64((int64(x) >> 63))))
 }
-func (m *DN) Unmarshal(dAtA []byte) error {
+func (m *DNShardRecord) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
 	for iNdEx < l {
@@ -281,17 +517,17 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 		fieldNum := int32(wire >> 3)
 		wireType := int(wire & 0x7)
 		if wireType == 4 {
-			return fmt.Errorf("proto: DN: wiretype end group for non-group")
+			return fmt.Errorf("proto: DNShardRecord: wiretype end group for non-group")
 		}
 		if fieldNum <= 0 {
-			return fmt.Errorf("proto: DN: illegal tag %d (wire type %d)", fieldNum, wire)
+			return fmt.Errorf("proto: DNShardRecord: illegal tag %d (wire type %d)", fieldNum, wire)
 		}
 		switch fieldNum {
 		case 1:
 			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ID", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field ShardID", wireType)
 			}
-			m.ID = 0
+			m.ShardID = 0
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowMetadata
@@ -301,16 +537,16 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ID |= uint64(b&0x7F) << shift
+				m.ShardID |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
 		case 2:
 			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field ReplicaID", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field LogShardID", wireType)
 			}
-			m.ReplicaID = 0
+			m.LogShardID = 0
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowMetadata
@@ -320,16 +556,16 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.ReplicaID |= uint64(b&0x7F) << shift
+				m.LogShardID |= uint64(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
 		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field LogShardID", wireType)
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Start", wireType)
 			}
-			m.LogShardID = 0
+			var byteLen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowMetadata
@@ -339,14 +575,29 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				m.LogShardID |= uint64(b&0x7F) << shift
+				byteLen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
+			if byteLen < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			postIndex := iNdEx + byteLen
+			if postIndex < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Start = append(m.Start[:0], dAtA[iNdEx:postIndex]...)
+			if m.Start == nil {
+				m.Start = []byte{}
+			}
+			iNdEx = postIndex
 		case 4:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Start", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field End", wireType)
 			}
 			var byteLen int
 			for shift := uint(0); ; shift += 7 {
@@ -373,16 +624,67 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.Start = append(m.Start[:0], dAtA[iNdEx:postIndex]...)
-			if m.Start == nil {
-				m.Start = []byte{}
+			m.End = append(m.End[:0], dAtA[iNdEx:postIndex]...)
+			if m.End == nil {
+				m.End = []byte{}
 			}
 			iNdEx = postIndex
-		case 5:
+		default:
+			iNdEx = preIndex
+			skippy, err := skipMetadata(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
+func (m *DNShard) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowMetadata
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: DNShard: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: DNShard: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
 			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field End", wireType)
+				return fmt.Errorf("proto: wrong wireType = %d for field DNShardRecord", wireType)
 			}
-			var byteLen int
+			var msglen int
 			for shift := uint(0); ; shift += 7 {
 				if shift >= 64 {
 					return ErrIntOverflowMetadata
@@ -392,27 +694,45 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 				}
 				b := dAtA[iNdEx]
 				iNdEx++
-				byteLen |= int(b&0x7F) << shift
+				msglen |= int(b&0x7F) << shift
 				if b < 0x80 {
 					break
 				}
 			}
-			if byteLen < 0 {
+			if msglen < 0 {
 				return ErrInvalidLengthMetadata
 			}
-			postIndex := iNdEx + byteLen
+			postIndex := iNdEx + msglen
 			if postIndex < 0 {
 				return ErrInvalidLengthMetadata
 			}
 			if postIndex > l {
 				return io.ErrUnexpectedEOF
 			}
-			m.End = append(m.End[:0], dAtA[iNdEx:postIndex]...)
-			if m.End == nil {
-				m.End = []byte{}
+			if err := m.DNShardRecord.Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
+				return err
 			}
 			iNdEx = postIndex
-		case 6:
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ReplicaID", wireType)
+			}
+			m.ReplicaID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMetadata
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ReplicaID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
 			if wireType != 2 {
 				return fmt.Errorf("proto: wrong wireType = %d for field Address", wireType)
 			}
@@ -466,6 +786,127 @@ func (m *DN) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
+func (m *LogShardRecord) Unmarshal(dAtA []byte) error {
+	l := len(dAtA)
+	iNdEx := 0
+	for iNdEx < l {
+		preIndex := iNdEx
+		var wire uint64
+		for shift := uint(0); ; shift += 7 {
+			if shift >= 64 {
+				return ErrIntOverflowMetadata
+			}
+			if iNdEx >= l {
+				return io.ErrUnexpectedEOF
+			}
+			b := dAtA[iNdEx]
+			iNdEx++
+			wire |= uint64(b&0x7F) << shift
+			if b < 0x80 {
+				break
+			}
+		}
+		fieldNum := int32(wire >> 3)
+		wireType := int(wire & 0x7)
+		if wireType == 4 {
+			return fmt.Errorf("proto: LogShardRecord: wiretype end group for non-group")
+		}
+		if fieldNum <= 0 {
+			return fmt.Errorf("proto: LogShardRecord: illegal tag %d (wire type %d)", fieldNum, wire)
+		}
+		switch fieldNum {
+		case 1:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field ShardID", wireType)
+			}
+			m.ShardID = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMetadata
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.ShardID |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 2:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field NumberOfReplicas", wireType)
+			}
+			m.NumberOfReplicas = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMetadata
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.NumberOfReplicas |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+		case 3:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field Name", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowMetadata
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.Name = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		default:
+			iNdEx = preIndex
+			skippy, err := skipMetadata(dAtA[iNdEx:])
+			if err != nil {
+				return err
+			}
+			if (skippy < 0) || (iNdEx+skippy) < 0 {
+				return ErrInvalidLengthMetadata
+			}
+			if (iNdEx + skippy) > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
+			iNdEx += skippy
+		}
+	}
+
+	if iNdEx > l {
+		return io.ErrUnexpectedEOF
+	}
+	return nil
+}
 func skipMetadata(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/pkg/pb/plan/plan.pb.go b/pkg/pb/plan/plan.pb.go
index ab1723d87..4fa59cab7 100644
--- a/pkg/pb/plan/plan.pb.go
+++ b/pkg/pb/plan/plan.pb.go
@@ -6785,8 +6785,6 @@ func (m *Node) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 			i--
 			dAtA[i] = 0x5a
 		}
-		i--
-		dAtA[i] = 0x6a
 	}
 	if len(m.GroupingSet) > 0 {
 		for iNdEx := len(m.GroupingSet) - 1; iNdEx >= 0; iNdEx-- {
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)
 }
diff --git a/pkg/sql/plan2/base_binder.go b/pkg/sql/plan2/base_binder.go
index 199bbd825..3d9180fbf 100644
--- a/pkg/sql/plan2/base_binder.go
+++ b/pkg/sql/plan2/base_binder.go
@@ -116,7 +116,7 @@ func (b *baseBinder) baseBindExpr(astExpr tree.Expr, depth int32) (expr *Expr, e
 	case *tree.MaxValue:
 		err = errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("expr max'%v' is not support now", exprImpl))
 	case *tree.VarExpr:
-		err = errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("expr var'%v' is not support now", exprImpl))
+		expr, err = b.baseBindVar(exprImpl, depth)
 	case *tree.StrVal:
 		err = errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("expr str'%v' is not support now", exprImpl))
 	case *tree.ExprList:
@@ -132,6 +132,123 @@ func (b *baseBinder) baseBindExpr(astExpr tree.Expr, depth int32) (expr *Expr, e
 	return
 }
 
+func (b *baseBinder) baseBindVar(astExpr *tree.VarExpr, depth int32) (expr *plan.Expr, err error) {
+	var getVal interface{}
+	getVal, err = b.builder.compCtx.ResolveVariable(astExpr.Name, astExpr.System, astExpr.Global)
+	if err != nil {
+		return nil, err
+	}
+	getIntExpr := func(data int64) *plan.Expr {
+		return &Expr{
+			Expr: &plan.Expr_C{
+				C: &Const{
+					Isnull: false,
+					Value: &plan.Const_Ival{
+						Ival: data,
+					},
+				},
+			},
+			Typ: &plan.Type{
+				Id:       plan.Type_INT64,
+				Nullable: false,
+				Size:     8,
+			},
+		}
+	}
+	getFloatExpr := func(data float64) *plan.Expr {
+		return &Expr{
+			Expr: &plan.Expr_C{
+				C: &Const{
+					Isnull: false,
+					Value: &plan.Const_Dval{
+						Dval: data,
+					},
+				},
+			},
+			Typ: &plan.Type{
+				Id:       plan.Type_FLOAT64,
+				Nullable: false,
+				Size:     8,
+			},
+		}
+	}
+
+	switch val := getVal.(type) {
+	case string:
+		expr = &Expr{
+			Expr: &plan.Expr_C{
+				C: &Const{
+					Isnull: false,
+					Value: &plan.Const_Sval{
+						Sval: val,
+					},
+				},
+			},
+			Typ: &plan.Type{
+				Id:       plan.Type_VARCHAR,
+				Nullable: false,
+				Size:     4,
+				Width:    math.MaxInt32,
+			},
+		}
+	case int:
+		expr = getIntExpr(int64(val))
+	case uint8:
+		expr = getIntExpr(int64(val))
+	case uint16:
+		expr = getIntExpr(int64(val))
+	case uint32:
+		expr = getIntExpr(int64(val))
+	case int8:
+		expr = getIntExpr(int64(val))
+	case int16:
+		expr = getIntExpr(int64(val))
+	case int32:
+		expr = getIntExpr(int64(val))
+	case int64:
+		expr = getIntExpr(val)
+	case uint64:
+		err = errors.New(errno.SyntaxErrororAccessRuleViolation, "decimal var not support now")
+	case float32:
+		expr = getFloatExpr(float64(val))
+	case float64:
+		expr = getFloatExpr(val)
+	case bool:
+		return &Expr{
+			Expr: &plan.Expr_C{
+				C: &Const{
+					Isnull: false,
+					Value: &plan.Const_Bval{
+						Bval: val,
+					},
+				},
+			},
+			Typ: &plan.Type{
+				Id:       plan.Type_BOOL,
+				Nullable: false,
+				Size:     1,
+			},
+		}, nil
+	case nil:
+		expr = &Expr{
+			Expr: &plan.Expr_C{
+				C: &Const{
+					Isnull: true,
+				},
+			},
+			Typ: &plan.Type{
+				Id:       plan.Type_ANY,
+				Nullable: true,
+			},
+		}
+	case types.Decimal64, types.Decimal128:
+		err = errors.New(errno.SyntaxErrororAccessRuleViolation, "decimal var not support now")
+	default:
+		err = errors.New(errno.SyntaxErrororAccessRuleViolation, fmt.Sprintf("type of var %q is not support now", astExpr.Name))
+	}
+	return
+}
+
 func (b *baseBinder) baseBindColRef(astExpr *tree.UnresolvedName, depth int32) (expr *plan.Expr, err error) {
 	col := astExpr.Parts[0]
 	table := astExpr.Parts[1]
@@ -483,24 +600,19 @@ func (b *baseBinder) bindFuncExprImplByPlanExpr(name string, args []*Expr) (*pla
 		if args[0].Typ.Id == plan.Type_DATE && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[0], args[1])
-		}
-		if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_DATE {
+		} else if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_DATE {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[1], args[0])
-		}
-		if args[0].Typ.Id == plan.Type_DATETIME && args[1].Typ.Id == plan.Type_INTERVAL {
+		} else if args[0].Typ.Id == plan.Type_DATETIME && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[0], args[1])
-		}
-		if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_DATETIME {
+		} else if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_DATETIME {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[1], args[0])
-		}
-		if args[0].Typ.Id == plan.Type_VARCHAR && args[1].Typ.Id == plan.Type_INTERVAL {
+		} else if args[0].Typ.Id == plan.Type_VARCHAR && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[0], args[1])
-		}
-		if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_VARCHAR {
+		} else if args[0].Typ.Id == plan.Type_INTERVAL && args[1].Typ.Id == plan.Type_VARCHAR {
 			name = "date_add"
 			args, err = resetDateFunctionArgs(args[1], args[0])
 		}
@@ -512,12 +624,10 @@ func (b *baseBinder) bindFuncExprImplByPlanExpr(name string, args []*Expr) (*pla
 		if args[0].Typ.Id == plan.Type_DATE && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_sub"
 			args, err = resetDateFunctionArgs(args[0], args[1])
-		}
-		if args[0].Typ.Id == plan.Type_DATETIME && args[1].Typ.Id == plan.Type_INTERVAL {
+		} else if args[0].Typ.Id == plan.Type_DATETIME && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_sub"
 			args, err = resetDateFunctionArgs(args[0], args[1])
-		}
-		if args[0].Typ.Id == plan.Type_VARCHAR && args[1].Typ.Id == plan.Type_INTERVAL {
+		} else if args[0].Typ.Id == plan.Type_VARCHAR && args[1].Typ.Id == plan.Type_INTERVAL {
 			name = "date_sub"
 			args, err = resetDateFunctionArgs(args[0], args[1])
 		}
diff --git a/pkg/sql/plan2/function/builtin/binary/extract.go b/pkg/sql/plan2/function/builtin/binary/extract.go
new file mode 100644
index 000000000..c5850e745
--- /dev/null
+++ b/pkg/sql/plan2/function/builtin/binary/extract.go
@@ -0,0 +1,110 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 binary
+
+import (
+	"errors"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/encoding"
+	"github.com/matrixorigin/matrixone/pkg/vectorize/extract"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+/*
+// when implicit cast from varchar to date is ready, get rid of this
+func ExtractFromString(vectors []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	left, right := vectors[0], vectors[1]
+	resultType := types.Type{Oid: types.T_uint32, Size: 4}
+	resultElementSize := int(resultType.Size)
+	switch {
+	case left.IsScalar() && right.IsScalar():
+		if left.ConstVectorIsNull() || right.ConstVectorIsNull() {
+			return proc.AllocScalarNullVector(resultType), nil
+		}
+		leftValues, rightValues := left.Col.(*types.Bytes), right.Col.(*types.Bytes)
+		resultVector := vector.NewConst(resultType)
+		resultValues := make([]uint32, 1)
+		unit := string(leftValues.Data)
+		inputDate, err := types.ParseDate(string(rightValues.Get(0)))
+		if err != nil {
+			return nil, errors.New("invalid input")
+		}
+		resultValues, err = extract.ExtractFromDate(unit, []types.Date{inputDate}, resultValues)
+		if err != nil {
+			return nil, errors.New("invalid input")
+		}
+		vector.SetCol(resultVector, resultValues)
+		return resultVector, nil
+	case left.IsScalar() && !right.IsScalar():
+		if left.ConstVectorIsNull() {
+			return proc.AllocScalarNullVector(resultType), nil
+		}
+		leftValues, rightValues := left.Col.(*types.Bytes), right.Col.(*types.Bytes)
+		unit := string(leftValues.Data)
+		resultValues, err := proc.AllocVector(resultType, int64(resultElementSize) * int64(len(rightValues.Lengths)))
+		if
+
+		result, resultNsp, err := extract.ExtractFromInputBytes(unit, rightValues, right.Nsp, )
+
+	default:
+		return nil, errors.New("invalid input")
+	}
+}
+*/
+
+func ExtractFromDate(vectors []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	left, right := vectors[0], vectors[1]
+	resultType := types.Type{Oid: types.T_uint32, Size: 4}
+	resultElementSize := int(resultType.Size)
+	switch {
+	case left.IsScalar() && right.IsScalar():
+		if left.ConstVectorIsNull() || right.ConstVectorIsNull() {
+			return proc.AllocScalarNullVector(resultType), nil
+		}
+		leftValues, rightValues := left.Col.(*types.Bytes), right.Col.([]types.Date)
+		resultVector := vector.NewConst(resultType)
+		resultValues := make([]uint32, 1)
+		unit := string(leftValues.Get(0))
+		results, err := extract.ExtractFromDate(unit, rightValues, resultValues)
+		if err != nil {
+			return nil, errors.New("invalid input")
+		}
+		vector.SetCol(resultVector, results)
+		return resultVector, nil
+	case left.IsScalar() && !right.IsScalar():
+		if left.ConstVectorIsNull() {
+			return proc.AllocScalarNullVector(resultType), nil
+		}
+		leftValues, rightValues := left.Col.(*types.Bytes), right.Col.([]types.Date)
+		resultVector, err := proc.AllocVector(resultType, int64(resultElementSize*len(rightValues)))
+		if err != nil {
+			return nil, err
+		}
+		resultValues := encoding.DecodeUint32Slice(resultVector.Data)
+		resultValues = resultValues[:len(rightValues)]
+		unit := string(leftValues.Get(0))
+		results, err := extract.ExtractFromDate(unit, rightValues, resultValues)
+		if err != nil {
+			return nil, err
+		}
+		nulls.Set(resultVector.Nsp, right.Nsp)
+		vector.SetCol(resultVector, results)
+		return resultVector, nil
+	default:
+		return nil, errors.New("invalid input")
+	}
+}
diff --git a/pkg/sql/plan2/function/buitins.go b/pkg/sql/plan2/function/buitins.go
index f95befd29..a06add0be 100644
--- a/pkg/sql/plan2/function/buitins.go
+++ b/pkg/sql/plan2/function/buitins.go
@@ -330,6 +330,17 @@ var builtins = map[int][]Function{
 			Fn:          unary.Exp[float64],
 		},
 	},
+	EXTRACT: {
+		{
+			Index:       0,
+			Flag:        plan.Function_STRICT,
+			Layout:      STANDARD_FUNCTION,
+			Args:        []types.T{types.T_varchar, types.T_date},
+			ReturnTyp:   types.T_uint32,
+			TypeCheckFn: strictTypeCheck,
+			Fn:          binary.ExtractFromDate,
+		},
+	},
 	LENGTH: {
 		{
 			Index:       0,
@@ -1711,26 +1722,29 @@ var builtins = map[int][]Function{
 			Fn:          multi.UTCTimestamp,
 		},
 	},
-	EXTRACT: {
-		{
-			Index:       0,
-			Flag:        plan.Function_STRICT,
-			Layout:      STANDARD_FUNCTION,
-			Args:        []types.T{types.T_varchar, types.T_date},
-			ReturnTyp:   types.T_float64,
-			TypeCheckFn: strictTypeCheck,
-			Fn:          nil,
+	/*
+		EXTRACT: {
+			{
+				Index:       0,
+				Flag:        plan.Function_STRICT,
+				Layout:      STANDARD_FUNCTION,
+				Args:        []types.T{types.T_varchar, types.T_date},
+				ReturnTyp:   types.T_float64,
+				TypeCheckFn: strictTypeCheck,
+				Fn:          nil,
+			},
+			{
+				Index:       1,
+				Flag:        plan.Function_STRICT,
+				Layout:      STANDARD_FUNCTION,
+				Args:        []types.T{types.T_varchar, types.T_datetime},
+				ReturnTyp:   types.T_float64,
+				TypeCheckFn: strictTypeCheck,
+				Fn:          nil,
+			},
 		},
-		{
-			Index:       1,
-			Flag:        plan.Function_STRICT,
-			Layout:      STANDARD_FUNCTION,
-			Args:        []types.T{types.T_varchar, types.T_datetime},
-			ReturnTyp:   types.T_float64,
-			TypeCheckFn: strictTypeCheck,
-			Fn:          nil,
-		},
-	},
+
+	*/
 	DATE_ADD: {
 		{
 			Index:       0,
diff --git a/pkg/sql/plan2/function/operator/cast.go b/pkg/sql/plan2/function/operator/cast.go
index 99f82cc35..2aa8ada65 100644
--- a/pkg/sql/plan2/function/operator/cast.go
+++ b/pkg/sql/plan2/function/operator/cast.go
@@ -414,6 +414,11 @@ func Cast(vs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
 	if lv.Typ.Oid == types.T_timestamp && rv.Typ.Oid == types.T_datetime {
 		return castTimeStampAsDatetime(lv, rv, proc)
 	}
+
+	if lv.Typ.Oid == types.T_timestamp && rv.Typ.Oid == types.T_varchar {
+		return castTimestampAsVarchar(lv, rv, proc)
+	}
+
 	return nil, errors.New(errno.SyntaxErrororAccessRuleViolation, "parameter types of cast function do not match")
 }
 
@@ -927,6 +932,43 @@ func castTimeStampAsDatetime(lv, rv *vector.Vector, proc *process.Process) (*vec
 	return vec, nil
 }
 
+//  castTimestampAsVarchar : Cast converts timestamp to varchar
+func castTimestampAsVarchar(lv, rv *vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	lvs := lv.Col.([]types.Timestamp)
+	resultType := rv.Typ
+	resultElementSize := int(resultType.Size)
+	if lv.IsScalar() {
+		vec := proc.AllocScalarVector(resultType)
+		rs := &types.Bytes{
+			Data:    []byte{},
+			Offsets: make([]uint32, 1),
+			Lengths: make([]uint32, 1),
+		}
+		if _, err := typecast.TimestampToVarchar(lvs, rs); err != nil {
+			return nil, err
+		}
+		nulls.Set(vec.Nsp, lv.Nsp)
+		vector.SetCol(vec, rs)
+		return vec, nil
+	}
+
+	vec, err := proc.AllocVector(resultType, int64(resultElementSize*len(lvs)))
+	if err != nil {
+		return nil, err
+	}
+	rs := &types.Bytes{
+		Data:    []byte{},
+		Offsets: make([]uint32, len(lvs)),
+		Lengths: make([]uint32, len(lvs)),
+	}
+	if _, err := typecast.TimestampToVarchar(lvs, rs); err != nil {
+		return nil, err
+	}
+	nulls.Set(vec.Nsp, lv.Nsp)
+	vector.SetCol(vec, rs)
+	return vec, nil
+}
+
 // CastStringAsDecimal64: onverts char/varchar as decimal64
 func CastStringAsDecimal64(lv, rv *vector.Vector, proc *process.Process) (*vector.Vector, error) {
 	resultType := rv.Typ
diff --git a/pkg/sql/plan2/function/operators.go b/pkg/sql/plan2/function/operators.go
index 893980a49..8be6c777c 100644
--- a/pkg/sql/plan2/function/operators.go
+++ b/pkg/sql/plan2/function/operators.go
@@ -3720,6 +3720,15 @@ var operators = map[int][]Function{
 			TypeCheckFn: strictTypeCheck,
 			Fn:          operator.Cast,
 		},
+		{
+			Index:       167,
+			Flag:        plan.Function_STRICT,
+			Layout:      CAST_EXPRESSION,
+			Args:        []types.T{types.T_timestamp, types.T_varchar},
+			ReturnTyp:   types.T_varchar,
+			TypeCheckFn: strictTypeCheck,
+			Fn:          operator.Cast,
+		},
 	},
 	CASE: {
 		{
diff --git a/pkg/sql/plan2/mock.go b/pkg/sql/plan2/mock.go
index d71720883..511a25934 100644
--- a/pkg/sql/plan2/mock.go
+++ b/pkg/sql/plan2/mock.go
@@ -15,8 +15,10 @@
 package plan2
 
 import (
+	"errors"
 	"strings"
 
+	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
 )
@@ -27,7 +29,20 @@ type MockCompilerContext struct {
 }
 
 func (m *MockCompilerContext) ResolveVariable(varName string, isSystemVar, isGlobalVar bool) (interface{}, error) {
-	return nil, nil
+	vars := make(map[string]interface{})
+	vars["str_var"] = "str"
+	vars["int_var"] = 20
+	vars["bool_var"] = false
+	vars["float_var"] = 20.20
+	dec, _ := types.ParseStringToDecimal128("200.001", 2, 2)
+	vars["decimal_var"] = dec
+	vars["null_var"] = nil
+
+	if result, ok := vars[varName]; ok {
+		return result, nil
+	}
+
+	return nil, errors.New("var not found")
 }
 
 type col struct {
diff --git a/pkg/sql/plan2/query_builder.go b/pkg/sql/plan2/query_builder.go
index 1b5c217ab..dbece2768 100644
--- a/pkg/sql/plan2/query_builder.go
+++ b/pkg/sql/plan2/query_builder.go
@@ -83,7 +83,10 @@ func (builder *QueryBuilder) resetNode(nodeId int32) (map[int64][2]int32, error)
 			returnMap[getColMapKey(tag, int32(idx))] = [2]int32{0, int32(idx)}
 		}
 		for _, expr := range node.WhereList {
-			builder.resetPosition(expr, returnMap)
+			err := builder.resetPosition(expr, returnMap)
+			if err != nil {
+				return nil, err
+			}
 		}
 	case plan.Node_JOIN:
 		// TODO deal with using
diff --git a/pkg/vectorize/extract/extract.go b/pkg/vectorize/extract/extract.go
new file mode 100644
index 000000000..5946ab230
--- /dev/null
+++ b/pkg/vectorize/extract/extract.go
@@ -0,0 +1,99 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 extract
+
+import (
+	"errors"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+)
+
+//func extractFromDate()
+
+var validDateUnit = map[string]struct{}{
+	"year":       {},
+	"month":      {},
+	"day":        {},
+	"year_month": {},
+	"quarter":    {},
+}
+
+/*
+func ExtractFromInputBytes(unit string, inputBytes *types.Bytes, inputNsp *nulls.Nulls, results []uint32) ([]uint32, *nulls.Nulls, error) {
+	resultNsp := new(nulls.Nulls)
+	if _, ok := validDateUnit[unit]; !ok {
+		return []uint32{}, nil, errors.New("invalid unit")
+	}
+	for i := range inputBytes.Lengths {
+		if nulls.Contains(inputNsp, uint64(i)) {
+			nulls.Add(resultNsp, uint64(i))
+			continue
+		}
+		inputValue := string(inputBytes.Get(int64(i)))
+		date, err := types.ParseDate(inputValue)
+		if err != nil {
+			return []uint32{}, nil, errors.New("invalid date string")
+		}
+		results[i] = ExtractFromOneDate(unit, date)
+	}
+	return results, resultNsp, nil
+}
+
+*/
+
+func ExtractFromOneDate(unit string, date types.Date) uint32 {
+	switch unit {
+	case "day":
+		return uint32(date.Day())
+	case "month":
+		return uint32(date.Month())
+	case "quarter":
+		return date.Quarter()
+	case "year":
+		return uint32(date.Year())
+	case "year_month":
+		return date.YearMonth()
+	default:
+		return 0
+	}
+}
+
+func ExtractFromDate(unit string, dates []types.Date, results []uint32) ([]uint32, error) {
+	if _, ok := validDateUnit[unit]; !ok {
+		return []uint32{}, errors.New("invalid unit")
+	}
+	switch unit {
+	case "day":
+		for i, d := range dates {
+			results[i] = uint32(d.Day())
+		}
+	case "month":
+		for i, d := range dates {
+			results[i] = uint32(d.Month())
+		}
+	case "year":
+		for i, d := range dates {
+			results[i] = uint32(d.Year())
+		}
+	case "year_month":
+		for i, d := range dates {
+			results[i] = d.YearMonth()
+		}
+	case "quarter":
+		for i, d := range dates {
+			results[i] = d.Quarter()
+		}
+	}
+	return results, nil
+}
diff --git a/pkg/vectorize/typecast/typecast.go b/pkg/vectorize/typecast/typecast.go
index 36c35d69f..eef94d371 100644
--- a/pkg/vectorize/typecast/typecast.go
+++ b/pkg/vectorize/typecast/typecast.go
@@ -156,6 +156,7 @@ var (
 	Uint64ToDecimal128 = UintToDecimal128[uint64]
 
 	TimestampToDatetime = timestampToDatetime
+	TimestampToVarchar  = timestampToVarchar
 )
 
 func NumericToNumeric[T1, T2 constraints.Integer | constraints.Float](xs []T1, rs []T2) ([]T2, error) {
@@ -243,3 +244,16 @@ func UintToDecimal128[T constraints.Integer](xs []T, rs []types.Decimal128) ([]t
 func timestampToDatetime(xs []types.Timestamp, rs []types.Datetime) ([]types.Datetime, error) {
 	return types.TimestampToDatetime(xs, rs)
 }
+
+func timestampToVarchar(xs []types.Timestamp, rs *types.Bytes) (*types.Bytes, error) {
+	oldLen := uint32(0)
+	for i, x := range xs {
+		//todo: pass precision arg?
+		rs.Data = append(rs.Data, []byte(x.String())...)
+		newLen := uint32(len(rs.Data))
+		rs.Offsets[i] = oldLen
+		rs.Lengths[i] = newLen - oldLen
+		oldLen = newLen
+	}
+	return rs, nil
+}
diff --git a/pkg/vm/engine/tae/db/replay.go b/pkg/vm/engine/tae/db/replay.go
index 9e7e8617e..49251277f 100644
--- a/pkg/vm/engine/tae/db/replay.go
+++ b/pkg/vm/engine/tae/db/replay.go
@@ -2,6 +2,10 @@ package db
 
 import (
 	"bytes"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
 	"sync"
 
 	"github.com/RoaringBitmap/roaring"
@@ -66,6 +70,26 @@ func (replayer *Replayer) PreReplayWal() {
 	}
 }
 
+func (replayer *Replayer) scanFiles() map[uint64]string {
+	files := make(map[uint64]string)
+	infos, err := ioutil.ReadDir(replayer.db.Dir)
+	if err != nil {
+		panic(err)
+	}
+	for _, info := range infos {
+		if info.IsDir() {
+			continue
+		}
+		name := info.Name()
+		id, err := replayer.db.FileFactory.DecodeName(name)
+		if err != nil {
+			continue
+		}
+		files[id] = path.Join(replayer.db.Dir, name)
+	}
+	return files
+}
+
 func (replayer *Replayer) Replay() {
 	if err := replayer.db.Wal.Replay(replayer.OnReplayEntry); err != nil {
 		panic(err)
@@ -77,6 +101,7 @@ func (replayer *Replayer) Replay() {
 }
 
 func (replayer *Replayer) PostReplayWal() {
+	activeSegs := make(map[uint64]*catalog.SegmentEntry)
 	processor := new(catalog.LoopProcessor)
 	processor.DatabaseFn = func(entry *catalog.DBEntry) (err error) {
 		if entry.IsActive() {
@@ -106,9 +131,15 @@ func (replayer *Replayer) PostReplayWal() {
 	}
 	processor.SegmentFn = func(entry *catalog.SegmentEntry) (err error) {
 		if entry.IsActive() {
+			if !entry.GetTable().IsVirtual() {
+				activeSegs[entry.ID] = entry
+			}
 			return
 		}
 		if entry.GetLogIndex().LSN > replayer.db.Wal.GetCheckpointed() {
+			if !entry.GetTable().IsVirtual() {
+				activeSegs[entry.ID] = entry
+			}
 			return
 		}
 		if err = gcSegmentClosure(entry, GCType_Segment)(); err != nil {
@@ -130,6 +161,23 @@ func (replayer *Replayer) PostReplayWal() {
 		return
 	}
 	_ = replayer.db.Catalog.RecurLoop(processor)
+
+	files := replayer.scanFiles()
+	for id := range activeSegs {
+		_, ok := files[id]
+		if !ok {
+			panic(fmt.Errorf("Cannot find segment file for: %d", id))
+		}
+		delete(files, id)
+	}
+	for _, file := range files {
+		logutil.Info("[Replay]", common.OperationField("clean-segment"),
+			common.OperandField(file))
+		if err := os.Remove(file); err != nil {
+			panic(err)
+		}
+	}
+
 	logutil.Info(replayer.db.Catalog.SimplePPString(common.PPL1))
 }
 
diff --git a/pkg/vm/engine/tae/moengine/relation.go b/pkg/vm/engine/tae/moengine/relation.go
index 84e64fe65..c4490b792 100644
--- a/pkg/vm/engine/tae/moengine/relation.go
+++ b/pkg/vm/engine/tae/moengine/relation.go
@@ -15,7 +15,6 @@
 package moengine
 
 import (
-	"errors"
 	"github.com/matrixorigin/matrixone/pkg/container/batch"
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
@@ -137,8 +136,23 @@ func (rel *txnRelation) Write(_ uint64, bat *batch.Batch, _ engine.Snapshot) err
 	return rel.handle.Append(bat)
 }
 
-func (rel *txnRelation) Update(_ uint64, bat *batch.Batch, _ engine.Snapshot) error {
-	return errors.New("doesn't support now")
+func (rel *txnRelation) Update(_ uint64, data *batch.Batch, _ engine.Snapshot) error {
+	schema := rel.handle.GetMeta().(*catalog.TableEntry).GetSchema()
+	hiddenIdx := catalog.GetAttrIdx(data.Attrs, schema.HiddenKey.Name)
+	for idx := 0; idx < vector.Length(data.Vecs[hiddenIdx]); idx++ {
+		v := compute.GetValue(data.Vecs[hiddenIdx], uint32(idx))
+		for i, attr := range data.Attrs {
+			if schema.HiddenKey.Name == attr {
+				continue
+			}
+			colIdx := schema.GetColIdx(attr)
+			err := rel.handle.UpdateByHiddenKey(v, colIdx, compute.GetValue(data.Vecs[i], uint32(idx)))
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
 }
 
 func (rel *txnRelation) Delete(_ uint64, data *vector.Vector, col string, _ engine.Snapshot) error {
diff --git a/proto/metadata.proto b/proto/metadata.proto
index ae6a01b14..1f0ef4424 100644
--- a/proto/metadata.proto
+++ b/proto/metadata.proto
@@ -17,20 +17,40 @@ option go_package = "github.com/matrixorigin/matrixone/pkg/pb/metadata";
 
 import "github.com/gogo/protobuf/gogoproto/gogo.proto";
 
-// DN dn shard metadata
-message DN {
-    // ID the id of the DN. The ID is not modified after the DN is created.
-    uint64 id         = 1 [(gogoproto.customname) = "ID"];
-    // ReplicaID only one replica for one DN. The ReplicaID is specified at
-    // the time the DN is created. DN restart ReplicaID will not change, when 
-    // a DN is migrated to another node, the ReplicaID will be reset.
-    uint64 replicaID  = 2;
-    // LogShardID a DN corresponds to a unique Shard of LogService.
-    uint64 logShardID = 3;
-    // Start A DN manages data within a range [Start, end)
-    bytes  start      = 4;
-    // End A DN manages data within a range [Start, end)
-    bytes  end        = 5;
-    // Address DN external service address.
-    string address    = 6;
-}
+// DNShardRecord is DN shard metadata describing what is a DN shard. It
+// is internally used by HAKeeper to maintain how many DNs available in
+// the system. 
+message DNShardRecord {
+  // ShardID the id of the DN shard.
+  uint64 ShardID    = 1;
+  // LogShardID a DN corresponds to a unique Shard of LogService.
+  uint64 LogShardID = 2;
+  // Start A DN manages data within a range [Start, end)
+  bytes  Start      = 3;
+  // End A DN manages data within a range [Start, end)
+  bytes  End        = 4;
+};
+
+// DNShard 
+message DNShard {
+  // DNShard extends DNShardRecord
+  DNShardRecord DNShardRecord = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+  // ReplicaID only one replica for one DN. The ReplicaID is specified at
+  // the time the DN is created. DN restart ReplicaID will not change, when
+  // a DN is migrated to another node, the ReplicaID will be reset.
+  uint64 ReplicaID  = 2;
+  // Address is DN's external service address.
+  string Address    = 3;
+};
+
+// LogShardRecord is Log Shard Metadata describing what is a Log shard. It is
+// internally used by the HAKeeper to maintain how many Log shards are available
+// in the system.
+message LogShardRecord {
+  // ShardID is the id of the Log Shard. 
+  uint64 ShardID = 1;
+  // NumberOfReplicas is the number of replicas in the shard.
+  uint64 NumberOfReplicas = 2;
+  // Name is the human readable name of the shard given by the DN.
+  string Name = 3;
+};


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[:])
diff --git a/pkg/sql/plan2/build_test.go b/pkg/sql/plan2/build_test.go
index 3efb2faab..d9d51c3db 100644
--- a/pkg/sql/plan2/build_test.go
+++ b/pkg/sql/plan2/build_test.go
@@ -30,7 +30,8 @@ func TestSingleSql(t *testing.T) {
 	// sql := `SELECT * FROM (SELECT relname as Tables_in_mo FROM mo_tables WHERE reldatabase = 'mo') a`
 	// sql := "SELECT nation2.* FROM nation2 natural join region"
 	// sql := `select n_name, avg(N_REGIONKEY) t from NATION where n_name != 'a' group by n_name having avg(N_REGIONKEY) > 10 order by t limit 20`
-	sql := `select distinct(n_name), ((abs(n_regionkey))) from nation`
+	// sql := `select date_add('1997-12-31 23:59:59',INTERVAL 100000 SECOND)`
+	sql := "select @str_var, @int_var, @bool_var, @@global.float_var, @@session.null_var"
 	// stmts, err := mysql.Parse(sql)
 	// if err != nil {
 	// 	t.Fatalf("%+v", err)
@@ -383,6 +384,7 @@ func TestSingleTableSqlBuilder(t *testing.T) {
 
 	// should pass
 	sqls := []string{
+		"SELECT '1900-01-01 00:00:00' + INTERVAL 2147483648 SECOND",
 		"SELECT N_NAME, N_REGIONKEY FROM NATION WHERE N_REGIONKEY > 0 AND N_NAME LIKE '%AA' ORDER BY N_NAME DESC, N_REGIONKEY LIMIT 10, 20",
 		"SELECT N_NAME, N_REGIONKEY a FROM NATION WHERE N_REGIONKEY > 0 ORDER BY a DESC", //test alias
 		"SELECT NATION.N_NAME FROM NATION",                                       //test alias
@@ -410,6 +412,10 @@ func TestSingleTableSqlBuilder(t *testing.T) {
 		"SELECT -1",
 		"select date_add('1997-12-31 23:59:59',INTERVAL 100000 SECOND)",
 		"select date_sub('1997-12-31 23:59:59',INTERVAL 2 HOUR)",
+		"select @str_var, @int_var, @bool_var, @float_var, @null_var",
+		"select @str_var, @@global.int_var, @@session.bool_var",
+		"select n_name from nation where n_name != @str_var and n_regionkey > @int_var",
+		"select n_name from nation where n_name != @@global.str_var and n_regionkey > @@session.int_var",
 		"select distinct(n_name), ((abs(n_regionkey))) from nation",
 	}
 	runTestShouldPass(mock, t, sqls, false, false)
@@ -423,6 +429,7 @@ func TestSingleTableSqlBuilder(t *testing.T) {
 		"SELECT N_NAME FROM NATION WHERE ffff(N_REGIONKEY) > 0",             //function name not exist
 		"SELECT NATION.N_NAME FROM NATION a",                                // mysql should error, but i don't think it is necesssary
 		"select n_nationkey, sum(n_nationkey) from nation",
+		"select n_name from nation where n_name != @not_exist_var",
 
 		"SELECT DISTINCT N_NAME FROM NATION GROUP BY N_REGIONKEY", //test distinct with group by
 	}
diff --git a/pkg/sql/plan2/function/builtin/binary/extract_test.go b/pkg/sql/plan2/function/builtin/binary/extract_test.go
new file mode 100644
index 000000000..648d4d09a
--- /dev/null
+++ b/pkg/sql/plan2/function/builtin/binary/extract_test.go
@@ -0,0 +1,81 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 binary
+
+import (
+	"fmt"
+	"github.com/matrixorigin/matrixone/pkg/container/nulls"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/sql/testutil"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestExtractFromDate(t *testing.T) {
+	vector0 := testutil.MakeScalarVarchar("year", 4)
+	vector1 := testutil.MakeDateVector([]string{"2020-01-01", "2021-02-03", "2024-03-04", ""}, []uint64{3})
+	inputVectors := make([]*vector.Vector, 2)
+	inputVectors[0] = vector0
+	inputVectors[1] = vector1
+	proc := testutil.NewProc()
+	outputVector, err := ExtractFromDate(inputVectors, proc)
+	require.NoError(t, err)
+	outputValues, ok := outputVector.Col.([]uint32)
+	require.True(t, ok)
+	require.Equal(t, []uint32{2020, 2021, 2024, 1}, outputValues)
+	require.True(t, nulls.Contains(outputVector.Nsp, uint64(3)))
+
+	vector0 = testutil.MakeScalarVarchar("month", 4)
+	vector1 = testutil.MakeDateVector([]string{"2020-01-01", "2021-02-03", "2024-03-04", ""}, []uint64{3})
+	inputVectors = make([]*vector.Vector, 2)
+	inputVectors[0] = vector0
+	inputVectors[1] = vector1
+	proc = testutil.NewProc()
+	outputVector, err = ExtractFromDate(inputVectors, proc)
+	require.NoError(t, err)
+	outputValues, ok = outputVector.Col.([]uint32)
+	fmt.Println(outputValues)
+	require.True(t, ok)
+	require.Equal(t, []uint32{1, 2, 3, 1}, outputValues)
+	require.True(t, nulls.Contains(outputVector.Nsp, uint64(3)))
+
+	vector0 = testutil.MakeScalarVarchar("day", 4)
+	vector1 = testutil.MakeDateVector([]string{"2020-01-01", "2021-02-03", "2024-03-04", ""}, []uint64{3})
+	inputVectors = make([]*vector.Vector, 2)
+	inputVectors[0] = vector0
+	inputVectors[1] = vector1
+	proc = testutil.NewProc()
+	outputVector, err = ExtractFromDate(inputVectors, proc)
+	require.NoError(t, err)
+	outputValues, ok = outputVector.Col.([]uint32)
+	fmt.Println(outputValues)
+	require.True(t, ok)
+	require.Equal(t, []uint32{1, 3, 4, 1}, outputValues)
+	require.True(t, nulls.Contains(outputVector.Nsp, uint64(3)))
+
+	vector0 = testutil.MakeScalarVarchar("year_month", 4)
+	vector1 = testutil.MakeDateVector([]string{"2020-01-01", "2021-02-03", "2024-03-04", ""}, []uint64{3})
+	inputVectors = make([]*vector.Vector, 2)
+	inputVectors[0] = vector0
+	inputVectors[1] = vector1
+	proc = testutil.NewProc()
+	outputVector, err = ExtractFromDate(inputVectors, proc)
+	require.NoError(t, err)
+	outputValues, ok = outputVector.Col.([]uint32)
+	fmt.Println(outputValues)
+	require.True(t, ok)
+	require.Equal(t, []uint32{202001, 202102, 202403, 101}, outputValues)
+	require.True(t, nulls.Contains(outputVector.Nsp, uint64(3)))
+}
diff --git a/pkg/sql/plan2/function/operator/cast_test.go b/pkg/sql/plan2/function/operator/cast_test.go
index 74294fe6e..69c3c01fe 100644
--- a/pkg/sql/plan2/function/operator/cast_test.go
+++ b/pkg/sql/plan2/function/operator/cast_test.go
@@ -15,6 +15,9 @@
 package operator
 
 import (
+	"testing"
+	"unsafe"
+
 	roaring "github.com/RoaringBitmap/roaring/roaring64"
 	"github.com/matrixorigin/matrixone/pkg/container/nulls"
 	"github.com/matrixorigin/matrixone/pkg/container/types"
@@ -23,8 +26,6 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/vm/process"
 	"github.com/smartystreets/goconvey/convey"
 	"github.com/stretchr/testify/require"
-	"testing"
-	"unsafe"
 )
 
 func TestCastSameType(t *testing.T) {
@@ -1916,6 +1917,63 @@ func TestCastVarcharAsDate(t *testing.T) {
 	})
 }
 
+func TestCastTimestampAsVarchar(t *testing.T) {
+	//Cast converts timestamp to varchar
+	//procs := testutil.NewProc()
+	cases := []struct {
+		name     string
+		vecs     []*vector.Vector
+		proc     *process.Process
+		input    []types.Timestamp
+		expected *types.Bytes
+		isScalar bool
+	}{
+		//{
+		//	name:  "01 - normal test",
+		//	proc:  procs,
+		//	input: []types.Timestamp{66823357574906480},
+		//	expected: &types.Bytes{
+		//		Data:    []byte("2020-06-14 16:24:15.230000"),
+		//		Offsets: []uint32{0},
+		//		Lengths: []uint32{26},
+		//	},
+		//	isScalar: false,
+		//},
+		//{
+		//	name:  "02 - scalar test",
+		//	proc:  procs,
+		//	input: []types.Timestamp{66823357574906480},
+		//	expected: &types.Bytes{
+		//		Data:    []byte("2020-06-14 16:24:15.230000"),
+		//		Offsets: []uint32{0},
+		//		Lengths: []uint32{26},
+		//	},
+		//	isScalar: true,
+		//},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			vecs := make([]*vector.Vector, 2)
+			if c.input != nil {
+				vecs[0] = vector.New(types.T_timestamp.ToType())
+				vecs[0].Col = c.input
+				vecs[0].IsConst = c.isScalar
+			} else {
+				vecs[0] = testutil.MakeScalarNull(0)
+			}
+			vecs[1] = vector.New(types.T_varchar.ToType())
+
+			result, err := Cast(vecs, c.proc)
+			if err != nil {
+				t.Fatal(err)
+			}
+			require.Equal(t, c.expected, result.Col.(*types.Bytes))
+			require.Equal(t, c.isScalar, result.IsScalar())
+		})
+	}
+}
+
 func TestCastDecimal64AsDecimal128(t *testing.T) {
 	//Cast converts decimal64 to decimal128
 	makeTempVector := func(left types.Decimal64, leftType types.Type, leftScalar bool, destType types.Type) []*vector.Vector {
diff --git a/pkg/vectorize/extract/extract_test.go b/pkg/vectorize/extract/extract_test.go
new file mode 100644
index 000000000..7366d167e
--- /dev/null
+++ b/pkg/vectorize/extract/extract_test.go
@@ -0,0 +1,63 @@
+// Copyright 2022 Matrix Origin
+//
+// Licensed 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 extract
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+func TestExtractFromDate(t *testing.T) {
+	a0, _ := types.ParseDate("2020-01-02")
+	a1, _ := types.ParseDate("2021-03-04")
+	inputs := make([]types.Date, 2)
+	inputs[0] = a0
+	inputs[1] = a1
+	results := make([]uint32, 2)
+	output, err := ExtractFromDate("year", inputs, results)
+	require.NoError(t, err)
+	require.Equal(t, []uint32{2020, 2021}, output)
+
+	a0, _ = types.ParseDate("2020-01-02")
+	a1, _ = types.ParseDate("2021-03-04")
+	inputs = make([]types.Date, 2)
+	inputs[0] = a0
+	inputs[1] = a1
+	results = make([]uint32, 2)
+	output, err = ExtractFromDate("month", inputs, results)
+	require.NoError(t, err)
+	require.Equal(t, []uint32{01, 03}, output)
+
+	a0, _ = types.ParseDate("2020-01-02")
+	a1, _ = types.ParseDate("2021-03-04")
+	inputs = make([]types.Date, 2)
+	inputs[0] = a0
+	inputs[1] = a1
+	results = make([]uint32, 2)
+	output, err = ExtractFromDate("year_month", inputs, results)
+	require.NoError(t, err)
+	require.Equal(t, []uint32{202001, 202103}, output)
+
+	a0, _ = types.ParseDate("2020-01-02")
+	a1, _ = types.ParseDate("2021-03-04")
+	inputs = make([]types.Date, 2)
+	inputs[0] = a0
+	inputs[1] = a1
+	results = make([]uint32, 2)
+	output, err = ExtractFromDate("quarter", inputs, results)
+	require.NoError(t, err)
+	require.Equal(t, []uint32{1, 1}, output)
+}
diff --git a/pkg/vm/engine/tae/moengine/engine_test.go b/pkg/vm/engine/tae/moengine/engine_test.go
index ea2a03c79..c5059ab98 100644
--- a/pkg/vm/engine/tae/moengine/engine_test.go
+++ b/pkg/vm/engine/tae/moengine/engine_test.go
@@ -2,6 +2,7 @@ package moengine
 
 import (
 	"github.com/matrixorigin/matrixone/pkg/vm/engine"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/iface/data"
 	"testing"
 
 	"github.com/matrixorigin/matrixone/pkg/container/vector"
@@ -167,5 +168,132 @@ func TestTxnRelation_GetHideKey(t *testing.T) {
 			assert.Equal(t, 0, vector.Length(bat.Vecs[0]))
 		}
 	}
+
 	t.Log(tae.Catalog.SimplePPString(common.PPL1))
 }
+
+func TestTxnRelation_Update(t *testing.T) {
+	tae := initDB(t, nil)
+	e := NewEngine(tae)
+	txn, err := e.StartTxn(nil)
+	assert.Nil(t, err)
+	err = e.Create(0, "db", 0, txn.GetCtx())
+	assert.Nil(t, err)
+	names := e.Databases(txn.GetCtx())
+	assert.Equal(t, 2, len(names))
+	dbase, err := e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+
+	schema := catalog.MockSchema(13, 2)
+	defs, err := SchemaToDefs(schema)
+	assert.NoError(t, err)
+	err = dbase.Create(0, schema.Name, defs, txn.GetCtx())
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err := dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	bat := catalog.MockData(schema, 4)
+	err = rel.Write(0, bat, txn.GetCtx())
+	assert.Nil(t, err)
+	assert.Nil(t, txn.Commit())
+	txn, err = e.StartTxn(nil)
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err = dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	assert.Nil(t, txn.Commit())
+	readers := rel.NewReader(10, nil, nil, nil)
+	update := bat
+	for _, reader := range readers {
+		bat1, err := reader.Read([]uint64{uint64(1), uint64(1), uint64(1)}, []string{schema.ColDefs[13].Name, schema.ColDefs[0].Name, schema.ColDefs[1].Name})
+		assert.Nil(t, err)
+		if bat1 != nil {
+			assert.Equal(t, 4, vector.Length(bat1.Vecs[0]))
+			update = bat1
+		}
+	}
+	assert.Equal(t, bat.Vecs[0].Col.([]int32)[0], update.Vecs[1].Col.([]int32)[0])
+	assert.Equal(t, bat.Vecs[0].Col.([]int32)[1], update.Vecs[1].Col.([]int32)[1])
+	assert.Equal(t, bat.Vecs[1].Col.([]int32)[0], update.Vecs[2].Col.([]int32)[0])
+	assert.Equal(t, bat.Vecs[1].Col.([]int32)[1], update.Vecs[2].Col.([]int32)[1])
+	update.Vecs[1].Col.([]int32)[0] = 5
+	update.Vecs[1].Col.([]int32)[1] = 6
+	update.Vecs[1].Col.([]int32)[3] = 8
+	update.Vecs[2].Col.([]int32)[0] = 9
+	update.Vecs[2].Col.([]int32)[1] = 10
+	txn, err = e.StartTxn(nil)
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err = dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	err = rel.Update(0, update, nil)
+	assert.Nil(t, err)
+	assert.Nil(t, txn.Commit())
+
+	txn, err = e.StartTxn(nil)
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err = dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	assert.Nil(t, txn.Commit())
+	readers = rel.NewReader(10, nil, nil, nil)
+	for _, reader := range readers {
+		bat, err := reader.Read([]uint64{uint64(1), uint64(1)}, []string{schema.ColDefs[0].Name, schema.ColDefs[1].Name})
+		assert.Nil(t, err)
+		if bat != nil {
+			assert.Equal(t, int32(5), bat.Vecs[0].Col.([]int32)[0])
+			assert.Equal(t, int32(6), bat.Vecs[0].Col.([]int32)[1])
+			assert.Equal(t, int32(8), bat.Vecs[0].Col.([]int32)[3])
+			assert.Equal(t, update.Vecs[1].Col.([]int32)[2], bat.Vecs[0].Col.([]int32)[2])
+			assert.Equal(t, int32(9), bat.Vecs[1].Col.([]int32)[0])
+			assert.Equal(t, int32(10), bat.Vecs[1].Col.([]int32)[1])
+			assert.Equal(t, update.Vecs[2].Col.([]int32)[2], bat.Vecs[1].Col.([]int32)[2])
+		}
+	}
+
+	tae.Close()
+	tae, err = db.Open(tae.Dir, nil)
+	defer tae.Close()
+	assert.NoError(t, err)
+	e = NewEngine(tae)
+	txn, err = e.StartTxn(nil)
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err = dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	assert.Nil(t, txn.Commit())
+	readers = rel.NewReader(10, nil, nil, nil)
+	updatePK := bat
+	for _, reader := range readers {
+		bat, err := reader.Read([]uint64{uint64(1), uint64(1), uint64(1), uint64(1)},
+			[]string{schema.ColDefs[0].Name,
+				schema.ColDefs[1].Name, schema.ColDefs[2].Name,
+				schema.ColDefs[13].Name})
+		assert.Nil(t, err)
+		if bat != nil {
+			assert.Equal(t, int32(5), bat.Vecs[0].Col.([]int32)[0])
+			assert.Equal(t, int32(6), bat.Vecs[0].Col.([]int32)[1])
+			assert.Equal(t, int32(8), bat.Vecs[0].Col.([]int32)[3])
+			assert.Equal(t, update.Vecs[1].Col.([]int32)[2], bat.Vecs[0].Col.([]int32)[2])
+			assert.Equal(t, int32(9), bat.Vecs[1].Col.([]int32)[0])
+			assert.Equal(t, int32(10), bat.Vecs[1].Col.([]int32)[1])
+			assert.Equal(t, update.Vecs[2].Col.([]int32)[2], bat.Vecs[1].Col.([]int32)[2])
+			updatePK = bat
+		}
+	}
+	updatePK.Vecs[2].Col.([]int32)[0] = 20
+	txn, err = e.StartTxn(nil)
+	assert.Nil(t, err)
+	dbase, err = e.Database("db", txn.GetCtx())
+	assert.Nil(t, err)
+	rel, err = dbase.Relation(schema.Name, txn.GetCtx())
+	assert.Nil(t, err)
+	err = rel.Update(0, updatePK, nil)
+	assert.NotNil(t, err)
+	assert.Equal(t, data.ErrUpdateUniqueKey, err)
+}