diff --git a/executor/mem_reader.go b/executor/mem_reader.go
index 56c2ccb309f14..23256e0ad5991 100644
--- a/executor/mem_reader.go
+++ b/executor/mem_reader.go
@@ -279,7 +279,7 @@ func (m *memTableReader) getRowData(handle kv.Handle, value []byte) ([][]byte, e
 		offset := colIDs[id]
 		if m.table.IsCommonHandle {
 			for i, colID := range m.pkColIDs {
-				if colID == col.ID {
+				if colID == col.ID && !types.CommonHandleNeedRestoredData(&col.FieldType) {
 					values[offset] = handle.EncodedCol(i)
 					break
 				}
diff --git a/expression/integration_test.go b/expression/integration_test.go
index fc1cf951fec03..5ce80eb7c9af1 100644
--- a/expression/integration_test.go
+++ b/expression/integration_test.go
@@ -7650,6 +7650,37 @@ func (s *testIntegrationSuite) TestIssue20180(c *C) {
 	tk.MustQuery("select * from t where a > 1  and a = \"b\";").Check(testkit.Rows("b"))
 }
 
+func (s *testIntegrationSerialSuite) TestClusteredIndexAndNewCollation(c *C) {
+	collate.SetNewCollationEnabledForTest(true)
+	defer collate.SetNewCollationEnabledForTest(false)
+
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("set @@tidb_enable_clustered_index = 1;")
+	tk.MustExec("CREATE TABLE `t` (" +
+		"`a` char(10) COLLATE utf8mb4_unicode_ci NOT NULL," +
+		"`b` char(20) COLLATE utf8mb4_general_ci NOT NULL," +
+		"`c` int(11) NOT NULL," +
+		"PRIMARY KEY (`a`,`b`,`c`)," +
+		"KEY `idx` (`a`))")
+
+	tk.MustExec("begin")
+	tk.MustExec("insert into t values ('a6', 'b6', 3)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("a6 b6 3"))
+	tk.MustQuery("select * from t where a='a6'").Check(testkit.Rows("a6 b6 3"))
+	tk.MustExec("delete from t")
+	tk.MustQuery("select * from t").Check(testkit.Rows())
+	tk.MustExec("commit")
+	tk.MustQuery("select * from t").Check(testkit.Rows())
+
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(`a` char(10) COLLATE utf8mb4_unicode_ci NOT NULL key)")
+	tk.MustExec("insert into t values ('&');")
+	tk.MustExec("replace into t values ('&');")
+	tk.MustQuery("select * from t").Check(testkit.Rows("&"))
+}
+
 func (s *testIntegrationSerialSuite) TestIssue20608(c *C) {
 	collate.SetNewCollationEnabledForTest(true)
 	defer collate.SetNewCollationEnabledForTest(false)
diff --git a/table/tables/tables.go b/table/tables/tables.go
index 18735e260cfa4..234c965ca77de 100644
--- a/table/tables/tables.go
+++ b/table/tables/tables.go
@@ -40,7 +40,6 @@ import (
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/codec"
-	"github.com/pingcap/tidb/util/collate"
 	"github.com/pingcap/tidb/util/generatedexpr"
 	"github.com/pingcap/tidb/util/logutil"
 	"github.com/pingcap/tidb/util/stringutil"
@@ -878,7 +877,7 @@ func DecodeRawRowData(ctx sessionctx.Context, meta *model.TableInfo, h kv.Handle
 			}
 			continue
 		}
-		if col.IsCommonHandleColumn(meta) {
+		if col.IsCommonHandleColumn(meta) && !types.CommonHandleNeedRestoredData(&col.FieldType) {
 			pkIdx := FindPrimaryIndex(meta)
 			var idxOfIdx int
 			for i, idxCol := range pkIdx.Columns {
@@ -910,7 +909,7 @@ func DecodeRawRowData(ctx sessionctx.Context, meta *model.TableInfo, h kv.Handle
 		if col == nil {
 			continue
 		}
-		if col.IsPKHandleColumn(meta) || col.IsCommonHandleColumn(meta) {
+		if col.IsPKHandleColumn(meta) || (col.IsCommonHandleColumn(meta) && !types.CommonHandleNeedRestoredData(&col.FieldType)) {
 			continue
 		}
 		ri, ok := rowMap[col.ID]
@@ -1409,10 +1408,7 @@ func CanSkip(info *model.TableInfo, col *table.Column, value *types.Datum) bool
 				continue
 			}
 			canSkip := idxCol.Length == types.UnspecifiedLength
-			isNewCollation := collate.NewCollationEnabled() &&
-				col.EvalType() == types.ETString &&
-				!mysql.HasBinaryFlag(col.Flag)
-			canSkip = canSkip && !isNewCollation
+			canSkip = canSkip && !types.CommonHandleNeedRestoredData(&col.FieldType)
 			return canSkip
 		}
 	}
diff --git a/tablecodec/tablecodec.go b/tablecodec/tablecodec.go
index 3f757a43cff98..957243cda5f42 100644
--- a/tablecodec/tablecodec.go
+++ b/tablecodec/tablecodec.go
@@ -474,6 +474,9 @@ func DecodeHandleToDatumMap(handle kv.Handle, handleColIDs []int64,
 			if id != hid {
 				continue
 			}
+			if types.CommonHandleNeedRestoredData(ft) {
+				continue
+			}
 			d, err := decodeHandleToDatum(handle, ft, idx)
 			if err != nil {
 				return row, err
diff --git a/types/field_type.go b/types/field_type.go
index a097a59130747..af7d6db4dcbc0 100644
--- a/types/field_type.go
+++ b/types/field_type.go
@@ -20,6 +20,7 @@ import (
 	"github.com/pingcap/parser/mysql"
 	ast "github.com/pingcap/parser/types"
 	"github.com/pingcap/tidb/types/json"
+	"github.com/pingcap/tidb/util/collate"
 	utilMath "github.com/pingcap/tidb/util/math"
 )
 
@@ -1266,3 +1267,11 @@ func SetBinChsClnFlag(ft *FieldType) {
 
 // VarStorageLen indicates this column is a variable length column.
 const VarStorageLen = ast.VarStorageLen
+
+// CommonHandleNeedRestoredData indicates whether the column can be decoded directly from the common handle.
+// If can, then returns false. Otherwise returns true.
+func CommonHandleNeedRestoredData(ft *FieldType) bool {
+	return collate.NewCollationEnabled() &&
+		ft.EvalType() == ETString &&
+		!mysql.HasBinaryFlag(ft.Flag)
+}
diff --git a/util/rowcodec/decoder.go b/util/rowcodec/decoder.go
index 1e74e4dbc3d7f..b9476eb794985 100644
--- a/util/rowcodec/decoder.go
+++ b/util/rowcodec/decoder.go
@@ -254,6 +254,9 @@ func (decoder *ChunkDecoder) tryAppendHandleColumn(colIdx int, col *ColInfo, han
 	}
 	for i, id := range decoder.handleColIDs {
 		if col.ID == id {
+			if types.CommonHandleNeedRestoredData(col.Ft) {
+				return false
+			}
 			coder := codec.NewDecoder(chk, decoder.loc)
 			_, err := coder.DecodeOne(handle.EncodedCol(i), colIdx, col.Ft)
 			if err != nil {
@@ -419,6 +422,9 @@ func (decoder *BytesDecoder) tryDecodeHandle(values [][]byte, offset int, col *C
 	if handle == nil {
 		return false
 	}
+	if types.CommonHandleNeedRestoredData(col.Ft) {
+		return false
+	}
 	if col.IsPKHandle || col.ID == model.ExtraHandleID {
 		handleData := cacheBytes
 		if mysql.HasUnsignedFlag(col.Ft.Flag) {