lightning: Fix panic when downstream table schema has changed (#36805)

close #34163

br/pkg/lightning/backend/local/local.go

@@ -58,6 +58,7 @@ import (
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/table"
 	"github.com/pingcap/tidb/tablecodec"
+	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/codec"
 	"github.com/pingcap/tidb/util/engine"
 	"github.com/pingcap/tidb/util/mathutil"
@@ -1693,6 +1694,10 @@ func (local *local) ResolveDuplicateRows(ctx context.Context, tbl table.Table, t
 				if err == nil {
 					return nil
 				}
+				if types.ErrBadNumber.Equal(err) {
+					logger.Warn("delete duplicate rows encounter error", log.ShortError(err))
+					return common.ErrResolveDuplicateRows.Wrap(err).GenWithStackByArgs(tableName)
+				}
 				if log.IsContextCanceledError(err) {
 					return err
 				}

br/pkg/lightning/common/errors.go

@@ -81,19 +81,20 @@ var (
 	ErrKVIngestFailed        = errors.Normalize("ingest tikv failed", errors.RFCCodeText("Lightning:KV:ErrKVIngestFailed"))
 	ErrKVRaftProposalDropped = errors.Normalize("raft proposal dropped", errors.RFCCodeText("Lightning:KV:ErrKVRaftProposalDropped"))
 
-	ErrUnknownBackend     = errors.Normalize("unknown backend %s", errors.RFCCodeText("Lightning:Restore:ErrUnknownBackend"))
-	ErrCheckLocalFile     = errors.Normalize("cannot find local file for table: %s engineDir: %s", errors.RFCCodeText("Lightning:Restore:ErrCheckLocalFile"))
-	ErrOpenDuplicateDB    = errors.Normalize("open duplicate db error", errors.RFCCodeText("Lightning:Restore:ErrOpenDuplicateDB"))
-	ErrSchemaNotExists    = errors.Normalize("table `%s`.`%s` schema not found", errors.RFCCodeText("Lightning:Restore:ErrSchemaNotExists"))
-	ErrInvalidSchemaStmt  = errors.Normalize("invalid schema statement: '%s'", errors.RFCCodeText("Lightning:Restore:ErrInvalidSchemaStmt"))
-	ErrCreateSchema       = errors.Normalize("create schema failed, table: %s, stmt: %s", errors.RFCCodeText("Lightning:Restore:ErrCreateSchema"))
-	ErrUnknownColumns     = errors.Normalize("unknown columns in header (%s) for table %s", errors.RFCCodeText("Lightning:Restore:ErrUnknownColumns"))
-	ErrChecksumMismatch   = errors.Normalize("checksum mismatched remote vs local => (checksum: %d vs %d) (total_kvs: %d vs %d) (total_bytes:%d vs %d)", errors.RFCCodeText("Lighting:Restore:ErrChecksumMismatch"))
-	ErrRestoreTable       = errors.Normalize("restore table %s failed", errors.RFCCodeText("Lightning:Restore:ErrRestoreTable"))
-	ErrEncodeKV           = errors.Normalize("encode kv error in file %s at offset %d", errors.RFCCodeText("Lightning:Restore:ErrEncodeKV"))
-	ErrAllocTableRowIDs   = errors.Normalize("allocate table row id error", errors.RFCCodeText("Lightning:Restore:ErrAllocTableRowIDs"))
-	ErrInvalidMetaStatus  = errors.Normalize("invalid meta status: '%s'", errors.RFCCodeText("Lightning:Restore:ErrInvalidMetaStatus"))
-	ErrTableIsChecksuming = errors.Normalize("table '%s' is checksuming", errors.RFCCodeText("Lightning:Restore:ErrTableIsChecksuming"))
+	ErrUnknownBackend       = errors.Normalize("unknown backend %s", errors.RFCCodeText("Lightning:Restore:ErrUnknownBackend"))
+	ErrCheckLocalFile       = errors.Normalize("cannot find local file for table: %s engineDir: %s", errors.RFCCodeText("Lightning:Restore:ErrCheckLocalFile"))
+	ErrOpenDuplicateDB      = errors.Normalize("open duplicate db error", errors.RFCCodeText("Lightning:Restore:ErrOpenDuplicateDB"))
+	ErrSchemaNotExists      = errors.Normalize("table `%s`.`%s` schema not found", errors.RFCCodeText("Lightning:Restore:ErrSchemaNotExists"))
+	ErrInvalidSchemaStmt    = errors.Normalize("invalid schema statement: '%s'", errors.RFCCodeText("Lightning:Restore:ErrInvalidSchemaStmt"))
+	ErrCreateSchema         = errors.Normalize("create schema failed, table: %s, stmt: %s", errors.RFCCodeText("Lightning:Restore:ErrCreateSchema"))
+	ErrUnknownColumns       = errors.Normalize("unknown columns in header (%s) for table %s", errors.RFCCodeText("Lightning:Restore:ErrUnknownColumns"))
+	ErrChecksumMismatch     = errors.Normalize("checksum mismatched remote vs local => (checksum: %d vs %d) (total_kvs: %d vs %d) (total_bytes:%d vs %d)", errors.RFCCodeText("Lighting:Restore:ErrChecksumMismatch"))
+	ErrRestoreTable         = errors.Normalize("restore table %s failed", errors.RFCCodeText("Lightning:Restore:ErrRestoreTable"))
+	ErrEncodeKV             = errors.Normalize("encode kv error in file %s at offset %d", errors.RFCCodeText("Lightning:Restore:ErrEncodeKV"))
+	ErrAllocTableRowIDs     = errors.Normalize("allocate table row id error", errors.RFCCodeText("Lightning:Restore:ErrAllocTableRowIDs"))
+	ErrInvalidMetaStatus    = errors.Normalize("invalid meta status: '%s'", errors.RFCCodeText("Lightning:Restore:ErrInvalidMetaStatus"))
+	ErrTableIsChecksuming   = errors.Normalize("table '%s' is checksuming", errors.RFCCodeText("Lightning:Restore:ErrTableIsChecksuming"))
+	ErrResolveDuplicateRows = errors.Normalize("resolve duplicate rows error on table '%s'", errors.RFCCodeText("Lightning:Restore:ErrResolveDuplicateRows"))
 )
 
 type withStack struct {

errors.toml

@@ -516,6 +516,11 @@ error = '''
 open duplicate db error
 '''
 
+["Lightning:Restore:ErrResolveDuplicateRows"]
+error = '''
+resolve duplicate rows error on table '%s'
+'''
+
 ["Lightning:Restore:ErrRestoreTable"]
 error = '''
 restore table %s failed

types/mydecimal.go

@@ -1406,7 +1406,13 @@ func (d *MyDecimal) FromBin(bin []byte, precision, frac int) (binSize int, err e
 	if bin[binIdx]&0x80 > 0 {
 		mask = 0
 	}
-	binSize = DecimalBinSize(precision, frac)
+	binSize, err = DecimalBinSize(precision, frac)
+	if err != nil {
+		return 0, err
+	}
+	if binSize < 0 || binSize > 40 {
+		return 0, ErrBadNumber
+	}
 	dCopy := make([]byte, 40)
 	dCopy = dCopy[:binSize]
 	copy(dCopy, bin)
@@ -1482,13 +1488,16 @@ func (d *MyDecimal) FromBin(bin []byte, precision, frac int) (binSize int, err e
 }
 
 // DecimalBinSize returns the size of array to hold a binary representation of a decimal.
-func DecimalBinSize(precision, frac int) int {
+func DecimalBinSize(precision, frac int) (int, error) {
 	digitsInt := precision - frac
 	wordsInt := digitsInt / digitsPerWord
 	wordsFrac := frac / digitsPerWord
 	xInt := digitsInt - wordsInt*digitsPerWord
 	xFrac := frac - wordsFrac*digitsPerWord
-	return wordsInt*wordSize + dig2bytes[xInt] + wordsFrac*wordSize + dig2bytes[xFrac]
+	if xInt < 0 || xInt >= len(dig2bytes) || xFrac < 0 || xFrac >= len(dig2bytes) {
+		return 0, ErrBadNumber
+	}
+	return wordsInt*wordSize + dig2bytes[xInt] + wordsFrac*wordSize + dig2bytes[xFrac], nil
 }
 
 func readWord(b []byte, size int) int32 {
@@ -2379,7 +2388,11 @@ func DecimalPeak(b []byte) (int, error) {
 	}
 	precision := int(b[0])
 	frac := int(b[1])
-	return DecimalBinSize(precision, frac) + 2, nil
+	binSize, err := DecimalBinSize(precision, frac)
+	if err != nil {
+		return 0, err
+	}
+	return binSize + 2, nil
 }
 
 // NewDecFromInt creates a MyDecimal from int.

types/mydecimal_test.go

@@ -476,17 +476,44 @@ func TestToBinFromBin(t *testing.T) {
 	var dec MyDecimal
 	dec.FromInt(1)
 	errTests := []struct {
-		prec int
-		frac int
+		prec       int
+		frac       int
+		ToBinErr   error
+		FromBinErr error
 	}{
-		{82, 1},
-		{-1, 1},
-		{10, 31},
-		{10, -1},
+		{82, 1, ErrBadNumber, ErrTruncated},
+		{-1, 1, ErrBadNumber, ErrBadNumber},
+		{10, 31, ErrBadNumber, ErrBadNumber},
+		{10, -1, ErrBadNumber, ErrBadNumber},
 	}
 	for _, tt := range errTests {
 		_, err := dec.ToBin(tt.prec, tt.frac)
-		require.True(t, ErrBadNumber.Equal(err))
+		require.Equal(t, tt.ToBinErr, err)
+		err = dec.FromString([]byte{'0'})
+		require.NoError(t, err)
+		buf, err := dec.ToBin(1, 0)
+		require.NoError(t, err)
+		_, err = dec.FromBin(buf, tt.prec, tt.frac)
+		require.Equal(t, tt.FromBinErr, err)
+	}
+}
+
+func TestDecimalBinSize(t *testing.T) {
+	type tcase struct {
+		precision int
+		frac      int
+		output    int
+		err       error
+	}
+	tests := []tcase{
+		{3, 1, 2, nil},
+		{-1, 0, 0, ErrBadNumber},
+		{3, 5, 0, ErrBadNumber},
+	}
+	for _, tt := range tests {
+		binSize, err := DecimalBinSize(tt.precision, tt.frac)
+		require.Equal(t, tt.output, binSize)
+		require.Equal(t, tt.err, err)
 	}
 }
 

util/codec/codec.go

@@ -157,7 +157,12 @@ func EstimateValueSize(sc *stmtctx.StatementContext, val types.Datum) (int, erro
 	case types.KindString, types.KindBytes:
 		l = valueSizeOfBytes(val.GetBytes())
 	case types.KindMysqlDecimal:
-		l = valueSizeOfDecimal(val.GetMysqlDecimal(), val.Length(), val.Frac()) + 1
+		var err error
+		l, err = valueSizeOfDecimal(val.GetMysqlDecimal(), val.Length(), val.Frac())
+		if err != nil {
+			return 0, err
+		}
+		l = l + 1
 	case types.KindMysqlEnum:
 		l = valueSizeOfUnsignedInt(val.GetMysqlEnum().Value)
 	case types.KindMysqlSet:

util/codec/decimal.go

@@ -34,11 +34,15 @@ func EncodeDecimal(b []byte, dec *types.MyDecimal, precision, frac int) ([]byte,
 	return b, errors.Trace(err)
 }
 
-func valueSizeOfDecimal(dec *types.MyDecimal, precision, frac int) int {
+func valueSizeOfDecimal(dec *types.MyDecimal, precision, frac int) (int, error) {
 	if precision == 0 {
 		precision, frac = dec.PrecisionAndFrac()
 	}
-	return types.DecimalBinSize(precision, frac) + 2
+	binSize, err := types.DecimalBinSize(precision, frac)
+	if err != nil {
+		return 0, err
+	}
+	return binSize + 2, nil
 }
 
 // DecodeDecimal decodes bytes to decimal.