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

#37232)

close #34163

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

@@ -56,6 +56,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"
 	tikverror "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/oracle"
@@ -1446,6 +1447,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 errors.Errorf("resolve duplicate rows error on table '%s'", tableName)
+				}
 				if log.IsContextCanceledError(err) {
 					return err
 				}

br/tests/br_other/run.sh

@@ -52,7 +52,7 @@ run_br debug backupmeta validate -s "local://$TEST_DIR/$DB" --offset 100
 run_br validate checksum -s "local://$TEST_DIR/$DB"
 
 # Test validate checksum
-for sst in $TEST_DIR/$DB/*.sst; do
+for sst in $TEST_DIR/$DB/*/*.sst; do
     echo "corrupted!" >> $sst
     echo "$sst corrupted!"
     break

types/mydecimal.go

@@ -1401,7 +1401,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)
@@ -1477,13 +1483,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 {
@@ -2340,7 +2349,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

@@ -475,17 +475,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.