Repository: pingcap/tidb (go)

Issue:
duration parameters don't have correct fsp
In `pkg/param/binary_param.go`, the `Duration` datum always uses the `MaxFsp`, which is actually not correct.

```go
case mysql.TypeDuration:
	switch len(binaryParams[i].Val) {
	case 0:
		tmp = "0"
	case 8:
		isNegative := binaryParams[i].Val[0]
		if isNegative > 1 {
			err = mysql.ErrMalformPacket
			return
		}
		_, tmp = binaryDuration(1, binaryParams[i].Val, isNegative)
	case 12:
		isNegative := binaryParams[i].Val[0]
		if isNegative > 1 {
			err = mysql.ErrMalformPacket
			return
		}
		_, tmp = binaryDurationWithMS(1, binaryParams[i].Val, isNegative)
	default:
		err = mysql.ErrMalformPacket
		return
	}
	// TODO: generate the duration datum directly
	var dur types.Duration
	dur, _, err = types.ParseDuration(typectx, tmp.(string), types.MaxFsp)
```

Pull request description:
pkg/param: add correct fsp for duration
### What problem does this PR solve?

Issue Number: close #48563

Problem Summary:

The `TIME` parameters will always use `us` precision.

### What is changed and how it works?

Only use `us` (6) precision when the parameter length is 12.

### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test
  > - [ ] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->  

### Release note

```release-note
Fix the issue that the `TIME` parameters always use the `us` precision.
```


Patch:
diff --git a/br/pkg/lightning/backend/external/BUILD.bazel b/br/pkg/lightning/backend/external/BUILD.bazel
index a9f2f06abd..a4390fbf03 100644
--- a/br/pkg/lightning/backend/external/BUILD.bazel
+++ b/br/pkg/lightning/backend/external/BUILD.bazel
@@ -65,7 +65,7 @@ go_test(
     ],
     embed = [":external"],
     flaky = True,
-    shard_count = 45,
+    shard_count = 46,
     deps = [
         "//br/pkg/lightning/backend/kv",
         "//br/pkg/lightning/common",
@@ -89,6 +89,7 @@ go_test(
         "@com_github_pingcap_kvproto//pkg/brpb",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_exp//rand",
+        "@org_golang_x_sync//errgroup",
         "@org_uber_go_atomic//:atomic",
     ],
 )
