add MySQL chaos test (#200)

* refine pick index logic
* add mysql connection chaos test

tests/chaos/run.sh

@@ -0,0 +1,28 @@
+#!/bin/sh
+
+set -eu
+cur=$(cd `dirname $0`; pwd)
+
+DB_NAME="chaos"
+TABLE_NAME="t"
+
+# drop database on mysql
+run_sql "drop database if exists \`$DB_NAME\`;"
+
+# build data on mysql
+run_sql "create database $DB_NAME;"
+run_sql "create table $DB_NAME.$TABLE_NAME (a int(255));"
+
+# insert 100 records
+run_sql "insert into $DB_NAME.$TABLE_NAME values $(seq -s, 100 | sed 's/,*$//g' | sed "s/[0-9]*/('1')/g");"
+
+# dumping with consistency none
+export DUMPLING_TEST_DATABASE=$DB_NAME
+export GO_FAILPOINTS="github.com/pingcap/dumpling/v4/export/ChaosBrokenMySQLConn=1*return"
+run_dumpling --consistency=none --loglevel debug
+
+# check data record count
+cnt=`grep -o "(1)" ${DUMPLING_OUTPUT_DIR}/${DB_NAME}.${TABLE_NAME}.000000000.sql|wc -l`
+echo "1st records count is ${cnt}"
+[ $cnt = 100 ]
+export GO_FAILPOINTS=""

v4/export/sql.go

@@ -342,31 +342,23 @@ func GetPrimaryKeyColumns(db *sql.Conn, database, table string) ([]string, error
 }
 
 func GetPrimaryKeyName(db *sql.Conn, database, table string) (string, error) {
-	priKeyQuery := "SELECT column_name FROM information_schema.columns " +
-		"WHERE table_schema = ? AND table_name = ? AND column_key = 'PRI';"
-	var colName string
-	row := db.QueryRowContext(context.Background(), priKeyQuery, database, table)
-	if err := row.Scan(&colName); err != nil {
-		if err == sql.ErrNoRows {
-			return "", nil
-		} else {
-			return "", errors.Annotatef(err, "sql: %s", priKeyQuery)
-		}
-	}
-	return colName, nil
+	return getNumericIndex(db, database, table, "PRI")
 }
 
 func GetUniqueIndexName(db *sql.Conn, database, table string) (string, error) {
-	uniKeyQuery := "SELECT column_name FROM information_schema.columns " +
-		"WHERE table_schema = ? AND table_name = ? AND column_key = 'UNI';"
+	return getNumericIndex(db, database, table, "UNI")
+}
+
+func getNumericIndex(db *sql.Conn, database, table, indexType string) (string, error) {
+	keyQuery := "SELECT column_name FROM information_schema.columns " +
+		"WHERE table_schema = ? AND table_name = ? AND column_key = ? AND data_type IN ('int', 'bigint');"
 	var colName string
-	row := db.QueryRowContext(context.Background(), uniKeyQuery, database, table)
-	if err := row.Scan(&colName); err != nil {
-		if err == sql.ErrNoRows {
-			return "", nil
-		} else {
-			return "", errors.Annotatef(err, "sql: %s", uniKeyQuery)
-		}
+	row := db.QueryRowContext(context.Background(), keyQuery, database, table, indexType)
+	err := row.Scan(&colName)
+	if err == sql.ErrNoRows {
+		return "", nil
+	} else if err != nil {
+		return "", errors.Annotatef(err, "sql: %s, indexType: %s", keyQuery, indexType)
 	}
 	return colName, nil
 }
@@ -633,28 +625,8 @@ func pickupPossibleField(dbName, tableName string, db *sql.Conn, conf *Config) (
 		}
 	}
 
-	// there is no proper index
-	if fieldName == "" {
-		return "", nil
-	}
-
-	query := "SELECT DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS " +
-		"WHERE TABLE_NAME = ? AND COLUMN_NAME = ?"
-	var fieldType string
-	row := db.QueryRowContext(context.Background(), query, tableName, fieldName)
-	err = row.Scan(&fieldType)
-	if err != nil {
-		if err == sql.ErrNoRows {
-			return "", nil
-		} else {
-			return "", errors.Annotatef(err, "sql: %s", query)
-		}
-	}
-	switch strings.ToLower(fieldType) {
-	case "int", "bigint":
-		return fieldName, nil
-	}
-	return "", nil
+	// if fieldName == "", there is no proper index
+	return fieldName, nil
 }
 
 func estimateCount(dbName, tableName string, db *sql.Conn, field string, conf *Config) uint64 {

v4/export/writer_util.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/pingcap/br/pkg/storage"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/dumpling/v4/log"
@@ -189,6 +190,10 @@ func WriteInsert(pCtx context.Context, tblIR TableDataIR, w storage.Writer, file
 			}
 			counter += 1
 			wp.AddFileSize(uint64(bf.Len()-lastBfSize) + 2) // 2 is for ",\n" and ";\n"
+			failpoint.Inject("ChaosBrokenMySQLConn", func(_ failpoint.Value) {
+				tblIR.Close()
+				failpoint.Return(errors.New("connection is closed"))
+			})
 
 			fileRowIter.Next()
 			shouldSwitch := wp.ShouldSwitchStatement()