lightning: fix panic due to row's column exceeds definition

br/pkg/lightning/backend/tidb/tidb.go

@@ -308,11 +308,21 @@ func (enc *tidbEncoder) Encode(logger log.Logger, row []types.Datum, _ int64, co
 	// See: tests/generated_columns/data/gencol.various_types.0.sql this sql has no columns, so encodeLoop will fill the
 	// column permutation with default, thus enc.columnCnt > len(row).
 	if len(row) < enc.columnCnt {
+		// 1. if len(row) < enc.columnCnt: data in row cannot populate the insert statement, because
+		// there are enc.columnCnt elements to insert but fewer columns in row
 		logger.Error("column count mismatch", zap.Ints("column_permutation", columnPermutation),
 			zap.Array("data", kv.RowArrayMarshaler(row)))
 		return emptyTiDBRow, errors.Errorf("column count mismatch, expected %d, got %d", enc.columnCnt, len(row))
 	}
 
+	if len(row) > len(enc.columnIdx) {
+		// 2. if len(row) > len(columnIdx): raw row data has more columns than those
+		// in the table
+		logger.Error("column count mismatch", zap.Ints("column_count", enc.columnIdx),
+			zap.Array("data", kv.RowArrayMarshaler(row)))
+		return emptyTiDBRow, errors.Errorf("column count mismatch, at most %d but got %d", len(enc.columnIdx), len(row))
+	}
+
 	var encoded strings.Builder
 	encoded.Grow(8 * len(row))
 	encoded.WriteByte('(')

br/pkg/lightning/backend/tidb/tidb_test.go

@@ -594,6 +594,14 @@ func encodeRowsTiDB(t *testing.T, b backend.Backend, tbl table.Table) kv.Rows {
 	require.NoError(t, err)
 
 	row.ClassifyAndAppend(&dataRows, &dataChecksum, &indexRows, &indexChecksum)
+
+	rawRow := make([]types.Datum, 0)
+	for i := 0; i < 15; i++ {
+		rawRow = append(rawRow, types.NewIntDatum(0))
+	}
+	row, err = encoder.Encode(logger, rawRow, 1, []int{0, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1}, "12.csv", 0)
+	require.NotNil(t, err)
+	require.Contains(t, err.Error(), "column count mismatch, at most")
 	return dataRows
 }
 

br/tests/lightning_routes/run.sh

@@ -21,5 +21,3 @@ check_contains 'sum(x): 1296'
 run_sql 'SHOW TABLES IN routes_a1;'
 check_not_contains 'Tables_in_routes_a1: t2'
 
-run_sql 'SHOW DATABASES;'
-check_not_contains 'Database: routes_a0'