Repository: matrixorigin/matrixone
Pull request #4387 (2022-08-04 15:13:58)

Adding a fault injection tool.
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #4349

## What this PR does / why we need it:

A falut injection tool.   See fault_test.go for how to use it in code, and fault.sql for how to call that in sql.
We will need to sprinkle the triggering of fault in our code.    


lni: Had a quick look at the PR. Seems like a lot of fun and profits if we can make it distributed, e.g. issue add_trigger_point/remove_trigger_point on CN nodes via sql, make the faultMap somehow remotely accessible and sprinkle the triggering of fault in CN/DN/Log nodes. 
fengttt: > 

I have not put lots of thought in distributed case yet -- so this one really is geared towards sql execution, most likely controlling CN.   Controlling DN is also not too hard by adding a CN -> DN rpc.   However, I am not sure about log/ha etc, a little bit too far away from sql.

Patch:
diff --git a/pkg/container/types/bytes.go b/pkg/container/types/bytes.go
index 5aa9460a3..a044031bb 100644
--- a/pkg/container/types/bytes.go
+++ b/pkg/container/types/bytes.go
@@ -59,6 +59,10 @@ func (a *Bytes) Get(n int64) []byte {
 	return a.Data[offset : offset+a.Lengths[n]]
 }
 
+func (a *Bytes) GetString(n int64) string {
+	return string(a.Get(n))
+}
+
 func (a *Bytes) Swap(i, j int64) {
 	a.Offsets[i], a.Offsets[j] = a.Offsets[j], a.Offsets[i]
 	a.Lengths[i], a.Lengths[j] = a.Lengths[j], a.Lengths[i]
diff --git a/pkg/sql/plan/function/builtin/multi/faultinj.go b/pkg/sql/plan/function/builtin/multi/faultinj.go
new file mode 100644
index 000000000..52232fde6
--- /dev/null
+++ b/pkg/sql/plan/function/builtin/multi/faultinj.go
@@ -0,0 +1,78 @@
+// 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 multi
+
+import (
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/types"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/util/fault"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+func EnableFaultInjection(vecs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	fault.Enable()
+	return proc.AllocBoolScalarVector(true), nil
+}
+
+func DisableFaultInjection(vecs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	fault.Disable()
+	return proc.AllocBoolScalarVector(true), nil
+}
+
+func AddFaultPoint(vecs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	for i := 0; i < 5; i++ {
+		if vecs[i].IsScalarNull() || !vecs[i].IsScalar() {
+			return nil, moerr.NewError(moerr.INVALID_ARGUMENT, "AddFaultPoint can only take scalar args")
+		}
+	}
+
+	name := vector.MustBytesCols(vecs[0]).GetString(0)
+	freq := vector.MustBytesCols(vecs[1]).GetString(0)
+	action := vector.MustBytesCols(vecs[2]).GetString(0)
+	iarg := vector.MustTCols[int64](vecs[3])[0]
+	sarg := vector.MustBytesCols(vecs[4]).GetString(0)
+
+	if err := fault.AddFaultPoint(name, freq, action, iarg, sarg); err != nil {
+		return nil, err
+	}
+
+	return proc.AllocBoolScalarVector(true), nil
+}
+
+func RemoveFaultPoint(vecs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	if vecs[0].IsScalarNull() || !vecs[0].IsScalar() {
+		return nil, moerr.NewError(moerr.INVALID_ARGUMENT, "RemoveFaultPoint can only take scalar args")
+	}
+
+	name := vector.MustBytesCols(vecs[0]).GetString(0)
+	if err := fault.RemoveFaultPoint(name); err != nil {
+		return nil, err
+	}
+	return proc.AllocBoolScalarVector(true), nil
+}
+
+func TriggerFaultPoint(vecs []*vector.Vector, proc *process.Process) (*vector.Vector, error) {
+	if vecs[0].IsScalarNull() || !vecs[0].IsScalar() {
+		return nil, moerr.NewError(moerr.INVALID_ARGUMENT, "TriggerFaultPoint can only take scalar args")
+	}
+
+	name := vector.MustBytesCols(vecs[0]).GetString(0)
+	iv, ok := fault.TriggerFault(name)
+	if !ok {
+		return proc.AllocScalarNullVector(types.T_int64.ToType()), nil
+	}
+	return proc.AllocInt64ScalarVector(iv), nil
+}
diff --git a/pkg/sql/plan/function/builtins.go b/pkg/sql/plan/function/builtins.go
index 256d038b2..a27823228 100644
--- a/pkg/sql/plan/function/builtins.go
+++ b/pkg/sql/plan/function/builtins.go
@@ -1795,4 +1795,75 @@ var builtins = map[int]Functions{
 			},
 		},
 	},
