tablecodec: fix the issue that decoding an index value might panic (#53034)

pkg/table/tables/BUILD.bazel

@@ -73,7 +73,7 @@ go_test(
     ],
     embed = [":tables"],
     flaky = True,
-    shard_count = 30,
+    shard_count = 31,
     deps = [
         "//pkg/ddl",
         "//pkg/ddl/util/callback",

pkg/table/tables/index_test.go

@@ -16,6 +16,7 @@ package tables_test
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/pingcap/tidb/pkg/ddl"
@@ -168,3 +169,72 @@ func buildTableInfo(t *testing.T, sql string) *model.TableInfo {
 	require.NoError(t, err)
 	return tblInfo
 }
+
+func TestGenIndexValueWithLargePaddingSize(t *testing.T) {
+	// ref https://github.com/pingcap/tidb/issues/47115
+	tblInfo := buildTableInfo(t, "create table t (a int, b int, k varchar(255), primary key (a, b), key (k))")
+	var idx table.Index
+	for _, idxInfo := range tblInfo.Indices {
+		if !idxInfo.Primary {
+			idx = tables.NewIndex(tblInfo.ID, tblInfo, idxInfo)
+			break
+		}
+	}
+	var a, b *model.ColumnInfo
+	for _, col := range tblInfo.Columns {
+		if col.Name.String() == "a" {
+			a = col
+		} else if col.Name.String() == "b" {
+			b = col
+		}
+	}
+	require.NotNil(t, a)
+	require.NotNil(t, b)
+
+	store := testkit.CreateMockStore(t)
+	txn, err := store.Begin()
+	require.NoError(t, err)
+	mockCtx := mock.NewContext()
+	sc := mockCtx.GetSessionVars().StmtCtx
+	padding := strings.Repeat(" ", 128)
+	idxColVals := types.MakeDatums("abc" + padding)
+	handleColVals := types.MakeDatums(1, 2)
+	encodedHandle, err := codec.EncodeKey(sc, nil, handleColVals...)
+	require.NoError(t, err)
+	commonHandle, err := kv.NewCommonHandle(encodedHandle)
+	require.NoError(t, err)
+
+	key, _, err := idx.GenIndexKey(sc, idxColVals, commonHandle, nil)
+	require.NoError(t, err)
+	_, err = idx.Create(mockCtx, txn, idxColVals, commonHandle, nil)
+	require.NoError(t, err)
+	val, err := txn.Get(context.Background(), key)
+	require.NoError(t, err)
+	colInfo := tables.BuildRowcodecColInfoForIndexColumns(idx.Meta(), tblInfo)
+	colInfo = append(colInfo, rowcodec.ColInfo{
+		ID:         a.ID,
+		IsPKHandle: false,
+		Ft:         rowcodec.FieldTypeFromModelColumn(a),
+	})
+	colInfo = append(colInfo, rowcodec.ColInfo{
+		ID:         b.ID,
+		IsPKHandle: false,
+		Ft:         rowcodec.FieldTypeFromModelColumn(b),
+	})
+	colVals, err := tablecodec.DecodeIndexKV(key, val, 1, tablecodec.HandleDefault, colInfo)
+	require.NoError(t, err)
+	require.Len(t, colVals, 3)
+	_, d, err := codec.DecodeOne(colVals[0])
+	require.NoError(t, err)
+	require.Equal(t, "abc"+padding, d.GetString())
+	_, d, err = codec.DecodeOne(colVals[1])
+	require.NoError(t, err)
+	require.Equal(t, int64(1), d.GetInt64())
+	_, d, err = codec.DecodeOne(colVals[2])
+	require.NoError(t, err)
+	require.Equal(t, int64(2), d.GetInt64())
+	handle, err := tablecodec.DecodeIndexHandle(key, val, 1)
+	require.NoError(t, err)
+	require.False(t, handle.IsInt())
+	require.Equal(t, commonHandle.Encoded(), handle.Encoded())
+}

pkg/tablecodec/tablecodec.go

@@ -877,7 +877,11 @@ func buildRestoredColumn(allCols []rowcodec.ColInfo) []rowcodec.ColInfo {
 		}
 		if collate.IsBinCollation(col.Ft.GetCollate()) {
 			// Change the fieldType from string to uint since we store the number of the truncated spaces.
+			// NOTE: the corresponding datum is generated as `types.NewUintDatum(paddingSize)`, and the raw data is
+			// encoded via `encodeUint`. Thus we should mark the field type as unsigened here so that the BytesDecoder
+			// can decode it correctly later. Otherwise there might be issues like #47115.
 			copyColInfo.Ft = types.NewFieldType(mysql.TypeLonglong)
+			copyColInfo.Ft.AddFlag(mysql.UnsignedFlag)
 		} else {
 			copyColInfo.Ft = allCols[i].Ft
 		}

tests/integrationtest/r/table/index.result

@@ -16,3 +16,7 @@ insert into t values (4, 3, 3);
 Error 1062 (23000): Duplicate entry '3' for key 't.v2'
 set @@tidb_txn_assertion_level=default;
 set @@tidb_constraint_check_in_place=default;
+drop table if exists t;
+create table t (a int, b int, k varchar(255), primary key (a, b), key k (k));
+insert into t values (1, 1, 'abc                                                                                                                                ');
+drop table t;

tests/integrationtest/t/table/index.test

@@ -20,3 +20,8 @@ insert into t values (4, 3, 3);
 set @@tidb_txn_assertion_level=default;
 set @@tidb_constraint_check_in_place=default;
 
+# Test Issue 47115.
+drop table if exists t;
+create table t (a int, b int, k varchar(255), primary key (a, b), key k (k));
+insert into t values (1, 1, 'abc                                                                                                                                ');
+drop table t;