Fix pingcap#34163

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

@@ -56,8 +56,10 @@ import (
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
+	"github.com/pingcap/tidb/parser/terror"
 	"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"
@@ -1681,6 +1683,12 @@ func (local *local) ResolveDuplicateRows(ctx context.Context, tbl table.Table, t
 				if err == nil {
 					return nil
 				}
+				if terror.ErrorEqual(err, types.ErrBadNumber) {
+					logger.Warn("delete duplicate rows encounter error", log.ShortError(err))
+					return errors.New(fmt.Sprintf(`delete duplicate rows encounter error: %s;
+RawCause: table schema has changed, expected decimal, received non-decimal type; 
+Workaround: please check whether table schemas in downstream are correct.`, err.Error()))
+				}
 				if log.IsContextCanceledError(err) {
 					return err
 				}

types/mydecimal.go

@@ -1403,8 +1403,14 @@ 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
+	}
 	dCopy := make([]byte, 40)
+	if binSize < 0 || binSize > 40 {
+		return 0, ErrBadNumber
+	}
 	dCopy = dCopy[:binSize]
 	copy(dCopy, bin)
 	dCopy[0] ^= 0x80
@@ -1479,13 +1485,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 {
@@ -2376,7 +2385,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.

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.