diff --git a/metrics/grafana/tidb.json b/metrics/grafana/tidb.json
index 9d4a451fafca9..7677a718e156f 100644
--- a/metrics/grafana/tidb.json
+++ b/metrics/grafana/tidb.json
@@ -17776,9 +17776,9 @@
       "targets": [
         {
           "exemplar": true,
-          "expr": "sum(tidb_server_ttl_job_status{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}) by (type)",
+          "expr": "sum(tidb_server_ttl_job_status{k8s_cluster=\"$k8s_cluster\",tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}) by (type, instance)",
           "interval": "",
-          "legendFormat": "{{ type }}",
+          "legendFormat": "{{ instance }} {{ type }}",
           "queryType": "randomWalk",
           "refId": "A"
         }
diff --git a/ttl/cache/split_test.go b/ttl/cache/split_test.go
index 391070fb3a9bd..1d2279eb8d0f9 100644
--- a/ttl/cache/split_test.go
+++ b/ttl/cache/split_test.go
@@ -233,6 +233,10 @@ func createTTLTable(t *testing.T, tk *testkit.TestKit, name string, option strin
 	return createTTLTableWithSQL(t, tk, name, fmt.Sprintf("create table test.%s(id %s primary key, t timestamp) TTL = `t` + interval 1 day", name, option))
 }
 
+func create2PKTTLTable(t *testing.T, tk *testkit.TestKit, name string, option string) *cache.PhysicalTable {
+	return createTTLTableWithSQL(t, tk, name, fmt.Sprintf("create table test.%s(id %s, id2 int, t timestamp, primary key(id, id2)) TTL = `t` + interval 1 day", name, option))
+}
+
 func createTTLTableWithSQL(t *testing.T, tk *testkit.TestKit, name string, sql string) *cache.PhysicalTable {
 	tk.MustExec(sql)
 	is, ok := tk.Session().GetDomainInfoSchema().(infoschema.InfoSchema)
@@ -273,6 +277,7 @@ func TestSplitTTLScanRangesWithSignedInt(t *testing.T) {
 		createTTLTable(t, tk, "t4", "int"),
 		createTTLTable(t, tk, "t5", "bigint"),
 		createTTLTable(t, tk, "t6", ""), // no clustered
+		create2PKTTLTable(t, tk, "t7", "tinyint"),
 	}
 
 	tikvStore := newMockTiKVStore(t)
@@ -334,6 +339,7 @@ func TestSplitTTLScanRangesWithUnsignedInt(t *testing.T) {
 		createTTLTable(t, tk, "t3", "mediumint unsigned"),
 		createTTLTable(t, tk, "t4", "int unsigned"),
 		createTTLTable(t, tk, "t5", "bigint unsigned"),
+		create2PKTTLTable(t, tk, "t6", "tinyint unsigned"),
 	}
 
 	tikvStore := newMockTiKVStore(t)
@@ -397,6 +403,7 @@ func TestSplitTTLScanRangesWithBytes(t *testing.T) {
 		createTTLTable(t, tk, "t2", "char(32) CHARACTER SET BINARY"),
 		createTTLTable(t, tk, "t3", "varchar(32) CHARACTER SET BINARY"),
 		createTTLTable(t, tk, "t4", "bit(32)"),
+		create2PKTTLTable(t, tk, "t5", "binary(32)"),
 	}
 
 	tikvStore := newMockTiKVStore(t)
@@ -446,6 +453,7 @@ func TestNoTTLSplitSupportTables(t *testing.T) {
 		createTTLTable(t, tk, "t2", "varchar(32) CHARACTER SET UTF8MB4"),
 		createTTLTable(t, tk, "t3", "double"),
 		createTTLTable(t, tk, "t4", "decimal(32, 2)"),
+		create2PKTTLTable(t, tk, "t5", "char(32)  CHARACTER SET UTF8MB4"),
 	}
 
 	tikvStore := newMockTiKVStore(t)
@@ -526,6 +534,14 @@ func TestGetNextBytesHandleDatum(t *testing.T) {
 			key:    buildBytesRowKey([]byte{1, 2, 3, 4, 5, 6, 7, 8, 0}),
 			result: []byte{1, 2, 3, 4, 5, 6, 7, 8, 0},
 		},
+		{
+			key:    append(buildBytesRowKey([]byte{1, 2, 3, 4, 5, 6, 7, 8, 0}), 0),
+			result: []byte{1, 2, 3, 4, 5, 6, 7, 8, 0, 0},
+		},
+		{
+			key:    append(buildBytesRowKey([]byte{1, 2, 3, 4, 5, 6, 7, 8, 0}), 1),
+			result: []byte{1, 2, 3, 4, 5, 6, 7, 8, 0, 0},
+		},
 		{
 			key:    []byte{},
 			result: []byte{},
@@ -613,7 +629,7 @@ func TestGetNextBytesHandleDatum(t *testing.T) {
 				bs[len(bs)-10] = 254
 				return bs
 			},
-			result: []byte{1, 2, 3, 4, 5, 6, 7},
+			result: []byte{1, 2, 3, 4, 5, 6, 7, 0},
 		},
 		{
 			// recordPrefix + bytesFlag + [1, 2, 3, 4, 5, 6, 7, 0, 253, 9, 0, 0, 0, 0, 0, 0, 0, 248]
@@ -718,6 +734,18 @@ func TestGetNextIntHandle(t *testing.T) {
 			key:    tablecodec.EncodeRowKeyWithHandle(tblID, kv.IntHandle(math.MinInt64)),
 			result: math.MinInt64,
 		},
+		{
+			key:    append(tablecodec.EncodeRowKeyWithHandle(tblID, kv.IntHandle(7)), 0),
+			result: 8,
+		},
+		{
+			key:    append(tablecodec.EncodeRowKeyWithHandle(tblID, kv.IntHandle(math.MaxInt64)), 0),
+			isNull: true,
+		},
+		{
+			key:    append(tablecodec.EncodeRowKeyWithHandle(tblID, kv.IntHandle(math.MinInt64)), 0),
+			result: math.MinInt64 + 1,
+		},
 		{
 			key:    []byte{},
 			result: math.MinInt64,
diff --git a/ttl/cache/table.go b/ttl/cache/table.go
index 0cf50d092e437..35aca7f8532b2 100644
--- a/ttl/cache/table.go
+++ b/ttl/cache/table.go
@@ -166,9 +166,9 @@ func NewPhysicalTable(schema model.CIStr, tbl *model.TableInfo, partition model.
 	}, nil
 }
 
-// ValidateKey validates a key
-func (t *PhysicalTable) ValidateKey(key []types.Datum) error {
-	if len(t.KeyColumns) != len(key) {
+// ValidateKeyPrefix validates a key prefix
+func (t *PhysicalTable) ValidateKeyPrefix(key []types.Datum) error {
+	if len(key) > len(t.KeyColumns) {
 		return errors.Errorf("invalid key length: %d, expected %d", len(key), len(t.KeyColumns))
 	}
 	return nil
@@ -198,7 +198,7 @@ func (t *PhysicalTable) EvalExpireTime(ctx context.Context, se session.Session,
 
 // SplitScanRanges split ranges for TTL scan
 func (t *PhysicalTable) SplitScanRanges(ctx context.Context, store kv.Storage, splitCnt int) ([]ScanRange, error) {
-	if len(t.KeyColumns) != 1 || splitCnt <= 1 {
+	if len(t.KeyColumns) < 1 || splitCnt <= 1 {
 		return []ScanRange{newFullRange()}, nil
 	}
 
@@ -431,7 +431,10 @@ func GetNextBytesHandleDatum(key kv.Key, recordPrefix []byte) (d types.Datum) {
 		return d
 	}
 
-	if _, v, err := codec.DecodeOne(encodedVal); err == nil {
+	if remain, v, err := codec.DecodeOne(encodedVal); err == nil {
+		if len(remain) > 0 {
+			v.SetBytes(kv.Key(v.GetBytes()).Next())
+		}
 		return v
 	}
 
diff --git a/ttl/sqlbuilder/sql.go b/ttl/sqlbuilder/sql.go
index 833327c7af404..c9e4181ccfdda 100644
--- a/ttl/sqlbuilder/sql.go
+++ b/ttl/sqlbuilder/sql.go
@@ -319,16 +319,12 @@ type ScanQueryGenerator struct {
 
 // NewScanQueryGenerator creates a new ScanQueryGenerator
 func NewScanQueryGenerator(tbl *cache.PhysicalTable, expire time.Time, rangeStart []types.Datum, rangeEnd []types.Datum) (*ScanQueryGenerator, error) {
-	if len(rangeStart) > 0 {
-		if err := tbl.ValidateKey(rangeStart); err != nil {
-			return nil, err
-		}
+	if err := tbl.ValidateKeyPrefix(rangeStart); err != nil {
+		return nil, err
 	}
 
-	if len(rangeEnd) > 0 {
-		if err := tbl.ValidateKey(rangeEnd); err != nil {
-			return nil, err
-		}
+	if err := tbl.ValidateKeyPrefix(rangeEnd); err != nil {
+		return nil, err
 	}
 
 	return &ScanQueryGenerator{
@@ -393,11 +389,11 @@ func (g *ScanQueryGenerator) setStack(key []types.Datum) error {
 		return nil
 	}
 
-	if err := g.tbl.ValidateKey(key); err != nil {
+	if err := g.tbl.ValidateKeyPrefix(key); err != nil {
 		return err
 	}
 
-	g.stack = g.stack[:cap(g.stack)]
+	g.stack = g.stack[:len(key)]
 	for i := 0; i < len(key); i++ {
 		g.stack[i] = key[0 : i+1]
 	}
@@ -440,7 +436,7 @@ func (g *ScanQueryGenerator) buildSQL() (string, error) {
 	}
 
 	if len(g.keyRangeEnd) > 0 {
-		if err := b.WriteCommonCondition(g.tbl.KeyColumns, "<", g.keyRangeEnd); err != nil {
+		if err := b.WriteCommonCondition(g.tbl.KeyColumns[0:len(g.keyRangeEnd)], "<", g.keyRangeEnd); err != nil {
 			return "", err
 		}
 	}
diff --git a/ttl/sqlbuilder/sql_test.go b/ttl/sqlbuilder/sql_test.go
index 3bc982a092014..ca7719d59574e 100644
--- a/ttl/sqlbuilder/sql_test.go
+++ b/ttl/sqlbuilder/sql_test.go
@@ -702,6 +702,54 @@ func TestScanQueryGenerator(t *testing.T) {
 				},
 			},
 		},
+		{
+			tbl:        t2,
+			expire:     time.UnixMilli(0).In(time.UTC),
+			rangeStart: d(1),
+			rangeEnd:   d(100),
+			path: [][]interface{}{
+				{
+					nil, 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` >= 1 AND `a` < 100 AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "x", []byte{0x1a}), 5), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` = 1 AND `b` = 'x' AND `c` > x'1a' AND `a` < 100 AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "x", []byte{0x20}), 4), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` = 1 AND `b` > 'x' AND `a` < 100 AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "y", []byte{0x0a}), 4), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` > 1 AND `a` < 100 AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+			},
+		},
+		{
+			tbl:        t2,
+			expire:     time.UnixMilli(0).In(time.UTC),
+			rangeStart: d(1, "x"),
+			rangeEnd:   d(100, "z"),
+			path: [][]interface{}{
+				{
+					nil, 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` = 1 AND `b` >= 'x' AND (`a`, `b`) < (100, 'z') AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "x", []byte{0x1a}), 5), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` = 1 AND `b` = 'x' AND `c` > x'1a' AND (`a`, `b`) < (100, 'z') AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "x", []byte{0x20}), 4), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` = 1 AND `b` > 'x' AND (`a`, `b`) < (100, 'z') AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+				{
+					result(d(1, "y", []byte{0x0a}), 4), 5,
+					"SELECT LOW_PRIORITY `a`, `b`, `c` FROM `test2`.`t2` WHERE `a` > 1 AND (`a`, `b`) < (100, 'z') AND `time` < '1970-01-01 00:00:00' ORDER BY `a`, `b`, `c` ASC LIMIT 5",
+				},
+			},
+		},
 	}
 
 	for i, c := range cases {