Repository: vanus-labs/vanus
Pull request #452 (2023-02-23 08:26:36)

feat(store): warm file after create
### What problem does this PR solve?

Improve file IO performance.

Issue Number: None

### Problem Summary

### What is changed and how does it work?

Warm file after create it.

### Check List

#### Tests

- [ ] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No code



Patch:
diff --git a/go.mod b/go.mod
index 4292b67..2d62ca2 100644
--- a/go.mod
+++ b/go.mod
@@ -51,6 +51,7 @@ require (
 	google.golang.org/genproto v0.0.0-20221027153422-115e99e71e1c
 	google.golang.org/grpc v1.51.0
 	google.golang.org/protobuf v1.28.1
+	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/apimachinery v0.25.0
 	k8s.io/client-go v0.25.0
@@ -143,7 +144,6 @@ require (
 	golang.org/x/text v0.5.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
-	gopkg.in/yaml.v2 v2.4.0 // indirect
 	sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect
 	sigs.k8s.io/yaml v1.2.0 // indirect
 )
diff --git a/internal/controller/eventbus/eventlog/eventlog.go b/internal/controller/eventbus/eventlog/eventlog.go
index 301a9e5..38045ed 100644
--- a/internal/controller/eventbus/eventlog/eventlog.go
+++ b/internal/controller/eventbus/eventlog/eventlog.go
@@ -52,14 +52,12 @@ type Manager interface {
 	AcquireEventLog(ctx context.Context, eventbusID vanus.ID, eventbusName string) (*metadata.Eventlog, error)
 	GetEventLog(ctx context.Context, id vanus.ID) *metadata.Eventlog
 	DeleteEventlog(ctx context.Context, id vanus.ID)
-	GetEventLogSegmentList(elID vanus.ID) []*Segment
-	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog,
-		num int) ([]*Segment, error)
-	UpdateSegment(ctx context.Context, m map[string][]Segment)
-	GetSegmentByBlockID(block *metadata.Block) (*Segment, error)
 	GetBlock(id vanus.ID) *metadata.Block
-	GetSegment(id vanus.ID) *Segment
 	UpdateSegmentReplicas(ctx context.Context, segID vanus.ID, term uint64) error
+	GetEventLogSegmentList(elID vanus.ID) []Segment
+	GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error)
+	UpdateSegment(ctx context.Context, m map[string][]Segment)
+	GetSegmentByBlockID(block *metadata.Block) (Segment, error)
 }
 
 var mgr = &eventlogManager{
@@ -135,15 +133,11 @@ func (mgr *eventlogManager) Run(ctx context.Context, kvClient kv.Client, startTa
 			return err2
 		}
 		mgr.eventLogMap.Store(elMD.ID.Key(), el)
-		ptr := el.segmentList.Front()
-		if ptr != nil {
-			seg, _ := ptr.Value.(*Segment)
-			for seg != nil {
-				mgr.globalSegmentMap.Store(seg.ID.Key(), seg)
-				for _, v := range seg.Replicas.Peers {
-					mgr.globalBlockMap.Store(v.ID.Key(), v)
-				}
-				seg = el.nextOf(seg)
+
+		for _, ptr := range el.getAllSegments() {
+			mgr.globalSegmentMap.Store(ptr.ID.Key(), ptr)
+			for _, v := range ptr.Replicas.Peers {
+				mgr.globalBlockMap.Store(v.ID.Key(), v)
 			}
 		}
 	}
@@ -250,9 +244,8 @@ func (mgr *eventlogManager) DeleteEventlog(ctx context.Context, id vanus.ID) {
 }
 
 func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
-	eli *metadata.Eventlog, num int) ([]*Segment, error) {
-	result := make([]*Segment, 0)
-
+	eli *metadata.Eventlog, num int) ([]Segment, error) {
+	result := make([]Segment, 0, num)
 	if eli == nil || num == 0 {
 		return result, nil
 	}
@@ -286,14 +279,13 @@ func (mgr *eventlogManager) GetAppendableSegment(ctx context.Context,
 		s = el.currentAppendableSegment()
 	}
 
-	for len(result) < num && s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
-	}
-	return result, nil
+	return el.listOfRight(s, true), nil
 }
 
 func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Segment) {
+	mgr.mutex.Lock()
+	defer mgr.mutex.Unlock()
+
 	// iterate eventlog
 	for eventlogID, segments := range m {
 		v, exist := mgr.eventLogMap.Load(eventlogID)
@@ -336,17 +328,20 @@ func (mgr *eventlogManager) UpdateSegment(ctx context.Context, m map[string][]Se
 	}
 }
 
-func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
-	result := make([]*Segment, 0)
+func (mgr *eventlogManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
+	result := make([]Segment, 0)
 	v, exist := mgr.eventLogMap.Load(elID.Key())
 	if !exist {
 		return result
 	}
+
 	el, _ := v.(*eventlog)
-	s := el.head()
-	for s != nil {
-		result = append(result, s)
-		s = el.nextOf(s)
+	list := el.getAllSegments()
+	el.lock()
+	defer el.unlock()
+
+	for _, v := range list {
+		result = append(result, v.Copy())
 	}
 	return result
 }
@@ -359,7 +354,7 @@ func (mgr *eventlogManager) GetBlock(id vanus.ID) *metadata.Block {
 	return v.(*metadata.Block)
 }
 
-func (mgr *eventlogManager) GetSegment(id vanus.ID) *Segment {
+func (mgr *eventlogManager) getSegment(id vanus.ID) *Segment {
 	v, exist := mgr.globalSegmentMap.Load(id.Key())
 	if !exist {
 		return nil
@@ -373,7 +368,7 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return errors.ErrBlockNotFound
 	}
 
-	seg := mgr.GetSegment(blk.SegmentID)
+	seg := mgr.getSegment(blk.SegmentID)
 	if seg == nil {
 		return errors.ErrSegmentNotFound
 	}
@@ -382,6 +377,13 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 		return nil
 	}
 
+	el := mgr.getEventLog(seg.EventLogID)
+	if el == nil {
+		return errors.ErrEventLogNotFound
+	}
+	el.lock()
+	defer el.unlock()
+
 	seg.Replicas.Leader = leaderID.Uint64()
 	seg.Replicas.Term = term
 	data, _ := json.Marshal(seg)
@@ -396,20 +398,22 @@ func (mgr *eventlogManager) UpdateSegmentReplicas(ctx context.Context, leaderID
 	return nil
 }
 
-func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (mgr *eventlogManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	v, exist := mgr.eventLogMap.Load(block.EventlogID.Key())
 	if !exist {
-		return nil, errors.ErrEventLogNotFound
+		return Segment{}, errors.ErrEventLogNotFound
 	}
 	el, _ := v.(*eventlog)
-	return el.get(block.SegmentID), nil
+	el.rLock()
+	defer el.rUnlock()
+	return el.get(block.SegmentID).Copy(), nil
 }
 
 func (mgr *eventlogManager) stop() {
 	mgr.cancel()
 }
 
-func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg *Segment) map[uint64]string {
+func (mgr *eventlogManager) getSegmentTopology(_ context.Context, seg Segment) map[uint64]string {
 	var addrs = map[uint64]string{}
 	for _, v := range seg.Replicas.Peers {
 		ins := mgr.volMgr.GetVolumeInstanceByID(v.VolumeID)
@@ -742,7 +746,7 @@ func (mgr *eventlogManager) createSegment(ctx context.Context, el *eventlog) (*S
 	_, err = srv.GetClient().ActivateSegment(ctx, &segment.ActivateSegmentRequest{
 		EventLogId:     seg.EventLogID.Uint64(),
 		ReplicaGroupId: seg.Replicas.ID.Uint64(),
-		Replicas:       mgr.getSegmentTopology(ctx, seg),
+		Replicas:       mgr.getSegmentTopology(ctx, *seg),
 	})
 
 	if err != nil {
@@ -945,12 +949,8 @@ func (el *eventlog) appendableSegmentNumber() int {
 	if cur == nil {
 		return 0
 	}
-	count := 0
-	for cur != nil {
-		count++
-		cur = el.nextOf(cur)
-	}
-	return count
+
+	return len(el.listOfRight(cur, true))
 }
 
 func (el *eventlog) currentAppendableSegment() *Segment {
@@ -959,6 +959,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 	if el.size() == 0 {
 		return nil
 	}
+
 	if el.writePtr == nil {
 		head := el.segmentList.Front()
 		for head != nil {
@@ -989,6 +990,7 @@ func (el *eventlog) currentAppendableSegment() *Segment {
 func (el *eventlog) add(ctx context.Context, seg *Segment) error {
 	el.mutex.Lock()
 	defer el.mutex.Unlock()
+
 	if !seg.isReady() {
 		return errors.ErrInvalidSegment
 	}
@@ -1091,37 +1093,69 @@ func (el *eventlog) size() int {
 	return el.segmentList.Len()
 }
 
-func (el *eventlog) nextOf(seg *Segment) *Segment {
-	if seg == nil {
-		return nil
-	}
+func (el *eventlog) getAllSegments() []*Segment {
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
 
-	node := el.segmentList.Get(seg.ID.Uint64())
-	if node == nil {
-		return nil
+	segs := make([]*Segment, 0)
+
+	if el.size() == 0 {
+		return segs
 	}
-	next := node.Next()
-	if next == nil {
-		return nil
+	ptr := el.segmentList.Front()
+	for ptr != nil {
+		next, _ := ptr.Value.(*Segment)
+		segs = append(segs, next)
+		ptr = ptr.Next()
 	}
-	return next.Value.(*Segment)
+	return segs
 }
 
-func (el *eventlog) previousOf(seg *Segment) *Segment {
+func (el *eventlog) listOfRight(seg *Segment, includeSelf bool) []Segment {
+	el.mutex.RLock()
+	defer el.mutex.RUnlock()
+	segs := make([]Segment, 0)
 	if seg == nil {
-		return nil
+		return segs
+	}
+	if includeSelf {
+		segs = append(segs, *seg)
 	}
+	node := el.segmentList.Get(seg.ID.Uint64())
+	if node == nil {
+		return segs
+	}
+	v := node.Next()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, *next)
+		v = v.Next()
+	}
+
+	return segs
+}
+
+func (el *eventlog) listOfPrevious(seg *Segment) []*Segment { //nolint:unused // ok
 	el.mutex.RLock()
 	defer el.mutex.RUnlock()
-
+	segs := make([]*Segment, 0)
 	node := el.segmentList.Get(seg.ID.Uint64())
-	prev := node.Prev()
-	if prev == nil {
-		return nil
+	if node == nil {
+		return segs
+	}
+	v := node.Prev()
+	for v != nil {
+		next, _ := v.Value.(*Segment)
+		segs = append(segs, next)
+		v = v.Prev()
+	}
+
+	// reverse slice
+	for i, j := 0, len(segs)-1; i < j; i, j = i+1, j-1 {
+		segs[i], segs[j] = segs[j], segs[i]
 	}
-	return prev.Value.(*Segment)
+
+	return segs
 }
 
 func (el *eventlog) deleteHead(ctx context.Context) error {
@@ -1192,3 +1226,11 @@ func (el *eventlog) lock() {
 func (el *eventlog) unlock() {
 	el.mutex.Unlock()
 }
+
+func (el *eventlog) rLock() {
+	el.mutex.RLock()
+}
+
+func (el *eventlog) rUnlock() {
+	el.mutex.RUnlock()
+}
diff --git a/internal/controller/eventbus/eventlog/mock_eventlog.go b/internal/controller/eventbus/eventlog/mock_eventlog.go
index bcac927..aeae26f 100644
--- a/internal/controller/eventbus/eventlog/mock_eventlog.go
+++ b/internal/controller/eventbus/eventlog/mock_eventlog.go
@@ -65,10 +65,10 @@ func (mr *MockManagerMockRecorder) DeleteEventlog(ctx, id interface{}) *gomock.C
 }
 
 // GetAppendableSegment mocks base method.
-func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]*Segment, error) {
+func (m *MockManager) GetAppendableSegment(ctx context.Context, eli *metadata.Eventlog, num int) ([]Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetAppendableSegment", ctx, eli, num)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
@@ -108,10 +108,10 @@ func (mr *MockManagerMockRecorder) GetEventLog(ctx, id interface{}) *gomock.Call
 }
 
 // GetEventLogSegmentList mocks base method.
-func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []*Segment {
+func (m *MockManager) GetEventLogSegmentList(elID vanus.ID) []Segment {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetEventLogSegmentList", elID)
-	ret0, _ := ret[0].([]*Segment)
+	ret0, _ := ret[0].([]Segment)
 	return ret0
 }
 
@@ -121,25 +121,11 @@ func (mr *MockManagerMockRecorder) GetEventLogSegmentList(elID interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEventLogSegmentList", reflect.TypeOf((*MockManager)(nil).GetEventLogSegmentList), elID)
 }
 
-// GetSegment mocks base method.
-func (m *MockManager) GetSegment(id vanus.ID) *Segment {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetSegment", id)
-	ret0, _ := ret[0].(*Segment)
-	return ret0
-}
-
-// GetSegment indicates an expected call of GetSegment.
-func (mr *MockManagerMockRecorder) GetSegment(id interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetSegment", reflect.TypeOf((*MockManager)(nil).GetSegment), id)
-}
-
 // GetSegmentByBlockID mocks base method.
-func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (*Segment, error) {
+func (m *MockManager) GetSegmentByBlockID(block *metadata.Block) (Segment, error) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetSegmentByBlockID", block)
-	ret0, _ := ret[0].(*Segment)
+	ret0, _ := ret[0].(Segment)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
diff --git a/internal/controller/eventbus/eventlog/segment.go b/internal/controller/eventbus/eventlog/segment.go
index c02c4fb..112fa20 100644
--- a/internal/controller/eventbus/eventlog/segment.go
+++ b/internal/controller/eventbus/eventlog/segment.go
@@ -143,7 +143,7 @@ type ReplicaGroup struct {
 	DestroyAt time.Time                  `json:"destroy_at"`
 }
 
-func Convert2ProtoSegment(ctx context.Context, ins ...*Segment) []*metapb.Segment {
+func Convert2ProtoSegment(ctx context.Context, ins ...Segment) []*metapb.Segment {
 	segs := make([]*metapb.Segment, len(ins))
 	for idx := 0; idx < len(ins); idx++ {
 		seg := ins[idx]
diff --git a/internal/primitive/container/conque/blocking/queue.go b/internal/primitive/container/conque/blocking/queue.go
index bb9841e..6dd21e7 100644
--- a/internal/primitive/container/conque/blocking/queue.go
+++ b/internal/primitive/container/conque/blocking/queue.go
@@ -15,33 +15,72 @@
 package blocking
 
 import (
+	// standard libraries.
+	stdsync "sync"
+	"sync/atomic"
+
 	// this project.
 	"github.com/linkall-labs/vanus/internal/primitive/container/conque/unbounded"
 	"github.com/linkall-labs/vanus/internal/primitive/sync"
 )
 
 type Queue[T any] struct {
-	q   unbounded.Queue[T]
-	sem sync.Semaphore
+	q     unbounded.Queue[T]
+	sem   sync.Semaphore
+	mu    stdsync.RWMutex
+	state int32
 }
 
-func New[T any]() *Queue[T] {
+func New[T any](handoff bool) *Queue[T] {
 	return new(Queue[T])
 }
 
+func (q *Queue[T]) Init(handoff bool) *Queue[T] {
+	q.sem.Init(handoff)
+	return q
+}
+
 func (q *Queue[T]) Close() {
-	// FIXME(james.yin): implements this.
+	atomic.StoreInt32(&q.state, 1)
+	q.sem.Release()
+}
+
+// Wait esures that all incoming Pushes observe that the queue is closed.
+func (q *Queue[T]) Wait() {
+	// Make sure no inflight Push.
+	q.mu.Lock()
+
+	// no op
+	_ = 1
+
+	q.mu.Unlock()
 }
 
-func (q *Queue[T]) Push(v T) {
+func (q *Queue[T]) Push(v T) bool {
+	// NOTE: no panic, avoid unlocking with defer.
+	q.mu.RLock()
+
+	// TODO: maybe atomic is unnecessary.
+	if atomic.LoadInt32(&q.state) != 0 {
+		q.mu.RUnlock()
+		return false
+	}
+
 	_ = q.q.Push(v)
 	q.sem.Release()
+	q.mu.RUnlock()
+	return true
 }
 
 func (q *Queue[T]) SharedPop() (T, bool) {
 	q.sem.Acquire()
 
-	// FIXME(james.yin): check close.
+	// Check close.
+	if atomic.LoadInt32(&q.state) != 0 {
+		q.sem.Release()
+		var v T
+		return v, false
+	}
 
 	for {
 		v, ok := q.q.SharedPop()
@@ -54,7 +93,12 @@ func (q *Queue[T]) SharedPop() (T, bool) {
 func (q *Queue[T]) UniquePop() (T, bool) {
 	q.sem.Acquire()
 
-	// FIXME(james.yin): check close.
+	// Check close.
+	if atomic.LoadInt32(&q.state) != 0 {
+		q.sem.Release()
+		var v T
+		return v, false
+	}
 
 	for {
 		v, _, ok := q.q.UniquePop()
@@ -63,3 +107,8 @@ func (q *Queue[T]) UniquePop() (T, bool) {
 		}
 	}
 }
+
+func (q *Queue[T]) RawPop() (T, bool) {
+	v, _, ok := q.q.UniquePop()
+	return v, ok
+}
diff --git a/internal/primitive/executor/multi_flow.go b/internal/primitive/executor/multi_flow.go
index b5ce33c..8eca3c7 100644
--- a/internal/primitive/executor/multi_flow.go
+++ b/internal/primitive/executor/multi_flow.go
@@ -72,12 +72,13 @@ type MultiFlow struct {
 	parallel int
 }
 
-func NewMultiFlow(parallel int, startImmediately bool) *MultiFlow {
-	return new(MultiFlow).Init(parallel, startImmediately)
+func NewMultiFlow(parallel int, handoff, startImmediately bool) *MultiFlow {
+	return new(MultiFlow).Init(parallel, handoff, startImmediately)
 }
 
-func (mf *MultiFlow) Init(parallel int, startImmediately bool) *MultiFlow {
+func (mf *MultiFlow) Init(parallel int, handoff, startImmediately bool) *MultiFlow {
 	mf.parallel = parallel
+	mf.q.Init(handoff)
 	if startImmediately {
 		mf.Start()
 	}
diff --git a/internal/primitive/sync/semaphore.go b/internal/primitive/sync/semaphore.go
index 8a84f70..3398bac 100644
--- a/internal/primitive/sync/semaphore.go
+++ b/internal/primitive/sync/semaphore.go
@@ -19,7 +19,13 @@ import (
 )
 
 type Semaphore struct {
-	sem uint32
+	sem     uint32
+	handoff bool
+}
+
+func (s *Semaphore) Init(handoff bool) *Semaphore {
+	s.handoff = handoff
+	return s
 }
 
 func (s *Semaphore) Acquire() {
@@ -27,7 +33,7 @@ func (s *Semaphore) Acquire() {
 }
 
 func (s *Semaphore) Release() {
-	semrelease(&s.sem, false, 0)
+	semrelease(&s.sem, s.handoff, 0)
 }
 
 //go:linkname semacquire sync.runtime_Semacquire
diff --git a/internal/primitive/transform/action/array/unfold_array.go b/internal/primitive/transform/action/array/unfold_array.go
new file mode 100644
index 0000000..7ff09db
--- /dev/null
+++ b/internal/primitive/transform/action/array/unfold_array.go
@@ -0,0 +1,75 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 array
+
+import (
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+)
+
+// ["unfold_array","sourceJsonPath", "targetPathPrefix"].
+type unfoldArrayAction struct {
+	action.CommonAction
+	targetPathPrefix string
+}
+
+func NewUnfoldArrayAction() action.Action {
+	a := &unfoldArrayAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "UNFOLD_ARRAY",
+		FixedArgs:  []arg.TypeList{[]arg.Type{arg.EventData}, []arg.Type{arg.EventData}},
+	}
+	return a
+}
+
+func (a *unfoldArrayAction) Init(args []arg.Arg) error {
+	a.targetPathPrefix = args[1].Original()
+	a.Args = args[:1]
+	a.ArgTypes = []common.Type{common.Array}
+	return nil
+}
+
+func (a *unfoldArrayAction) Execute(ceCtx *context.EventContext) error {
+	args, err := a.RunArgs(ceCtx)
+	if err != nil {
+		return err
+	}
+	arrayValue, _ := args[0].([]interface{})
+	errs := make([]string, 0, len(arrayValue))
+	for i, v := range arrayValue {
+		argument := a.targetPathPrefix + "-" + strconv.Itoa(i)
+		targetArg, err := arg.NewArg(argument)
+		if err != nil {
+			errs = append(errs, fmt.Sprintf("failed to NewArg, arg: %v, err: %v", argument, err))
+			continue
+		}
+		err = targetArg.SetValue(ceCtx, v)
+		if err != nil {
+			errs = append(errs, fmt.Sprintf("failed to SetValue, v: %v, err: %v", v, err))
+			continue
+		}
+	}
+	if len(errs) > 0 {
+		return errors.New(strings.Join(errs, ";"))
+	}
+	return nil
+}
diff --git a/internal/primitive/transform/action/condition/condition_if_action.go b/internal/primitive/transform/action/condition/condition_if_action.go
index 49801c1..c49b7f2 100644
--- a/internal/primitive/transform/action/condition/condition_if_action.go
+++ b/internal/primitive/transform/action/condition/condition_if_action.go
@@ -57,7 +57,7 @@ func (a *conditionIfAction) Execute(ceCtx *context.EventContext) error {
 	} else {
 		switch op {
 		case ">=", ">", "<=", "<":
-			a.ArgTypes = []common.Type{common.Number, common.String, common.Number, common.Any, common.Any}
+			a.ArgTypes = []common.Type{common.Float, common.String, common.Float, common.Any, common.Any}
 		default:
 			return errors.Errorf("not support op [%s]", op)
 		}
diff --git a/internal/primitive/transform/action/strings/capitalize_word.go b/internal/primitive/transform/action/strings/capitalize_word.go
new file mode 100644
index 0000000..b8946d7
--- /dev/null
+++ b/internal/primitive/transform/action/strings/capitalize_word.go
@@ -0,0 +1,32 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 strings
+
+import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/function"
+)
+
+// NewCapitalizeWord ["capitalize_word", "key"].
+func NewCapitalizeWordAction() action.Action {
+	a := &action.SourceTargetSameAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "CAPITALIZE_WORD",
+		FixedArgs:  []arg.TypeList{arg.EventList},
+		Fn:         function.CapitalizeWord,
+	}
+	return a
+}
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter.go b/internal/primitive/transform/action/strings/split_with_delimiter.go
new file mode 100644
index 0000000..ac05f4d
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter.go
@@ -0,0 +1,44 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 strings
+
+import (
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/arg"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/function"
+)
+
+type splitWithDelimiterAction struct {
+	action.FunctionAction
+}
+
+// NewSplitWithDelimiterAction ["split_with_delimiter","sourceJsonPath", "delimiter", "targetJsonPath"].
+func NewSplitWithDelimiterAction() action.Action {
+	a := &splitWithDelimiterAction{}
+	a.CommonAction = action.CommonAction{
+		ActionName: "SPLIT_WITH_DELIMITER",
+		FixedArgs:  []arg.TypeList{arg.EventList, []arg.Type{arg.Constant}, []arg.Type{arg.EventData}},
+		Fn:         function.SplitWithSepFunction,
+	}
+	return a
+}
+
+func (a *splitWithDelimiterAction) Init(args []arg.Arg) error {
+	a.TargetArg = args[2]
+	a.Args = args[:2]
+	a.ArgTypes = []common.Type{common.String, common.String}
+	return nil
+}
diff --git a/internal/primitive/transform/common/cast.go b/internal/primitive/transform/common/cast.go
index 7218e83..3f562d6 100644
--- a/internal/primitive/transform/common/cast.go
+++ b/internal/primitive/transform/common/cast.go
@@ -29,7 +29,7 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 		switch value := val.(type) {
 		case int32: // ce attribute
 			return strconv.FormatInt(int64(value), 10), nil
-		case int64: // ce attribute
+		case int64:
 			return strconv.FormatInt(value, 10), nil
 		case float64: // ce data json marshal
 			return fmt.Sprintf("%v", val), nil
@@ -38,7 +38,7 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 		}
 		// Casting to string is always defined
 		return fmt.Sprintf("%v", val), nil
-	case Number:
+	case Float:
 		switch value := val.(type) {
 		case string:
 			v, err := strconv.ParseFloat(value, 64)
@@ -54,6 +54,22 @@ func Cast(val interface{}, target Type) (interface{}, error) {
 			return float64(value), nil
 		}
 		return 0, fmt.Errorf("undefined cast from %v to %v", TypeFromVal(val), target)
+	case Int:
+		switch value := val.(type) {
+		case string:
+			v, err := strconv.ParseInt(value, 10, 64)
+			if err != nil {
+				err = fmt.Errorf("cannot cast from String to Int: %w", err)
+			}
+			return int(v), err
+		case float64:
+			return int(value), nil
+		case int32:
+			return int(value), nil
+		case int64:
+			return int(value), nil
+		}
+		return 0, fmt.Errorf("undefined cast from %v to %v", TypeFromVal(val), target)
 	case Bool:
 		if value, ok := val.(string); ok {
 			lowerCase := strings.ToLower(value)
diff --git a/internal/primitive/transform/common/types.go b/internal/primitive/transform/common/types.go
index 8698cef..4599963 100644
--- a/internal/primitive/transform/common/types.go
+++ b/internal/primitive/transform/common/types.go
@@ -18,7 +18,8 @@ type Type uint8
 
 const (
 	String Type = iota
-	Number
+	Float
+	Int
 	Bool
 	Object
 	Array
@@ -30,8 +31,10 @@ func (t Type) String() string {
 	switch t {
 	case String:
 		return "string"
-	case Number:
-		return "number"
+	case Float:
+		return "float"
+	case Int:
+		return "int"
 	case Bool:
 		return "bool"
 	case Object:
@@ -59,7 +62,9 @@ func TypeFromVal(val interface{}) Type {
 	case string:
 		return String
 	case float64:
-		return Number
+		return Float
+	case int:
+		return Int
 	case bool:
 		return Bool
 	case map[string]interface{}:
diff --git a/internal/primitive/transform/function/datatime_functions.go b/internal/primitive/transform/function/datatime_functions.go
index 908befb..157a852 100644
--- a/internal/primitive/transform/function/datatime_functions.go
+++ b/internal/primitive/transform/function/datatime_functions.go
@@ -44,10 +44,10 @@ var DateFormatFunction = function{
 
 var UnixTimeFormatFunction = function{
 	name:         "UNIX_TIME_FORMAT",
-	fixedArgs:    []common.Type{common.Number, common.String},
+	fixedArgs:    []common.Type{common.Int, common.String},
 	variadicArgs: common.TypePtr(common.String),
 	fn: func(args []interface{}) (interface{}, error) {
-		t := time.Unix(int64(args[0].(float64)), 0)
+		t := time.Unix(int64(args[0].(int)), 0)
 		loc := time.UTC
 		if len(args) > 2 && args[2].(string) != "" {
 			var err error
diff --git a/internal/primitive/transform/function/math_functions.go b/internal/primitive/transform/function/math_functions.go
index e18dd46..ea0e306 100644
--- a/internal/primitive/transform/function/math_functions.go
+++ b/internal/primitive/transform/function/math_functions.go
@@ -22,8 +22,8 @@ import (
 
 var MathAddFunction = function{
 	name:         "MATH_ADD",
-	fixedArgs:    []common.Type{common.Number, common.Number},
-	variadicArgs: common.TypePtr(common.Number),
+	fixedArgs:    []common.Type{common.Float, common.Float},
+	variadicArgs: common.TypePtr(common.Float),
 	fn: func(args []interface{}) (interface{}, error) {
 		var sum float64
 		for i := 0; i < len(args); i++ {
@@ -36,7 +36,7 @@ var MathAddFunction = function{
 
 var MathSubFunction = function{
 	name:      "MATH_SUB",
-	fixedArgs: []common.Type{common.Number, common.Number},
+	fixedArgs: []common.Type{common.Float, common.Float},
 	fn: func(args []interface{}) (interface{}, error) {
 		return args[0].(float64) - args[1].(float64), nil
 	},
@@ -44,8 +44,8 @@ var MathSubFunction = function{
 
 var MathMulFunction = function{
 	name:         "MATH_MUL",
-	fixedArgs:    []common.Type{common.Number, common.Number},
-	variadicArgs: common.TypePtr(common.Number),
+	fixedArgs:    []common.Type{common.Float, common.Float},
+	variadicArgs: common.TypePtr(common.Float),
 	fn: func(args []interface{}) (interface{}, error) {
 		sum := float64(1)
 		for i := 0; i < len(args); i++ {
@@ -58,7 +58,7 @@ var MathMulFunction = function{
 
 var MathDivFunction = function{
 	name:      "MATH_DIV",
-	fixedArgs: []common.Type{common.Number, common.Number},
+	fixedArgs: []common.Type{common.Float, common.Float},
 	fn: func(args []interface{}) (interface{}, error) {
 		if args[1].(float64) == 0 {
 			return nil, fmt.Errorf("division by zero")
diff --git a/internal/primitive/transform/function/strings_functions.go b/internal/primitive/transform/function/strings_functions.go
index da8bd6f..2846d27 100644
--- a/internal/primitive/transform/function/strings_functions.go
+++ b/internal/primitive/transform/function/strings_functions.go
@@ -17,6 +17,7 @@ package function
 import (
 	"fmt"
 	"strings"
+	"unicode"
 
 	"github.com/linkall-labs/vanus/internal/primitive/transform/common"
 )
@@ -72,24 +73,24 @@ var AddSuffixFunction = function{
 var SplitWithSepFunction = function{
 	name:         "SPLIT_WITH_SEP",
 	fixedArgs:    []common.Type{common.String, common.String},
-	variadicArgs: common.TypePtr(common.Number),
+	variadicArgs: common.TypePtr(common.Int),
 	fn: func(args []interface{}) (interface{}, error) {
 		s, _ := args[0].(string)
 		sep, _ := args[1].(string)
 		if len(args) == 2 {
 			return strings.Split(s, sep), nil
 		}
-		return strings.SplitN(s, sep, int(args[2].(float64))), nil
+		return strings.SplitN(s, sep, args[2].(int)), nil
 	},
 }
 
 var ReplaceBetweenPositionsFunction = function{
 	name:      "REPLACE_BETWEEN_POSITIONS",
-	fixedArgs: []common.Type{common.String, common.Number, common.Number, common.String},
+	fixedArgs: []common.Type{common.String, common.Int, common.Int, common.String},
 	fn: func(args []interface{}) (interface{}, error) {
 		path, _ := args[0].(string)
-		startPosition := int(args[1].(float64))
-		endPosition := int(args[2].(float64))
+		startPosition, _ := args[1].(int)
+		endPosition, _ := args[2].(int)
 		targetValue, _ := args[3].(string)
 		if startPosition >= len(path) {
 			return nil, fmt.Errorf("start position must be less than the length of the string")
@@ -118,3 +119,24 @@ var CapitalizeSentence = function{
 		return strings.ToUpper(string(value[0])) + value[1:], nil
 	},
 }
+
+var CapitalizeWord = function{
+	name:      "CAPITALIZE_WORD",
+	fixedArgs: []common.Type{common.String},
+	fn: func(args []interface{}) (interface{}, error) {
+		value, _ := args[0].(string)
+		rs := []rune(value)
+		inWord := false
+		for i, r := range rs {
+			if !unicode.IsSpace(r) {
+				if !inWord {
+					rs[i] = unicode.ToTitle(r)
+				}
+				inWord = true
+			} else {
+				inWord = false
+			}
+		}
+		return string(rs), nil
+	},
+}
diff --git a/internal/primitive/transform/runtime/init.go b/internal/primitive/transform/runtime/init.go
index 6b3747e..d9d775f 100644
--- a/internal/primitive/transform/runtime/init.go
+++ b/internal/primitive/transform/runtime/init.go
@@ -53,11 +53,14 @@ func init() {
 		strings.NewReplaceBetweenPositionsAction,
 		strings.NewCapitalizeSentenceAction,
 		strings.NewCheckCustomValuesAction,
+		strings.NewCapitalizeWordAction,
+		strings.NewSplitWithDelimiterAction,
 		// condition
 		condition.NewConditionIfAction,
 		// array
 		array.NewRenderArrayAction,
 		array.NewArrayForeachAction,
+		array.NewUnfoldArrayAction,
 		// common
 		common.NewLengthAction,
 		// source
diff --git a/internal/store/io/engine/psync/psync.go b/internal/store/io/engine/psync/psync.go
index 3c93808..02f5255 100644
--- a/internal/store/io/engine/psync/psync.go
+++ b/internal/store/io/engine/psync/psync.go
@@ -45,6 +45,7 @@ func New(opts ...Option) engine.Interface {
 }
 
 func (e *psync) init(cfg config) *psync {
+	e.q.Init(false)
 	for i := 0; i < cfg.parallel; i++ {
 		go e.run()
 	}
diff --git a/internal/store/io/engine/uring/uring.go b/internal/store/io/engine/uring/uring.go
index 69066cc..69d66fb 100644
--- a/internal/store/io/engine/uring/uring.go
+++ b/internal/store/io/engine/uring/uring.go
@@ -20,7 +20,6 @@ package uring
 import (
 	// standard libraries.
 	"context"
-	"sort"
 
 	// third-party libraries.
 	"github.com/iceber/iouring-go"
@@ -33,15 +32,12 @@ import (
 )
 
 const (
-	defaultResultBufferSize   = 64
-	defaultInflightBufferSize = defaultResultBufferSize
+	defaultResultBufferSize = 64
 )
 
 type uRing struct {
-	ring      *iouring.IOURing
-	resultC   chan iouring.Result
-	inflightC chan uint64
-	seq       uint64
+	ring    *iouring.IOURing
+	resultC chan iouring.Result
 }
 
 // Make sure uRing implements engine.Interface.
@@ -57,9 +53,8 @@ func New() engine.Interface {
 	}
 
 	e := &uRing{
-		ring:      ring,
-		resultC:   make(chan iouring.Result, defaultResultBufferSize),
-		inflightC: make(chan uint64, defaultInflightBufferSize),
+		ring:    ring,
+		resultC: make(chan iouring.Result, defaultResultBufferSize),
 	}
 
 	go e.runCallback()
@@ -76,71 +71,15 @@ func (e *uRing) Close() {
 	close(e.resultC)
 }
 
-type pendingResult struct {
-	seq uint64
-	re  iouring.Result
-}
-
 func (e *uRing) runCallback() {
-	results := make(map[uint64]iouring.Result)
-	pendingResults := make([]pendingResult, 0, defaultInflightBufferSize)
-
 	for result := range e.resultC {
-		seq, _ := result.GetRequestInfo().(uint64)
-		need := <-e.inflightC
-
-		if seq == need {
-			if len(pendingResults) == 0 {
-				_ = result.Callback()
-			} else {
-				pendingResults = append(pendingResults, pendingResult{
-					seq: need,
-					re:  result,
-				})
-			}
-			continue
-		}
-
-		// Buffer and reorder before trigger callback.
-
-		re := results[need]
-		if re != nil {
-			delete(results, need)
-		}
-		pendingResults = append(pendingResults, pendingResult{
-			seq: need,
-			re:  re,
-		})
-
-		if seq > need {
-			results[seq] = result
-		} else {
-			if seq == pendingResults[0].seq {
-				_ = result.Callback()
-				pendingResults = pendingResults[1:]
-				for len(pendingResults) != 0 {
-					if pendingResults[0].re == nil {
-						break
-					}
-					_ = pendingResults[0].re.Callback()
-					pendingResults = pendingResults[1:]
-				}
-			} else {
-				i := sort.Search(len(pendingResults)-1, func(i int) bool {
-					return pendingResults[i].seq >= seq
-				})
-				pendingResults[i].re = result
-			}
-		}
+		_ = result.Callback()
 	}
 }
 
 func (e *uRing) WriteAt(z zone.Interface, b []byte, off int64, so, eo int, cb io.WriteCallback) {
-	seq := e.generateSeq()
-
 	f, offset := z.Raw(off)
 	pr := iouring.Pwrite(int(f.Fd()), b, uint64(offset)).
-		WithInfo(seq).
 		WithCallback(func(result iouring.Result) error {
 			cb(result.ReturnInt())
 			return nil
@@ -151,12 +90,4 @@ func (e *uRing) WriteAt(z zone.Interface, b []byte, off int64, so, eo int, cb io
 		cb(0, err)
 		return
 	}
-
-	e.inflightC <- seq
-}
-
-func (e *uRing) generateSeq() uint64 {
-	seq := e.seq
-	e.seq++
-	return seq
 }
diff --git a/internal/store/io/file.go b/internal/store/io/file.go
index 70771a4..796839b 100644
--- a/internal/store/io/file.go
+++ b/internal/store/io/file.go
@@ -14,6 +14,139 @@
 
 package io
 
+import (
+	// standard libraries.
+	"os"
+
+	// third-party libraries.
+	"github.com/ncw/directio"
+
+	// first-party libraries.
+	"github.com/linkall-labs/vanus/pkg/errors"
+)
+
 const (
 	defaultFilePerm = 0o644
+	baseFillSize    = 4 * 1024 // 4KB
+	fastFillVecSize = 11       // Up to 4MB
 )
+
+var fastFillVec []fastFillTemplate
+
+func init() { //nolint:gochecknoinits // Initialize fastFileVec.
+	fastFillVec = make([]fastFillTemplate, fastFillVecSize)
+
+	n := fastFillVecSize - 1
+	for i := 0; i <= n; i++ {
+		size := (1 << (n - i)) * baseFillSize
+		fastFillVec[i] = fastFillTemplate{
+			size: int64(size),
+			data: directio.AlignedBlock(size),
+		}
+	}
+}
+
+type fastFillTemplate struct {
+	size int64
+	data []byte
+}
+
+func OpenFile(path string, flag int, sync bool, direct bool) (*os.File, error) {
+	return openFile(path, openFileFlag|flag, sync, direct)
+}
+
+func CreateFile(path string, size int64, flag int, sync bool, direct bool) (*os.File, error) {
+	return createFile(path, size, flag, sync, direct)
+}
+
+func ResizeFile(f *os.File, size int64) error {
+	return resizeFile(f, size)
+}
+
+func doCreateFile(path string, size int64, flag int, sync bool, direct bool) (*os.File, error) {
+	f, err := openFile(path, createFileFlag|flag, sync, direct)
+	if err != nil {
+		return nil, err
+	}
+
+	if size <= 0 {
+		return f, nil
+	}
+
+	// Resize file.
+	if err = resizeFile(f, size); err != nil {
+		if err2 := f.Close(); err2 != nil {
+			return f, errors.Chain(err, err2)
+		}
+		return nil, err
+	}
+
+	return f, nil
+}
+
+func doCreateFileAndWarm( //nolint:unused,nolintlint // use in linux.
+	path string, size int64, flag int, sync bool, direct bool,
+) (*os.File, error) {
+	// Create file.
+	f, err := openFile(path, createFileFlag|os.O_WRONLY, false, true)
+	if err != nil {
+		return nil, err
+	}
+
+	// Resize file.
+	if err = resizeFile(f, size); err != nil {
+		if err2 := f.Close(); err2 != nil {
+			return f, errors.Chain(err, err2)
+		}
+		return nil, err
+	}
+
+	// Warm file.
+	if err = warmFile(f, size); err != nil {
+		if err2 := f.Close(); err2 != nil {
+			return f, errors.Chain(err, err2)
+		}
+		return nil, err
+	}
+
+	if err = f.Close(); err != nil {
+		return f, err
+	}
+
+	f, err = openFile(path, openFileFlag|flag, sync, direct)
+	if err != nil {
+		return nil, err
+	}
+
+	return f, nil
+}
+
+func openFile(path string, flag int, sync bool, direct bool) (*os.File, error) {
+	if direct {
+		return directio.OpenFile(path, makeFlag(flag, sync), defaultFilePerm)
+	}
+	return os.OpenFile(path, makeFlag(flag, sync), defaultFilePerm)
+}
+
+func makeFlag(flag int, sync bool) int {
+	if sync {
+		flag |= syncFlag
+	}
+	return flag
+}
+
+func warmFile(f *os.File, size int64) error { //nolint:unused,nolintlint // use in linux.
+	for i, off := 0, int64(0); i < len(fastFillVec); i++ {
+		for fill := &fastFillVec[i]; off+fill.size <= size; off += fill.size {
+			if _, err := f.WriteAt(fill.data, off); err != nil {
+				return err
+			}
+		}
+	}
+
+	if err := f.Sync(); err != nil {
+		return err
+	}
+
+	return nil
+}
diff --git a/internal/store/io/file_linux.go b/internal/store/io/file_linux.go
index 8c43f6a..649877f 100644
--- a/internal/store/io/file_linux.go
+++ b/internal/store/io/file_linux.go
@@ -21,43 +21,21 @@ import (
 	// standard libraries.
 	"os"
 	"syscall"
-
-	// third-party libraries.
-	"github.com/ncw/directio"
-
-	// first-party libraries.
-	"github.com/linkall-labs/vanus/pkg/errors"
 )
 
-func OpenFile(path string, wronly bool, sync bool) (*os.File, error) {
-	flag := makeFlag(syscall.O_NOATIME, wronly, sync)
-	return directio.OpenFile(path, flag, 0)
-}
+const (
+	openFileFlag   = syscall.O_NOATIME
+	createFileFlag = os.O_CREATE | os.O_EXCL | syscall.O_NOATIME
+	syncFlag       = syscall.O_DSYNC
+)
 
-func CreateFile(path string, size int64, wronly bool, sync bool) (*os.File, error) {
-	flag := makeFlag(os.O_CREATE|os.O_EXCL|syscall.O_NOATIME, wronly, sync)
-	f, err := directio.OpenFile(path, flag, defaultFilePerm)
-	if err != nil {
-		return nil, err
+func createFile(path string, size int64, flag int, sync bool, direct bool) (*os.File, error) {
+	if size > 0 && sync {
+		return doCreateFileAndWarm(path, size, flag, sync, direct)
 	}
-	// Resize file.
-	if err = f.Truncate(size); err != nil {
-		if err2 := f.Close(); err2 != nil {
-			return f, errors.Chain(err, err2)
-		}
-		return nil, err
-	}
-	return f, nil
+	return doCreateFile(path, size, flag, sync, direct)
 }
 
-func makeFlag(flag int, wronly bool, sync bool) int {
-	if wronly {
-		flag |= os.O_WRONLY
-	} else {
-		flag |= os.O_RDWR
-	}
-	if sync {
-		flag |= syscall.O_DSYNC
-	}
-	return flag
+func resizeFile(f *os.File, size int64) error {
+	return syscall.Fallocate(int(f.Fd()), 0, 0, size)
 }
diff --git a/internal/store/io/file_other.go b/internal/store/io/file_other.go
index f561781..e3ca721 100644
--- a/internal/store/io/file_other.go
+++ b/internal/store/io/file_other.go
@@ -20,43 +20,18 @@ package io
 import (
 	// standard libraries.
 	"os"
-
-	// third-party libraries.
-	"github.com/ncw/directio"
-
-	// first-party libraries.
-	"github.com/linkall-labs/vanus/pkg/errors"
 )
 
-func OpenFile(path string, wronly bool, sync bool) (*os.File, error) {
-	flag := makeFlag(0, wronly, sync)
-	return directio.OpenFile(path, flag, 0)
-}
+const (
+	openFileFlag   = 0
+	createFileFlag = os.O_CREATE | os.O_EXCL
+	syncFlag       = os.O_SYNC
+)
 
-func CreateFile(path string, size int64, wronly bool, sync bool) (*os.File, error) {
-	flag := makeFlag(os.O_CREATE|os.O_EXCL, wronly, sync)
-	f, err := directio.OpenFile(path, flag, defaultFilePerm)
-	if err != nil {
-		return nil, err
-	}
-	// Resize file.
-	if err = f.Truncate(size); err != nil {
-		if err2 := f.Close(); err2 != nil {
-			return f, errors.Chain(err, err2)
-		}
-		return nil, err
-	}
-	return f, nil
+func createFile(path string, size int64, flag int, sync bool, direct bool) (*os.File, error) {
+	return doCreateFile(path, size, flag, sync, direct)
 }
 
-func makeFlag(flag int, wronly bool, sync bool) int {
-	if wronly {
-		flag |= os.O_WRONLY
-	} else {
-		flag |= os.O_RDWR
-	}
-	if sync {
-		flag |= os.O_SYNC
-	}
-	return flag
+func resizeFile(f *os.File, size int64) error {
+	return f.Truncate(size)
 }
diff --git a/internal/store/io/stream/scheduler.go b/internal/store/io/stream/scheduler.go
index 56eefb0..1c07561 100644
--- a/internal/store/io/stream/scheduler.go
+++ b/internal/store/io/stream/scheduler.go
@@ -50,7 +50,7 @@ func (s *scheduler) init(e engine.Interface, cfg config) *scheduler {
 	s.e = e
 	s.bp = block.NewBufferPool(cfg.flushBatchSize)
 	s.pq.init(cfg.flushDelayTime)
-	s.callbackExecutor.Init(cfg.callbackParallel, true)
+	s.callbackExecutor.Init(cfg.callbackParallel, false, true)
 	return s
 }
 
diff --git a/internal/store/io/zone/segmentedfile/recovery.go b/internal/store/io/zone/segmentedfile/recovery.go
index 60b130e..dd4ffc3 100644
--- a/internal/store/io/zone/segmentedfile/recovery.go
+++ b/internal/store/io/zone/segmentedfile/recovery.go
@@ -100,7 +100,7 @@ func scanSegmentFiles(dir, ext string, segmentSize int64) (segments []*Segment,
 			size = truncated
 		}
 
-		f, err2 := io.OpenFile(path, false, true)
+		f, err2 := io.OpenFile(path, os.O_RDWR, true, true)
 		if err2 != nil {
 			return nil, nil, err2
 		}
diff --git a/internal/store/io/zone/segmentedfile/segmented_file.go b/internal/store/io/zone/segmentedfile/segmented_file.go
index edfeda0..7f92de5 100644
--- a/internal/store/io/zone/segmentedfile/segmented_file.go
+++ b/internal/store/io/zone/segmentedfile/segmented_file.go
@@ -132,7 +132,7 @@ func (sf *SegmentedFile) createNextSegment(last *Segment) *Segment {
 		off = last.eo
 	}
 
-	next, err := createSegment(sf.dir, sf.ext, off, sf.segmentSize, true)
+	next, err := createSegment(sf.dir, sf.ext, off, sf.segmentSize)
 	if err != nil {
 		panic(err)
 	}
@@ -144,9 +144,9 @@ func (sf *SegmentedFile) createNextSegment(last *Segment) *Segment {
 	return next
 }
 
-func createSegment(dir, ext string, so, size int64, sync bool) (*Segment, error) {
+func createSegment(dir, ext string, so, size int64) (*Segment, error) {
 	path := filepath.Join(dir, fmt.Sprintf("%020d%s", so, ext))
-	f, err := io.CreateFile(path, size, true, sync)
+	f, err := io.CreateFile(path, size, os.O_WRONLY, true, true)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/store/raft/block/executor.go b/internal/store/raft/block/executor.go
index 8eb8943..d32a036 100644
--- a/internal/store/raft/block/executor.go
+++ b/internal/store/raft/block/executor.go
@@ -33,12 +33,12 @@ func newExecutorFactory(cfg executorConfig, startImmediately bool) *executorFact
 }
 
 func (k *executorFactory) init(cfg executorConfig, startImmediately bool) *executorFactory {
-	k.raftExecutor.Init(cfg.raftExecutorParallel, startImmediately)
-	k.appendExecutor.Init(cfg.appendExecutorParallel, startImmediately)
-	k.commitExecutor.Init(cfg.commitExecutorParallel, startImmediately)
-	k.persistExecutor.Init(cfg.persistExecutorParallel, startImmediately)
-	k.applyExecutor.Init(cfg.applyExecutorParallel, startImmediately)
-	k.transportExecutor.Init(cfg.transportExecutorParallel, startImmediately)
+	k.raftExecutor.Init(cfg.raftExecutorParallel, false, startImmediately)
+	k.appendExecutor.Init(cfg.appendExecutorParallel, true, startImmediately)
+	k.commitExecutor.Init(cfg.commitExecutorParallel, false, startImmediately)
+	k.persistExecutor.Init(cfg.persistExecutorParallel, false, startImmediately)
+	k.applyExecutor.Init(cfg.applyExecutorParallel, false, startImmediately)
+	k.transportExecutor.Init(cfg.transportExecutorParallel, false, startImmediately)
 	return k
 }
 
diff --git a/internal/store/vsb/block_open.go b/internal/store/vsb/block_open.go
index b08da6a..3de4247 100644
--- a/internal/store/vsb/block_open.go
+++ b/internal/store/vsb/block_open.go
@@ -18,7 +18,7 @@ import (
 	// standard libraries.
 	"context"
 	stderr "errors"
-	"io"
+	stdio "io"
 	"math"
 	"os"
 
@@ -27,6 +27,7 @@ import (
 
 	// this project.
 	"github.com/linkall-labs/vanus/internal/store/block"
+	"github.com/linkall-labs/vanus/internal/store/io"
 	ceschema "github.com/linkall-labs/vanus/internal/store/schema/ce"
 	"github.com/linkall-labs/vanus/internal/store/vsb/codec"
 	"github.com/linkall-labs/vanus/internal/store/vsb/index"
@@ -43,7 +44,7 @@ func (b *vsBlock) Open(ctx context.Context) error {
 	}
 
 	// TODO(james.yin): use direct IO
-	f, err := os.OpenFile(b.path, os.O_RDWR|os.O_SYNC, 0)
+	f, err := io.OpenFile(b.path, os.O_RDWR, true, false)
 	if err != nil {
 		return err
 	}
@@ -95,7 +96,7 @@ func (b *vsBlock) repairMeta() error {
 	// Scan entries.
 	indexes := make([]index.Index, 0)
 	// Note: use math.MaxInt64-off to avoid overflow.
-	r := io.NewSectionReader(b.f, off, math.MaxInt64-off)
+	r := stdio.NewSectionReader(b.f, off, math.MaxInt64-off)
 	if full {
 		n, entry, err = b.dec.UnmarshalReader(r)
 		if err != nil || ceschema.EntryType(entry) != ceschema.End {
@@ -172,7 +173,7 @@ func (b *vsBlock) rebuildIndexes(num int, tail []index.Index) error {
 
 	// Scan entries.
 	off := b.dataOffset
-	r := io.NewSectionReader(b.f, off, b.fm.entryLength)
+	r := stdio.NewSectionReader(b.f, off, b.fm.entryLength)
 	for {
 		n, entry, err := b.dec.UnmarshalReader(r)
 		if err != nil {
diff --git a/internal/store/vsb/engine_open.go b/internal/store/vsb/engine_open.go
index c1f8c2d..ab7ca8a 100644
--- a/internal/store/vsb/engine_open.go
+++ b/internal/store/vsb/engine_open.go
@@ -27,6 +27,7 @@ import (
 	// this project.
 	"github.com/linkall-labs/vanus/internal/primitive/vanus"
 	"github.com/linkall-labs/vanus/internal/store/block"
+	"github.com/linkall-labs/vanus/internal/store/io"
 	"github.com/linkall-labs/vanus/internal/store/io/zone/file"
 	"github.com/linkall-labs/vanus/internal/store/vsb/codec"
 )
@@ -39,15 +40,11 @@ const (
 func (e *engine) Create(ctx context.Context, id vanus.ID, capacity int64) (block.Raw, error) {
 	path := e.resolvePath(id)
 
-	f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_RDWR|os.O_SYNC, defaultFilePerm)
+	f, err := io.CreateFile(path, capacity, os.O_RDWR, true, false)
 	if err != nil {
 		return nil, err
 	}
 
-	if err = f.Truncate(capacity); err != nil {
-		return nil, processError(err, f, path)
-	}
-
 	dec, _ := codec.NewDecoder(false, codec.IndexSize)
 	b := &vsBlock{
 		id:         id,
diff --git a/internal/store/wal/config.go b/internal/store/wal/config.go
index 5b488df..de69532 100644
--- a/internal/store/wal/config.go
+++ b/internal/store/wal/config.go
@@ -32,13 +32,12 @@ const (
 )
 
 type config struct {
-	pos              int64
-	cb               OnEntryCallback
-	blockSize        int
-	fileSize         int64
-	flushDelayTime   time.Duration // default: 3 * time.Millisecond
-	appendBufferSize int
-	engine           ioengine.Interface
+	pos            int64
+	cb             OnEntryCallback
+	blockSize      int
+	fileSize       int64
+	flushDelayTime time.Duration // default: 3 * time.Millisecond
+	engine         ioengine.Interface
 }
 
 func (cfg *config) segmentedFileOptions() []segmentedfile.Option {
@@ -62,9 +61,8 @@ func (cfg *config) streamSchedulerOptions() []stream.Option {
 
 func defaultConfig() config {
 	cfg := config{
-		blockSize:        defaultBlockSize,
-		fileSize:         defaultFileSize,
-		appendBufferSize: defaultAppendBufferSize,
+		blockSize: defaultBlockSize,
+		fileSize:  defaultFileSize,
 	}
 	return cfg
 }
@@ -112,12 +110,6 @@ func WithFlushDelayTime(d time.Duration) Option {
 	}
 }
 
-func WithAppendBufferSize(size int) Option {
-	return func(cfg *config) {
-		cfg.appendBufferSize = size
-	}
-}
-
 func WithIOEngine(engine ioengine.Interface) Option {
 	return func(cfg *config) {
 		cfg.engine = engine
diff --git a/internal/store/wal/recovery.go b/internal/store/wal/recovery.go
index a5f1037..0a19b79 100644
--- a/internal/store/wal/recovery.go
+++ b/internal/store/wal/recovery.go
@@ -47,6 +47,10 @@ func scanLogEntries(sf *segmentedfile.SegmentedFile, blockSize int, from int64,
 		return -1, ErrOutOfRange
 	}
 
+	if cb == nil {
+		cb = noopOnEntry
+	}
+
 	sc := scanner{
 		blockSize: int64(blockSize),
 		buf:       directio.AlignedBlock(blockSize),
@@ -201,3 +205,7 @@ func onRecord(ctx *scanner, r record.Record, eo int64) error {
 
 	return nil
 }
+
+func noopOnEntry(entry []byte, r Range) error {
+	return nil
+}
diff --git a/internal/store/wal/wal.go b/internal/store/wal/wal.go
index 59cc0b5..bff24d9 100644
--- a/internal/store/wal/wal.go
+++ b/internal/store/wal/wal.go
@@ -24,6 +24,7 @@ import (
 	"github.com/linkall-labs/vanus/observability/log"
 
 	// this project.
+	"github.com/linkall-labs/vanus/internal/primitive/container/conque/blocking"
 	"github.com/linkall-labs/vanus/internal/store/io/engine"
 	"github.com/linkall-labs/vanus/internal/store/io/stream"
 	"github.com/linkall-labs/vanus/internal/store/io/zone/segmentedfile"
@@ -54,13 +55,11 @@ type WAL struct {
 
 	blockSize int
 
-	appendC chan *appender
+	appendQ blocking.Queue[*appender]
 
-	closeMu  sync.RWMutex
 	appendWg sync.WaitGroup
 
-	closeC chan struct{}
-	doneC  chan struct{}
+	doneC chan struct{}
 }
 
 func Open(ctx context.Context, dir string, opts ...Option) (*WAL, error) {
@@ -106,11 +105,10 @@ func open(ctx context.Context, dir string, cfg config) (*WAL, error) {
 		scheduler: scheduler,
 		blockSize: cfg.blockSize,
 
-		appendC: make(chan *appender, cfg.appendBufferSize),
-		closeC:  make(chan struct{}),
-		doneC:   make(chan struct{}),
+		doneC: make(chan struct{}),
 	}
 
+	w.appendQ.Init(false)
 	go w.runAppend()
 
 	return w, nil
@@ -121,15 +119,7 @@ func (w *WAL) Dir() string {
 }
 
 func (w *WAL) Close() {
-	w.closeMu.Lock()
-	defer w.closeMu.Unlock()
-
-	select {
-	case <-w.closeC:
-	default:
-		close(w.closeC)
-		close(w.appendC)
-	}
+	w.appendQ.Close()
 }
 
 func (w *WAL) doClose() {
@@ -164,20 +154,31 @@ func (w *WAL) append(ctx context.Context, entries [][]byte, direct bool, cb Appe
 		cb(nil, nil)
 	}
 
-	// NOTE: Can not close the WAL while writing to appendC.
-	w.closeMu.RLock()
-	select {
-	case <-w.closeC:
+	if !w.appendQ.Push(w.newAppender(ctx, entries, direct, cb)) {
 		// TODO(james.yin): invoke callback in another goroutine.
 		cb(nil, ErrClosed)
-	default:
-		w.appendC <- w.newAppender(ctx, entries, direct, cb)
 	}
-	w.closeMu.RUnlock()
 }
 
 func (w *WAL) runAppend() {
-	for task := range w.appendC {
+	for {
+		task, ok := w.appendQ.UniquePop()
+		if !ok {
+			break
+		}
+
+		task.invoke()
+	}
+
+	w.appendQ.Wait()
+
+	// Invoke remaind tasks in w.appendQ.
+	for {
+		task, ok := w.appendQ.RawPop()
+		if !ok {
+			break
+		}
+
 		task.invoke()
 	}
 
diff --git a/vsctl/command/connector.go b/vsctl/command/connector.go
index 9d032bd..17400fb 100644
--- a/vsctl/command/connector.go
+++ b/vsctl/command/connector.go
@@ -237,7 +237,7 @@ func uninstallConnectorCommand() *cobra.Command {
 			}
 
 			client := &http.Client{}
-			url := fmt.Sprintf("%s%s%s/connectors?name=%s", HttpPrefix, operatorEndpoint, BaseUrl, name)
+                        url := fmt.Sprintf("%s%s%s/connectors/%s", HttpPrefix, operatorEndpoint, BaseUrl, name)
 			req, err := http.NewRequest("DELETE", url, &bytes.Reader{})
 			if err != nil {
 				cmdFailedf(cmd, "new http request failed: %s", err)
@@ -358,7 +358,7 @@ func getConnectorCommand() *cobra.Command {
 			}
 
 			client := &http.Client{}
-			url := fmt.Sprintf("%s%s%s/connector/%s", HttpPrefix, operatorEndpoint, BaseUrl, name)
+			url := fmt.Sprintf("%s%s%s/connectors/%s", HttpPrefix, operatorEndpoint, BaseUrl, name)
 			req, err := http.NewRequest("GET", url, &bytes.Reader{})
 			if err != nil {
 				cmdFailedf(cmd, "new http request failed: %s", err)


Test patch:
diff --git a/internal/controller/eventbus/eventlog/eventlog_test.go b/internal/controller/eventbus/eventlog/eventlog_test.go
index 57fe3b1..96d766e 100644
--- a/internal/controller/eventbus/eventlog/eventlog_test.go
+++ b/internal/controller/eventbus/eventlog/eventlog_test.go
@@ -287,8 +287,11 @@ func TestEventlogManager_ScaleSegmentTask(t *testing.T) {
 		So(util.MapLen(&utMgr.globalBlockMap), ShouldEqual, (defaultAppendableSegmentNumber*2+1)*3)
 
 		utMgr.stop()
+		// avoid data race during UT
 		head = el2.head()
+		el2.lock()
 		head.State = StateFrozen
+		el2.unlock()
 		So(err, ShouldBeNil)
 		time.Sleep(50 * time.Millisecond)
 		So(el.size(), ShouldEqual, defaultAppendableSegmentNumber+1)
@@ -511,14 +514,14 @@ func TestEventlogManager_CreateAndGetEventlog(t *testing.T) {
 			So(blockObj.VolumeID, ShouldEqual, vol1.ID)
 			So(blockObj.SegmentID, ShouldEqual, segments[0].ID)
 
-			seg := utMgr.GetSegment(segments2[1].ID)
+			seg := utMgr.getSegment(segments2[1].ID)
 			So(seg, ShouldNotBeNil)
 			So(seg.EventLogID, ShouldEqual, logMD.ID)
 
 			blockObj = utMgr.GetBlock(vanus.NewIDFromUint64(segments[1].Replicas.Leader))
 			segAnother, err := utMgr.GetSegmentByBlockID(blockObj)
 			So(err, ShouldBeNil)
-			So(segAnother, ShouldEqual, seg)
+			So(segAnother, ShouldResemble, *seg)
 		})
 	})
 }
@@ -783,21 +786,30 @@ func TestEventlogManager_UpdateSegmentReplicas(t *testing.T) {
 		kvCli := kv.NewMockClient(ctrl)
 		utMgr.kvClient = kvCli
 
+		ctx := stdCtx.Background()
+		el, err := newEventlog(ctx, &metadata.Eventlog{
+			ID:           vanus.NewTestID(),
+			EventbusID:   vanus.NewTestID(),
+			EventbusName: "ut",
+		}, kvCli, false)
+		So(err, ShouldBeNil)
+		utMgr.eventLogMap.Store(el.md.ID.Key(), el)
+
 		blk := &metadata.Block{
-			ID:        vanus.NewTestID(),
-			Capacity:  64 * 1024 * 1024,
-			SegmentID: 0,
+			ID:         vanus.NewTestID(),
+			Capacity:   64 * 1024 * 1024,
+			EventlogID: el.md.ID,
 		}
 		seg := createTestSegment(vanus.NewTestID())
+		seg.EventLogID = el.md.ID
 		seg.Replicas.Term = 3
 		utMgr.globalSegmentMap.Store(seg.ID.Key(), seg)
 		utMgr.globalBlockMap.Store(blk.ID.Key(), blk)
 
-		ctx := stdCtx.Background()
 		kvCli.EXPECT().Set(ctx, filepath.Join(metadata.SegmentKeyPrefixInKVStore, seg.ID.String()),
 			gomock.Any()).Times(1).Return(nil)
 
-		err := utMgr.UpdateSegmentReplicas(ctx, vanus.NewTestID(), 3)
+		err = utMgr.UpdateSegmentReplicas(ctx, vanus.NewTestID(), 3)
 		So(err, ShouldEqual, errors.ErrBlockNotFound)
 
 		err = utMgr.UpdateSegmentReplicas(ctx, blk.ID, 3)
@@ -1064,15 +1076,12 @@ func TestEventlog_All(t *testing.T) {
 			So(el.indexAt(4), ShouldBeNil)
 			So(el.indexAt(999), ShouldBeNil)
 
-			So(el.nextOf(el.indexAt(0)), ShouldEqual, seg2)
-			So(el.nextOf(el.indexAt(1)), ShouldEqual, seg3)
-			So(el.nextOf(el.indexAt(2)), ShouldEqual, seg4)
-			So(el.nextOf(el.indexAt(3)), ShouldBeNil)
+			list := el.getAllSegments()
 
-			So(el.previousOf(el.indexAt(0)), ShouldBeNil)
-			So(el.previousOf(el.indexAt(1)), ShouldEqual, seg1)
-			So(el.previousOf(el.indexAt(2)), ShouldEqual, seg2)
-			So(el.previousOf(el.indexAt(3)), ShouldEqual, seg3)
+			So(list[0], ShouldEqual, seg1)
+			So(list[1], ShouldEqual, seg2)
+			So(list[2], ShouldEqual, seg3)
+			So(list[3], ShouldEqual, seg4)
 		})
 	})
 }
diff --git a/internal/controller/eventbus/eventlog/segment_test.go b/internal/controller/eventbus/eventlog/segment_test.go
index e7a9594..0bada72 100644
--- a/internal/controller/eventbus/eventlog/segment_test.go
+++ b/internal/controller/eventbus/eventlog/segment_test.go
@@ -38,7 +38,7 @@ func TestConvert2ProtoSegment(t *testing.T) {
 		block1 := vanus.NewTestID()
 		block2 := vanus.NewTestID()
 		block3 := vanus.NewTestID()
-		seg := &Segment{
+		seg := Segment{
 			ID:                segID,
 			Capacity:          64 * 1024 * 1024,
 			EventLogID:        eID,
diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go
index 5d464be..c0de7ca 100644
--- a/internal/gateway/gateway_test.go
+++ b/internal/gateway/gateway_test.go
@@ -166,7 +166,6 @@ func TestGateway_EventID(t *testing.T) {
 		ControllerAddr: controllers,
 	}
 	ga := NewGateway(cfg)
-	defer ga.Stop()
 
 	ga.client = mockClient
 	ctx, cancel := context.WithCancel(context.Background())
diff --git a/internal/primitive/transform/action/array/unfold_array_test.go b/internal/primitive/transform/action/array/unfold_array_test.go
new file mode 100644
index 0000000..5d73963
--- /dev/null
+++ b/internal/primitive/transform/action/array/unfold_array_test.go
@@ -0,0 +1,63 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 array_test
+
+import (
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/array"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestUnfoldArrayAction(t *testing.T) {
+	funcName := array.NewUnfoldArrayAction().Name()
+	Convey("test unfold array valid", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.data.array", "$.data.target"})
+		So(err, ShouldBeNil)
+
+		e := cetest.MinEvent()
+		data := map[string]interface{}{
+			"array": []interface{}{"one", 2, []string{"three"}},
+		}
+		err = a.Execute(&context.EventContext{
+			Event: &e,
+			Data:  data,
+		})
+		So(err, ShouldBeNil)
+		So(data, ShouldHaveLength, 4)
+		So(data["target-0"], ShouldEqual, "one")
+		So(data["target-1"], ShouldEqual, 2)
+		So(data["target-2"], ShouldResemble, []interface{}{"three"})
+	})
+	Convey("test unfold array string input", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.data.array", "$.data.target"})
+		So(err, ShouldBeNil)
+
+		e := cetest.MinEvent()
+		data := map[string]interface{}{
+			"array": "one",
+		}
+		err = a.Execute(&context.EventContext{
+			Event: &e,
+			Data:  data,
+		})
+		So(err, ShouldBeNil)
+		So(data, ShouldHaveLength, 2)
+		So(data["target-0"], ShouldEqual, "one")
+	})
+}
diff --git a/internal/primitive/transform/action/strings/capitalize_word_test.go b/internal/primitive/transform/action/strings/capitalize_word_test.go
new file mode 100644
index 0000000..8108e4d
--- /dev/null
+++ b/internal/primitive/transform/action/strings/capitalize_word_test.go
@@ -0,0 +1,101 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 strings_test
+
+import (
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/strings"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestCapitalizeWordAction(t *testing.T) {
+	funcName := strings.NewCapitalizeWordAction().Name()
+	Convey("test capitalize word: several words", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "one two three")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "One Two Three")
+	})
+	Convey("test capitalize word: extra spaces, numbers, other symbols", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", " . one, two,   three q four 111 плюс минус  ")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, " . One, Two,   Three Q Four 111 Плюс Минус  ")
+	})
+	Convey("test capitalize word: empty string", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "")
+	})
+	Convey("test capitalize word: unicode ♬", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "♬")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "♬")
+	})
+	Convey("test capitalize word: a", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "a")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "A")
+	})
+	Convey("test capitalize word: leading symbols", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		e.SetExtension("test", "let 'em go")
+		ceCtx := &context.EventContext{
+			Event: &e,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		So(e.Extensions()["test"], ShouldEqual, "Let 'em Go")
+	})
+}
diff --git a/internal/primitive/transform/action/strings/split_with_delimiter_test.go b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
new file mode 100644
index 0000000..19af4f2
--- /dev/null
+++ b/internal/primitive/transform/action/strings/split_with_delimiter_test.go
@@ -0,0 +1,45 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 strings_test
+
+import (
+	"testing"
+
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/action/strings"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/context"
+	"github.com/linkall-labs/vanus/internal/primitive/transform/runtime"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestSplitWithDelimiterAction(t *testing.T) {
+	funcName := strings.NewSplitWithDelimiterAction().Name()
+	Convey("test split with delimiter", t, func() {
+		a, err := runtime.NewAction([]interface{}{funcName, "$.test", ",", "$.data.target"})
+		So(err, ShouldBeNil)
+		e := cetest.MinEvent()
+		data := map[string]interface{}{}
+		e.SetExtension("test", "one,two,three")
+		ceCtx := &context.EventContext{
+			Event: &e,
+			Data:  data,
+		}
+		err = a.Execute(ceCtx)
+		So(err, ShouldBeNil)
+		res, ok := data["target"]
+		So(ok, ShouldBeTrue)
+		So(res, ShouldResemble, []string{"one", "two", "three"})
+	})
+}
diff --git a/internal/primitive/transform/runtime/action_bench_test.go b/internal/primitive/transform/runtime/action_bench_test.go
index b54f5f4..ae8c9f5 100644
--- a/internal/primitive/transform/runtime/action_bench_test.go
+++ b/internal/primitive/transform/runtime/action_bench_test.go
@@ -79,4 +79,6 @@ func BenchmarkAction(b *testing.B) {
 	b.Run("replace_with_regex", actionBenchmark([]interface{}{"replace_with_regex", "$.data.str", "a", "Aa"}))
 	b.Run("capitalize_sentence", actionBenchmark([]interface{}{"capitalize_sentence", "$.data.str"}))
 	b.Run("check_custom_values", actionBenchmark([]interface{}{"check_custom_values", "$.data.str", "value", "$.data.target", "true", "false"}))
+	b.Run("split_with_delimiter", actionBenchmark([]interface{}{"split_with_delimiter", "$.data.str", "a", "$.data.target"}))
+	b.Run("unfold_array", actionBenchmark([]interface{}{"unfold_array", "$.data.str", "$.data.target"}))
 }
diff --git a/internal/store/wal/config_test.go b/internal/store/wal/config_test.go
index 9905fa8..c7f0af2 100644
--- a/internal/store/wal/config_test.go
+++ b/internal/store/wal/config_test.go
@@ -36,7 +36,6 @@ func TestConfig(t *testing.T) {
 			WithBlockSize(1024),
 			WithFileSize(4*1024*1024),
 			WithFlushDelayTime(5*time.Millisecond),
-			WithAppendBufferSize(1),
 			WithIOEngine(engine),
 		)
 
@@ -44,7 +43,6 @@ func TestConfig(t *testing.T) {
 		So(cfg.blockSize, ShouldEqual, 1024)
 		So(cfg.fileSize, ShouldEqual, 4*1024*1024)
 		So(cfg.flushDelayTime, ShouldEqual, 5*time.Millisecond)
-		So(cfg.appendBufferSize, ShouldEqual, 1)
 		So(cfg.engine, ShouldEqual, engine)
 	})
 }
diff --git a/test/regression/pubsub/main.go b/test/regression/pubsub/main.go
index 3995149..0246e0e 100644
--- a/test/regression/pubsub/main.go
+++ b/test/regression/pubsub/main.go
@@ -32,7 +32,7 @@ var (
 	sentFailed     int64
 	receivedNumber int64
 
-	totalNumber        = int64(1 << 15) // 1024*1024
+	totalNumber        = int64(1 << 20) // 1024*1024
 	parallelism        = 4
 	batchSize          = 8
 	maximumPayloadSize = 4 * 1024
diff --git a/test/wal/walbench.go b/test/wal/walbench.go
new file mode 100644
index 0000000..581db4c
--- /dev/null
+++ b/test/wal/walbench.go
@@ -0,0 +1,125 @@
+// Copyright 2023 Linkall Inc.
+//
+// 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 main
+
+import (
+	// standard libraries.
+	"context"
+	"fmt"
+	"log"
+	stdsync "sync"
+	"sync/atomic"
+	"time"
+
+	// this project.
+	"github.com/linkall-labs/vanus/internal/primitive/sync"
+	walog "github.com/linkall-labs/vanus/internal/store/wal"
+)
+
+const (
+	N           = 500000
+	PayloadSize = 1024
+)
+
+func main() {
+	ctx := context.Background()
+
+	wal, err := walog.Open(
+		ctx,
+		"wal-test",
+		// walog.WithBlockSize(4*1024),
+		// walog.WithFileSize(1024*1024*1024),
+		walog.WithFlushDelayTime(10*time.Millisecond),
+		// walog.WithIOEngine(uring.New()),
+		// walog.WithIOEngine(psync.New(psync.WithParallel(8))),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	defer func() {
+		wal.Close()
+		wal.Wait()
+	}()
+
+	payload := generatePayload(PayloadSize)
+
+	var count, last int64
+	var totalCost, lastCost int64
+	var totalWrite, lastWrite int64
+
+	go func() {
+		for {
+			time.Sleep(time.Second)
+			cu := atomic.LoadInt64(&count)
+			ct := atomic.LoadInt64(&totalCost)
+			cw := atomic.LoadInt64(&totalWrite)
+			if n := cu - last; n != 0 {
+				log.Printf("TPS: %d\tlatency: %.3fms\tthroughput: %.2fMiB/s\n",
+					n, float64(ct-lastCost)/float64(n)/1000, float64(cw-lastWrite)/1024/1024)
+			} else {
+				log.Printf("TPS: %d\tlatency: NaN\n", n)
+			}
+			last = cu
+			lastCost = ct
+			lastWrite = cw
+		}
+	}()
+
+	start := time.Now()
+
+	var sema sync.Semaphore
+	for i := 0; i < 256; i++ {
+		sema.Release()
+	}
+
+	var wg stdsync.WaitGroup
+	wg.Add(N)
+	for i := 0; i < N; i++ {
+		sema.Acquire()
+		st := time.Now()
+		wal.AppendOne(ctx, payload, func(r walog.Range, err error) {
+			cost := time.Since(st)
+			if err != nil {
+				log.Printf("err: %v", err)
+			} else {
+				atomic.AddInt64(&count, 1)
+				atomic.AddInt64(&totalCost, cost.Microseconds())
+				atomic.AddInt64(&totalWrite, r.EO-r.SO)
+			}
+			wg.Done()
+			sema.Release()
+		})
+	}
+	wg.Wait()
+
+	cost := time.Since(start)
+	fmt.Printf("write: %.2f MiB\n", float64(totalCost)/1024/1024)
+	fmt.Printf("cost: %d ms\n", cost.Milliseconds())
+	fmt.Printf("failed: %d\n", N-count)
+	fmt.Printf("tps: %f\n", float64(count)/cost.Seconds())
+}
+
+func generatePayload(size int) []byte {
+	data := func() string {
+		str := ""
+		for idx := 0; idx < size-1; idx++ {
+			str += "a"
+		}
+		str += "\n"
+		return str
+	}()
+	return []byte(data)
+}