+
+	ENABLE_FAULT_INJECTION: {
+		Id: ENABLE_FAULT_INJECTION,
+		Overloads: []Function{
+			{
+				Index:     0,
+				Volatile:  true,
+				Flag:      plan.Function_INTERNAL,
+				Layout:    STANDARD_FUNCTION,
+				Args:      []types.T{},
+				ReturnTyp: types.T_bool,
+				Fn:        multi.EnableFaultInjection,
+			},
+		},
+	},
+	DISABLE_FAULT_INJECTION: {
+		Id: DISABLE_FAULT_INJECTION,
+		Overloads: []Function{
+			{
+				Index:     0,
+				Volatile:  true,
+				Flag:      plan.Function_INTERNAL,
+				Layout:    STANDARD_FUNCTION,
+				Args:      []types.T{},
+				ReturnTyp: types.T_bool,
+				Fn:        multi.DisableFaultInjection,
+			},
+		},
+	},
+	ADD_FAULT_POINT: {
+		Id: ADD_FAULT_POINT,
+		Overloads: []Function{
+			{
+				Index:     0,
+				Volatile:  true,
+				Flag:      plan.Function_INTERNAL,
+				Layout:    STANDARD_FUNCTION,
+				Args:      []types.T{types.T_varchar, types.T_varchar, types.T_varchar, types.T_int64, types.T_varchar},
+				ReturnTyp: types.T_bool,
+				Fn:        multi.AddFaultPoint,
+			},
+		},
+	},
+	REMOVE_FAULT_POINT: {
+		Id: REMOVE_FAULT_POINT,
+		Overloads: []Function{
+			{
+				Index:     0,
+				Volatile:  true,
+				Flag:      plan.Function_INTERNAL,
+				Layout:    STANDARD_FUNCTION,
+				Args:      []types.T{types.T_varchar},
+				ReturnTyp: types.T_bool,
+				Fn:        multi.RemoveFaultPoint,
+			},
+		},
+	},
+	TRIGGER_FAULT_POINT: {
+		Id: REMOVE_FAULT_POINT,
+		Overloads: []Function{
+			{
+				Index:     0,
+				Volatile:  true,
+				Flag:      plan.Function_INTERNAL,
+				Layout:    STANDARD_FUNCTION,
+				Args:      []types.T{types.T_varchar},
+				ReturnTyp: types.T_int64,
+				Fn:        multi.TriggerFaultPoint,
+			},
+		},
+	},
 }
