tiny update for restoring verification (pingcap#8)

* refine codes on verfication after restoring finished

* adjust comment

* correct gramma mistake

ingest/restore/restore.go

@@ -524,7 +524,7 @@ func (tr *TableRestore) onFinished() {
 	tr.ingestKV()
 
 	// verify table data
-	tr.verifyTable()
+	tr.verifyTable(tableRows)
 
 	return
 }
@@ -582,32 +582,64 @@ func (tr *TableRestore) ingestKV() error {
 	return nil
 }
 
-func (tr *TableRestore) verifyTable() error {
+func (tr *TableRestore) verifyTable(rows uint64) error {
 	table := tr.tableInfo.Name
 	log.Infof("[%s] verifying table ...", table)
 
 	start := time.Now()
 	defer func() {
 		metrics.MarkTiming(fmt.Sprintf("[%s]_verify", table), start)
+		log.Infof("[%s] finish verification", table)
 	}()
 
-	/*
-		TODO : compare executed rows == count(*)
-	*/
+	if err := tr.verifyQuantity(rows); err != nil {
+		log.Errorf("[%s] verify quantity failed : %s", table, err.Error())
+		return err
+	}
+	log.Infof("[%s] owns %d rows integrallty !", table, rows)
 
-	/*{
-		dsn := tr.cfg.TiDB
-		tidb := ConnectDB(dsn.Host, dsn.Port, dsn.User, dsn.Pwd)
-		defer tidb.Close()
+	if tr.cfg.Verify.RunCheckTable {
+		if err := tr.excCheckTable(); err != nil {
+			log.Errorf("[%s] verify check table failed : %s", table, err.Error())
+			return err
+		}
+	}
+
+	return nil
+}
 
-		tidb.Exec("USE " + tr.tableMeta.DB)
-		_, err := tidb.Exec("ADMIN CHECK TABLE " + tr.tableMeta.Name)
+func (tr *TableRestore) verifyQuantity(expectRows uint64) error {
+	dsn := tr.cfg.TiDB
+	db := common.ConnectDB(dsn.Host, dsn.Port, dsn.User, dsn.Pwd)
+	defer db.Close()
+
+	rows := uint64(0)
+	r := db.QueryRow(
+		fmt.Sprintf("SELECT COUNT(*) FROM `%s.%s`", tr.tableMeta.DB, tr.tableInfo.Name))
+	if err := r.Scan(&rows); err != nil {
 		return err
-	}*/
+	}
+
+	if rows != expectRows {
+		return errors.Errorf("[verify] Rows num not equal %d (expect = %d)", rows, expectRows)
+	}
 
 	return nil
 }
 
+func (tr *TableRestore) excCheckTable() error {
+	log.Infof("Verify by execute `admin check table` : %s", tr.tableMeta.Name)
+
+	dsn := tr.cfg.TiDB
+	db := common.ConnectDB(dsn.Host, dsn.Port, dsn.User, dsn.Pwd)
+	defer db.Close()
+
+	// verify datas completion via command "admin check table"
+	_, err := db.Exec(
+		fmt.Sprintf("ADMIN CHECK TABLE `%s.%s`", tr.tableMeta.DB, tr.tableMeta.Name))
+	return err
+}
+
 ////////////////////////////////////////////////////////////////
 
 type RegionRestoreExectuor struct {