push-down-test: make test case parallel (#22)

Signed-off-by: Lonng <heng@lonng.org>

README.md

@@ -69,6 +69,13 @@ export tidb_src_dir=`realpath ../tidb`
 make push-down-test
 ```
 
+If you want to filter some test cases:
+
+```
+include=1_arith_1.sql make push-down-test
+exclude=1_arith_1.sql make push-down-test
+```
+
 [failpoint]: https://github.com/pingcap/failpoint
 [randgen]: https://github.com/MariaDB/randgen
 [TiDB]: https://github.com/pingcap/community/blob/master/CONTRIBUTING.md#building-tidb-on-a-local-osshell-environment

push-down-test/run-tests.sh

@@ -20,6 +20,8 @@ tidb_src_dir=$tidb_src_dir
 tidb_bin=""
 pd_bin=$pd_bin
 tikv_bin=$tikv_bin
+include=$include
+exclude=$exclude
 # the hash of tidb_master_bin and pd_master_bin
 #tidb_master_sha=$tidb_master_sha
 #pd_master_sha=$pd_master_sha
@@ -92,11 +94,11 @@ function run_tidbs() {
   "$tidb_bin" -log-file "$no_push_down_tidb_log_file" -config "$no_push_down_config_dir"/tidb.toml -L ${log_level} &
   tidb_no_push_down_pid=$!
 
-  export GO_FAILPOINTS="github.com/pingcap/tidb/expression/PushDownTestSwitcher=return(\"$push_down_func_list\")"
+  export GO_FAILPOINTS="github.com/pingcap/tidb/expression/PushDownTestSwitcher=return(\"$push_down_func_list\");github.com/pingcap/tidb/expression/PanicIfPbCodeUnspecified=return(true)"
   "$tidb_bin" -log-file "$push_down_no_batch_tidb_log_file" -config "$push_down_no_batch_config_dir"/tidb.toml -L ${log_level} &
   tidb_push_down_no_batch_pid=$!
 
-  export GO_FAILPOINTS="github.com/pingcap/tidb/expression/PushDownTestSwitcher=return(\"$push_down_func_list\")"
+  export GO_FAILPOINTS="github.com/pingcap/tidb/expression/PushDownTestSwitcher=return(\"$push_down_func_list\");github.com/pingcap/tidb/expression/PanicIfPbCodeUnspecified=return(true)"
   "$tidb_bin" -log-file "$push_down_with_batch_tidb_log_file" -config "$push_down_with_batch_config_dir"/tidb.toml -L ${log_level} &
   tidb_push_down_with_batch_pid=$!
 }
@@ -254,7 +256,9 @@ echo "+ Start test"
 ./$push_down_test_bin \
   -conn-no-push "${tidb_user}@tcp(${tidb_host}:${no_push_down_tidb_port})/{db}?allowNativePasswords=true" \
   -conn-push "${tidb_user}@tcp(${tidb_host}:${push_down_tidb_port})/{db}?allowNativePasswords=true" \
-  -conn-push-with-batch "${tidb_user}@tcp(${tidb_host}:${push_down_with_batch_tidb_port})/{db}?allowNativePasswords=true"
+  -conn-push-with-batch "${tidb_user}@tcp(${tidb_host}:${push_down_with_batch_tidb_port})/{db}?allowNativePasswords=true" \
+  -include "${include}" \
+  -exclude "${exclude}"
 readonly exit_code=$?
 
 echo "+ Test finished"

push-down-test/src/main.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/parser"
@@ -19,13 +20,15 @@ import (
 )
 
 const testCaseDir = "./sql"
+const testPrepDir = "./prepare"
 
 var connStrNoPush *string
 var connStrPush *string
 var connStrPushWithBatch *string
 var outputSuccessQueries *bool
 var dbName *string
 var verboseOutput *bool
+var fileFilter func(file string) bool
 
 type statementLog struct {
 	output    *bytes.Buffer
@@ -64,43 +67,78 @@ func prepareDB(connString string) {
 	mustDBClose(db)
 }
 
-func iterateTestCases() {
+func iterateTestCases(dir string, parallel bool) {
 	successCases := 0
 	failedCases := 0
 
-	err := filepath.Walk(testCaseDir, func(path string, info os.FileInfo, err error) error {
+	var files []string
+	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
 		if info.IsDir() {
 			return nil
 		}
 		if !strings.HasSuffix(info.Name(), ".sql") {
 			return nil
 		}
-		log.Printf("Testing [%s]...", path)
-		if runTestCase(path) {
-			successCases++
-		} else {
-			failedCases++
+		if fileFilter != nil && !fileFilter(path) {
+			return nil
 		}
+		files = append(files, path)
 		return nil
 	})
 	if err != nil {
 		log.Panicf("Failed to read test case directory [%s]: %v\n", testCaseDir, err)
 	}
+
+	if !parallel {
+		for _, path := range files {
+			log.Printf("Serial Testing [%s]...", path)
+			if runTestCase(path) {
+				successCases++
+			} else {
+				failedCases++
+			}
+		}
+	} else {
+		wg := &sync.WaitGroup{}
+		ch := make(chan bool, len(files))
+		for _, path := range files {
+			wg.Add(1)
+			go func(path string) {
+				defer wg.Done()
+				log.Printf("Parallel Testing [%s]...", path)
+				if runTestCase(path) {
+					ch <- true
+				} else {
+					ch <- false
+				}
+			}(path)
+		}
+		wg.Wait()
+		close(ch)
+		for succ := range ch {
+			if succ {
+				successCases++
+			} else {
+				failedCases++
+			}
+		}
+	}
+
 	log.Printf("All test finished: pass cases: %d, fail cases: %d", successCases, failedCases)
 	if failedCases > 0 {
 		os.Exit(2) // Diff fail results in exit code 2 to distinguish with panic
 	}
 }
 
 func runTestCase(testCasePath string) bool {
-	log.Printf("Parsing...")
+	log.Printf("Parsing...[%s]", testCasePath)
 	stmtAsts := readAndParseSQLText(testCasePath)
 	statements := make([]string, 0, len(stmtAsts))
 	for _, stmt := range stmtAsts {
 		statements = append(statements, stmt.Text())
 	}
 
-	log.Printf("Running...")
+	log.Printf("Running...[%s]", testCasePath)
 	noPushDownLogChan := make(chan *statementLog)
 	pushDownLogChan := make(chan *statementLog)
 	pushDownWithBatchLogChan := make(chan *statementLog)
@@ -157,7 +195,10 @@ func diffRunResult(
 	execOkStatements := 0
 	execFailStatements := 0
 	diffFailStatements := 0
+
 	successQueries := new(bytes.Buffer)
+	output := new(bytes.Buffer)
+	logger := log.New(output, "", log.LstdFlags)
 
 	for {
 		noPushDownLog, ok1 := <-noPushDownLogChan
@@ -169,22 +210,22 @@ func diffRunResult(
 			break
 		}
 		if !ok1 {
-			log.Panicf("Internal error: NoPushDown channel drained\n")
+			logger.Panicf("Internal error: NoPushDown channel drained\n")
 		}
 		if !ok2 {
-			log.Panicf("Internal error: PushDownWithoutVec channel drained\n")
+			logger.Panicf("Internal error: PushDownWithoutVec channel drained\n")
 		}
 		if !ok3 {
-			log.Panicf("Internal error: PushDownWithVec channel drained\n")
+			logger.Panicf("Internal error: PushDownWithVec channel drained\n")
 		}
 		if noPushDownLog.stmt != pushDownLog.stmt ||
 			pushDownLog.stmt != pushDownWithBatchLog.stmt {
-			log.Panicln("Internal error: Pre-check failed, stmt should be identical",
+			logger.Panicln("Internal error: Pre-check failed, stmt should be identical",
 				noPushDownLog.stmt, pushDownLog.stmt, pushDownWithBatchLog.stmt)
 		}
 		if noPushDownLog.stmtIndex != pushDownLog.stmtIndex ||
 			pushDownLog.stmtIndex != pushDownWithBatchLog.stmtIndex {
-			log.Panicln("Internal error: Pre-check failed, stmtIndex should be identical",
+			logger.Panicln("Internal error: Pre-check failed, stmtIndex should be identical",
 				noPushDownLog.stmtIndex, pushDownLog.stmtIndex, pushDownWithBatchLog.stmtIndex)
 		}
 
@@ -212,7 +253,7 @@ func diffRunResult(
 
 		if diffFail {
 			diffFailStatements++
-			log.Printf("Test fail: Outputs are not matching.\n"+
+			logger.Printf("Test fail: Outputs are not matching.\n"+
 				"Test case: %s\n"+
 				"Statement: #%d - %s\n"+
 				"NoPushDown Output: \n%s\n"+
@@ -226,7 +267,7 @@ func diffRunResult(
 				string(pushDownWithBatchLog.output.Bytes()))
 		} else if hasError {
 			if *verboseOutput {
-				log.Printf("Warn: Execute fail, diff skipped.\n"+
+				logger.Printf("Warn: Execute fail, diff skipped.\n"+
 					"Test case: %s\n"+
 					"Statement: #%d - %s\n"+
 					"NoPushDown Output: \n%s\n"+
@@ -242,28 +283,31 @@ func diffRunResult(
 		} else {
 			if *verboseOutput {
 				// Output is same and there is no errors
-				log.Printf("Info: SQL result is idential: \n%s\n", noPushDownLog.stmt)
+				logger.Printf("Info: SQL result is idential: \n%s\n", noPushDownLog.stmt)
 			}
 
 			successQueries.WriteString(noPushDownLog.stmt)
 			successQueries.WriteByte('\n')
 		}
 	}
 
-	log.Printf("Test summary: non-matching queries: %d, success queries: %d, skipped queries: %d",
+	logger.Printf("Test summary: non-matching queries: %d, success queries: %d, skipped queries: %d",
 		diffFailStatements,
 		execOkStatements,
 		execFailStatements)
 
 	if diffFailStatements == 0 {
-		log.Printf("Test summary: Test case PASS")
+		logger.Printf("Test summary(%s): Test case PASS", testCasePath)
 	} else {
-		log.Printf("Test summary: Test case FAIL")
+		logger.Printf("Test summary(%s): Test case FAIL", testCasePath)
 	}
 
+	// combine all output
+	log.Println(output.String())
+
 	if *outputSuccessQueries {
 		outputFilePath := testCasePath + ".success"
-		log.Printf("Output success queries to [%s]", outputFilePath)
+		logger.Printf("Output success queries to [%s]", outputFilePath)
 		expectNoErr(ioutil.WriteFile(outputFilePath, successQueries.Bytes(), 0644))
 	}
 
@@ -281,13 +325,44 @@ func main() {
 	outputSuccessQueries = flag.Bool("output-success", false, "Output success queries of test cases to a file ends with '.success' along with the original test case")
 	dbName = flag.String("db", "push_down_test_db", "The database name to run test cases")
 	verboseOutput = flag.Bool("verbose", false, "Verbose output")
+	includeFiles := flag.String("include", "", "Test cases included in this test (file lists separated by comma)")
+	excludeFiles := flag.String("exclude", "", "Test cases excluded in this test (file lists separated by comma)")
 
 	flag.Parse()
 
 	prepareDB(*connStrNoPush)
 	prepareDB(*connStrPush)
 	prepareDB(*connStrPushWithBatch)
 
+	// Prepare SQL does not apply the filter
+	iterateTestCases(testPrepDir, false)
+
 	log.Printf("Prepare finished, start testing...")
-	iterateTestCases()
+
+	// Build the filter
+	var includeList, excludeList []string
+	if len(*includeFiles) > 0 {
+		includeList = strings.Split(*includeFiles, ",")
+	}
+	if len(*excludeFiles) > 0 {
+		excludeList = strings.Split(*excludeFiles, ",")
+	}
+	fileFilter = func(file string) bool {
+		if len(includeList) == 0 && len(excludeList) == 0 {
+			return true
+		}
+		base := filepath.Base(file)
+		for _, ex := range excludeList {
+			if ex == base {
+				return false
+			}
+		}
+		for _, in := range includeList {
+			if in == base {
+				return true
+			}
+		}
+		return false
+	}
+	iterateTestCases(testCaseDir, true)
 }