diff --git a/pkg/sql/plan/function/function_id.go b/pkg/sql/plan/function/function_id.go
index 1eba70cb6..286bf94c6 100644
--- a/pkg/sql/plan/function/function_id.go
+++ b/pkg/sql/plan/function/function_id.go
@@ -222,6 +222,13 @@ const (
 	TIMESTAMP    // TIMESTAMP
 	DATE_FORMAT  // DATE_FORMAT
 	JSON_EXTRACT // JSON_EXTRACT
+
+	ENABLE_FAULT_INJECTION
+	DISABLE_FAULT_INJECTION
+	ADD_FAULT_POINT     // Add a fault point
+	REMOVE_FAULT_POINT  // Remove
+	TRIGGER_FAULT_POINT // Trigger.
+
 	// FUNCTION_END_NUMBER is not a function, just a flag to record the max number of function.
 	// TODO: every one should put the new function id in front of this one if you want to make a new function.
 	FUNCTION_END_NUMBER
@@ -310,57 +317,62 @@ var functionIdRegister = map[string]int32{
 	"from_unixtime":     FROM_UNIXTIME,
 	// unary functions
 	// whoever edit this, please follow the lexical order, or come up with a better ordering method
-	"abs":            ABS,
-	"acos":           ACOS,
-	"bit_length":     BIT_LENGTH,
-	"date":           DATE,
-	"day":            DAY,
-	"dayofyear":      DAYOFYEAR,
-	"exp":            EXP,
-	"empty":          EMPTY,
-	"length":         LENGTH,
-	"lengthutf8":     LENGTH_UTF8,
-	"char_length":    LENGTH_UTF8,
-	"ln":             LN,
-	"log":            LOG,
-	"ltrim":          LTRIM,
-	"month":          MONTH,
-	"oct":            OCT,
-	"reverse":        REVERSE,
-	"rtrim":          RTRIM,
-	"sin":            SIN,
-	"sinh":           SINH,
-	"space":          SPACE,
-	"tan":            TAN,
-	"week":           WEEK,
-	"weekday":        WEEKDAY,
-	"year":           YEAR,
-	"extract":        EXTRACT,
-	"if":             IFF,
-	"iff":            IFF,
-	"date_add":       DATE_ADD,
-	"date_sub":       DATE_SUB,
-	"atan":           ATAN,
-	"cos":            COS,
-	"cot":            COT,
-	"timestamp":      TIMESTAMP,
-	"database":       DATABASE,
-	"schema":         DATABASE,
-	"user":           USER,
-	"system_user":    USER,
-	"session_user":   USER,
-	"current_user":   USER,
-	"connection_id":  CONNECTION_ID,
-	"charset":        CHARSET,
-	"current_role":   CURRENT_ROLE,
-	"found_rows":     FOUND_ROWS,
-	"icu_version":    ICULIBVERSION,
-	"last_insert_id": LAST_INSERT_ID,
-	"roles_graphml":  ROLES_GRAPHML,
-	"row_count":      ROW_COUNT,
-	"version":        VERSION,
-	"collation":      COLLATION,
-	"json_extract":   JSON_EXTRACT,
+	"abs":                     ABS,
+	"acos":                    ACOS,
+	"bit_length":              BIT_LENGTH,
+	"date":                    DATE,
+	"day":                     DAY,
+	"dayofyear":               DAYOFYEAR,
+	"exp":                     EXP,
+	"empty":                   EMPTY,
+	"length":                  LENGTH,
+	"lengthutf8":              LENGTH_UTF8,
+	"char_length":             LENGTH_UTF8,
+	"ln":                      LN,
+	"log":                     LOG,
+	"ltrim":                   LTRIM,
+	"month":                   MONTH,
+	"oct":                     OCT,
+	"reverse":                 REVERSE,
+	"rtrim":                   RTRIM,
+	"sin":                     SIN,
+	"sinh":                    SINH,
+	"space":                   SPACE,
+	"tan":                     TAN,
+	"week":                    WEEK,
+	"weekday":                 WEEKDAY,
+	"year":                    YEAR,
+	"extract":                 EXTRACT,
+	"if":                      IFF,
+	"iff":                     IFF,
+	"date_add":                DATE_ADD,
+	"date_sub":                DATE_SUB,
+	"atan":                    ATAN,
+	"cos":                     COS,
+	"cot":                     COT,
+	"timestamp":               TIMESTAMP,
+	"database":                DATABASE,
+	"schema":                  DATABASE,
+	"user":                    USER,
+	"system_user":             USER,
+	"session_user":            USER,
+	"current_user":            USER,
+	"connection_id":           CONNECTION_ID,
+	"charset":                 CHARSET,
+	"current_role":            CURRENT_ROLE,
+	"found_rows":              FOUND_ROWS,
+	"icu_version":             ICULIBVERSION,
+	"last_insert_id":          LAST_INSERT_ID,
+	"roles_graphml":           ROLES_GRAPHML,
+	"row_count":               ROW_COUNT,
+	"version":                 VERSION,
+	"collation":               COLLATION,
+	"json_extract":            JSON_EXTRACT,
+	"enable_fault_injection":  ENABLE_FAULT_INJECTION,
+	"disable_fault_injection": DISABLE_FAULT_INJECTION,
+	"add_fault_point":         ADD_FAULT_POINT,
+	"remove_fault_point":      REMOVE_FAULT_POINT,
+	"trigger_fault_point":     TRIGGER_FAULT_POINT,
 }
 
 func GetFunctionIsWinfunByName(name string) bool {
diff --git a/pkg/util/fault/fault.go b/pkg/util/fault/fault.go
new file mode 100644
index 000000000..3ea607d5d
--- /dev/null
+++ b/pkg/util/fault/fault.go
@@ -0,0 +1,321 @@
+// 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.
+
+//
+// A very simple fault injection tool.
+package fault
+
+import (
+	"math"
+	"math/rand"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+)
+
+const (
+	STOP = iota
+	LOOKUP
+	ADD
+	REMOVE
+	TRIGGER
+)
+
+const (
+	RETURN = iota
+	GETCOUNT
+	SLEEP
+	WAIT
+	GETWAITERS
+	NOTIFY
+	NOTIFYALL
+	PANIC
+)
+
+// faultEntry describes how we shall fail
+type faultEntry struct {
+	cmd              int     // command
+	name             string  // name of the fault
+	cnt              int     // count how many times we run into this
+	start, end, skip int     // start, end, skip
+	prob             float64 // probability of failure
+	action           int
+	iarg             int64  // int arg
+	sarg             string // string arg
+
+	nWaiters int
+	mutex    sync.Mutex
+	cond     *sync.Cond
+}
+
+type faultMap struct {
+	faultPoints map[string]*faultEntry
+	chIn        chan *faultEntry
+	chOut       chan *faultEntry
+}
+
+var enabled atomic.Value
+var gfm *faultMap
+
+func (fm *faultMap) run() {
+	for {
+		e := <-fm.chIn
+		switch e.cmd {
+		case STOP:
+			return
+		case ADD:
+			if _, ok := fm.faultPoints[e.name]; ok {
+				fm.chOut <- nil
+			} else {
+				fm.faultPoints[e.name] = e
+				fm.chOut <- e
+			}
+		case REMOVE:
+			if v, ok := fm.faultPoints[e.name]; ok {
+				delete(fm.faultPoints, e.name)
+				fm.chOut <- v
+			} else {
+				fm.chOut <- nil
+			}
+		case TRIGGER:
+			var out *faultEntry
+			if v, ok := fm.faultPoints[e.name]; ok {
+				v.cnt += 1
+				if v.cnt >= v.start && v.cnt <= v.end && (v.cnt-v.start)%v.skip == 0 {
+					if v.prob == 1 || rand.Float64() < v.prob {
+						out = v
+					}
+				}
+			}
+			fm.chOut <- out
+		case LOOKUP:
+			fm.chOut <- fm.faultPoints[e.sarg]
+		default:
+			fm.chOut <- nil
+		}
+	}
+}
+
+func (e *faultEntry) do() int64 {
+	switch e.action {
+	case RETURN: // no op
+	case SLEEP:
+		time.Sleep(time.Duration(e.iarg) * time.Second)
+	case GETCOUNT:
+		if ee := lookup(e.sarg); ee != nil {
+			return int64(ee.cnt)
+		}
+	case WAIT:
+		e.mutex.Lock()
+		e.nWaiters += 1
+		e.cond.Wait()
+		e.nWaiters -= 1
+		e.mutex.Unlock()
+	case GETWAITERS:
+		if ee := lookup(e.sarg); ee != nil {
+			ee.mutex.Lock()
+			nw := ee.nWaiters
+			ee.mutex.Unlock()
+			return int64(nw)
+		}
+	case NOTIFY:
+		if ee := lookup(e.sarg); ee != nil {
+			ee.cond.Signal()
+		}
+	case NOTIFYALL:
+		if ee := lookup(e.sarg); ee != nil {
+			ee.cond.Broadcast()
+		}
+	case PANIC:
+		panic(e.sarg)
+	}
+	return 0
+}
+
+func startFaultMap() {
+	gfm = new(faultMap)
+	gfm.faultPoints = make(map[string]*faultEntry)
+	gfm.chIn = make(chan *faultEntry)
+	gfm.chOut = make(chan *faultEntry)
+	go gfm.run()
+}
+
+func stopFaultMap() {
+	var msg faultEntry
+	msg.cmd = STOP
+	gfm.chIn <- &msg
+	gfm = nil
+}
+
+// Enable fault injection
+func Enable() {
+	if !IsEnabled() {
+		startFaultMap()
+		enabled.Store(gfm)
+	}
+}
+
+// Disable fault injection
+func Disable() {
+	if IsEnabled() {
+		stopFaultMap()
+		enabled.Store(gfm)
+	}
+}
+
+func IsEnabled() bool {
+	ld := enabled.Load()
+	if ld == nil {
+		return false
+	}
+	return ld.(*faultMap) != nil
+}
+
+// Trigger a fault point.
+func TriggerFault(name string) (int64, bool) {
+	if !IsEnabled() {
+		return 0, false
+	}
+	var msg faultEntry
+	msg.cmd = TRIGGER
+	msg.name = name
+	gfm.chIn <- &msg
+	out := <-gfm.chOut
+
+	if out == nil {
+		return 0, false
+	}
+	return out.do(), true
+}
+
+func AddFaultPoint(name string, freq string, action string, iarg int64, sarg string) error {
+	if !IsEnabled() {
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "fault injection not enabled")
+	}
+
+	var err error
+
+	// Build msg from input.
+	var msg faultEntry
+	msg.cmd = ADD
+	msg.name = name
+
+	// freq is start:end:skip:prob
+	sesp := strings.Split(freq, ":")
+	if len(sesp) != 4 {
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault freq")
+	}
+
+	if sesp[0] == "" {
+		msg.start = 1
+	} else {
+		msg.start, err = strconv.Atoi(sesp[0])
+		if err != nil {
+			return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault freq")
+		}
+	}
+	if sesp[1] == "" {
+		msg.end = math.MaxInt
+	} else {
+		msg.end, err = strconv.Atoi(sesp[1])
+		if err != nil || msg.end < msg.start {
+			return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault freq")
+		}
+	}
+	if sesp[2] == "" {
+		msg.skip = 1
+	} else {
+		msg.skip, err = strconv.Atoi(sesp[2])
+		if err != nil || msg.skip <= 0 {
+			return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault freq")
+		}
+	}
+	if sesp[3] == "" {
+		msg.prob = 1.0
+	} else {
+		msg.prob, err = strconv.ParseFloat(sesp[3], 64)
+		if err != nil || msg.prob <= 0 || msg.prob >= 1 {
+			return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault freq")
+		}
+	}
+
+	// Action
+	switch strings.ToUpper(action) {
+	case "RETURN":
+		msg.action = RETURN
+	case "SLEEP":
+		msg.action = SLEEP
+	case "GETCOUNT":
+		msg.action = GETCOUNT
+	case "WAIT":
+		msg.action = WAIT
+	case "GETWAITERS":
+		msg.action = GETWAITERS
+	case "NOTIFY":
+		msg.action = NOTIFY
+	case "NOTIFYALL":
+		msg.action = NOTIFYALL
+	case "PANIC":
+		msg.action = PANIC
+	default:
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "invalid fault action")
+	}
+
+	msg.iarg = iarg
+	msg.sarg = sarg
+
+	if msg.action == WAIT {
+		msg.cond = sync.NewCond(&msg.mutex)
+	}
+
+	gfm.chIn <- &msg
+	out := <-gfm.chOut
+	if out == nil {
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "add fault injection point failed.")
+	}
+	return nil
+}
+
+func RemoveFaultPoint(name string) error {
+	if !IsEnabled() {
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "fault injection not enabled")
+	}
+
+	var msg faultEntry
+	msg.cmd = REMOVE
+	msg.name = name
+	gfm.chIn <- &msg
+	out := <-gfm.chOut
+	if out == nil {
+		return moerr.NewError(moerr.INVALID_ARGUMENT, "fault injection point does not exist")
+	}
+	return nil
+}
+
+func lookup(name string) *faultEntry {
+	if !IsEnabled() {
+		return nil
+	}
+
+	var msg faultEntry
+	msg.cmd = LOOKUP
+	msg.sarg = name
+	gfm.chIn <- &msg
+	out := <-gfm.chOut
+	return out
+}
diff --git a/pkg/vm/process/process.go b/pkg/vm/process/process.go
index df6627f19..1d5df6fa9 100644
--- a/pkg/vm/process/process.go
+++ b/pkg/vm/process/process.go
@@ -134,3 +134,21 @@ func (proc *Process) AllocConstNullVector(typ types.Type, cnt int) *vector.Vecto
 	nulls.Add(vec.Nsp, 0)
 	return vec
 }
