address commnet

Author: lcwangchao

pkg/ttl/cache/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "//pkg/infoschema",
         "//pkg/kv",
         "//pkg/parser/ast",
+        "//pkg/parser/charset",
         "//pkg/parser/model",
         "//pkg/parser/mysql",
         "//pkg/parser/terror",

pkg/ttl/cache/split_test.go

@@ -517,8 +517,8 @@ func TestSplitTTLScanRangesWithBytes(t *testing.T) {
 		createTTLTable(t, tk, "t3", "varchar(32) CHARACTER SET BINARY"),
 		createTTLTable(t, tk, "t4", "bit(32)"),
 		create2PKTTLTable(t, tk, "t5", "binary(32)"),
-		createTTLTable(t, tk, "t6", "char(32) CHARACTER SET UTF8MB4"),
-		create2PKTTLTable(t, tk, "t7", "char(32) CHARACTER SET gbk"),
+		createTTLTable(t, tk, "t6", "char(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin"),
+		create2PKTTLTable(t, tk, "t7", "char(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin"),
 	}
 
 	cases := []struct {
@@ -656,6 +656,8 @@ func TestNoTTLSplitSupportTables(t *testing.T) {
 		createTTLTable(t, tk, "t2", "date"),
 		createTTLTable(t, tk, "t3", "datetime"),
 		createTTLTable(t, tk, "t4", "timestamp"),
+		createTTLTable(t, tk, "t5", "varchar(32) character set utf8mb4 collate utf8mb4_general_ci"),
+		createTTLTable(t, tk, "t6", "varchar(32) character set utf8mb4 collate utf8mb4_0900_ai_ci"),
 	}
 
 	tikvStore := newMockTiKVStore(t)

pkg/ttl/cache/table.go

@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression/contextstatic"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/terror"
@@ -287,13 +288,31 @@ func (t *PhysicalTable) SplitScanRanges(ctx context.Context, store kv.Storage, s
 	switch ft.GetType() {
 	case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeLonglong, mysql.TypeInt24:
 		if len(t.KeyColumns) > 1 {
-			return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, true, mysql.HasUnsignedFlag(ft.GetFlag()), false)
+			return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, true, mysql.HasUnsignedFlag(ft.GetFlag()), nil)
 		}
 		return t.splitIntRanges(ctx, tikvStore, splitCnt)
 	case mysql.TypeBit:
-		return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, false, false, true)
+		return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, false, false, nil)
 	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
-		return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, false, false, mysql.HasBinaryFlag(ft.GetFlag()))
+		var decode func([]byte) types.Datum
+		if !mysql.HasBinaryFlag(ft.GetFlag()) {
+			switch ft.GetCharset() {
+			case charset.CharsetASCII, charset.CharsetLatin1:
+				// ASCII and Latin1 are 8-bit charset, we can use GetASCIIPrefixDatumFromBytes to decode it.
+				decode = GetASCIIPrefixDatumFromBytes
+			case charset.CharsetUTF8, charset.CharsetUTF8MB4:
+				switch ft.GetCollate() {
+				case charset.CollationUTF8, charset.CollationUTF8MB4, "utf8mb4_0900_bin":
+					// We can only use GetASCIIPrefixDatumFromBytes to decode UTF8 and UTF8MB4 when they are
+					// "utf8_bin" or "utf8mb4_bin" collation.
+					decode = GetASCIIPrefixDatumFromBytes
+				}
+			}
+			if decode == nil {
+				return []ScanRange{newFullRange()}, nil
+			}
+		}
+		return t.splitCommonHandleRanges(ctx, tikvStore, splitCnt, false, false, decode)
 	}
 	return []ScanRange{newFullRange()}, nil
 }
@@ -363,7 +382,7 @@ func (t *PhysicalTable) splitIntRanges(ctx context.Context, store tikv.Storage,
 }
 
 func (t *PhysicalTable) splitCommonHandleRanges(
-	ctx context.Context, store tikv.Storage, splitCnt int, isInt bool, unsigned bool, binary bool,
+	ctx context.Context, store tikv.Storage, splitCnt int, isInt bool, unsigned bool, decode func([]byte) types.Datum,
 ) ([]ScanRange, error) {
 	recordPrefix := tablecodec.GenTableRecordPrefix(t.ID)
 	startKey, endKey := recordPrefix, recordPrefix.PrefixNext()
@@ -385,8 +404,8 @@ func (t *PhysicalTable) splitCommonHandleRanges(
 				curScanEnd = GetNextIntDatumFromCommonHandle(keyRange.EndKey, recordPrefix, unsigned)
 			} else {
 				curScanEnd = GetNextBytesHandleDatum(keyRange.EndKey, recordPrefix)
-				if !binary {
-					curScanEnd = GetASCIIPrefixDatumFromBytes(curScanEnd.GetBytes())
+				if decode != nil {
+					curScanEnd = decode(curScanEnd.GetBytes())
 				}
 
 				// "" is the smallest value for string/[]byte, skip to add it to ranges.