support plan replayer for tpch (#131)

* support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> support plan replayer Signed-off-by: yisaer <disxiaofei@163.com> * try new test Signed-off-by: yisaer <disxiaofei@163.com> Signed-off-by: yisaer <disxiaofei@163.com>
pingcap · Aug 18, 2022 · 956083c · 956083c
1 parent 062d2d7
commit 956083c
Show file tree

Hide file tree

Showing 12 changed files with 228 additions and 17 deletions.
diff --git a/.gitignore b/.gitignore
@@ -3,3 +3,4 @@ bin
 dist/
 .vscode/
 .DS_Store
+vendor/
diff --git a/ch/workload.go b/ch/workload.go
@@ -278,3 +278,15 @@ func (w Workloader) OutputStats(ifSummaryReport bool) {
 func (w Workloader) DBName() string {
 	return w.cfg.DBName
 }
+
+func (w Workloader) IsPlanReplayerDumpEnabled() bool {
+	return false
+}
+
+func (w Workloader) PreparePlanReplayerDump() error {
+	return nil
+}
+
+func (w Workloader) FinishPlanReplayerDump() error {
+	return nil
+}
diff --git a/cmd/go-tpc/main.go b/cmd/go-tpc/main.go
@@ -21,6 +21,7 @@ var (
 	dbName         string
 	host           string
 	port           int
+	statusPort     int
 	user           string
 	password       string
 	threads        int
@@ -102,6 +103,7 @@ func main() {
 	rootCmd.PersistentFlags().StringVarP(&user, "user", "U", "root", "Database user")
 	rootCmd.PersistentFlags().StringVarP(&password, "password", "p", "", "Database password")
 	rootCmd.PersistentFlags().IntVarP(&port, "port", "P", 4000, "Database port")
+	rootCmd.PersistentFlags().IntVarP(&statusPort, "statusPort", "S", 10080, "Database status port")
 	rootCmd.PersistentFlags().IntVarP(&threads, "threads", "T", 1, "Thread concurrency")
 	rootCmd.PersistentFlags().IntVarP(&acThreads, "acThreads", "t", 1, "OLAP client concurrency, only for CH-benCHmark")
 	rootCmd.PersistentFlags().StringVarP(&driver, "driver", "d", "", "Database driver: mysql")

diff --git a/cmd/go-tpc/misc.go b/cmd/go-tpc/misc.go
@@ -58,6 +58,21 @@ func execute(ctx context.Context, w workload.Workloader, action string, threads,
 		return w.Check(ctx, index)
 	}
 
+	enabledDumpPlanReplayer := w.IsPlanReplayerDumpEnabled()
+	if enabledDumpPlanReplayer {
+		err := w.PreparePlanReplayerDump()
+		if err != nil {
+			return err
+		}
+		defer func() {
+			err := w.FinishPlanReplayerDump()
+			if err != nil {
+				fmt.Printf("[%s] dump plan replayer failed, err%v\n",
+					time.Now().Format("2006-01-02 15:04:05"), err)
+			}
+		}()
+	}
+
 	for i := 0; i < count || count <= 0; i++ {
 		err := w.Run(ctx, index)
 

diff --git a/cmd/go-tpc/tpch.go b/cmd/go-tpc/tpch.go
@@ -6,9 +6,8 @@ import (
 	"os"
 	"strings"
 
-	"github.com/spf13/cobra"
-
 	"github.com/pingcap/go-tpc/tpch"
+	"github.com/spf13/cobra"
 )
 
 var tpchConfig tpch.Config
@@ -22,12 +21,14 @@ func executeTpch(action string) {
 		os.Exit(1)
 	}
 
+	tpchConfig.Host = host
+	tpchConfig.StatusPort = statusPort
+
 	tpchConfig.OutputStyle = outputStyle
 	tpchConfig.DBName = dbName
 	tpchConfig.PrepareThreads = threads
 	tpchConfig.QueryNames = strings.Split(tpchConfig.RawQueries, ",")
 	w := tpch.NewWorkloader(globalDB, &tpchConfig)
-
 	timeoutCtx, cancel := context.WithTimeout(globalCtx, totalTime)
 	defer cancel()
 
@@ -61,6 +62,21 @@ func registerTpch(root *cobra.Command) {
 		false,
 		"Check output data, only when the scale factor equals 1")
 
+	cmd.PersistentFlags().BoolVar(&tpchConfig.EnablePlanReplayer,
+		"use-plan-replayer",
+		false,
+		"Use Plan Replayer to dump stats and variables before running queries")
+
+	cmd.PersistentFlags().StringVar(&tpchConfig.PlanReplayerDir,
+		"plan-replayer-dir",
+		"",
+		"Dir of Plan Replayer file dumps")
+
+	cmd.PersistentFlags().StringVar(&tpchConfig.PlanReplayerFileName,
+		"plan-replayer-file",
+		"",
+		"Name of plan Replayer file dumps")
+
 	var cmdPrepare = &cobra.Command{
 		Use:   "prepare",
 		Short: "Prepare data for the workload",

diff --git a/pkg/workload/workload.go b/pkg/workload/workload.go
@@ -16,4 +16,8 @@ type Workloader interface {
 	Check(ctx context.Context, threadID int) error
 	OutputStats(ifSummaryReport bool)
 	DBName() string
+
+	IsPlanReplayerDumpEnabled() bool
+	PreparePlanReplayerDump() error
+	FinishPlanReplayerDump() error
 }
diff --git a/rawsql/workload.go b/rawsql/workload.go
@@ -171,3 +171,15 @@ func (w *Workloader) Cleanup(ctx context.Context, threadID int) error {
 func (w *Workloader) Check(ctx context.Context, threadID int) error {
 	panic("not implemented") // TODO: Implement
 }
+
+func (w Workloader) IsPlanReplayerDumpEnabled() bool {
+	return false
+}
+
+func (w Workloader) PreparePlanReplayerDump() error {
+	return nil
+}
+
+func (w Workloader) FinishPlanReplayerDump() error {
+	return nil
+}
diff --git a/tpcc/csv.go b/tpcc/csv.go
@@ -487,3 +487,15 @@ func (c *CSVWorkLoader) loadOrderLine(ctx context.Context, warehouse int,
 
 	return l.Flush(ctx)
 }
+
+func (c *CSVWorkLoader) IsPlanReplayerDumpEnabled() bool {
+	return false
+}
+
+func (c *CSVWorkLoader) PreparePlanReplayerDump() error {
+	return nil
+}
+
+func (c *CSVWorkLoader) FinishPlanReplayerDump() error {
+	return nil
+}
diff --git a/tpcc/workload.go b/tpcc/workload.go
@@ -472,3 +472,15 @@ func closeStmts(stmts map[string]*sql.Stmt) {
 		stmt.Close()
 	}
 }
+
+func (w *Workloader) IsPlanReplayerDumpEnabled() bool {
+	return false
+}
+
+func (w *Workloader) PreparePlanReplayerDump() error {
+	return nil
+}
+
+func (w *Workloader) FinishPlanReplayerDump() error {
+	return nil
+}
diff --git a/tpch/check.go b/tpch/check.go
@@ -51,7 +51,7 @@ var queryColPrecisions = map[string][]precision{
 	"q22": {num, cnt, sum},
 }
 
-func (w Workloader) scanQueryResult(queryName string, rows *sql.Rows) error {
+func (w *Workloader) scanQueryResult(queryName string, rows *sql.Rows) error {
 	var got [][]string
 
 	cols, err := rows.Columns()

diff --git a/tpch/ddl.go b/tpch/ddl.go
@@ -28,7 +28,7 @@ func (w *Workloader) createTableDDL(ctx context.Context, query string, tableName
 }
 
 // createTables creates tables schema.
-func (w Workloader) createTables(ctx context.Context) error {
+func (w *Workloader) createTables(ctx context.Context) error {
 	query := `
 CREATE TABLE IF NOT EXISTS nation (
     N_NATIONKEY BIGINT NOT NULL,