+
+func (proc *Process) AllocBoolScalarVector(v bool) *vector.Vector {
+	typ := types.T_bool.ToType()
+	vec := proc.AllocScalarVector(typ)
+	bvec := make([]bool, 1)
+	bvec[0] = v
+	vec.Col = bvec
+	return vec
+}
+
+func (proc *Process) AllocInt64ScalarVector(v int64) *vector.Vector {
+	typ := types.T_int64.ToType()
+	vec := proc.AllocScalarVector(typ)
+	ivec := make([]int64, 1)
+	ivec[0] = v
+	vec.Col = ivec
+	return vec
+}


Test patch:
diff --git a/pkg/util/fault/fault_test.go b/pkg/util/fault/fault_test.go
new file mode 100644
index 000000000..bd30c6f0f
--- /dev/null
+++ b/pkg/util/fault/fault_test.go
@@ -0,0 +1,164 @@
+// 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 fault
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCount(t *testing.T) {
+	var ok bool
+	var cnt int64
+
+	Enable()
+	AddFaultPoint("a", ":5::", "return", 0, "")
+	AddFaultPoint("aa", ":::", "getcount", 0, "a")
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(3), cnt)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(4), cnt)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(5), cnt)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(6), cnt)
+
+	RemoveFaultPoint("a")
+	RemoveFaultPoint("aa")
+
+	AddFaultPoint("a", "3:8:2:", "return", 0, "")
+	AddFaultPoint("aa", ":::", "getcount", 0, "a")
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(1), cnt)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+	cnt, _ = TriggerFault("aa")
+	require.Equal(t, int64(3), cnt)
+
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+	cnt, _ = TriggerFault("aa")
+	require.Equal(t, int64(4), cnt)
+
+	// 5
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+
+	// 6
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+
+	// 7
+	_, ok = TriggerFault("a")
+	require.Equal(t, true, ok)
+
+	//8
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+
+	//9
+	_, ok = TriggerFault("a")
+	require.Equal(t, false, ok)
+
+	cnt, ok = TriggerFault("aa")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(9), cnt)
+	Disable()
+	Disable()
+}
+
+func wait(t *testing.T) {
+	_, ok := TriggerFault("w")
+	require.Equal(t, true, ok)
+}
+
+func TestWait(t *testing.T) {
+	var ok bool
+	var cnt int64
+
+	Enable()
+
+	AddFaultPoint("w", ":::", "wait", 0, "")
+	AddFaultPoint("n1", ":::", "notify", 0, "w")
+	AddFaultPoint("nall", ":::", "notifyall", 0, "w")
+	AddFaultPoint("gc", ":::", "getcount", 0, "w")
+	AddFaultPoint("gw", ":::", "getwaiters", 0, "w")
+	AddFaultPoint("s", ":::", "sleep", 1, "w")
+
+	for i := 0; i < 10; i++ {
+		go wait(t)
+	}
+
+	TriggerFault("s")
+
+	cnt, ok = TriggerFault("gc")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(10), cnt)
+
+	cnt, ok = TriggerFault("gw")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(10), cnt)
+
+	TriggerFault("n1")
+	TriggerFault("s")
+
+	cnt, ok = TriggerFault("gc")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(10), cnt)
+
+	cnt, ok = TriggerFault("gw")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(9), cnt)
+
+	TriggerFault("nall")
+	TriggerFault("s")
+
+	cnt, ok = TriggerFault("gc")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(10), cnt)
+
+	cnt, ok = TriggerFault("gw")
+	require.Equal(t, true, ok)
+	require.Equal(t, int64(0), cnt)
+
+	Disable()
+}
diff --git a/test/cases/util/fault.sql b/test/cases/util/fault.sql
new file mode 100644
index 000000000..979552d5e
--- /dev/null
+++ b/test/cases/util/fault.sql
@@ -0,0 +1,11 @@
+select disable_fault_injection();
+select enable_fault_injection();
+select add_fault_point('ret', ':::', 'return', 0, '');
+select trigger_fault_point('ret');
+select trigger_fault_point('ret');
+select trigger_fault_point('ret');
+select add_fault_point('cnt', ':::', 'getcount', 0, 'ret');
+select trigger_fault_point('cnt');
+-- select add_fault_point('panic', ':::', 'panic', 0, '');
+-- select trigger_fault_point('panic');
+select disable_fault_injection();
diff --git a/test/result/util/fault.result b/test/result/util/fault.result
new file mode 100644
index 000000000..59337f585
--- /dev/null
+++ b/test/result/util/fault.result
@@ -0,0 +1,27 @@
+select disable_fault_injection();
+disable_fault_injection()
+true
+select enable_fault_injection();
+enable_fault_injection()
+true
+select add_fault_point('ret', ':::', 'return', 0, '');
+add_fault_point(ret, :::, return, 0, )
+true
+select trigger_fault_point('ret');
+trigger_fault_point(ret)
+0
+select trigger_fault_point('ret');
+trigger_fault_point(ret)
+0
+select trigger_fault_point('ret');
+trigger_fault_point(ret)
+0
+select add_fault_point('cnt', ':::', 'getcount', 0, 'ret');
+add_fault_point(cnt, :::, getcount, 0, ret)
+true
+select trigger_fault_point('cnt');
+trigger_fault_point(cnt)
+3
+select disable_fault_injection();
+disable_fault_injection()
+true