diff --git a/DEPS.bzl b/DEPS.bzl
index 686aef1f3c028..0333a2dc70184 100644
--- a/DEPS.bzl
+++ b/DEPS.bzl
@@ -390,8 +390,8 @@ def go_deps():
         name = "com_github_cespare_xxhash_v2",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/cespare/xxhash/v2",
-        sum = "h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=",
-        version = "v2.1.2",
+        sum = "h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=",
+        version = "v2.2.0",
     )
     go_repository(
         name = "com_github_charithe_durationcheck",
@@ -1143,8 +1143,8 @@ def go_deps():
         name = "com_github_go_kit_log",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/go-kit/log",
-        sum = "h1:7i2K3eKTos3Vc0enKCfnVcgHh2olr/MyfboYq7cAcFw=",
-        version = "v0.2.0",
+        sum = "h1:MRVx0/zhvdseW+Gza6N9rVzU/IVzaeE1SFI4raAhmBU=",
+        version = "v0.2.1",
     )
     go_repository(
         name = "com_github_go_logfmt_logfmt",
@@ -2182,8 +2182,8 @@ def go_deps():
         name = "com_github_klauspost_compress",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/klauspost/compress",
-        sum = "h1:y9FcTHGyrebwfP0ZZqFiaxTaiDnUrGkJkI+f583BL1A=",
-        version = "v1.15.1",
+        sum = "h1:NFn1Wr8cfnenSJSA46lLq4wHCcBzKTSjnBIexDMMOV0=",
+        version = "v1.15.13",
     )
     go_repository(
         name = "com_github_klauspost_cpuid",
@@ -2435,8 +2435,8 @@ def go_deps():
         name = "com_github_matttproud_golang_protobuf_extensions",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/matttproud/golang_protobuf_extensions",
-        sum = "h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=",
-        version = "v1.0.1",
+        sum = "h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=",
+        version = "v1.0.4",
     )
     go_repository(
         name = "com_github_maxatome_go_testdeep",
@@ -2881,8 +2881,8 @@ def go_deps():
         name = "com_github_pingcap_badger",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/pingcap/badger",
-        sum = "h1:MKVFZuqFvAMiDtv3AbihOQ6rY5IE8LWflI1BuZ/hF0Y=",
-        version = "v1.5.1-0.20220314162537-ab58fbf40580",
+        sum = "h1:QB16qn8wx5X4SRn3/5axrjPMNS3WRt87+5Bfrnmt6IA=",
+        version = "v1.5.1-0.20221229114011-ddffaa0fff7a",
     )
     go_repository(
         name = "com_github_pingcap_check",
@@ -2895,8 +2895,8 @@ def go_deps():
         name = "com_github_pingcap_errors",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/pingcap/errors",
-        sum = "h1:3Dm0DWeQlwV8LbpQxP2tojHhxd9aY59KI+QN0ns6bBo=",
-        version = "v0.11.5-0.20220729040631-518f63d66278",
+        sum = "h1:m5ZsBa5o/0CkzZXfXLaThzKuR85SnHHetqBCpzQ30h8=",
+        version = "v0.11.5-0.20221009092201-b66cddb77c32",
     )
     go_repository(
         name = "com_github_pingcap_failpoint",
@@ -3001,29 +3001,29 @@ def go_deps():
         name = "com_github_prometheus_client_golang",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/client_golang",
-        sum = "h1:b71QUfeo5M8gq2+evJdTPfZhYMAU0uKPkyPJ7TPsloU=",
-        version = "v1.13.0",
+        sum = "h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=",
+        version = "v1.14.0",
     )
     go_repository(
         name = "com_github_prometheus_client_model",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/client_model",
-        sum = "h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M=",
-        version = "v0.2.0",
+        sum = "h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4=",
+        version = "v0.3.0",
     )
     go_repository(
         name = "com_github_prometheus_common",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/common",
-        sum = "h1:ccBbHCgIiT9uSoFY0vX8H3zsNR5eLt17/RQLUvn8pXE=",
-        version = "v0.37.0",
+        sum = "h1:oOyhkDq05hPZKItWVBkJ6g6AtGxi+fy7F4JvUV8uhsI=",
+        version = "v0.39.0",
     )
     go_repository(
         name = "com_github_prometheus_procfs",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/prometheus/procfs",
-        sum = "h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=",
-        version = "v0.8.0",
+        sum = "h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI=",
+        version = "v0.9.0",
     )
     go_repository(
         name = "com_github_prometheus_prometheus",
@@ -4608,8 +4608,8 @@ def go_deps():
         name = "org_uber_go_multierr",
         build_file_proto_mode = "disable_global",
         importpath = "go.uber.org/multierr",
-        sum = "h1:dg6GjLku4EH+249NNmoIciG9N/jURbDG+pFlTkhzIC8=",
-        version = "v1.8.0",
+        sum = "h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=",
+        version = "v1.9.0",
     )
     go_repository(
         name = "org_uber_go_tools",
@@ -4622,6 +4622,6 @@ def go_deps():
         name = "org_uber_go_zap",
         build_file_proto_mode = "disable_global",
         importpath = "go.uber.org/zap",
-        sum = "h1:OjGQ5KQDEUawVHxNwQgPpiypGHOxo2mNZsOqTak4fFY=",
-        version = "v1.23.0",
+        sum = "h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=",
+        version = "v1.24.0",
     )
diff --git a/executor/admin.go b/executor/admin.go
index a0484ce957b30..21378b21b1677 100644
--- a/executor/admin.go
+++ b/executor/admin.go
@@ -163,7 +163,7 @@ func (e *CheckIndexRangeExec) constructIndexScanPB() *tipb.Executor {
 	idxExec := &tipb.IndexScan{
 		TableId: e.table.ID,
 		IndexId: e.index.ID,
-		Columns: util.ColumnsToProto(e.cols, e.table.PKIsHandle),
+		Columns: util.ColumnsToProto(e.cols, e.table.PKIsHandle, true),
 	}
 	return &tipb.Executor{Tp: tipb.ExecType_TypeIndexScan, IdxScan: idxExec}
 }