diff --git a/pkg/executor/sortexec/BUILD.bazel b/pkg/executor/sortexec/BUILD.bazel
index 81d3caed95..a5498953bd 100644
--- a/pkg/executor/sortexec/BUILD.bazel
+++ b/pkg/executor/sortexec/BUILD.bazel
@@ -2,7 +2,10 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "sortexec",
-    srcs = ["sort.go"],
+    srcs = [
+        "sort.go",
+        "topn.go",
+    ],
     importpath = "github.com/pingcap/tidb/pkg/executor/sortexec",
     visibility = ["//visibility:public"],
     deps = [
diff --git a/pkg/executor/sortexec/sort.go b/pkg/executor/sortexec/sort.go
index 030ec2b426..ce26bac26b 100644
--- a/pkg/executor/sortexec/sort.go
+++ b/pkg/executor/sortexec/sort.go
@@ -18,12 +18,10 @@ import (
 	"container/heap"
 	"context"
 	"errors"
-	"slices"
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/executor/internal/exec"
 	"github.com/pingcap/tidb/pkg/expression"
-	plannercore "github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/util/chunk"
@@ -328,223 +326,3 @@ func (h *multiWayMerge) Pop() interface{} {
 func (h *multiWayMerge) Swap(i, j int) {
 	h.elements[i], h.elements[j] = h.elements[j], h.elements[i]
 }
-
-// TopNExec implements a Top-N algorithm and it is built from a SELECT statement with ORDER BY and LIMIT.
-// Instead of sorting all the rows fetched from the table, it keeps the Top-N elements only in a heap to reduce memory usage.
-type TopNExec struct {
-	SortExec
-	Limit      *plannercore.PhysicalLimit
-	totalLimit uint64
-
-	// rowChunks is the chunks to store row values.
-	rowChunks *chunk.List
-	// rowPointer store the chunk index and row index for each row.
-	rowPtrs []chunk.RowPtr
-
-	chkHeap *topNChunkHeap
-}
-
-// topNChunkHeap implements heap.Interface.
-type topNChunkHeap struct {
-	*TopNExec
-}
-
-// Less implement heap.Interface, but since we mantains a max heap,
-// this function returns true if row i is greater than row j.
-func (h *topNChunkHeap) Less(i, j int) bool {
-	rowI := h.rowChunks.GetRow(h.rowPtrs[i])
-	rowJ := h.rowChunks.GetRow(h.rowPtrs[j])
-	return h.greaterRow(rowI, rowJ)
-}
-
-func (h *topNChunkHeap) greaterRow(rowI, rowJ chunk.Row) bool {
-	for i, colIdx := range h.keyColumns {
-		cmpFunc := h.keyCmpFuncs[i]
-		cmp := cmpFunc(rowI, colIdx, rowJ, colIdx)
-		if h.ByItems[i].Desc {
-			cmp = -cmp
-		}
-		if cmp > 0 {
-			return true
-		} else if cmp < 0 {
-			return false
-		}
-	}
-	return false
-}
-
-func (h *topNChunkHeap) Len() int {
-	return len(h.rowPtrs)
-}
-
-func (*topNChunkHeap) Push(interface{}) {
-	// Should never be called.
-}
-
-func (h *topNChunkHeap) Pop() interface{} {
-	h.rowPtrs = h.rowPtrs[:len(h.rowPtrs)-1]
-	// We don't need the popped value, return nil to avoid memory allocation.
-	return nil
-}
-
-func (h *topNChunkHeap) Swap(i, j int) {
-	h.rowPtrs[i], h.rowPtrs[j] = h.rowPtrs[j], h.rowPtrs[i]
-}
-
-// keyColumnsLess is the less function for key columns.
-func (e *TopNExec) keyColumnsLess(i, j chunk.RowPtr) bool {
-	rowI := e.rowChunks.GetRow(i)
-	rowJ := e.rowChunks.GetRow(j)
-	return e.lessRow(rowI, rowJ)
-}
-
-func (e *TopNExec) keyColumnsCompare(i, j chunk.RowPtr) int {
-	rowI := e.rowChunks.GetRow(i)
-	rowJ := e.rowChunks.GetRow(j)
-	return e.compressRow(rowI, rowJ)
-}
-
-func (e *TopNExec) initPointers() {
-	e.rowPtrs = make([]chunk.RowPtr, 0, e.rowChunks.Len())
-	e.memTracker.Consume(int64(8 * e.rowChunks.Len()))
-	for chkIdx := 0; chkIdx < e.rowChunks.NumChunks(); chkIdx++ {
-		rowChk := e.rowChunks.GetChunk(chkIdx)
-		for rowIdx := 0; rowIdx < rowChk.NumRows(); rowIdx++ {
-			e.rowPtrs = append(e.rowPtrs, chunk.RowPtr{ChkIdx: uint32(chkIdx), RowIdx: uint32(rowIdx)})
-		}
-	}
-}
-
-// Open implements the Executor Open interface.
-func (e *TopNExec) Open(ctx context.Context) error {
-	e.memTracker = memory.NewTracker(e.ID(), -1)
-	e.memTracker.AttachTo(e.Ctx().GetSessionVars().StmtCtx.MemTracker)
-
-	e.fetched = false
-	e.Idx = 0
-
-	return exec.Open(ctx, e.Children(0))
-}
-
-// Next implements the Executor Next interface.
-func (e *TopNExec) Next(ctx context.Context, req *chunk.Chunk) error {
-	req.Reset()
-	if !e.fetched {
-		e.totalLimit = e.Limit.Offset + e.Limit.Count
-		e.Idx = int(e.Limit.Offset)
-		err := e.loadChunksUntilTotalLimit(ctx)
-		if err != nil {
-			return err
-		}
-		err = e.executeTopN(ctx)
-		if err != nil {
-			return err
-		}
-		e.fetched = true
-	}
-	if e.Idx >= len(e.rowPtrs) {
-		return nil
-	}
-	if !req.IsFull() {
-		numToAppend := min(len(e.rowPtrs)-e.Idx, req.RequiredRows()-req.NumRows())
-		rows := make([]chunk.Row, numToAppend)
-		for index := 0; index < numToAppend; index++ {
-			rows[index] = e.rowChunks.GetRow(e.rowPtrs[e.Idx])
-			e.Idx++
-		}
-		req.AppendRows(rows)
-	}
-	return nil
-}
-
-func (e *TopNExec) loadChunksUntilTotalLimit(ctx context.Context) error {
-	e.chkHeap = &topNChunkHeap{e}
-	e.rowChunks = chunk.NewList(exec.RetTypes(e), e.InitCap(), e.MaxChunkSize())
-	e.rowChunks.GetMemTracker().AttachTo(e.memTracker)
-	e.rowChunks.GetMemTracker().SetLabel(memory.LabelForRowChunks)
-	for uint64(e.rowChunks.Len()) < e.totalLimit {
-		srcChk := exec.TryNewCacheChunk(e.Children(0))
-		// adjust required rows by total limit
-		srcChk.SetRequiredRows(int(e.totalLimit-uint64(e.rowChunks.Len())), e.MaxChunkSize())
-		err := exec.Next(ctx, e.Children(0), srcChk)
-		if err != nil {
-			return err
-		}
-		if srcChk.NumRows() == 0 {
-			break
-		}
-		e.rowChunks.Add(srcChk)
-	}
-	e.initPointers()
-	e.initCompareFuncs()
-	e.buildKeyColumns()
-	return nil
-}
-
-const topNCompactionFactor = 4
-
-func (e *TopNExec) executeTopN(ctx context.Context) error {
-	heap.Init(e.chkHeap)
-	for uint64(len(e.rowPtrs)) > e.totalLimit {
-		// The number of rows we loaded may exceeds total limit, remove greatest rows by Pop.
-		heap.Pop(e.chkHeap)
-	}
-	childRowChk := exec.TryNewCacheChunk(e.Children(0))
-	for {
-		err := exec.Next(ctx, e.Children(0), childRowChk)
-		if err != nil {
-			return err
-		}
-		if childRowChk.NumRows() == 0 {
-			break
-		}
-		err = e.processChildChk(childRowChk)
-		if err != nil {
-			return err
-		}
-		if e.rowChunks.Len() > len(e.rowPtrs)*topNCompactionFactor {
-			err = e.doCompaction()
-			if err != nil {
-				return err
-			}
-		}
-	}
-	slices.SortFunc(e.rowPtrs, e.keyColumnsCompare)
-	return nil
-}
-
-func (e *TopNExec) processChildChk(childRowChk *chunk.Chunk) error {
-	for i := 0; i < childRowChk.NumRows(); i++ {
-		heapMaxPtr := e.rowPtrs[0]
-		var heapMax, next chunk.Row
-		heapMax = e.rowChunks.GetRow(heapMaxPtr)
-		next = childRowChk.GetRow(i)
-		if e.chkHeap.greaterRow(heapMax, next) {
-			// Evict heap max, keep the next row.
-			e.rowPtrs[0] = e.rowChunks.AppendRow(childRowChk.GetRow(i))
-			heap.Fix(e.chkHeap, 0)
-		}
-	}
-	return nil
-}
-
-// doCompaction rebuild the chunks and row pointers to release memory.
-// If we don't do compaction, in a extreme case like the child data is already ascending sorted
-// but we want descending top N, then we will keep all data in memory.
-// But if data is distributed randomly, this function will be called log(n) times.
-func (e *TopNExec) doCompaction() error {
-	newRowChunks := chunk.NewList(exec.RetTypes(e), e.InitCap(), e.MaxChunkSize())
-	newRowPtrs := make([]chunk.RowPtr, 0, e.rowChunks.Len())
-	for _, rowPtr := range e.rowPtrs {
-		newRowPtr := newRowChunks.AppendRow(e.rowChunks.GetRow(rowPtr))
-		newRowPtrs = append(newRowPtrs, newRowPtr)
-	}
-	newRowChunks.GetMemTracker().SetLabel(memory.LabelForRowChunks)
-	e.memTracker.ReplaceChild(e.rowChunks.GetMemTracker(), newRowChunks.GetMemTracker())
-	e.rowChunks = newRowChunks
-
-	e.memTracker.Consume(int64(-8 * len(e.rowPtrs)))
-	e.memTracker.Consume(int64(8 * len(newRowPtrs)))
-	e.rowPtrs = newRowPtrs
-	return nil
-}
diff --git a/pkg/executor/sortexec/topn.go b/pkg/executor/sortexec/topn.go
new file mode 100644
index 0000000000..a52de50c42
--- /dev/null
+++ b/pkg/executor/sortexec/topn.go
@@ -0,0 +1,246 @@
+// Copyright 2023 PingCAP, 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 sortexec
+
+import (
+	"container/heap"
+	"context"
+	"slices"
+
+	"github.com/pingcap/tidb/pkg/executor/internal/exec"
+	plannercore "github.com/pingcap/tidb/pkg/planner/core"
+	"github.com/pingcap/tidb/pkg/util/chunk"
+	"github.com/pingcap/tidb/pkg/util/memory"
+)
+
+// TopNExec implements a Top-N algorithm and it is built from a SELECT statement with ORDER BY and LIMIT.
+// Instead of sorting all the rows fetched from the table, it keeps the Top-N elements only in a heap to reduce memory usage.
+type TopNExec struct {
+	SortExec
+	Limit      *plannercore.PhysicalLimit
+	totalLimit uint64
+
+	// rowChunks is the chunks to store row values.
+	rowChunks *chunk.List
+	// rowPointer store the chunk index and row index for each row.
+	rowPtrs []chunk.RowPtr
+
+	chkHeap *topNChunkHeap
+}
+
+// topNChunkHeap implements heap.Interface.
+type topNChunkHeap struct {
+	*TopNExec
+}
+
+// Less implement heap.Interface, but since we mantains a max heap,
+// this function returns true if row i is greater than row j.
+func (h *topNChunkHeap) Less(i, j int) bool {
+	rowI := h.rowChunks.GetRow(h.rowPtrs[i])
+	rowJ := h.rowChunks.GetRow(h.rowPtrs[j])
+	return h.greaterRow(rowI, rowJ)
+}
+
+func (h *topNChunkHeap) greaterRow(rowI, rowJ chunk.Row) bool {
+	for i, colIdx := range h.keyColumns {
+		cmpFunc := h.keyCmpFuncs[i]
+		cmp := cmpFunc(rowI, colIdx, rowJ, colIdx)
+		if h.ByItems[i].Desc {
+			cmp = -cmp
+		}
+		if cmp > 0 {
+			return true
+		} else if cmp < 0 {
+			return false
+		}
+	}
+	return false
+}
+
+func (h *topNChunkHeap) Len() int {
+	return len(h.rowPtrs)
+}
+
+func (*topNChunkHeap) Push(interface{}) {
+	// Should never be called.
+}
+
+func (h *topNChunkHeap) Pop() interface{} {
+	h.rowPtrs = h.rowPtrs[:len(h.rowPtrs)-1]
+	// We don't need the popped value, return nil to avoid memory allocation.
+	return nil
+}
+
+func (h *topNChunkHeap) Swap(i, j int) {
+	h.rowPtrs[i], h.rowPtrs[j] = h.rowPtrs[j], h.rowPtrs[i]
+}
+
+// keyColumnsLess is the less function for key columns.
+func (e *TopNExec) keyColumnsLess(i, j chunk.RowPtr) bool {
+	rowI := e.rowChunks.GetRow(i)
+	rowJ := e.rowChunks.GetRow(j)
+	return e.lessRow(rowI, rowJ)
+}
+
+func (e *TopNExec) keyColumnsCompare(i, j chunk.RowPtr) int {
+	rowI := e.rowChunks.GetRow(i)
+	rowJ := e.rowChunks.GetRow(j)
+	return e.compressRow(rowI, rowJ)
+}
+
+func (e *TopNExec) initPointers() {
+	e.rowPtrs = make([]chunk.RowPtr, 0, e.rowChunks.Len())
+	e.memTracker.Consume(int64(8 * e.rowChunks.Len()))
+	for chkIdx := 0; chkIdx < e.rowChunks.NumChunks(); chkIdx++ {
+		rowChk := e.rowChunks.GetChunk(chkIdx)
+		for rowIdx := 0; rowIdx < rowChk.NumRows(); rowIdx++ {
+			e.rowPtrs = append(e.rowPtrs, chunk.RowPtr{ChkIdx: uint32(chkIdx), RowIdx: uint32(rowIdx)})
+		}
+	}
+}
+
+// Open implements the Executor Open interface.
+func (e *TopNExec) Open(ctx context.Context) error {
+	e.memTracker = memory.NewTracker(e.ID(), -1)
+	e.memTracker.AttachTo(e.Ctx().GetSessionVars().StmtCtx.MemTracker)
+
+	e.fetched = false
+	e.Idx = 0
+
+	return exec.Open(ctx, e.Children(0))
+}
+
+// Next implements the Executor Next interface.
+func (e *TopNExec) Next(ctx context.Context, req *chunk.Chunk) error {
+	req.Reset()
+	if !e.fetched {
+		e.totalLimit = e.Limit.Offset + e.Limit.Count
+		e.Idx = int(e.Limit.Offset)
+		err := e.loadChunksUntilTotalLimit(ctx)
+		if err != nil {
+			return err
+		}
+		err = e.executeTopN(ctx)
+		if err != nil {
+			return err
+		}
+		e.fetched = true
+	}
+	if e.Idx >= len(e.rowPtrs) {
+		return nil
+	}
+	if !req.IsFull() {
+		numToAppend := min(len(e.rowPtrs)-e.Idx, req.RequiredRows()-req.NumRows())
+		rows := make([]chunk.Row, numToAppend)
+		for index := 0; index < numToAppend; index++ {
+			rows[index] = e.rowChunks.GetRow(e.rowPtrs[e.Idx])
+			e.Idx++
+		}
+		req.AppendRows(rows)
+	}
+	return nil
+}
+
+func (e *TopNExec) loadChunksUntilTotalLimit(ctx context.Context) error {
+	e.chkHeap = &topNChunkHeap{e}
+	e.rowChunks = chunk.NewList(exec.RetTypes(e), e.InitCap(), e.MaxChunkSize())
+	e.rowChunks.GetMemTracker().AttachTo(e.memTracker)
+	e.rowChunks.GetMemTracker().SetLabel(memory.LabelForRowChunks)
+	for uint64(e.rowChunks.Len()) < e.totalLimit {
+		srcChk := exec.TryNewCacheChunk(e.Children(0))
+		// adjust required rows by total limit
+		srcChk.SetRequiredRows(int(e.totalLimit-uint64(e.rowChunks.Len())), e.MaxChunkSize())
+		err := exec.Next(ctx, e.Children(0), srcChk)
+		if err != nil {
+			return err
+		}
+		if srcChk.NumRows() == 0 {
+			break
+		}
+		e.rowChunks.Add(srcChk)
+	}
+	e.initPointers()
+	e.initCompareFuncs()
+	e.buildKeyColumns()
+	return nil
+}
+
+const topNCompactionFactor = 4
+
+func (e *TopNExec) executeTopN(ctx context.Context) error {
+	heap.Init(e.chkHeap)
+	for uint64(len(e.rowPtrs)) > e.totalLimit {
+		// The number of rows we loaded may exceeds total limit, remove greatest rows by Pop.
+		heap.Pop(e.chkHeap)
+	}
+	childRowChk := exec.TryNewCacheChunk(e.Children(0))
+	for {
+		err := exec.Next(ctx, e.Children(0), childRowChk)
+		if err != nil {
+			return err
+		}
+		if childRowChk.NumRows() == 0 {
+			break
+		}
+		err = e.processChildChk(childRowChk)
+		if err != nil {
+			return err
+		}
+		if e.rowChunks.Len() > len(e.rowPtrs)*topNCompactionFactor {
+			err = e.doCompaction()
+			if err != nil {
+				return err
+			}
+		}
+	}
+	slices.SortFunc(e.rowPtrs, e.keyColumnsCompare)
+	return nil
+}
+
+func (e *TopNExec) processChildChk(childRowChk *chunk.Chunk) error {
+	for i := 0; i < childRowChk.NumRows(); i++ {
+		heapMaxPtr := e.rowPtrs[0]
+		var heapMax, next chunk.Row
+		heapMax = e.rowChunks.GetRow(heapMaxPtr)
+		next = childRowChk.GetRow(i)
+		if e.chkHeap.greaterRow(heapMax, next) {
+			// Evict heap max, keep the next row.
+			e.rowPtrs[0] = e.rowChunks.AppendRow(childRowChk.GetRow(i))
+			heap.Fix(e.chkHeap, 0)
+		}
+	}
+	return nil
+}
+
+// doCompaction rebuild the chunks and row pointers to release memory.
+// If we don't do compaction, in a extreme case like the child data is already ascending sorted
+// but we want descending top N, then we will keep all data in memory.
+// But if data is distributed randomly, this function will be called log(n) times.
+func (e *TopNExec) doCompaction() error {
+	newRowChunks := chunk.NewList(exec.RetTypes(e), e.InitCap(), e.MaxChunkSize())
+	newRowPtrs := make([]chunk.RowPtr, 0, e.rowChunks.Len())
+	for _, rowPtr := range e.rowPtrs {
+		newRowPtr := newRowChunks.AppendRow(e.rowChunks.GetRow(rowPtr))
+		newRowPtrs = append(newRowPtrs, newRowPtr)
+	}
+	newRowChunks.GetMemTracker().SetLabel(memory.LabelForRowChunks)
+	e.memTracker.ReplaceChild(e.rowChunks.GetMemTracker(), newRowChunks.GetMemTracker())
+	e.rowChunks = newRowChunks
+
+	e.memTracker.Consume(int64(-8 * len(e.rowPtrs)))
+	e.memTracker.Consume(int64(8 * len(newRowPtrs)))
+	e.rowPtrs = newRowPtrs
+	return nil
+}
diff --git a/pkg/param/binary_params.go b/pkg/param/binary_params.go
index ce016f5016..602f41e9aa 100644
--- a/pkg/param/binary_params.go
+++ b/pkg/param/binary_params.go
@@ -138,6 +138,7 @@ func ExecArgs(typectx types.Context, binaryParams []BinaryParam) (params []expre
 			continue
 
 		case mysql.TypeDuration:
+			fsp := 0
 			switch len(binaryParams[i].Val) {
 			case 0:
 				tmp = "0"
@@ -155,13 +156,14 @@ func ExecArgs(typectx types.Context, binaryParams []BinaryParam) (params []expre
 					return
 				}
 				_, tmp = binaryDurationWithMS(1, binaryParams[i].Val, isNegative)
+				fsp = types.MaxFsp
 			default:
 				err = mysql.ErrMalformPacket
 				return
 			}
 			// TODO: generate the duration datum directly
 			var dur types.Duration
-			dur, _, err = types.ParseDuration(typectx, tmp.(string), types.MaxFsp)
+			dur, _, err = types.ParseDuration(typectx, tmp.(string), fsp)
 			err = typectx.HandleTruncate(err)
 			if err != nil {
 				return
diff --git a/pkg/sessionctx/stmtctx/stmtctx.go b/pkg/sessionctx/stmtctx/stmtctx.go
index 2216d05ef3..d7fa9eed2f 100644
--- a/pkg/sessionctx/stmtctx/stmtctx.go
+++ b/pkg/sessionctx/stmtctx/stmtctx.go
@@ -431,7 +431,7 @@ func NewStmtCtx() *StatementContext {
 
 // NewStmtCtxWithTimeZone creates a new StatementContext with the given timezone
 func NewStmtCtxWithTimeZone(tz *time.Location) *StatementContext {
-	intest.Assert(tz)
+	intest.AssertNotNil(tz)
 	sc := &StatementContext{}
 	sc.typeCtx = types.NewContext(types.DefaultStmtFlags, tz, sc.AppendWarning)
 	return sc
@@ -451,7 +451,7 @@ func (sc *StatementContext) TimeZone() *time.Location {
 
 // SetTimeZone sets the timezone
 func (sc *StatementContext) SetTimeZone(tz *time.Location) {
-	intest.Assert(tz)
+	intest.AssertNotNil(tz)
 	sc.typeCtx = sc.typeCtx.WithLocation(tz)
 }
 
diff --git a/pkg/types/context.go b/pkg/types/context.go
index b969e4801f..af678ebc67 100644
--- a/pkg/types/context.go
+++ b/pkg/types/context.go
@@ -211,7 +211,7 @@ func (c *Context) WithFlags(f Flags) Context {
 
 // WithLocation returns a new context with the given location
 func (c *Context) WithLocation(loc *time.Location) Context {
-	intest.Assert(loc)
+	intest.AssertNotNil(loc)
 	ctx := *c
 	ctx.loc = loc
 	return ctx
@@ -219,7 +219,7 @@ func (c *Context) WithLocation(loc *time.Location) Context {
 
 // Location returns the location of the context
 func (c *Context) Location() *time.Location {
-	intest.Assert(c.loc)
+	intest.AssertNotNil(c.loc)
 	if c.loc == nil {
 		// c.loc should always not be nil, just make the code safe here.
 		return time.UTC


Test patch:
diff --git a/br/pkg/lightning/backend/external/bench_test.go b/br/pkg/lightning/backend/external/bench_test.go
index 55b7fea3ea..798a81eae8 100644
--- a/br/pkg/lightning/backend/external/bench_test.go
+++ b/br/pkg/lightning/backend/external/bench_test.go
@@ -18,14 +18,17 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"os"
 	"runtime/pprof"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/util/intest"
+	"golang.org/x/sync/errgroup"
 )
 
 var testingStorageURI = flag.String("testing-storage-uri", "", "the URI of the storage used for testing")
@@ -93,8 +96,7 @@ func (s *ascendingKeySource) outputSize() int {
 	return s.totalSize
 }
 
-type testSuite struct {
-	t                  *testing.T
+type writeTestSuite struct {
 	store              storage.ExternalStorage
 	source             kvSource
 	memoryLimit        int
@@ -103,7 +105,7 @@ type testSuite struct {
 	afterWriterClose   func()
 }
 
-func writePlainFile(s *testSuite) {
+func writePlainFile(s *writeTestSuite) {
 	ctx := context.Background()
 	buf := make([]byte, s.memoryLimit)
 	offset := 0
@@ -139,7 +141,7 @@ func writePlainFile(s *testSuite) {
 	}
 }
 
-func writeExternalFile(s *testSuite) {
+func writeExternalFile(s *writeTestSuite) {
 	ctx := context.Background()
 	builder := NewWriterBuilder().
 		SetMemorySizeLimit(uint64(s.memoryLimit))
@@ -164,7 +166,7 @@ func writeExternalFile(s *testSuite) {
 	}
 }
 
-func TestCompare(t *testing.T) {
+func TestCompareWriter(t *testing.T) {
 	store := openTestingStorage(t)
 	source := newAscendingKeySource(20, 100, 10000000)
 	memoryLimit := 64 * 1024 * 1024
@@ -195,8 +197,7 @@ func TestCompare(t *testing.T) {
 		pprof.StopCPUProfile()
 	}
 
-	suite := &testSuite{
-		t:                  t,
+	suite := &writeTestSuite{
 		store:              store,
 		source:             source,
 		memoryLimit:        memoryLimit,
@@ -214,3 +215,218 @@ func TestCompare(t *testing.T) {
 	writerSpeed := float64(source.outputSize()) / elapsed.Seconds() / 1024 / 1024
 	t.Logf("writer speed for %d bytes: %.2f MB/s", source.outputSize(), writerSpeed)
 }
+
+type readTestSuite struct {
+	store              storage.ExternalStorage
+	totalKVCnt         int
+	concurrency        int
+	memoryLimit        int
+	beforeCreateReader func()
+	beforeReaderClose  func()
+	afterReaderClose   func()
+}
+
+func readFileSequential(s *readTestSuite) {
+	ctx := context.Background()
+	files, _, err := GetAllFileNames(ctx, s.store, "evenly_distributed")
+	intest.Assert(err == nil)
+
+	buf := make([]byte, s.memoryLimit)
+	if s.beforeCreateReader != nil {
+		s.beforeCreateReader()
+	}
+	for i, file := range files {
+		reader, err := s.store.Open(ctx, file, nil)
+		intest.Assert(err == nil)
+		_, err = reader.Read(buf)
+		for err == nil {
+			_, err = reader.Read(buf)
+		}
+		intest.Assert(err == io.EOF)
+		if i == len(files)-1 {
+			if s.beforeReaderClose != nil {
+				s.beforeReaderClose()
+			}
+		}
+		err = reader.Close()
+		intest.Assert(err == nil)
+	}
+	if s.afterReaderClose != nil {
+		s.afterReaderClose()
+	}
+}
+
+func readFileConcurrently(s *readTestSuite) {
+	ctx := context.Background()
+	files, _, err := GetAllFileNames(ctx, s.store, "evenly_distributed")
+	intest.Assert(err == nil)
+
+	conc := min(s.concurrency, len(files))
+	var eg errgroup.Group
+	eg.SetLimit(conc)
+	var once sync.Once
+
+	if s.beforeCreateReader != nil {
+		s.beforeCreateReader()
+	}
+	for _, file := range files {
+		eg.Go(func() error {
+			buf := make([]byte, s.memoryLimit/conc)
+			reader, err := s.store.Open(ctx, file, nil)
+			intest.Assert(err == nil)
+			_, err = reader.Read(buf)
+			for err == nil {
+				_, err = reader.Read(buf)
+			}
+			intest.Assert(err == io.EOF)
+			once.Do(func() {
+				if s.beforeReaderClose != nil {
+					s.beforeReaderClose()
+				}
+			})
+			err = reader.Close()
+			intest.Assert(err == nil)
+			return nil
+		})
+	}
+	err = eg.Wait()
+	intest.Assert(err == nil)
+	if s.afterReaderClose != nil {
+		s.afterReaderClose()
+	}
+}
+
+func createEvenlyDistributedFiles(
+	t *testing.T,
+	fileSize, fileCount int,
+) (storage.ExternalStorage, int) {
+	store := openTestingStorage(t)
+	ctx := context.Background()
+
+	files, statFiles, err := GetAllFileNames(ctx, store, "evenly_distributed")
+	intest.Assert(err == nil)
+	err = store.DeleteFiles(ctx, files)
+	intest.Assert(err == nil)
+	err = store.DeleteFiles(ctx, statFiles)
+	intest.Assert(err == nil)
+
+	value := make([]byte, 100)
+	kvCnt := 0
+	for i := 0; i < fileCount; i++ {
+		builder := NewWriterBuilder().
+			SetMemorySizeLimit(uint64(float64(fileSize) * 1.1))
+		writer := builder.Build(
+			store,
+			"evenly_distributed",
+			fmt.Sprintf("%d", i),
+		)
+
+		keyIdx := i
+		totalSize := 0
+		for totalSize < fileSize {
+			key := fmt.Sprintf("key_%09d", keyIdx)
+			err := writer.WriteRow(ctx, []byte(key), value, nil)
+			intest.Assert(err == nil)
+			keyIdx += fileCount
+			totalSize += len(key) + len(value)
+			kvCnt++
+		}
+		err := writer.Close(ctx)
+		intest.Assert(err == nil)
+	}
+	return store, kvCnt
+}
+
+func readMergeIter(s *readTestSuite) {
+	ctx := context.Background()
+	files, _, err := GetAllFileNames(ctx, s.store, "evenly_distributed")
+	intest.Assert(err == nil)
+
+	if s.beforeCreateReader != nil {
+		s.beforeCreateReader()
+	}
+
+	readBufSize := s.memoryLimit / len(files)
+	zeroOffsets := make([]uint64, len(files))
+	iter, err := NewMergeKVIter(ctx, files, zeroOffsets, s.store, readBufSize, false)
+	intest.Assert(err == nil)
+
+	kvCnt := 0
+	for iter.Next() {
+		kvCnt++
+		if kvCnt == s.totalKVCnt/2 {
+			if s.beforeReaderClose != nil {
+				s.beforeReaderClose()
+			}
+		}
+	}
+	intest.Assert(kvCnt == s.totalKVCnt)
+	err = iter.Close()
+	intest.Assert(err == nil)
+	if s.afterReaderClose != nil {
+		s.afterReaderClose()
+	}
+}
+
+func TestCompareReader(t *testing.T) {
+	fileSize := 50 * 1024 * 1024
+	fileCnt := 24
+	store, kvCnt := createEvenlyDistributedFiles(t, fileSize, fileCnt)
+	memoryLimit := 64 * 1024 * 1024
+	fileIdx := 0
+	var (
+		now     time.Time
+		elapsed time.Duration
+		file    *os.File
+		err     error
+	)
+	beforeTest := func() {
+		fileIdx++
+		file, err = os.Create(fmt.Sprintf("cpu-profile-%d.prof", fileIdx))
+		intest.Assert(err == nil)
+		err = pprof.StartCPUProfile(file)
+		intest.Assert(err == nil)
+		now = time.Now()
+	}
+	beforeClose := func() {
+		file, err = os.Create(fmt.Sprintf("heap-profile-%d.prof", fileIdx))
+		intest.Assert(err == nil)
+		// check heap profile to see the memory usage is expected
+		err = pprof.WriteHeapProfile(file)
+		intest.Assert(err == nil)
+	}
+	afterClose := func() {
+		elapsed = time.Since(now)
+		pprof.StopCPUProfile()
+	}
+
+	suite := &readTestSuite{
+		store:              store,
+		totalKVCnt:         kvCnt,
+		concurrency:        100,
+		memoryLimit:        memoryLimit,
+		beforeCreateReader: beforeTest,
+		beforeReaderClose:  beforeClose,
+		afterReaderClose:   afterClose,
+	}
+	readFileSequential(suite)
+	t.Logf(
+		"sequential read speed for %d bytes: %.2f MB/s",
+		fileSize*fileCnt,
+		float64(fileSize*fileCnt)/elapsed.Seconds()/1024/1024,
+	)
+
+	readFileConcurrently(suite)
+	t.Logf(
+		"concurrent read speed for %d bytes: %.2f MB/s",
+		fileSize*fileCnt,
+		float64(fileSize*fileCnt)/elapsed.Seconds()/1024/1024,
+	)
+
+	readMergeIter(suite)
+	t.Logf(
+		"merge iter read speed for %d bytes: %.2f MB/s",
+		fileSize*fileCnt,
+		float64(fileSize*fileCnt)/elapsed.Seconds()/1024/1024,
+	)
+}
diff --git a/pkg/server/conn_stmt_params_test.go b/pkg/server/conn_stmt_params_test.go
index beaa275692..61dd33ffcc 100644
--- a/pkg/server/conn_stmt_params_test.go
+++ b/pkg/server/conn_stmt_params_test.go
@@ -214,7 +214,7 @@ func TestParseExecArgs(t *testing.T) {
 			},
 			nil,
 			types.ErrTruncatedWrongVal,
-			types.Duration{Duration: types.MinTime, Fsp: types.MaxFsp},
+			types.Duration{Duration: types.MinTime, Fsp: 0},
 		},
 		{
 			args{
@@ -226,7 +226,7 @@ func TestParseExecArgs(t *testing.T) {
 			},
 			nil,
 			nil,
-			types.Duration{Duration: time.Duration(0), Fsp: types.MaxFsp},
+			types.Duration{Duration: time.Duration(0), Fsp: 0},
 		},
 		// For error test
 		{
diff --git a/pkg/util/intest/BUILD.bazel b/pkg/util/intest/BUILD.bazel
index bab02008d0..d34d4b1cbd 100644
--- a/pkg/util/intest/BUILD.bazel
+++ b/pkg/util/intest/BUILD.bazel
@@ -18,6 +18,7 @@ go_test(
     flaky = True,
     deps = [
         ":intest",
+        "@com_github_pingcap_errors//:errors",
         "@com_github_stretchr_testify//require",
     ],
 )
diff --git a/pkg/util/intest/assert.go b/pkg/util/intest/assert.go
index c40d6cc71a..1d96c38ff1 100644
--- a/pkg/util/intest/assert.go
+++ b/pkg/util/intest/assert.go
@@ -21,63 +21,66 @@ import (
 	"reflect"
 )
 
-// Assert asserts a condition. It only works in test (intest.InTest == true).
-// You can assert a condition like this to assert a variable `foo` is not nil: `assert.Assert(foo != nil)`.
-// Or you can pass foo as a parameter directly for simple: `assert.Assert(foo)`
-// You can also assert a function that returns a bool: `intest.Assert(func() bool { return foo != nil })`
-// If you pass a function without a signature `func() bool`, the function will always panic.
-func Assert(cond any, msgAndArgs ...any) {
+// Assert asserts a condition is true
+func Assert(cond bool, msgAndArgs ...any) {
+	if InTest && !cond {
+		doPanic("", msgAndArgs...)
+	}
+}
+
+// AssertNoError asserts an error is nil
+func AssertNoError(err error, msgAndArgs ...any) {
+	if InTest && err != nil {
+		doPanic(fmt.Sprintf("error is not nil: %+v", err), msgAndArgs...)
+	}
+}
+
+// AssertNotNil asserts an object is not nil
+func AssertNotNil(obj any, msgAndArgs ...any) {
 	if InTest {
-		assert(cond, msgAndArgs...)
+		Assert(obj != nil, msgAndArgs...)
+		value := reflect.ValueOf(obj)
+		switch value.Kind() {
+		case reflect.Func, reflect.Chan, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice, reflect.UnsafePointer:
+			Assert(!value.IsNil(), msgAndArgs...)
+		}
 	}
 }
 
 // AssertFunc asserts a function condition
 func AssertFunc(fn func() bool, msgAndArgs ...any) {
 	if InTest {
-		assert(fn(), msgAndArgs...)
+		Assert(fn != nil, msgAndArgs...)
+		Assert(fn(), msgAndArgs...)
 	}
 }
 
-func assert(cond any, msgAndArgs ...any) {
-	if !checkAssertObject(cond) {
-		doPanic(msgAndArgs...)
-	}
+func doPanic(extraMsg string, userMsgAndArgs ...any) {
+	panic(assertionFailedMsg(extraMsg, userMsgAndArgs...))
 }
 
-func doPanic(msgAndArgs ...any) {
-	panic(assertionFailedMsg(msgAndArgs...))
-}
+func assertionFailedMsg(extraMsg string, userMsgAndArgs ...any) string {
+	msg := "assert failed"
+	if len(userMsgAndArgs) == 0 {
+		if extraMsg != "" {
+			msg = fmt.Sprintf("%s, %s", msg, extraMsg)
+		}
+		return msg
+	}
 
-func assertionFailedMsg(msgAndArgs ...any) string {
-	if len(msgAndArgs) == 0 {
-		return "assert failed"
+	if len(userMsgAndArgs) == 0 {
+		return fmt.Sprintf("assert failed, %s", extraMsg)
 	}
 
-	msg, ok := msgAndArgs[0].(string)
+	userMsg, ok := userMsgAndArgs[0].(string)
 	if !ok {
-		msg = fmt.Sprintf("%+v", msgAndArgs[0])
+		userMsg = fmt.Sprintf("%+v", userMsgAndArgs[0])
 	}
 
-	msg = fmt.Sprintf("assert failed: %s", msg)
-	return fmt.Sprintf(msg, msgAndArgs[1:]...)
-}
-
-func checkAssertObject(obj any) bool {
-	if obj == nil {
-		return false
+	msg = fmt.Sprintf("%s, %s", msg, userMsg)
+	if extraMsg != "" {
+		msg = fmt.Sprintf("%s, %s", msg, extraMsg)
 	}
 
-	value := reflect.ValueOf(obj)
-	switch value.Kind() {
-	case reflect.Bool:
-		return obj.(bool)
-	case reflect.Func:
-		panic("you should use `intest.Assert(fn != nil)` to assert a function is not nil, " +
-			"or use `intest.AssertFunc(fn)` to assert a function's return value is true")
-	case reflect.Chan, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice, reflect.UnsafePointer:
-		return !value.IsNil()
-	default:
-		return true
-	}
+	return fmt.Sprintf(msg, userMsgAndArgs[1:]...)
 }
diff --git a/pkg/util/intest/assert_test.go b/pkg/util/intest/assert_test.go
index c85fc8d5b1..6a14cf7939 100644
--- a/pkg/util/intest/assert_test.go
+++ b/pkg/util/intest/assert_test.go
@@ -15,8 +15,10 @@
 package intest_test
 
 import (
+	"strings"
 	"testing"
 
+	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/util/intest"
 	"github.com/stretchr/testify/require"
 )
@@ -26,37 +28,55 @@ type foo struct{}
 func TestAssert(t *testing.T) {
 	require.True(t, intest.InTest)
 	checkAssert(t, true, true)
-	checkAssert(t, "", true)
-	checkAssert(t, "abc", true)
-	checkAssert(t, 0, true)
-	checkAssert(t, 123, true)
-	checkAssert(t, foo{}, true)
-	checkAssert(t, &foo{}, true)
 	checkAssert(t, false, false)
-	checkAssert(t, false, false, "assert failed: msg1", "msg1")
-	checkAssert(t, false, false, "assert failed: msg2 a b 1", "msg2 %s %s %d", "a", "b", 1)
-	checkAssert(t, false, false, "assert failed: 123", 123)
-	checkAssert(t, nil, false)
+	checkAssert(t, false, false, "assert failed, msg1", "msg1")
+	checkAssert(t, false, false, "assert failed, msg2 a b 1", "msg2 %s %s %d", "a", "b", 1)
+	checkAssert(t, false, false, "assert failed, 123", 123)
+	checkAssertNotNil(t, "", true)
+	checkAssertNotNil(t, "abc", true)
+	checkAssertNotNil(t, 0, true)
+	checkAssertNotNil(t, 123, true)
+	checkAssertNotNil(t, foo{}, true)
+	checkAssertNotNil(t, &foo{}, true)
+	checkAssertNotNil(t, nil, false)
+	checkAssertNotNil(t, true, true)
+	checkAssertNotNil(t, false, true)
 	var f *foo
-	checkAssert(t, f, false)
-	checkAssert(t, func() bool { return true }, false, "you should use `intest.Assert(fn != nil)` to assert a function is not nil, or use `intest.AssertFunc(fn)` to assert a function's return value is true")
-	checkAssert(t, func(_ string) bool { return true }, false, "you should use `intest.Assert(fn != nil)` to assert a function is not nil, or use `intest.AssertFunc(fn)` to assert a function's return value is true")
+	checkAssertNotNil(t, f, false)
+	checkAssertNotNil(t, func() bool { return true }, true)
+	checkAssertNotNil(t, func() bool { return false }, true)
+	checkAssertNotNil(t, func(_ string) bool { return true }, true)
+	checkAssertNotNil(t, nil, false, "assert failed, msg1", "msg1")
+	checkAssertNotNil(t, nil, false, "assert failed, msg2 a b 1", "msg2 %s %s %d", "a", "b", 1)
 	checkFuncAssert(t, func() bool { panic("inner panic1") }, false, "inner panic1")
 	checkFuncAssert(t, func() bool { return true }, true)
 	checkFuncAssert(t, func() bool { return false }, false)
-	checkFuncAssert(t, func() bool { return false }, false, "assert failed: msg3", "msg3")
-	checkFuncAssert(t, func() bool { return false }, false, "assert failed: msg4 c d 2", "msg4 %s %s %d", "c", "d", 2)
+	checkFuncAssert(t, func() bool { return false }, false, "assert failed, msg3", "msg3")
+	checkFuncAssert(t, func() bool { return false }, false, "assert failed, msg4 c d 2", "msg4 %s %s %d", "c", "d", 2)
 	checkFuncAssert(t, func() bool { panic("inner panic2") }, false, "inner panic2")
+	checkFuncAssert(t, nil, false)
+	checkAssertNoError(t, nil, true)
+	checkAssertNoError(t, errors.New("mock err1"), false, "assert failed, error is not nil: mock err1")
+	var err error
+	checkAssertNoError(t, err, true)
 }
 
 func checkFuncAssert(t *testing.T, fn func() bool, pass bool, msgAndArgs ...any) {
 	doCheckAssert(t, intest.AssertFunc, fn, pass, msgAndArgs...)
 }
 
-func checkAssert(t *testing.T, cond any, pass bool, msgAndArgs ...any) {
+func checkAssert(t *testing.T, cond bool, pass bool, msgAndArgs ...any) {
 	doCheckAssert(t, intest.Assert, cond, pass, msgAndArgs...)
 }
 
+func checkAssertNotNil(t *testing.T, obj any, pass bool, msgAndArgs ...any) {
+	doCheckAssert(t, intest.AssertNotNil, obj, pass, msgAndArgs...)
+}
+
+func checkAssertNoError(t *testing.T, err error, pass bool, msgAndArgs ...any) {
+	doCheckAssert(t, intest.AssertNoError, err, pass, msgAndArgs...)
+}
+
 func doCheckAssert(t *testing.T, fn any, cond any, pass bool, msgAndArgs ...any) {
 	expectMsg := "assert failed"
 	if len(msgAndArgs) > 0 {
@@ -64,32 +84,39 @@ func doCheckAssert(t *testing.T, fn any, cond any, pass bool, msgAndArgs ...any)
 		msgAndArgs = msgAndArgs[1:]
 	}
 
+	fail := ""
+	onlyCheckPrefix := false
 	if !pass {
 		defer func() {
+			if fail != "" {
+				require.FailNow(t, fail)
+			}
 			r := recover()
 			require.NotNil(t, r)
-			require.Equal(t, expectMsg, r)
+			if onlyCheckPrefix {
+				require.True(t, strings.HasPrefix(r.(string), expectMsg), "%q\nshould have prefix \n%q", r, expectMsg)
+				require.Contains(t, r.(string), expectMsg)
+			} else {
+				require.Equal(t, expectMsg, r)
+			}
 		}()
 	}
 
-	testFn, ok := fn.(func(any, ...any))
-	if !ok {
-		if fnAssert, ok := fn.(func(func() bool, ...any)); ok {
-			testFn = func(any, ...any) {
-				fnAssert(cond.(func() bool), msgAndArgs...)
-			}
+	switch assertFn := fn.(type) {
+	case func(bool, ...any):
+		assertFn(cond.(bool), msgAndArgs...)
+	case func(any, ...any):
+		assertFn(cond, msgAndArgs...)
+	case func(func() bool, ...any):
+		assertFn(cond.(func() bool), msgAndArgs...)
+	case func(error, ...any):
+		if cond == nil {
+			assertFn(nil, msgAndArgs...)
 		} else {
-			require.FailNow(t, "invalid assert function")
+			onlyCheckPrefix = true
+			assertFn(cond.(error), msgAndArgs...)
 		}
+	default:
+		fail = "invalid input assert function"
 	}
-
-	if len(msgAndArgs) == 0 {
-		testFn(cond)
-	}
-
-	if len(msgAndArgs) == 1 {
-		testFn(cond, msgAndArgs[0])
-	}
-
-	testFn(cond, msgAndArgs...)
 }
diff --git a/pkg/util/intest/common.go b/pkg/util/intest/common.go
index 56b928bc9e..a2ebccd4b3 100644
--- a/pkg/util/intest/common.go
+++ b/pkg/util/intest/common.go
@@ -21,7 +21,15 @@ const InTest = false
 
 // Assert is a stub function in release build.
 // See the same function in `util/intest/assert.go` for the real implement in test.
-func Assert(_ any, _ ...any) {}
+func Assert(_ bool, _ ...any) {}
+
+// AssertNotNil is a stub function in release build.
+// See the same function in `util/intest/assert.go` for the real implement in test.
+func AssertNotNil(_ any, _ ...any) {}
+
+// AssertNoError is a stub function in release build.
+// See the same function in `util/intest/assert.go` for the real implement in test.
+func AssertNoError(_ error, _ ...any) {}
 
 // AssertFunc is a stub function in release build.
 // See the same function `util/intest/assert.go` for the real implement in test.