@@ -814,7 +814,7 @@ func (e *CleanupIndexExec) constructIndexScanPB() *tipb.Executor {
 	idxExec := &tipb.IndexScan{
 		TableId:          e.physicalID,
 		IndexId:          e.index.Meta().ID,
-		Columns:          util.ColumnsToProto(e.columns, e.table.Meta().PKIsHandle),
+		Columns:          util.ColumnsToProto(e.columns, e.table.Meta().PKIsHandle, true),
 		PrimaryColumnIds: tables.TryGetCommonPkColumnIds(e.table.Meta()),
 	}
 	return &tipb.Executor{Tp: tipb.ExecType_TypeIndexScan, IdxScan: idxExec}
diff --git a/executor/builder.go b/executor/builder.go
index c6baaf0932849..f771c706bfa9b 100644
--- a/executor/builder.go
+++ b/executor/builder.go
@@ -2580,7 +2580,7 @@ func (b *executorBuilder) buildAnalyzeSamplingPushdown(task plannercore.AnalyzeC
 		SampleSize:   int64(opts[ast.AnalyzeOptNumSamples]),
 		SampleRate:   sampleRate,
 		SketchSize:   maxSketchSize,
-		ColumnsInfo:  util.ColumnsToProto(task.ColsInfo, task.TblInfo.PKIsHandle),
+		ColumnsInfo:  util.ColumnsToProto(task.ColsInfo, task.TblInfo.PKIsHandle, false),
 		ColumnGroups: colGroups,
 	}
 	if task.TblInfo != nil {
@@ -2741,7 +2741,7 @@ func (b *executorBuilder) buildAnalyzeColumnsPushdown(task plannercore.AnalyzeCo
 		BucketSize:    int64(opts[ast.AnalyzeOptNumBuckets]),
 		SampleSize:    MaxRegionSampleSize,
 		SketchSize:    maxSketchSize,
-		ColumnsInfo:   util.ColumnsToProto(cols, task.HandleCols != nil && task.HandleCols.IsInt()),
+		ColumnsInfo:   util.ColumnsToProto(cols, task.HandleCols != nil && task.HandleCols.IsInt(), false),
 		CmsketchDepth: &depth,
 		CmsketchWidth: &width,
 	}
diff --git a/expression/expr_to_pb_test.go b/expression/expr_to_pb_test.go
index 1025f3c7fdcb1..341c03b552804 100644
--- a/expression/expr_to_pb_test.go
+++ b/expression/expr_to_pb_test.go
@@ -1239,6 +1239,11 @@ func TestExprPushDownToFlash(t *testing.T) {
 	require.Equal(t, tipb.ScalarFuncSig_CastTimeAsDuration, function.(*ScalarFunction).Function.PbCode())
 	exprs = append(exprs, function)
 
+	// Unhex
+	function, err = NewFunction(mock.NewContext(), ast.Unhex, types.NewFieldType(mysql.TypeString), stringColumn)
+	require.NoError(t, err)
+	exprs = append(exprs, function)
+
 	pushed, remained = PushDownExprs(sc, exprs, client, kv.TiFlash)
 	require.Len(t, pushed, len(exprs))
 	require.Len(t, remained, 0)
diff --git a/expression/expression.go b/expression/expression.go
index c7ce0b22c1c0e..00697c2df68ea 100644
--- a/expression/expression.go
+++ b/expression/expression.go
@@ -1267,7 +1267,7 @@ func scalarExprSupportedByFlash(function *ScalarFunction) bool {
 		}
 	case ast.IsTruthWithNull, ast.IsTruthWithoutNull, ast.IsFalsity:
 		return true
-	case ast.Hex, ast.Bin:
+	case ast.Hex, ast.Unhex, ast.Bin:
 		return true
 	case ast.GetFormat:
 		return true
diff --git a/expression/expression_test.go b/expression/expression_test.go
index e80cf5c41d575..4a24cb5eef759 100644
--- a/expression/expression_test.go
+++ b/expression/expression_test.go
@@ -77,7 +77,7 @@ func TestEvaluateExprWithNullAndParameters(t *testing.T) {
 	res = EvaluateExprWithNull(ctx, schema, ltWithParam)
 	_, isConst := res.(*Constant)
 	require.True(t, isConst) // this expression is evaluated and skip-plan cache flag is set.
-	require.True(t, ctx.GetSessionVars().StmtCtx.SkipPlanCache)
+	require.True(t, !ctx.GetSessionVars().StmtCtx.UseCache)
 }
 
 func TestEvaluateExprWithNullNoChangeRetType(t *testing.T) {
diff --git a/expression/multi_valued_index_test.go b/expression/multi_valued_index_test.go
index 33557726a1073..97e59993e23b4 100644
--- a/expression/multi_valued_index_test.go
+++ b/expression/multi_valued_index_test.go
@@ -15,11 +15,20 @@
 package expression_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
 	"github.com/pingcap/tidb/errno"
+	"github.com/pingcap/tidb/kv"
+	"github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/sessiontxn"
+	"github.com/pingcap/tidb/table"
+	"github.com/pingcap/tidb/tablecodec"
 	"github.com/pingcap/tidb/testkit"
+	"github.com/pingcap/tidb/types"
+	"github.com/pingcap/tidb/util/codec"
+	"github.com/stretchr/testify/require"
 )
 
 func TestMultiValuedIndexDDL(t *testing.T) {
@@ -225,3 +234,251 @@ func TestMultiValuedIndexDML(t *testing.T) {
 		tk.MustGetErrCode(`insert into t values (json_array(cast('{"a":1}' as json)));`, errno.ErrInvalidJSONValueForFuncIndex)
 	}
 }
+
+func TestWriteMultiValuedIndex(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1(pk int primary key, a json, index idx((cast(a as signed array))))")
+	tk.MustExec("insert into t1 values (1, '[1,2,2,3]')")
+	tk.MustExec("insert into t1 values (2, '[1,2,3]')")
+	tk.MustExec("insert into t1 values (3, '[]')")
+	tk.MustExec("insert into t1 values (4, '[2,3,4]')")
+	tk.MustExec("insert into t1 values (5, null)")
+
+	t1, err := dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	require.NoError(t, err)
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 10)
+			checkKey(t, t1.IndexPrefix(), index, store, [][]types.Datum{
+				{types.NewDatum(nil), types.NewIntDatum(5)},
+				{types.NewIntDatum(1), types.NewIntDatum(1)},
+				{types.NewIntDatum(1), types.NewIntDatum(2)},
+				{types.NewIntDatum(2), types.NewIntDatum(1)},
+				{types.NewIntDatum(2), types.NewIntDatum(2)},
+				{types.NewIntDatum(2), types.NewIntDatum(4)},
+				{types.NewIntDatum(3), types.NewIntDatum(1)},
+				{types.NewIntDatum(3), types.NewIntDatum(2)},
+				{types.NewIntDatum(3), types.NewIntDatum(4)},
+				{types.NewIntDatum(4), types.NewIntDatum(4)},
+			})
+		}
+	}
+	tk.MustExec("delete from t1")
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 0)
+		}
+	}
+
+	tk.MustExec("drop table t1")
+	tk.MustExec("create table t1(pk int primary key, a json, index idx((cast(a as char(5) array))))")
+	tk.MustExec("insert into t1 values (1, '[\"abc\", \"abc \"]')")
+	tk.MustExec("insert into t1 values (2, '[\"b\"]')")
+	tk.MustExec("insert into t1 values (3, '[\"b   \"]')")
+
+	t1, err = dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	require.NoError(t, err)
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 4)
+			checkKey(t, t1.IndexPrefix(), index, store, [][]types.Datum{
+				{types.NewBytesDatum([]byte("abc")), types.NewIntDatum(1)},
+				{types.NewBytesDatum([]byte("abc ")), types.NewIntDatum(1)},
+				{types.NewBytesDatum([]byte("b")), types.NewIntDatum(2)},
+				{types.NewBytesDatum([]byte("b   ")), types.NewIntDatum(3)},
+			})
+		}
+	}
+
+	tk.MustExec("update t1 set a = json_array_append(a, '$', 'bcd') where pk = 1")
+	tk.MustExec("update t1 set a = '[]' where pk = 2")
+	tk.MustExec("update t1 set a = '[\"abc\"]' where pk = 3")
+
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 4)
+			checkKey(t, t1.IndexPrefix(), index, store, [][]types.Datum{
+				{types.NewBytesDatum([]byte("abc")), types.NewIntDatum(1)},
+				{types.NewBytesDatum([]byte("abc")), types.NewIntDatum(3)},
+				{types.NewBytesDatum([]byte("abc ")), types.NewIntDatum(1)},
+				{types.NewBytesDatum([]byte("bcd")), types.NewIntDatum(1)},
+			})
+		}
+	}
+
+	tk.MustExec("delete from t1")
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 0)
+		}
+	}
+}
+
+func TestWriteMultiValuedIndexPartitionTable(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec(`create table t1
+(
+    pk int primary key,
+    a  json,
+    index idx ((cast(a as signed array)))
+) partition by range columns (pk) (partition p0 values less than (10), partition p1 values less than (20));`)
+	tk.MustExec("insert into t1 values (1, '[1,2,2,3]')")
+	tk.MustExec("insert into t1 values (11, '[1,2,3]')")
+	tk.MustExec("insert into t1 values (2, '[]')")
+	tk.MustExec("insert into t1 values (12, '[2,3,4]')")
+	tk.MustExec("insert into t1 values (3, null)")
+	tk.MustExec("insert into t1 values (13, null)")
+
+	t1, err := dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	require.NoError(t, err)
+
+	expect := map[string]struct {
+		count int
+		vals  [][]types.Datum
+	}{
+		"p0": {4, [][]types.Datum{
+			{types.NewDatum(nil), types.NewIntDatum(3)},
+			{types.NewIntDatum(1), types.NewIntDatum(1)},
+			{types.NewIntDatum(2), types.NewIntDatum(1)},
+			{types.NewIntDatum(3), types.NewIntDatum(1)},
+		}},
+		"p1": {7, [][]types.Datum{
+			{types.NewDatum(nil), types.NewIntDatum(13)},
+			{types.NewIntDatum(1), types.NewIntDatum(11)},
+			{types.NewIntDatum(2), types.NewIntDatum(11)},
+			{types.NewIntDatum(2), types.NewIntDatum(12)},
+			{types.NewIntDatum(3), types.NewIntDatum(11)},
+			{types.NewIntDatum(3), types.NewIntDatum(12)},
+			{types.NewIntDatum(4), types.NewIntDatum(12)},
+		}},
+	}
+
+	for _, def := range t1.Meta().GetPartitionInfo().Definitions {
+		partition := t1.(table.PartitionedTable).GetPartition(def.ID)
+		for _, index := range partition.Indices() {
+			if index.Meta().MVIndex {
+				checkCount(t, partition.IndexPrefix(), index, store, expect[def.Name.L].count)
+				checkKey(t, partition.IndexPrefix(), index, store, expect[def.Name.L].vals)
+			}
+		}
+	}
+
+	tk.MustExec("delete from t1")
+	for _, def := range t1.Meta().GetPartitionInfo().Definitions {
+		partition := t1.(table.PartitionedTable).GetPartition(def.ID)
+		for _, index := range partition.Indices() {
+			if index.Meta().MVIndex {
+				checkCount(t, partition.IndexPrefix(), index, store, 0)
+			}
+		}
+	}
+}
+
+func TestWriteMultiValuedIndexUnique(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1(pk int primary key, a json, unique index idx((cast(a as signed array))))")
+	tk.MustExec("insert into t1 values (1, '[1,2,2]')")
+	tk.MustGetErrCode("insert into t1 values (2, '[1]')", errno.ErrDupEntry)
+	tk.MustExec("insert into t1 values (3, '[3,3,4]')")
+
+	t1, err := dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	require.NoError(t, err)
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 4)
+			checkKey(t, t1.IndexPrefix(), index, store, [][]types.Datum{
+				{types.NewIntDatum(1)},
+				{types.NewIntDatum(2)},
+				{types.NewIntDatum(3)},
+				{types.NewIntDatum(4)},
+			})
+		}
+	}
+}
+
+func TestWriteMultiValuedIndexComposite(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1(pk int primary key, a json, c int, d int, index idx(c, (cast(a as signed array)), d))")
+	tk.MustExec("insert into t1 values (1, '[1,2,2]', 1, 1)")
+	tk.MustExec("insert into t1 values (2, '[2,2,2]', 2, 2)")
+	tk.MustExec("insert into t1 values (3, '[3,3,4]', 3, 3)")
+	tk.MustExec("insert into t1 values (4, null, 4, 4)")
+	tk.MustExec("insert into t1 values (5, '[]', 5, 5)")
+
+	t1, err := dom.InfoSchema().TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	require.NoError(t, err)
+	for _, index := range t1.Indices() {
+		if index.Meta().MVIndex {
+			checkCount(t, t1.IndexPrefix(), index, store, 6)
+			checkKey(t, t1.IndexPrefix(), index, store, [][]types.Datum{
+				{types.NewIntDatum(1), types.NewIntDatum(1), types.NewIntDatum(1), types.NewIntDatum(1)},
+				{types.NewIntDatum(1), types.NewIntDatum(2), types.NewIntDatum(1), types.NewIntDatum(1)},
+				{types.NewIntDatum(2), types.NewIntDatum(2), types.NewIntDatum(2), types.NewIntDatum(2)},
+				{types.NewIntDatum(3), types.NewIntDatum(3), types.NewIntDatum(3), types.NewIntDatum(3)},
+				{types.NewIntDatum(3), types.NewIntDatum(4), types.NewIntDatum(3), types.NewIntDatum(3)},
+				{types.NewIntDatum(4), types.NewDatum(nil), types.NewIntDatum(4), types.NewIntDatum(4)},
+			})
+		}
+	}
+}
+
+func checkCount(t *testing.T, prefix kv.Key, index table.Index, store kv.Storage, except int) {
+	c := 0
+	checkIndex(t, prefix, index, store, func(it kv.Iterator) {
+		c++
+	})
+	require.Equal(t, except, c)
+}
+
+func checkKey(t *testing.T, prefix kv.Key, index table.Index, store kv.Storage, except [][]types.Datum) {
+	idx := 0
+	checkIndex(t, prefix, index, store, func(it kv.Iterator) {
+		indexKey := decodeIndexKey(t, it.Key())
+		require.Equal(t, except[idx], indexKey)
+		idx++
+	})
+}
+
+func checkIndex(t *testing.T, prefix kv.Key, index table.Index, store kv.Storage, fn func(kv.Iterator)) {
+	startKey := codec.EncodeInt(prefix, index.Meta().ID)
+	prefix.Next()
+	se := testkit.NewTestKit(t, store).Session()
+	err := sessiontxn.NewTxn(context.Background(), se)
+	require.NoError(t, err)
+	txn, err := se.Txn(true)
+	require.NoError(t, err)
+	it, err := txn.Iter(startKey, prefix.PrefixNext())
+	require.NoError(t, err)
+	for it.Valid() && it.Key().HasPrefix(prefix) {
+		fn(it)
+		err = it.Next()
+		require.NoError(t, err)
+	}
+	it.Close()
+	se.Close()
+}
+
+func decodeIndexKey(t *testing.T, key kv.Key) []types.Datum {
+	var idLen = 8
+	var prefixLen = 1 + idLen /*tableID*/ + 2
+	_, _, isRecord, err := tablecodec.DecodeKeyHead(key)
+	require.NoError(t, err)
+	require.False(t, isRecord)
+	indexKey := key[prefixLen+idLen:]
+	var datumValues []types.Datum
+	for len(indexKey) > 0 {
+		remain, d, err := codec.DecodeOne(indexKey)
+		require.NoError(t, err)
+		datumValues = append(datumValues, d)
+		indexKey = remain
+	}
+	return datumValues
+}
diff --git a/go.mod b/go.mod
index b7375dddbe3dc..d87619f7dc67c 100644
--- a/go.mod
+++ b/go.mod
@@ -57,7 +57,7 @@ require (
 	github.com/jingyugao/rowserrcheck v1.1.1
 	github.com/joho/sqltocsv v0.0.0-20210428211105-a6d6801d59df
 	github.com/kisielk/errcheck v1.6.2
-	github.com/klauspost/compress v1.15.1
+	github.com/klauspost/compress v1.15.13
 	github.com/kyoh86/exportloopref v0.1.8
 	github.com/lestrrat-go/jwx/v2 v2.0.6
 	github.com/mgechev/revive v1.2.4
@@ -66,8 +66,8 @@ require (
 	github.com/opentracing/basictracer-go v1.0.0
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
-	github.com/pingcap/badger v1.5.1-0.20220314162537-ab58fbf40580
-	github.com/pingcap/errors v0.11.5-0.20220729040631-518f63d66278
+	github.com/pingcap/badger v1.5.1-0.20221229114011-ddffaa0fff7a
+	github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32
 	github.com/pingcap/failpoint v0.0.0-20220423142525-ae43b7f4e5c3
 	github.com/pingcap/fn v0.0.0-20200306044125-d5540d389059
 	github.com/pingcap/kvproto v0.0.0-20221213093948-9ccc6beaf0aa
@@ -76,9 +76,9 @@ require (
 	github.com/pingcap/tidb/parser v0.0.0-20211011031125-9b13dc409c5e
 	github.com/pingcap/tipb v0.0.0-20221123081521-2fb828910813
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.13.0
-	github.com/prometheus/client_model v0.2.0
-	github.com/prometheus/common v0.37.0
+	github.com/prometheus/client_golang v1.14.0
+	github.com/prometheus/client_model v0.3.0
+	github.com/prometheus/common v0.39.0
 	github.com/prometheus/prometheus v0.0.0-20190525122359-d20e84d0fb64
 	github.com/shirou/gopsutil/v3 v3.22.9
 	github.com/shurcooL/httpgzip v0.0.0-20190720172056-320755c1c1b0
@@ -107,8 +107,8 @@ require (
 	go.uber.org/atomic v1.10.0
 	go.uber.org/automaxprocs v1.4.0
 	go.uber.org/goleak v1.2.0
-	go.uber.org/multierr v1.8.0
-	go.uber.org/zap v1.23.0
+	go.uber.org/multierr v1.9.0
+	go.uber.org/zap v1.24.0
 	golang.org/x/exp v0.0.0-20221023144134-a1e5550cf13e
 	golang.org/x/net v0.4.0
 	golang.org/x/oauth2 v0.3.0
@@ -140,7 +140,7 @@ require (
 	github.com/benbjohnson/clock v1.3.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash v1.1.0 // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/chavacava/garif v0.0.0-20220630083739-93517212f375 // indirect
 	github.com/cockroachdb/logtags v0.0.0-20190617123548-eb05cc24525f // indirect
 	github.com/cockroachdb/redact v1.0.8 // indirect
@@ -195,7 +195,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mattn/go-runewidth v0.0.14 // indirect
-	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/nbutton23/zxcvbn-go v0.0.0-20210217022336-fa2cb2858354 // indirect
@@ -209,7 +209,7 @@ require (
 	github.com/pkg/browser v0.0.0-20180916011732-0a3d74bf9ce4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
-	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/prometheus/procfs v0.9.0 // indirect
 	github.com/prometheus/tsdb v0.8.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
diff --git a/go.sum b/go.sum
index 2a6c8d4fc308e..efeb0a0fb5537 100644
--- a/go.sum
+++ b/go.sum
@@ -151,8 +151,9 @@ github.com/certifi/gocertifi v0.0.0-20200922220541-2c3bb06c6054/go.mod h1:sGbDF6
 github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
-github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/charithe/durationcheck v0.0.9 h1:mPP4ucLrf/rKZiIG/a9IPXHGlh8p4CzgpyTy6EEutYk=
 github.com/charithe/durationcheck v0.0.9/go.mod h1:SSbRIBVfMjCi/kEB6K65XEA83D6prSM8ap1UCpNKtgg=
 github.com/chavacava/garif v0.0.0-20220630083739-93517212f375 h1:E7LT642ysztPWE0dfz43cWOvMiF42DyTRC+eZIaO4yI=
@@ -317,7 +318,6 @@ github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2
 github.com/go-kit/kit v0.9.0 h1:wDJmvq38kDhkVxi50ni9ykkdUr1PKgqKOoi01fa0Mdk=
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
-github.com/go-kit/log v0.2.0/go.mod h1:NwTd00d/i8cPZ3xOwwiv2PO5MOcx78fFErGNcVmBjv0=
 github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE=
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
@@ -614,8 +614,8 @@ github.com/klauspost/compress v1.9.7/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
 github.com/klauspost/compress v1.10.5/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/compress v1.11.7/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
 github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
-github.com/klauspost/compress v1.15.1 h1:y9FcTHGyrebwfP0ZZqFiaxTaiDnUrGkJkI+f583BL1A=
-github.com/klauspost/compress v1.15.1/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.15.13 h1:NFn1Wr8cfnenSJSA46lLq4wHCcBzKTSjnBIexDMMOV0=
+github.com/klauspost/compress v1.15.13/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
 github.com/klauspost/cpuid v1.3.1 h1:5JNjFYYQrZeKRJ0734q51WCEEn2huer72Dc7K+R/b6s=
 github.com/klauspost/cpuid v1.3.1/go.mod h1:bYW4mA6ZgKPob1/Dlai2LviZJO7KGI3uoWLd42rAQw4=
@@ -674,8 +674,9 @@ github.com/mattn/go-runewidth v0.0.13/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh
 github.com/mattn/go-runewidth v0.0.14 h1:+xnbZSEeDbOIg5/mE6JF0w6n9duR1l3/WmbinWVwUuU=
 github.com/mattn/go-runewidth v0.0.14/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/mattn/goveralls v0.0.2/go.mod h1:8d1ZMHsd7fW6IRPKQh46F2WRpyib5/X4FOpevwGNQEw=
-github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
+github.com/matttproud/golang_protobuf_extensions v1.0.4 h1:mmDVorXM7PCGKw94cs5zkfA9PSy5pEvNWRP0ET0TIVo=
+github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
 github.com/maxatome/go-testdeep v1.11.0 h1:Tgh5efyCYyJFGUYiT0qxBSIDeXw0F5zSoatlou685kk=
 github.com/mediocregopher/mediocre-go-lib v0.0.0-20181029021733-cb65787f37ed/go.mod h1:dSsfyI2zABAdhcbvkXqgxOxrCsbYeHCPgrZkku60dSg=
 github.com/mediocregopher/radix/v3 v3.3.0/go.mod h1:EmfVyvspXz1uZEyPBMyGK+kjWiKQGvsUt6O3Pj+LDCQ=
@@ -760,8 +761,8 @@ github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2/go.mod h1:iIss55rK
 github.com/pierrec/lz4 v2.6.0+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
 github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
-github.com/pingcap/badger v1.5.1-0.20220314162537-ab58fbf40580 h1:MKVFZuqFvAMiDtv3AbihOQ6rY5IE8LWflI1BuZ/hF0Y=
-github.com/pingcap/badger v1.5.1-0.20220314162537-ab58fbf40580/go.mod h1:upwDfet29M5y5koWilbWWA6ca3Lr0YVuzwX/DK58Vdk=
+github.com/pingcap/badger v1.5.1-0.20221229114011-ddffaa0fff7a h1:QB16qn8wx5X4SRn3/5axrjPMNS3WRt87+5Bfrnmt6IA=
+github.com/pingcap/badger v1.5.1-0.20221229114011-ddffaa0fff7a/go.mod h1:p8QnkZnmyV8L/M/jzYb8rT7kv3bz9m7bn1Ju94wDifs=
 github.com/pingcap/check v0.0.0-20190102082844-67f458068fc8/go.mod h1:B1+S9LNcuMyLH/4HMTViQOJevkGiik3wW2AN9zb2fNQ=
 github.com/pingcap/check v0.0.0-20191107115940-caf2b9e6ccf4/go.mod h1:PYMCGwN0JHjoqGr3HrZoD+b8Tgx8bKnArhSq8YVzUMc=
 github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712 h1:R8gStypOBmpnHEx1qi//SaqxJVI4inOqljg/Aj5/390=
@@ -770,8 +771,8 @@ github.com/pingcap/errors v0.11.0/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTw
 github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pingcap/errors v0.11.5-0.20190809092503-95897b64e011/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
 github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
-github.com/pingcap/errors v0.11.5-0.20220729040631-518f63d66278 h1:3Dm0DWeQlwV8LbpQxP2tojHhxd9aY59KI+QN0ns6bBo=
-github.com/pingcap/errors v0.11.5-0.20220729040631-518f63d66278/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
+github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32 h1:m5ZsBa5o/0CkzZXfXLaThzKuR85SnHHetqBCpzQ30h8=
+github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32/go.mod h1:X2r9ueLEUZgtx2cIogM0v4Zj5uvvzhuuiu7Pn8HzMPg=
 github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew=
 github.com/pingcap/failpoint v0.0.0-20220423142525-ae43b7f4e5c3 h1:kJolJWbyadVeL8RKBlqmXQR7FRKPsIeU85TUYyhbhiQ=
 github.com/pingcap/failpoint v0.0.0-20220423142525-ae43b7f4e5c3/go.mod h1:4qGtCB0QK0wBzKtFEGDhxXnSnbQApw1gc9siScUl8ew=
@@ -784,8 +785,8 @@ github.com/pingcap/kvproto v0.0.0-20221026112947-f8d61344b172/go.mod h1:OYtxs078
 github.com/pingcap/kvproto v0.0.0-20221213093948-9ccc6beaf0aa h1:v0Z0nC0knwWHn3e9br8EMNfLBB14QDULn142UGjiTMQ=
 github.com/pingcap/kvproto v0.0.0-20221213093948-9ccc6beaf0aa/go.mod h1:OYtxs0786qojVTmkVeufx93xe+jUgm56GUYRIKnmaGI=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
-github.com/pingcap/log v0.0.0-20200511115504-543df19646ad/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
+github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.1-0.20221015072633-39906604fb81/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.1-0.20221116035753-734d527bc87c h1:crhkw6DD+07Bg1wYhW5Piw+kYNKZqFQqfC2puUf6gMI=
 github.com/pingcap/log v1.1.1-0.20221116035753-734d527bc87c/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
@@ -811,31 +812,29 @@ github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDf
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
 github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
-github.com/prometheus/client_golang v1.12.1/go.mod h1:3Z9XVyYiZYEO+YQWt3RD2R3jrbd179Rt297l4aS6nDY=
-github.com/prometheus/client_golang v1.13.0 h1:b71QUfeo5M8gq2+evJdTPfZhYMAU0uKPkyPJ7TPsloU=
-github.com/prometheus/client_golang v1.13.0/go.mod h1:vTeo+zgvILHsnnj/39Ou/1fPN5nJFOEMgftOUOmlvYQ=
+github.com/prometheus/client_golang v1.14.0 h1:nJdhIvne2eSX/XRAFV9PcvFFRbrjbcTUj0VP62TMhnw=
+github.com/prometheus/client_golang v1.14.0/go.mod h1:8vpkKitgIVNcqrRBWh1C4TIUQgYNtG/XQE4E/Zae36Y=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/client_model v0.2.0 h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
+github.com/prometheus/client_model v0.3.0 h1:UBgGFHqYdG/TPFD1B1ogZywDqEkwp3fBMvqdiQ7Xew4=
+github.com/prometheus/client_model v0.3.0/go.mod h1:LDGWKZIo7rky3hgvBe+caln+Dr3dPggB5dvjtD7w9+w=
 github.com/prometheus/common v0.0.0-20181020173914-7e9e6cabbd39/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
 github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc=
-github.com/prometheus/common v0.32.1/go.mod h1:vu+V0TpY+O6vW9J44gczi3Ap/oXXR10b+M/gUGO4Hls=
-github.com/prometheus/common v0.37.0 h1:ccBbHCgIiT9uSoFY0vX8H3zsNR5eLt17/RQLUvn8pXE=
-github.com/prometheus/common v0.37.0/go.mod h1:phzohg0JFMnBEFGxTDbfu3QyL5GI8gTQJFhYO5B3mfA=
+github.com/prometheus/common v0.39.0 h1:oOyhkDq05hPZKItWVBkJ6g6AtGxi+fy7F4JvUV8uhsI=
+github.com/prometheus/common v0.39.0/go.mod h1:6XBZ7lYdLCbkAVhwRsWTZn+IN5AB9F/NXd5w0BbEX0Y=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
-github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
-github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5mo=
-github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
+github.com/prometheus/procfs v0.9.0 h1:wzCHvIvM5SxWqYvwgVL7yJY8Lz3PKn49KQtpgMYJfhI=
+github.com/prometheus/procfs v0.9.0/go.mod h1:+pB4zwohETzFnmlpe6yd2lSc+0/46IYZRB/chUwxUZY=
 github.com/prometheus/prometheus v0.0.0-20190525122359-d20e84d0fb64 h1:3DyLm+sTAJkfLyR/1pJ3L+fU2lFufWbpcgMFlGtqeyA=
 github.com/prometheus/prometheus v0.0.0-20190525122359-d20e84d0fb64/go.mod h1:oYrT4Vs22/NcnoVYXt5m4cIHP+znvgyusahVpyETKTw=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
@@ -1055,8 +1054,9 @@ go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+
 go.uber.org/multierr v1.4.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/multierr v1.7.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak=
-go.uber.org/multierr v1.8.0 h1:dg6GjLku4EH+249NNmoIciG9N/jURbDG+pFlTkhzIC8=
 go.uber.org/multierr v1.8.0/go.mod h1:7EAYxJLBy9rStEaz58O2t4Uvip6FSURkq8/ppBp95ak=
+go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
+go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
@@ -1064,8 +1064,9 @@ go.uber.org/zap v1.12.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM=
 go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
 go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
 go.uber.org/zap v1.20.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw=
-go.uber.org/zap v1.23.0 h1:OjGQ5KQDEUawVHxNwQgPpiypGHOxo2mNZsOqTak4fFY=
-go.uber.org/zap v1.23.0/go.mod h1:D+nX8jyLsMHMYrln8A0rJjFt/T/9/bGgIhAqxv5URuY=
+go.uber.org/zap v1.21.0/go.mod h1:wjWOCqI0f2ZZrJF/UufIOkiC8ii6tm1iqIsLo76RfJw=
+go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
+go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
 golang.org/x/crypto v0.0.0-20180723164146-c126467f60eb/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -1184,7 +1185,6 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210427231257-85d9c07bbe3a/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210610132358-84b48f89b13b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
diff --git a/planner/core/integration_test.go b/planner/core/integration_test.go
index 31cd5aa9c346b..307971e2d77e1 100644
--- a/planner/core/integration_test.go
+++ b/planner/core/integration_test.go
@@ -7520,6 +7520,39 @@ func TestCastTimeAsDurationToTiFlash(t *testing.T) {
 	tk.MustQuery("explain select cast(a as time), cast(b as time) from t;").CheckAt([]int{0, 2, 4}, rows)
 }
 
+func TestUnhexPushDownToTiFlash(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int, b varchar(20));")
+	tk.MustExec("insert into t values(6162, '7469666C617368');")
+	tk.MustExec("set @@tidb_allow_mpp=1; set @@tidb_enforce_mpp=1;")
+	tk.MustExec("set @@tidb_isolation_read_engines = 'tiflash'")
+
+	tbl, err := dom.InfoSchema().TableByName(model.CIStr{O: "test", L: "test"}, model.CIStr{O: "t", L: "t"})
+	require.NoError(t, err)
+	// Set the hacked TiFlash replica for explain tests.
+	tbl.Meta().TiFlashReplica = &model.TiFlashReplicaInfo{Count: 1, Available: true}
+
+	rows := [][]interface{}{
+		{"TableReader_9", "root", "data:ExchangeSender_8"},
+		{"└─ExchangeSender_8", "mpp[tiflash]", "ExchangeType: PassThrough"},
+		{"  └─Projection_4", "mpp[tiflash]", "unhex(cast(test.t.a, var_string(20)))->Column#4"},
+		{"    └─TableFullScan_7", "mpp[tiflash]", "keep order:false, stats:pseudo"},
+	}
+	tk.MustQuery("explain select unhex(a) from t;").CheckAt([]int{0, 2, 4}, rows)
+
+	rows = [][]interface{}{
+		{"TableReader_9", "root", "data:ExchangeSender_8"},
+		{"└─ExchangeSender_8", "mpp[tiflash]", "ExchangeType: PassThrough"},
+		{"  └─Projection_4", "mpp[tiflash]", "unhex(test.t.b)->Column#4"},
+		{"    └─TableFullScan_7", "mpp[tiflash]", "keep order:false, stats:pseudo"},
+	}
+	tk.MustQuery("explain select unhex(b) from t;").CheckAt([]int{0, 2, 4}, rows)
+}
+
 func TestPartitionTableFallBackStatic(t *testing.T) {
 	store, _ := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index f60c1f617e467..592bb55f79619 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -5739,7 +5739,10 @@ func (b *PlanBuilder) buildUpdateLists(ctx context.Context, tableList []*ast.Tab
 				}
 			}
 
+			o := b.allowBuildCastArray
+			b.allowBuildCastArray = true
 			newExpr, np, err = b.rewriteWithPreprocess(ctx, assign.Expr, p, nil, nil, false, rewritePreprocess(assign))
+			b.allowBuildCastArray = o
 			if err != nil {
 				return nil, nil, false, err
 			}
diff --git a/planner/core/plan_to_pb.go b/planner/core/plan_to_pb.go
index a68fdae38f5de..71a7f11c09a65 100644
--- a/planner/core/plan_to_pb.go
+++ b/planner/core/plan_to_pb.go
@@ -370,7 +370,7 @@ func (p *PhysicalIndexScan) ToPB(_ sessionctx.Context, _ kv.StoreType) (*tipb.Ex
 	idxExec := &tipb.IndexScan{
 		TableId:          p.Table.ID,
 		IndexId:          p.Index.ID,
-		Columns:          util.ColumnsToProto(columns, p.Table.PKIsHandle),
+		Columns:          util.ColumnsToProto(columns, p.Table.PKIsHandle, true),
 		Desc:             p.Desc,
 		PrimaryColumnIds: pkColIds,
 	}
diff --git a/planner/core/plan_to_pb_test.go b/planner/core/plan_to_pb_test.go
index 9ea55a23babdd..cb108678629f2 100644
--- a/planner/core/plan_to_pb_test.go
+++ b/planner/core/plan_to_pb_test.go
@@ -35,16 +35,16 @@ func TestColumnToProto(t *testing.T) {
 	col := &model.ColumnInfo{
 		FieldType: *tp,
 	}
-	pc := util.ColumnToProto(col)
+	pc := util.ColumnToProto(col, false)
 	expect := &tipb.ColumnInfo{ColumnId: 0, Tp: 3, Collation: 83, ColumnLen: 11, Decimal: 0, Flag: 10, Elems: []string(nil), DefaultVal: []uint8(nil), PkHandle: false, XXX_unrecognized: []uint8(nil)}
 	require.Equal(t, expect, pc)
 
 	cols := []*model.ColumnInfo{col, col}
-	pcs := util.ColumnsToProto(cols, false)
+	pcs := util.ColumnsToProto(cols, false, false)
 	for _, v := range pcs {
 		require.Equal(t, int32(10), v.GetFlag())
 	}
-	pcs = util.ColumnsToProto(cols, true)
+	pcs = util.ColumnsToProto(cols, true, false)
 	for _, v := range pcs {
 		require.Equal(t, int32(10), v.GetFlag())
 	}
@@ -56,19 +56,19 @@ func TestColumnToProto(t *testing.T) {
 	col1 := &model.ColumnInfo{
 		FieldType: *tp,
 	}
-	pc = util.ColumnToProto(col1)
+	pc = util.ColumnToProto(col1, false)
 	require.Equal(t, int32(8), pc.Collation)
 
 	collate.SetNewCollationEnabledForTest(true)
 
-	pc = util.ColumnToProto(col)
+	pc = util.ColumnToProto(col, false)
 	expect = &tipb.ColumnInfo{ColumnId: 0, Tp: 3, Collation: -83, ColumnLen: 11, Decimal: 0, Flag: 10, Elems: []string(nil), DefaultVal: []uint8(nil), PkHandle: false, XXX_unrecognized: []uint8(nil)}
 	require.Equal(t, expect, pc)
-	pcs = util.ColumnsToProto(cols, true)
+	pcs = util.ColumnsToProto(cols, true, false)
 	for _, v := range pcs {
 		require.Equal(t, int32(-83), v.Collation)
 	}
-	pc = util.ColumnToProto(col1)
+	pc = util.ColumnToProto(col1, false)
 	require.Equal(t, int32(-8), pc.Collation)
 
 	tp = types.NewFieldType(mysql.TypeEnum)
@@ -77,6 +77,6 @@ func TestColumnToProto(t *testing.T) {
 	col2 := &model.ColumnInfo{
 		FieldType: *tp,
 	}
-	pc = util.ColumnToProto(col2)
+	pc = util.ColumnToProto(col2, false)
 	require.Len(t, pc.Elems, 2)
 }
diff --git a/table/tables/index.go b/table/tables/index.go
index 446a2e7288595..607afb9640aad 100644
--- a/table/tables/index.go
+++ b/table/tables/index.go
@@ -103,10 +103,61 @@ func (c *index) GenIndexValue(sc *stmtctx.StatementContext, distinct bool, index
 	return tablecodec.GenIndexValuePortal(sc, c.tblInfo, c.idxInfo, c.needRestoredData, distinct, false, indexedValues, h, c.phyTblID, restoredData)
 }
 
+// getIndexedValue will produce the result like:
+// 1. If not multi-valued index, return directly.
+// 2. (i1, [m1,m2], i2, ...) ==> [(i1, m1, i2, ...), (i1, m2, i2, ...)]
+// 3. (i1, null, i2, ...) ==> [(i1, null, i2, ...)]
+// 4. (i1, [], i2, ...) ==> nothing.
+func (c *index) getIndexedValue(indexedValues []types.Datum) [][]types.Datum {
+	if !c.idxInfo.MVIndex {
+		return [][]types.Datum{indexedValues}
+	}
+
+	vals := make([][]types.Datum, 0, 16)
+	jsonIdx := 0
+	jsonIsNull := false
+	existsVals := make(map[string]struct{})
+	var buf []byte
+	for !jsonIsNull {
+		val := make([]types.Datum, 0, len(indexedValues))
+		for i, v := range indexedValues {
+			if !c.tblInfo.Columns[c.idxInfo.Columns[i].Offset].FieldType.IsArray() {
+				val = append(val, v)
+			} else {
+				if v.IsNull() {
+					val = append(val, v)
+					jsonIsNull = true
+					continue
+				}
+				elemCount := v.GetMysqlJSON().GetElemCount()
+				for {
+					// JSON cannot be indexed, if the value is JSON type, it must be multi-valued index.
+					if jsonIdx >= elemCount {
+						goto out
+					}
+					binaryJSON := v.GetMysqlJSON().ArrayGetElem(jsonIdx)
+					jsonIdx++
+					buf = buf[:0]
+					key := string(binaryJSON.HashValue(buf))
+					if _, exists := existsVals[key]; exists {
+						continue
+					}
+					existsVals[key] = struct{}{}
+					val = append(val, types.NewDatum(binaryJSON.GetValue()))
+					break
+				}
+			}
+		}
+		vals = append(vals, val)
+	}
+out:
+	return vals
+}
+
 // Create creates a new entry in the kvIndex data.
 // If the index is unique and there is an existing entry with the same key,
 // Create will return the existing entry's handle as the first return value, ErrKeyExists as the second return value.
-func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValues []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...table.CreateIdxOptFunc) (kv.Handle, error) {
+func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValue []types.Datum, h kv.Handle, handleRestoreData []types.Datum, opts ...table.CreateIdxOptFunc) (kv.Handle, error) {
 	if c.Meta().Unique {
 		txn.CacheTableInfo(c.phyTblID, c.tblInfo)
 	}
@@ -114,225 +165,241 @@ func (c *index) Create(sctx sessionctx.Context, txn kv.Transaction, indexedValue
 	for _, fn := range opts {
 		fn(&opt)
 	}
+
+	indexedValues := c.getIndexedValue(indexedValue)
+	ctx := opt.Ctx
+	if ctx != nil {
+		if span := opentracing.SpanFromContext(ctx); span != nil && span.Tracer() != nil {
+			span1 := span.Tracer().StartSpan("index.Create", opentracing.ChildOf(span.Context()))
+			defer span1.Finish()
+			ctx = opentracing.ContextWithSpan(ctx, span1)
+		}
+	} else {
+		ctx = context.TODO()
+	}
 	vars := sctx.GetSessionVars()
 	writeBufs := vars.GetWriteStmtBufs()
 	skipCheck := vars.StmtCtx.BatchCheck
-	key, distinct, err := c.GenIndexKey(vars.StmtCtx, indexedValues, h, writeBufs.IndexKeyBuf)
-	if err != nil {
-		return nil, err
-	}
-
-	var (
-		tempKey         []byte
-		keyVer          byte
-		keyIsTempIdxKey bool
-	)
-	if !opt.FromBackFill {
-		key, tempKey, keyVer = GenTempIdxKeyByState(c.idxInfo, key)
-		if keyVer == TempIndexKeyTypeBackfill || keyVer == TempIndexKeyTypeDelete {
-			key, tempKey = tempKey, nil
-			keyIsTempIdxKey = true
+	for _, value := range indexedValues {
+		key, distinct, err := c.GenIndexKey(vars.StmtCtx, value, h, writeBufs.IndexKeyBuf)
+		if err != nil {
+			return nil, err
 		}
-	}
 
-	ctx := opt.Ctx
-	if opt.Untouched {
-		txn, err1 := sctx.Txn(true)
-		if err1 != nil {
-			return nil, err1
+		var (
+			tempKey         []byte
+			keyVer          byte
+			keyIsTempIdxKey bool
+		)
+		if !opt.FromBackFill {
+			key, tempKey, keyVer = GenTempIdxKeyByState(c.idxInfo, key)
+			if keyVer == TempIndexKeyTypeBackfill || keyVer == TempIndexKeyTypeDelete {
+				key, tempKey = tempKey, nil
+				keyIsTempIdxKey = true
+			}
 		}
-		// If the index kv was untouched(unchanged), and the key/value already exists in mem-buffer,
-		// should not overwrite the key with un-commit flag.
-		// So if the key exists, just do nothing and return.
-		v, err := txn.GetMemBuffer().Get(ctx, key)
-		if err == nil {
-			if len(v) != 0 {
-				return nil, nil
+
+		if opt.Untouched {
+			txn, err1 := sctx.Txn(true)
+			if err1 != nil {
+				return nil, err1
 			}
-			// The key is marked as deleted in the memory buffer, as the existence check is done lazily
-			// for optimistic transactions by default. The "untouched" key could still exist in the store,
-			// it's needed to commit this key to do the existence check so unset the untouched flag.
-			if !txn.IsPessimistic() {
-				keyFlags, err := txn.GetMemBuffer().GetFlags(key)
-				if err != nil {
-					return nil, err
+			// If the index kv was untouched(unchanged), and the key/value already exists in mem-buffer,
+			// should not overwrite the key with un-commit flag.
+			// So if the key exists, just do nothing and return.
+			v, err := txn.GetMemBuffer().Get(ctx, key)
+			if err == nil {
+				if len(v) != 0 {
+					continue
 				}
-				if keyFlags.HasPresumeKeyNotExists() {
-					opt.Untouched = false
+				// The key is marked as deleted in the memory buffer, as the existence check is done lazily
+				// for optimistic transactions by default. The "untouched" key could still exist in the store,
+				// it's needed to commit this key to do the existence check so unset the untouched flag.
+				if !txn.IsPessimistic() {
+					keyFlags, err := txn.GetMemBuffer().GetFlags(key)
+					if err != nil {
+						return nil, err
+					}
+					if keyFlags.HasPresumeKeyNotExists() {
+						opt.Untouched = false
+					}
 				}
 			}
 		}
-	}
-
-	// save the key buffer to reuse.
-	writeBufs.IndexKeyBuf = key
-	c.initNeedRestoreData.Do(func() {
-		c.needRestoredData = NeedRestoredData(c.idxInfo.Columns, c.tblInfo.Columns)
-	})
-	idxVal, err := tablecodec.GenIndexValuePortal(sctx.GetSessionVars().StmtCtx, c.tblInfo, c.idxInfo, c.needRestoredData, distinct, opt.Untouched, indexedValues, h, c.phyTblID, handleRestoreData)
-	if err != nil {
-		return nil, err
-	}
-
-	opt.IgnoreAssertion = opt.IgnoreAssertion || c.idxInfo.State != model.StatePublic
 
-	if !distinct || skipCheck || opt.Untouched {
-		if keyIsTempIdxKey && !opt.Untouched { // Untouched key-values never occur in the storage.
-			idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
-		}
-		err = txn.GetMemBuffer().Set(key, idxVal)
+		// save the key buffer to reuse.
+		writeBufs.IndexKeyBuf = key
+		c.initNeedRestoreData.Do(func() {
+			c.needRestoredData = NeedRestoredData(c.idxInfo.Columns, c.tblInfo.Columns)
+		})
+		idxVal, err := tablecodec.GenIndexValuePortal(sctx.GetSessionVars().StmtCtx, c.tblInfo, c.idxInfo, c.needRestoredData, distinct, opt.Untouched, value, h, c.phyTblID, handleRestoreData)
 		if err != nil {
 			return nil, err
 		}
-		if len(tempKey) > 0 {
-			if !opt.Untouched { // Untouched key-values never occur in the storage.
+
+		opt.IgnoreAssertion = opt.IgnoreAssertion || c.idxInfo.State != model.StatePublic
+
+		if !distinct || skipCheck || opt.Untouched {
+			if keyIsTempIdxKey && !opt.Untouched { // Untouched key-values never occur in the storage.
 				idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
 			}
-			err = txn.GetMemBuffer().Set(tempKey, idxVal)
+			err = txn.GetMemBuffer().Set(key, idxVal)
 			if err != nil {
 				return nil, err
 			}
-		}
-		if !opt.IgnoreAssertion && (!opt.Untouched) {
-			if sctx.GetSessionVars().LazyCheckKeyNotExists() && !txn.IsPessimistic() {
-				err = txn.SetAssertion(key, kv.SetAssertUnknown)
-			} else {
-				err = txn.SetAssertion(key, kv.SetAssertNotExist)
+			if len(tempKey) > 0 {
+				if !opt.Untouched { // Untouched key-values never occur in the storage.
+					idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
+				}
+				err = txn.GetMemBuffer().Set(tempKey, idxVal)
+				if err != nil {
+					return nil, err
+				}
 			}
+			if !opt.IgnoreAssertion && (!opt.Untouched) {
+				if sctx.GetSessionVars().LazyCheckKeyNotExists() && !txn.IsPessimistic() {
+					err = txn.SetAssertion(key, kv.SetAssertUnknown)
+				} else {
+					err = txn.SetAssertion(key, kv.SetAssertNotExist)
+				}
+			}
+			if err != nil {
+				return nil, err
+			}
+			continue
 		}
-		return nil, err
-	}
 
-	if ctx != nil {
-		if span := opentracing.SpanFromContext(ctx); span != nil && span.Tracer() != nil {
-			span1 := span.Tracer().StartSpan("index.Create", opentracing.ChildOf(span.Context()))
-			defer span1.Finish()
-			ctx = opentracing.ContextWithSpan(ctx, span1)
+		var value []byte
+		if c.tblInfo.TempTableType != model.TempTableNone {
+			// Always check key for temporary table because it does not write to TiKV
+			value, err = txn.Get(ctx, key)
+		} else if sctx.GetSessionVars().LazyCheckKeyNotExists() {
+			value, err = txn.GetMemBuffer().Get(ctx, key)
+		} else {
+			value, err = txn.Get(ctx, key)
 		}
-	} else {
-		ctx = context.TODO()
-	}
-
-	var value []byte
-	if c.tblInfo.TempTableType != model.TempTableNone {
-		// Always check key for temporary table because it does not write to TiKV
-		value, err = txn.Get(ctx, key)
-	} else if sctx.GetSessionVars().LazyCheckKeyNotExists() {
-		value, err = txn.GetMemBuffer().Get(ctx, key)
-	} else {
-		value, err = txn.Get(ctx, key)
-	}
-	if err != nil && !kv.IsErrNotFound(err) {
-		return nil, err
-	}
-	if err != nil || len(value) == 0 || (keyIsTempIdxKey && tablecodec.CheckTempIndexValueIsDelete(value)) {
-		lazyCheck := sctx.GetSessionVars().LazyCheckKeyNotExists() && err != nil
-		var needPresumeKey TempIndexKeyState
-		if keyIsTempIdxKey {
-			idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
-			needPresumeKey, _, err = KeyExistInTempIndex(ctx, txn, key, distinct, h, c.tblInfo.IsCommonHandle)
+		if err != nil && !kv.IsErrNotFound(err) {
+			return nil, err
+		}
+		if err != nil || len(value) == 0 || (keyIsTempIdxKey && tablecodec.CheckTempIndexValueIsDelete(value)) {
+			lazyCheck := sctx.GetSessionVars().LazyCheckKeyNotExists() && err != nil
+			var needPresumeKey TempIndexKeyState
+			if keyIsTempIdxKey {
+				idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
+				needPresumeKey, _, err = KeyExistInTempIndex(ctx, txn, key, distinct, h, c.tblInfo.IsCommonHandle)
+				if err != nil {
+					return nil, err
+				}
+			} else {
+				if len(tempKey) > 0 {
+					needPresumeKey, _, err = KeyExistInTempIndex(ctx, txn, tempKey, distinct, h, c.tblInfo.IsCommonHandle)
+					if err != nil {
+						return nil, err
+					}
+				}
+			}
+			if lazyCheck {
+				var flags []kv.FlagsOp
+				if needPresumeKey != KeyInTempIndexIsDeleted {
+					flags = []kv.FlagsOp{kv.SetPresumeKeyNotExists}
+				}
+				if !vars.ConstraintCheckInPlacePessimistic && vars.TxnCtx.IsPessimistic && vars.InTxn() &&
+					!vars.InRestrictedSQL && vars.ConnectionID > 0 {
+					flags = append(flags, kv.SetNeedConstraintCheckInPrewrite)
+				}
+				err = txn.GetMemBuffer().SetWithFlags(key, idxVal, flags...)
+			} else {
+				err = txn.GetMemBuffer().Set(key, idxVal)
+			}
 			if err != nil {
 				return nil, err
 			}
-		} else {
 			if len(tempKey) > 0 {
-				needPresumeKey, _, err = KeyExistInTempIndex(ctx, txn, tempKey, distinct, h, c.tblInfo.IsCommonHandle)
+				idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
+				if lazyCheck && needPresumeKey != KeyInTempIndexIsDeleted {
+					err = txn.GetMemBuffer().SetWithFlags(tempKey, idxVal, kv.SetPresumeKeyNotExists)
+				} else {
+					err = txn.GetMemBuffer().Set(tempKey, idxVal)
+				}
 				if err != nil {
 					return nil, err
 				}
 			}
-		}
-		if lazyCheck {
-			var flags []kv.FlagsOp
-			if needPresumeKey != KeyInTempIndexIsDeleted {
-				flags = []kv.FlagsOp{kv.SetPresumeKeyNotExists}
+			if opt.IgnoreAssertion {
+				continue
 			}
-			if !vars.ConstraintCheckInPlacePessimistic && vars.TxnCtx.IsPessimistic && vars.InTxn() &&
-				!vars.InRestrictedSQL && vars.ConnectionID > 0 {
-				flags = append(flags, kv.SetNeedConstraintCheckInPrewrite)
-			}
-			err = txn.GetMemBuffer().SetWithFlags(key, idxVal, flags...)
-		} else {
-			err = txn.GetMemBuffer().Set(key, idxVal)
-		}
-		if err != nil {
-			return nil, err
-		}
-		if len(tempKey) > 0 {
-			idxVal = tablecodec.EncodeTempIndexValue(idxVal, keyVer)
-			if lazyCheck && needPresumeKey != KeyInTempIndexIsDeleted {
-				err = txn.GetMemBuffer().SetWithFlags(tempKey, idxVal, kv.SetPresumeKeyNotExists)
+			if lazyCheck && !txn.IsPessimistic() {
+				err = txn.SetAssertion(key, kv.SetAssertUnknown)
 			} else {
-				err = txn.GetMemBuffer().Set(tempKey, idxVal)
+				err = txn.SetAssertion(key, kv.SetAssertNotExist)
 			}
 			if err != nil {
 				return nil, err
 			}
+			continue
 		}
-		if opt.IgnoreAssertion {
-			return nil, nil
+
+		if keyIsTempIdxKey {
+			value = tablecodec.DecodeTempIndexOriginValue(value)
 		}
-		if lazyCheck && !txn.IsPessimistic() {
-			err = txn.SetAssertion(key, kv.SetAssertUnknown)
-		} else {
-			err = txn.SetAssertion(key, kv.SetAssertNotExist)
+		handle, err := tablecodec.DecodeHandleInUniqueIndexValue(value, c.tblInfo.IsCommonHandle)
+		if err != nil {
+			return nil, err
 		}
-		return nil, err
-	}
-
-	if keyIsTempIdxKey {
-		value = tablecodec.DecodeTempIndexOriginValue(value)
-	}
-	handle, err := tablecodec.DecodeHandleInUniqueIndexValue(value, c.tblInfo.IsCommonHandle)
-	if err != nil {
-		return nil, err
+		return handle, kv.ErrKeyExists
 	}
-	return handle, kv.ErrKeyExists
+	return nil, nil
 }
 
 // Delete removes the entry for handle h and indexedValues from KV index.
-func (c *index) Delete(sc *stmtctx.StatementContext, txn kv.Transaction, indexedValues []types.Datum, h kv.Handle) error {
-	key, distinct, err := c.GenIndexKey(sc, indexedValues, h, nil)
-	if err != nil {
-		return err
-	}
+func (c *index) Delete(sc *stmtctx.StatementContext, txn kv.Transaction, indexedValue []types.Datum, h kv.Handle) error {
+	indexedValues := c.getIndexedValue(indexedValue)
+	for _, value := range indexedValues {
+		key, distinct, err := c.GenIndexKey(sc, value, h, nil)
+		if err != nil {
+			return err
+		}
 
-	key, tempKey, tempKeyVer := GenTempIdxKeyByState(c.idxInfo, key)
+		key, tempKey, tempKeyVer := GenTempIdxKeyByState(c.idxInfo, key)
 
-	if distinct {
-		if len(key) > 0 {
-			err = txn.GetMemBuffer().DeleteWithFlags(key, kv.SetNeedLocked)
-			if err != nil {
-				return err
+		if distinct {
+			if len(key) > 0 {
+				err = txn.GetMemBuffer().DeleteWithFlags(key, kv.SetNeedLocked)
+				if err != nil {
+					return err
+				}
 			}
-		}
-		if len(tempKey) > 0 {
-			tempVal := tablecodec.EncodeTempIndexValueDeletedUnique(h, tempKeyVer)
-			err = txn.GetMemBuffer().Set(tempKey, tempVal)
-			if err != nil {
-				return err
+			if len(tempKey) > 0 {
+				tempVal := tablecodec.EncodeTempIndexValueDeletedUnique(h, tempKeyVer)
+				err = txn.GetMemBuffer().Set(tempKey, tempVal)
+				if err != nil {
+					return err
+				}
 			}
-		}
-	} else {
-		if len(key) > 0 {
-			err = txn.GetMemBuffer().Delete(key)
-			if err != nil {
-				return err
+		} else {
+			if len(key) > 0 {
+				err = txn.GetMemBuffer().Delete(key)
+				if err != nil {
+					return err
+				}
 			}
-		}
-		if len(tempKey) > 0 {
-			tempVal := tablecodec.EncodeTempIndexValueDeleted(tempKeyVer)
-			err = txn.GetMemBuffer().Set(tempKey, tempVal)
-			if err != nil {
-				return err
+			if len(tempKey) > 0 {
+				tempVal := tablecodec.EncodeTempIndexValueDeleted(tempKeyVer)
+				err = txn.GetMemBuffer().Set(tempKey, tempVal)
+				if err != nil {
+					return err
+				}
 			}
 		}
+		if c.idxInfo.State == model.StatePublic {
+			// If the index is in public state, delete this index means it must exists.
+			err = txn.SetAssertion(key, kv.SetAssertExist)
+		}
+		if err != nil {
+			return err
+		}
 	}
-	if c.idxInfo.State == model.StatePublic {
-		// If the index is in public state, delete this index means it must exists.
-		err = txn.SetAssertion(key, kv.SetAssertExist)
-	}
-	return err
+	return nil
 }
 
 const (
diff --git a/table/tables/mutation_checker.go b/table/tables/mutation_checker.go
index e4513b8cae409..8445a266e3d2d 100644
--- a/table/tables/mutation_checker.go
+++ b/table/tables/mutation_checker.go
@@ -236,7 +236,7 @@ func checkIndexKeys(
 		}
 
 		for i, v := range decodedIndexValues {
-			fieldType := &t.Columns[indexInfo.Columns[i].Offset].FieldType
+			fieldType := t.Columns[indexInfo.Columns[i].Offset].FieldType.ArrayType()
 			datum, err := tablecodec.DecodeColumnValue(v, fieldType, sessVars.Location())
 			if err != nil {
 				return errors.Trace(err)
@@ -347,9 +347,27 @@ func compareIndexData(
 			cols[indexInfo.Columns[i].Offset].ColumnInfo,
 		)
 
-		comparison, err := decodedMutationDatum.Compare(sc, &expectedDatum, collate.GetCollator(decodedMutationDatum.Collation()))
-		if err != nil {
-			return errors.Trace(err)
+		var comparison int
+		var err error
+		// If it is multi-valued index, we should check the JSON contains the indexed value.
+		if cols[indexInfo.Columns[i].Offset].ColumnInfo.FieldType.IsArray() && expectedDatum.Kind() == types.KindMysqlJSON {
+			bj := expectedDatum.GetMysqlJSON()
+			count := bj.GetElemCount()
+			for elemIdx := 0; elemIdx < count; elemIdx++ {
+				jsonDatum := types.NewJSONDatum(bj.ArrayGetElem(elemIdx))
+				comparison, err = jsonDatum.Compare(sc, &decodedMutationDatum, collate.GetBinaryCollator())
+				if err != nil {
+					return errors.Trace(err)
+				}
+				if comparison == 0 {
+					break
+				}
+			}
+		} else {
+			comparison, err = decodedMutationDatum.Compare(sc, &expectedDatum, collate.GetCollator(decodedMutationDatum.Collation()))
+			if err != nil {
+				return errors.Trace(err)
+			}
 		}
 
 		if comparison != 0 {
diff --git a/table/tables/tables.go b/table/tables/tables.go
index 46d037c3197ac..709318a42332c 100644
--- a/table/tables/tables.go
+++ b/table/tables/tables.go
@@ -1948,7 +1948,7 @@ func BuildTableScanFromInfos(tableInfo *model.TableInfo, columnInfos []*model.Co
 	pkColIds := TryGetCommonPkColumnIds(tableInfo)
 	tsExec := &tipb.TableScan{
 		TableId:          tableInfo.ID,
-		Columns:          util.ColumnsToProto(columnInfos, tableInfo.PKIsHandle),
+		Columns:          util.ColumnsToProto(columnInfos, tableInfo.PKIsHandle, false),
 		PrimaryColumnIds: pkColIds,
 	}
 	if tableInfo.IsCommonHandle {
@@ -1962,7 +1962,7 @@ func BuildPartitionTableScanFromInfos(tableInfo *model.TableInfo, columnInfos []
 	pkColIds := TryGetCommonPkColumnIds(tableInfo)
 	tsExec := &tipb.PartitionTableScan{
 		TableId:          tableInfo.ID,
-		Columns:          util.ColumnsToProto(columnInfos, tableInfo.PKIsHandle),
+		Columns:          util.ColumnsToProto(columnInfos, tableInfo.PKIsHandle, false),
 		PrimaryColumnIds: pkColIds,
 		IsFastScan:       &fastScan,
 	}
diff --git a/types/convert.go b/types/convert.go
index 96a12f64ca641..1b97b7336ef66 100644
--- a/types/convert.go
+++ b/types/convert.go
@@ -758,6 +758,8 @@ func ToString(value interface{}) (string, error) {
 		return v.String(), nil
 	case Set:
 		return v.String(), nil
+	case BinaryJSON:
+		return v.String(), nil
 	default:
 		return "", errors.Errorf("cannot convert %v(type %T) to string", value, value)
 	}
diff --git a/types/datum.go b/types/datum.go
index 34639cdf10bdb..231de4dd9a069 100644
--- a/types/datum.go
+++ b/types/datum.go
@@ -2016,6 +2016,10 @@ func (d *Datum) ToMysqlJSON() (j BinaryJSON, err error) {
 		in = d.GetBinaryLiteral().ToString()
 	case KindNull:
 		in = nil
+	case KindMysqlTime:
+		in = d.GetMysqlTime()
+	case KindMysqlDuration:
+		in = d.GetMysqlDuration()
 	default:
 		in, err = d.ToString()
 	}
diff --git a/types/json_binary.go b/types/json_binary.go
index 6fe01d2b4f28e..cacf5b69b025a 100644
--- a/types/json_binary.go
+++ b/types/json_binary.go
@@ -31,6 +31,8 @@ import (
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/terror"
 	"github.com/pingcap/tidb/util/hack"
+	"github.com/pingcap/tidb/util/logutil"
+	"go.uber.org/zap"
 	"golang.org/x/exp/slices"
 )
 
@@ -578,6 +580,26 @@ func (bj BinaryJSON) HashValue(buf []byte) []byte {
 	return buf
 }
 
+// GetValue return the primitive value of the JSON.
+func (bj BinaryJSON) GetValue() any {
+	switch bj.TypeCode {
+	case JSONTypeCodeInt64:
+		return bj.GetInt64()
+	case JSONTypeCodeUint64:
+		return bj.GetUint64()
+	case JSONTypeCodeDuration:
+		return bj.GetDuration()
+	case JSONTypeCodeFloat64:
+		return bj.GetFloat64()
+	case JSONTypeCodeString:
+		return bj.GetString()
+	case JSONTypeCodeDate, JSONTypeCodeDatetime:
+		return bj.GetTime()
+	}
+	logutil.BgLogger().Error("unreachable JSON type", zap.Any("type", bj.TypeCode))
+	return nil
+}
+
 // CreateBinaryJSON creates a BinaryJSON from interface.
 func CreateBinaryJSON(in interface{}) BinaryJSON {
 	bj, err := CreateBinaryJSONWithCheck(in)
diff --git a/util/misc.go b/util/misc.go
index 0e28baa5f62fc..d336aa5765451 100644
--- a/util/misc.go
+++ b/util/misc.go
@@ -394,10 +394,10 @@ func TLSCipher2String(n uint16) string {
 }
 
 // ColumnsToProto converts a slice of model.ColumnInfo to a slice of tipb.ColumnInfo.
-func ColumnsToProto(columns []*model.ColumnInfo, pkIsHandle bool) []*tipb.ColumnInfo {
+func ColumnsToProto(columns []*model.ColumnInfo, pkIsHandle bool, forIndex bool) []*tipb.ColumnInfo {
 	cols := make([]*tipb.ColumnInfo, 0, len(columns))
 	for _, c := range columns {
-		col := ColumnToProto(c)
+		col := ColumnToProto(c, forIndex)
 		// TODO: Here `PkHandle`'s meaning is changed, we will change it to `IsHandle` when tikv's old select logic
 		// is abandoned.
 		if (pkIsHandle && mysql.HasPriKeyFlag(c.GetFlag())) || c.ID == model.ExtraHandleID {
@@ -411,7 +411,7 @@ func ColumnsToProto(columns []*model.ColumnInfo, pkIsHandle bool) []*tipb.Column
 }
 
 // ColumnToProto converts model.ColumnInfo to tipb.ColumnInfo.
-func ColumnToProto(c *model.ColumnInfo) *tipb.ColumnInfo {
+func ColumnToProto(c *model.ColumnInfo, forIndex bool) *tipb.ColumnInfo {
 	pc := &tipb.ColumnInfo{
 		ColumnId:  c.ID,
 		Collation: collate.RewriteNewCollationIDIfNeeded(int32(mysql.CollationNames[c.GetCollate()])),
@@ -420,7 +420,12 @@ func ColumnToProto(c *model.ColumnInfo) *tipb.ColumnInfo {
 		Flag:      int32(c.GetFlag()),
 		Elems:     c.GetElems(),
 	}
-	pc.Tp = int32(c.GetType())
+	if forIndex {
+		// Use array type for read the multi-valued index.
+		pc.Tp = int32(c.FieldType.ArrayType().GetType())
+	} else {
+		pc.Tp = int32(c.GetType())
+	}
 	return pc
 }
 
diff --git a/util/misc_test.go b/util/misc_test.go
index c1510625593f0..7a5baeb197d28 100644
--- a/util/misc_test.go
+++ b/util/misc_test.go
@@ -174,8 +174,8 @@ func TestToPB(t *testing.T) {
 	}
 	column2.SetCollate("utf8mb4_bin")
 
-	assert.Equal(t, "column_id:1 collation:-45 columnLen:-1 decimal:-1 ", ColumnToProto(column).String())
-	assert.Equal(t, "column_id:1 collation:-45 columnLen:-1 decimal:-1 ", ColumnsToProto([]*model.ColumnInfo{column, column2}, false)[0].String())
+	assert.Equal(t, "column_id:1 collation:-45 columnLen:-1 decimal:-1 ", ColumnToProto(column, false).String())
+	assert.Equal(t, "column_id:1 collation:-45 columnLen:-1 decimal:-1 ", ColumnsToProto([]*model.ColumnInfo{column, column2}, false, false)[0].String())
 }
 
 func TestComposeURL(t *testing.T) {