diff --git a/framework/certchecker.go b/framework/certchecker.go index 2846b2a..3134863 100644 --- a/framework/certchecker.go +++ b/framework/certchecker.go @@ -1,11 +1,13 @@ package framework import ( + "strings" + "github.com/PingCAP-QE/schrddl/pinolo/stage2" + "github.com/PingCAP-QE/schrddl/util" "github.com/pingcap/errors" "github.com/pingcap/tidb/pkg/util/logutil" "go.uber.org/zap" - "strings" ) type certChecker struct { @@ -42,7 +44,7 @@ func (n *certChecker) check(sql string, isReduce bool) (ok bool, err error) { rs1, err := n.c.execQueryForPlanEstCnt(querySQL) //println(fmt.Sprintf("%s;", querySQL)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -81,7 +83,7 @@ func (n *certChecker) check(sql string, isReduce bool) (ok bool, err error) { rs2, err := n.c.execQueryForPlanEstCnt(r.Sql) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { //logutil.BgLogger().Warn("ignore error", zap.String("query", r.Sql), zap.Error(err)) return false, nil } else { diff --git a/framework/mainloop.go b/framework/mainloop.go index d4fb047..4fed7fa 100644 --- a/framework/mainloop.go +++ b/framework/mainloop.go @@ -3,30 +3,19 @@ package framework import ( "context" "database/sql" - "encoding/json" "fmt" - crc322 "hash/crc32" - "math/rand" "os" - "os/exec" "path/filepath" - "strings" "sync" "sync/atomic" "time" - "unsafe" - "github.com/PingCAP-QE/schrddl/dump" - "github.com/PingCAP-QE/schrddl/norec" - "github.com/PingCAP-QE/schrddl/reduce" - "github.com/pingcap/tidb/pkg/parser" "github.com/pingcap/tidb/pkg/util/logutil" "go.uber.org/zap" "github.com/PingCAP-QE/schrddl/sqlgenerator" "github.com/juju/errors" "github.com/ngaut/log" - "github.com/pingcap/tidb/pkg/parser/model" _ "github.com/pingcap/tidb/pkg/types/parser_driver" ) @@ -57,10 +46,12 @@ type CaseConfig struct { MySQLCompatible bool TablesToCreate int TestTp DDLTestType - dbAddr string + DBAddr string + DBName string + TestPrepare bool } -var globalBugSeqNum int64 = 0 +var globalBugSeqNum atomic.Int64 var globalRunQueryCnt atomic.Int64 var globalSuccessQueryCnt atomic.Int64 @@ -82,43 +73,40 @@ func (c *DDLCase) String() string { func (c *DDLCase) statloop() { tick := time.NewTicker(10 * time.Second) - for { - select { - case <-tick.C: - subcaseStat := make([]string, len(c.cases)) - subcaseUseMvindex := make([]string, len(c.cases)) - subcaseUseCERT := make([]string, len(c.cases)) - subcaseUseAggIndexJoin := make([]string, len(c.cases)) - for _, c := range c.cases { - subcaseStat = append(subcaseStat, fmt.Sprintf("%d", len(c.queryPlanMap))) - subcaseUseMvindex = append(subcaseUseMvindex, fmt.Sprintf("%d", c.planUseMvIndex)) - subcaseUseCERT = append(subcaseUseCERT, fmt.Sprintf("%d", c.checkCERTCnt)) - subcaseUseAggIndexJoin = append(subcaseUseAggIndexJoin, fmt.Sprintf("%d", c.aggregationAsInnerSideOfIndexJoin)) - - //i := 0 - //for k, v := range c.queryPlanMap { - // logutil.BgLogger().Warn("sample query plan", zap.String("plan", k), zap.String("query", v)) - // i++ - // if i >= 10 { - // break - // } - //} - } - - logutil.BgLogger().Info("stat", zap.Int64("run query:", globalRunQueryCnt.Load()), - zap.Int64("success:", globalSuccessQueryCnt.Load()), - zap.Int64("fetch json row val:", sqlgenerator.GlobalFetchJsonRowValCnt.Load()), - zap.Strings("unique query plan", subcaseStat), - zap.Strings("use mv index", subcaseUseMvindex), - zap.Strings("use CERT", subcaseUseCERT), - zap.Strings("use agg index join", subcaseUseAggIndexJoin), - ) - } + for range tick.C { + subcaseStat := make([]string, len(c.cases)) + subcaseUseMvindex := make([]string, len(c.cases)) + subcaseUseCERT := make([]string, len(c.cases)) + subcaseUseAggIndexJoin := make([]string, len(c.cases)) + for _, c := range c.cases { + subcaseStat = append(subcaseStat, fmt.Sprintf("%d", len(c.queryPlanMap))) + subcaseUseMvindex = append(subcaseUseMvindex, fmt.Sprintf("%d", c.planUseMvIndex)) + subcaseUseCERT = append(subcaseUseCERT, fmt.Sprintf("%d", c.checkCERTCnt)) + subcaseUseAggIndexJoin = append(subcaseUseAggIndexJoin, fmt.Sprintf("%d", c.aggregationAsInnerSideOfIndexJoin)) + + //i := 0 + //for k, v := range c.queryPlanMap { + // logutil.BgLogger().Warn("sample query plan", zap.String("plan", k), zap.String("query", v)) + // i++ + // if i >= 10 { + // break + // } + //} + } + + logutil.BgLogger().Info("stat", zap.Int64("run query:", globalRunQueryCnt.Load()), + zap.Int64("success:", globalSuccessQueryCnt.Load()), + zap.Int64("fetch json row val:", sqlgenerator.GlobalFetchJsonRowValCnt.Load()), + zap.Strings("unique query plan", subcaseStat), + zap.Strings("use mv index", subcaseUseMvindex), + zap.Strings("use CERT", subcaseUseCERT), + zap.Strings("use agg index join", subcaseUseAggIndexJoin), + ) } } // Execute executes each goroutine (i.e. `testCase`) concurrently. -func (c *DDLCase) Execute(ctx context.Context, dbss [][]*sql.DB) error { +func (c *DDLCase) Execute(ctx context.Context) error { log.Infof("[%s] start to test...", c) go func() { c.statloop() @@ -132,14 +120,20 @@ func (c *DDLCase) Execute(ctx context.Context, dbss [][]*sql.DB) error { go func(i int) { defer wg.Done() for { - err := c.cases[i].execute(ctx) + var err error + switch c.cases[i].caseType { + case CaseTypeNormal: + err = c.cases[i].execute(ctx) + case CaseTypePlanCache: + err = c.cases[i].testPlanCache(ctx) + default: + log.Fatalf("Unknown case type %d", c.cases[i].caseType) + } + if err != nil { - for _, dbs := range dbss { - for _, db := range dbs { - disableTiKVGC(db) - } + for _, tc := range c.cases { + tc.DisableKVGC() } - // os.Exit(-1) log.Fatalf("[error] [instance %d] ERROR: %s", i, errors.ErrorStack(err)) } select { @@ -158,13 +152,8 @@ func (c *DDLCase) Execute(ctx context.Context, dbss [][]*sql.DB) error { // Initialize initializes all supported charsets, collates and each concurrent // goroutine (i.e. `testCase`). func (c *DDLCase) Initialize(ctx context.Context, dbss [][]*sql.DB, initDB string) error { - charsets, charsetsCollates, err := getAllCharsetAndCollates(dbss[0][0]) - if err != nil { - return errors.Trace(err) - } for i := 0; i < c.cfg.Concurrency; i++ { - c.cases[i].initDB = initDB - c.cases[i].setCharsetsAndCollates(charsets, charsetsCollates) + c.cases[i].dbname = initDB err := c.cases[i].initialize(dbss[i]) if err != nil { return errors.Trace(err) @@ -173,34 +162,6 @@ func (c *DDLCase) Initialize(ctx context.Context, dbss [][]*sql.DB, initDB strin return nil } -// getAllCharsetAndCollates returns all allowable charsets and collates by executing a -// simple SQL query: `show charset`. -func getAllCharsetAndCollates(db *sql.DB) ([]string, map[string][]string, error) { - sql := "show charset" - rows, err := db.Query(sql) - if err != nil { - return nil, nil, err - } - defer rows.Close() - charsets := make([]string, 0) - charsetsCollates := make(map[string][]string) - for rows.Next() { - var collate, charset, description string - var maxLen int - err := rows.Scan(&charset, &description, &collate, &maxLen) - if err != nil { - return nil, nil, err - } - if collates, ok := charsetsCollates[charset]; ok { - charsetsCollates[charset] = append(collates, collate) - } else { - charsets = append(charsets, charset) - charsetsCollates[charset] = []string{collate} - } - } - return charsets, charsetsCollates, nil -} - // NewDDLCase returns a DDLCase, which contains specified `testCase`s. func NewDDLCase(cfg *CaseConfig) *DDLCase { cases := make([]*testCase, cfg.Concurrency) @@ -213,9 +174,16 @@ func NewDDLCase(cfg *CaseConfig) *DDLCase { if err != nil { log.Fatal(err) } + + caseType := CaseTypeNormal + if cfg.TestPrepare { + caseType = CaseTypePlanCache + } + for i := 0; i < cfg.Concurrency; i++ { cases[i] = &testCase{ cfg: cfg, + caseType: caseType, tables: make(map[string]*ddlTestTable), schemas: make(map[string]*ddlTestSchema), views: make(map[string]*ddlTestView), @@ -226,698 +194,17 @@ func NewDDLCase(cfg *CaseConfig) *DDLCase { queryPlanMap: make(map[string]string), } } - b := &DDLCase{ + + return &DDLCase{ cfg: cfg, cases: cases, } - return b } const ( - ddlTestValueNull string = "NULL" - ddlTestValueInvalid int32 = -99 -) - -type DMLKind int - -const ( - dmlInsert DMLKind = iota - dmlUpdate - dmlDelete - dmlSelect + ddlTestValueNull string = "NULL" ) -type dmlJobArg unsafe.Pointer - -type dmlJobTask struct { - k DMLKind - tblInfo *ddlTestTable - sql string - assigns []*ddlTestColumnDescriptor - whereColumns []*ddlTestColumnDescriptor - err error -} - -// initialize generates possible DDL and DML operations for one `testCase`. -// Different `testCase`s will be run in parallel according to the concurrent configuration. -func (c *testCase) initialize(dbs []*sql.DB) error { - //var err error - c.dbs = dbs - c.tidbParser = parser.New() - return nil -} - -// setCharsetsAndCollates sets the allowable character sets and associated collates for this testCase. -func (c *testCase) setCharsetsAndCollates(charsets []string, charsetsCollates map[string][]string) { - c.charsets = charsets - c.charsetsCollates = charsetsCollates -} - -func (c *testCase) checkError(err error) error { - if err != nil { - if c.cfg.MySQLCompatible { - if strings.Contains(err.Error(), "Duplicate entry") { - return nil - } - } - return errors.Trace(err) - } - return nil -} - -func (c *testCase) execSQL(sql string) error { - _, err := c.dbs[0].Exec(sql) - if err != nil && dmlIgnoreError(err) || ddlIgnoreError(err) { - return nil - } - if strings.Contains(err.Error(), "plan not match") { - _, err = c.dbs[0].Exec(sql) - return err - } - return errors.Trace(err) -} - -func (c *testCase) execQueryForPlanEstCnt(sql string) (float64, error) { - sql = "explain format='brief' " + sql - rows, err := c.dbs[0].Query(sql) - if err != nil { - return 0, err - } - defer func() { - rows.Close() - }() - var id string - var estRows float64 - var task string - var accessObject string - var operatorInfo string - rows.Next() - err = rows.Scan(&id, &estRows, &task, &accessObject, &operatorInfo) - return estRows, err -} - -func (c *testCase) execQueryForCnt(sql string) (int, error) { - rows, err := c.dbs[0].Query(sql) - if err != nil { - return 0, err - } - defer func() { - rows.Close() - }() - rs := 0 - for rows.Next() { - rs++ - } - if rows.Err() != nil { - return 0, rows.Err() - } - return rs, nil -} - -func (c *testCase) execQueryAndCheckEmpty(sql string) bool { - rows, err := c.dbs[0].Query(sql) - if err != nil { - logutil.BgLogger().Error("unexpected error", zap.Error(err)) - return false - } - defer func() { - rows.Close() - }() - - result := !rows.Next() - - cnt2 := 0 - if !result { - cols, _ := rows.Columns() - rawResult := make([][]byte, len(cols)) - dest := make([]interface{}, len(cols)) - for i := range rawResult { - dest[i] = &rawResult[i] - } - - err := rows.Scan(dest...) - if err != nil { - panic(err) - } - - for rows.Next() { - cnt2++ - } - - for _, r := range rawResult { - c.outputWriter.WriteString(fmt.Sprintf("%s\n", string(r))) - } - - cnt, _ := c.execQueryForCnt(sql) - - rs, _ := c.execQuery(sql) - - c.outputWriter.WriteString(fmt.Sprintf("sql: %s, cnt: %d, cnt2: %d, cnt3: %d \n", sql, cnt, cnt2+1, len(rs))) - if rows.Err() != nil { - c.outputWriter.WriteString(fmt.Sprintf("err %s \n", rows.Err().Error())) - } - } - return result -} - -func (c *testCase) execQueryForCRC32(sql string) (map[uint32]struct{}, error) { - rows, err := c.dbs[0].Query(sql) - if err != nil { - return nil, err - } - defer func() { - rows.Close() - }() - - // Read all rows. - crc32s := make(map[uint32]struct{}, 0) - - for rows.Next() { - cols, err1 := rows.Columns() - if err1 != nil { - return nil, err - } - - //log.Infof("[ddl] [instance %d] rows.Columns():%v, len(cols):%v", c.caseIndex, cols, len(cols)) - - // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go - rawResult := make([][]byte, len(cols)) - result := make([]string, len(cols)) - dest := make([]interface{}, len(cols)) - ct, _ := rows.ColumnTypes() - for i := range rawResult { - dest[i] = &rawResult[i] - } - - err1 = rows.Scan(dest...) - if err1 != nil { - return nil, err - } - - for i, raw := range rawResult { - if raw == nil { - result[i] = ddlTestValueNull - } else { - //logutil.BgLogger().Warn("type to debug", zap.String("type", ct[i].DatabaseTypeName())) - if strings.EqualFold(ct[i].DatabaseTypeName(), "double") { - result[i] = fmt.Sprintf("'%s'", RoundToOneDecimals(string(raw))) - } else { - result[i] = fmt.Sprintf("'%s'", strings.ToLower(string(raw))) - } - //if typeNeedQuota(metaCols[i].k) { - // result[i] = fmt.Sprintf("'%s'", string(raw)) - //} - //result[i] = string(raw) - } - } - - crc32 := crc322.NewIEEE() - for _, r := range result { - crc32.Write([]byte(r)) - } - - sum := crc32.Sum32() - //logutil.BgLogger().Info("crc32", zap.Uint32("crc32", sum), zap.String("data", fmt.Sprintf("%v", result))) - - crc32s[sum] = struct{}{} - } - if rows.Err() != nil { - return nil, rows.Err() - } - - return crc32s, nil -} - -func (c *testCase) execQuery(sql string) ([][]string, error) { - rows, err := c.dbs[0].Query(sql) - if err != nil { - return nil, err - } - defer func() { - rows.Close() - }() - - // Read all rows. - var actualRows [][]string - for rows.Next() { - cols, err1 := rows.Columns() - if err1 != nil { - return nil, err - } - - //log.Infof("[ddl] [instance %d] rows.Columns():%v, len(cols):%v", c.caseIndex, cols, len(cols)) - - // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go - rawResult := make([][]byte, len(cols)) - result := make([]string, len(cols)) - dest := make([]interface{}, len(cols)) - for i := range rawResult { - dest[i] = &rawResult[i] - } - - err1 = rows.Scan(dest...) - if err1 != nil { - return nil, err - } - - for i, raw := range rawResult { - if raw == nil { - result[i] = ddlTestValueNull - } else { - result[i] = fmt.Sprintf("'%s'", string(raw)) - //if typeNeedQuota(metaCols[i].k) { - // result[i] = fmt.Sprintf("'%s'", string(raw)) - //} - //result[i] = string(raw) - } - } - - actualRows = append(actualRows, result) - } - if rows.Err() != nil { - return nil, rows.Err() - } - - return actualRows, nil -} - -func mapLess(m1, m2 map[uint32]struct{}) bool { - for k := range m2 { - if _, ok := m1[k]; !ok { - return true - } - } - return false -} - -func mapMore(m1, m2 map[uint32]struct{}) bool { - for k := range m1 { - if _, ok := m2[k]; !ok { - return true - } - } - return false -} - type checker interface { check(sql string, isReduce bool) (bool, error) } - -// execute iterates over two list of operations concurrently, one is -// ddl operations, one is dml operations. -// When one list completes, it starts over from the beginning again. -// When both of them ONCE complete, it exits. -func (c *testCase) execute(ctx context.Context) error { - state := sqlgenerator.NewState() - state.SetWeight(sqlgenerator.RenameColumn, 0) - state.SetWeight(sqlgenerator.WindowFunction, 0) - state.SetWeight(sqlgenerator.WindowClause, 0) - state.SetWeight(sqlgenerator.WindowFunctionOverW, 0) - state.SetWeight(sqlgenerator.WhereClause, 1) - //state.SetWeight(sqlgenerator.Limit, 0) - state.SetWeight(sqlgenerator.UnionSelect, 0) - //state.SetWeight(sqlgenerator.PartitionDefinitionHash, 1000) - state.SetWeight(sqlgenerator.PartitionDefinitionKey, 0) - //state.SetWeight(sqlgenerator.PartitionDefinitionRange, 1000) - - //state.SetWeight(sqlgenerator.AggSelect, 0) - - // Sub query is hard for NoREC - //state.SetWeight(sqlgenerator.SubSelect, 0) - - if !EnableApproximateQuerySynthesis { - state.SetWeight(sqlgenerator.ScalarSubQuery, 0) - state.SetWeight(sqlgenerator.SubSelect, 0) - } - - // bug - state.SetWeight(sqlgenerator.ColumnDefinitionTypesEnum, 0) - state.SetWeight(sqlgenerator.ColumnDefinitionTypesSet, 0) - state.SetWeight(sqlgenerator.ColumnDefinitionTypesYear, 0) - state.SetWeight(sqlgenerator.ColumnDefinitionTypesBit, 0) - - //state.Hook().Append(sqlgenerator.NewFnHookDebug()) - - //state.SetWeight(sqlgenerator.ColumnDefinitionTypesJSON, 0) - //state.SetWeight(sqlgenerator.JSONPredicate, 0) - - prepareStmtCnt := 50 - for i := 0; i < prepareStmtCnt; i++ { - startSQL, err := sqlgenerator.Start.Eval(state) - if err != nil { - return err - } - err = c.execSQL(startSQL) - //println(fmt.Sprintf("%s;", startSQL)) - if err != nil { - return err - } - } - - err := c.execSQL("set @@max_execution_time=800000") - if err != nil { - return err - } - err = c.execSQL("set @@group_concat_max_len=10240000") - if err != nil { - return err - } - - tableMetas := make([]*model.TableInfo, 0) - for i := 0; i < len(state.Tables); i++ { - addr := c.cfg.dbAddr - stateAddr := strings.Replace(addr, "4000", "10080", -1) - path := fmt.Sprintf("%s/schema/%s/%s", stateAddr, c.initDB, state.Tables[i].Name) - rawMeta, err := exec.Command("curl", path).Output() - if err != nil { - log.Infof("curl error %s", err.Error()) - continue - } - var meta model.TableInfo - err = json.Unmarshal(rawMeta, &meta) - if err != nil { - logutil.BgLogger().Warn("unmarshal error", zap.Error(err), zap.String("table name", state.Tables[i].Name)) - state.Tables = append(state.Tables[:i], state.Tables[i+1:]...) - i-- - continue - } - tableMetas = append(tableMetas, &meta) - } - for _, table := range tableMetas { - log.Infof("table %s", table.Name.O) - } - log.Infof("tableMetas %d", len(tableMetas)) - state.SetTableMeta(tableMetas) - sqlgenerator.PrepareIndexJoinColumns(state) - defer sqlgenerator.RemoveIndexJoinColumns(state) - - cnt := 0 - - for { - cnt++ - if cnt%10000 == 0 { - err := c.executeAdminCheck() - if err != nil { - return errors.Trace(err) - } - err = c.readDataFromTiDB() - if err != nil { - if !dmlIgnoreError(err) { - return errors.Trace(err) - } - } - } - if cnt%10000 == 0 && rand.Intn(2) == 0 { - break - } - - // NoREC - rewriter := &norec.NoRecRewriter{} - var sb strings.Builder - - doDML := rand.Intn(2) == 0 - if doDML { - dmlSQL, err := sqlgenerator.DMLStmt.Eval(state) - //println(fmt.Sprintf("%s;", dmlSQL)) - if err != nil { - return err - } - if rand.Intn(100) == 0 { - dmlSQL, err = sqlgenerator.SetTiFlashReplica.Eval(state) - if err != nil { - return err - } - } else if rand.Intn(15) == 0 { - dmlSQL, err = sqlgenerator.SetVariable.Eval(state) - if err != nil { - return err - } - } - if rand.Intn(100) == 0 { - dmlSQL, err = sqlgenerator.AnalyzeTable.Eval(state) - if err != nil { - return err - } - } - err = c.execSQL(dmlSQL) - if err != nil { - return err - } - } else { - var ck checker - if EnableApproximateQuerySynthesis { - ck = &pinoloChecker{c: c} - } else if EnableCERT { - ck = &certChecker{c: c} - } else if EnableTLP { - ck = &tlpChecker{c: c} - } else { - ck = &norecChecker{c: c, rewriter: *rewriter, sb: sb} - } - - fn := sqlgenerator.QueryOrCTE - if EnableCERT { - fn = sqlgenerator.Query - } - querySQL, err := fn.Eval(state) - if err != nil { - return errors.Trace(err) - } - - //println(fmt.Sprintf("%s;", querySQL)) - - found, err := ck.check(querySQL, false) - if err != nil { - if !dmlIgnoreError(err) { - if strings.Contains(err.Error(), "plan not match") { - _, err = c.dbs[0].Exec(querySQL) - if err == nil { - continue - } - } - return errors.Trace(err) - } else { - continue - } - } - if found { - reduceSQL := reduce.ReduceSQL(ck.check, querySQL) - if EnableTLP { - _, err = c.outputWriter.WriteString( - fmt.Sprintf("old count of predicate reduce SQL:%d, count of non-predicate reduce SQL:%d,\ncount of null-predicate reduce SQL:%d, count of all-predicate reduce SQL:%d\n"+ - "query of orginal SQL: %s\n"+ - "predicate reduce query: %s\n"+ - "negative predicate reduce query: %s\n"+ - "isnull predicate reduce query: %s\n"+ - "all predicate reduce query: %s\n"+ - "\n\n", - c.cntOfP, c.cntOfN, c.cntOfNull, c.cntOfAll, c.originalSQL, c.reduceSQL, c.nQuery, c.nullQuery, c.allQuery)) - } else if EnableCERT { - _, err = c.outputWriter.WriteString( - fmt.Sprintf("old count of orginal SQL:%f, new count of orginal SQL:%f,\nold count of reduce SQL:%f, new count of reduce SQL:%f,\nold query of orginal SQL: %s\nnew query of reduce SQL: %s\nreduce query: %s\n\n\n", - c.oldEstCntOriginal, c.newEstCntOriginal, c.oldEstCntReduce, c.newEstCntReduce, c.originalSQL, c.reduceChangedSQL, c.reduceSQL)) - } else { - _, err = c.outputWriter.WriteString( - fmt.Sprintf("old count of orginal SQL:%d, new count of orginal SQL:%d,\nold count of reduce SQL:%d, new count of reduce SQL:%d,\nold query of orginal SQL: %s\nnew query of reduce SQL: %s\nreduce query: %s\n\n\n", - c.cntOfOldOriginal, c.cntOfNewOriginal, c.cntOfOldReduce, c.cntOfNewReduce, c.originalSQL, c.reduceChangedSQL, c.reduceSQL)) - } - globalBugSeqNum++ - num := globalBugSeqNum - - // Dump data. - tblNames, err := dump.ExtraFromSQL(reduceSQL) - if err != nil { - return err - } - pwd := os.Getenv("PWD") - if GlobalOutPut != "" { - pwd = filepath.Dir(GlobalOutPut) - } - err = dump.DumpToFile("test", tblNames, fmt.Sprintf("local://%s/bug-%s-%d", pwd, time.Now().Format("2006-01-02-15-04-05"), num), c.cfg.dbAddr) - if err != nil { - return err - } - TestFail = true - break - } - } - } - - log.Infof("[ddl] [instance %d] Round completed", c.caseIndex) - log.Infof("[ddl] [instance %d] Executing post round operations...", c.caseIndex) - - if !c.cfg.MySQLCompatible { - err := c.executeAdminCheck() - if err != nil { - return errors.Trace(err) - } - err = c.readDataFromTiDB() - if err != nil { - if !dmlIgnoreError(err) { - return errors.Trace(err) - } - } - } - - return nil -} - -func (c *testCase) readDataFromTiDB() error { - if len(c.tables) == 0 { - return nil - } - - sql := "select * from " - for _, table := range c.tables { - readSql := sql + fmt.Sprintf("`%s`", table.name) - dbIdx := rand.Intn(len(c.dbs)) - db := c.dbs[dbIdx] - rows, err := db.Query(readSql) - if err != nil { - return err - } - defer func() { - rows.Close() - }() - metaCols := make([]*ddlTestColumn, 0) - for ite := table.columns.Iterator(); ite.Next(); { - metaCols = append(metaCols, ite.Value().(*ddlTestColumn)) - } - // Read all rows. - var actualRows [][]string - for rows.Next() { - cols, err1 := rows.Columns() - if err1 != nil { - return errors.Trace(err) - } - - log.Infof("[ddl] [instance %d] rows.Columns():%v, len(cols):%v", c.caseIndex, cols, len(cols)) - - // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go - rawResult := make([][]byte, len(cols)) - result := make([]string, len(cols)) - dest := make([]interface{}, len(cols)) - for i := range rawResult { - dest[i] = &rawResult[i] - } - - err1 = rows.Scan(dest...) - if err1 != nil { - return errors.Trace(err) - } - - for i, raw := range rawResult { - if raw == nil { - result[i] = ddlTestValueNull - } else { - result[i] = fmt.Sprintf("'%s'", string(raw)) - //if typeNeedQuota(metaCols[i].k) { - // result[i] = fmt.Sprintf("'%s'", string(raw)) - //} - //result[i] = string(raw) - } - } - - actualRows = append(actualRows, result) - } - c.tableMap[table.name].Values = actualRows - } - - return nil -} - -func readData(ctx context.Context, conn *sql.Conn, query string) ([][]string, error) { - rows, err := conn.QueryContext(ctx, query) - if err != nil { - return nil, errors.Annotatef(err, "Error when executing SQL: %s\n", query) - } - defer func() { - rows.Close() - }() - //metaCols := make([]*ddlTestColumn, 0) - //for ite := table.columns.Iterator(); ite.Next(); { - // metaCols = append(metaCols, ite.Value().(*ddlTestColumn)) - //} - // Read all rows. - var actualRows [][]string - for rows.Next() { - cols, err1 := rows.Columns() - if err1 != nil { - return nil, errors.Trace(err) - } - - // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go - rawResult := make([][]byte, len(cols)) - result := make([]string, len(cols)) - dest := make([]interface{}, len(cols)) - for i := range rawResult { - dest[i] = &rawResult[i] - } - - err1 = rows.Scan(dest...) - if err1 != nil { - return nil, errors.Trace(err) - } - - for i, raw := range rawResult { - if raw == nil { - result[i] = ddlTestValueNull - } else { - result[i] = fmt.Sprintf("'%s'", string(raw)) - //if typeNeedQuota(metaCols[i].k) { - // result[i] = fmt.Sprintf("'%s'", string(raw)) - //} - //result[i] = string(raw) - } - } - - actualRows = append(actualRows, result) - } - return actualRows, err -} - -func trimValue(tp int, val []byte) string { - // a='{"DnOJQOlx":52,"ZmvzPtdm":82}' - // eg: set a={"a":"b","b":"c"} - // get a={"a": "b", "b": "c"} , so have to remove the space - if tp == KindJSON { - for i := 1; i < len(val)-2; i++ { - if val[i-1] == '"' && val[i] == ':' && val[i+1] == ' ' { - val = append(val[:i+1], val[i+2:]...) - } - if val[i-1] == ',' && val[i] == ' ' && val[i+1] == '"' { - val = append(val[:i], val[i+1:]...) - } - } - } - return string(val) -} - -func (c *testCase) executeAdminCheck() error { - if len(c.tables) == 0 { - return nil - } - - // build SQL - sql := "ADMIN CHECK TABLE " - i := 0 - for _, table := range c.tables { - if i > 0 { - sql += ", " - } - sql += fmt.Sprintf("`%s`", table.name) - i++ - } - dbIdx := rand.Intn(len(c.dbs)) - db := c.dbs[dbIdx] - // execute - log.Infof("[ddl] [instance %d] %s", c.caseIndex, sql) - _, err := db.Exec(sql) - if err != nil { - if dmlIgnoreError(err) { - return nil - } - return errors.Annotatef(err, "Error when executing SQL: %s", sql) - } - return nil -} diff --git a/framework/meta.go b/framework/meta.go index 1073af9..e1bead5 100644 --- a/framework/meta.go +++ b/framework/meta.go @@ -2,80 +2,19 @@ package framework import ( "bytes" - "database/sql" "encoding/json" "fmt" "math/rand" - "os" "sort" "strings" "sync" "sync/atomic" "time" - "github.com/PingCAP-QE/schrddl/sqlgenerator" "github.com/emirpasic/gods/lists/arraylist" - "github.com/pingcap/tidb/pkg/parser" "github.com/twinj/uuid" ) -type testCase struct { - cfg *CaseConfig - initDB string - dbs []*sql.DB - caseIndex int - tables map[string]*ddlTestTable - schemas map[string]*ddlTestSchema - views map[string]*ddlTestView - tablesLock sync.RWMutex - schemasLock sync.Mutex - stop int32 - lastDDLID int - charsets []string - charsetsCollates map[string][]string - - tableMap map[string]*sqlgenerator.Table - outputWriter *os.File - - tidbParser *parser.Parser - - // stat info - queryPlanMap map[string]string - originalSQL string - reduceChangedSQL string - reduceSQL string - aggregationAsInnerSideOfIndexJoin int - planUseMvIndex int - cntOfOldOriginal int - cntOfNewOriginal int - cntOfOldReduce int - cntOfNewReduce int - - // cert - oldEstCntOriginal float64 - newEstCntOriginal float64 - oldEstCntReduce float64 - newEstCntReduce float64 - checkCERTCnt int - - // tlp - cntOfP int - cntOfN int - cntOfNull int - cntOfAll int - nQuery string - nullQuery string - allQuery string -} - -func (c *testCase) stopTest() { - atomic.StoreInt32(&c.stop, 1) -} - -func (c *testCase) isStop() bool { - return atomic.LoadInt32(&c.stop) == 1 -} - // schema type, this might be modified to support other operations, but it's not clear // currently for me. type ddlTestSchema struct { diff --git a/framework/norecchecker.go b/framework/norecchecker.go index 4de5dc7..7af5d76 100644 --- a/framework/norecchecker.go +++ b/framework/norecchecker.go @@ -6,6 +6,7 @@ import ( "strings" "github.com/PingCAP-QE/schrddl/norec" + "github.com/PingCAP-QE/schrddl/util" "github.com/pingcap/errors" "github.com/pingcap/tidb/pkg/parser/format" "github.com/pingcap/tidb/pkg/util/logutil" @@ -44,7 +45,7 @@ func (n *norecChecker) check(sql string, isReduce bool) (ok bool, err error) { rs1, err := n.c.execQueryForCnt(querySQL) //println(fmt.Sprintf("%s;", querySQL)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -92,7 +93,7 @@ func (n *norecChecker) check(sql string, isReduce bool) (ok bool, err error) { rs2, err := n.c.execQuery(newQuery) //println(fmt.Sprintf("%s;", newQuery)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) diff --git a/framework/pinolochecker.go b/framework/pinolochecker.go index 9b241a9..f040d8a 100644 --- a/framework/pinolochecker.go +++ b/framework/pinolochecker.go @@ -3,11 +3,30 @@ package framework import ( "github.com/PingCAP-QE/schrddl/pinolo" "github.com/PingCAP-QE/schrddl/pinolo/stage2" + "github.com/PingCAP-QE/schrddl/util" "github.com/pingcap/errors" "github.com/pingcap/tidb/pkg/util/logutil" "go.uber.org/zap" ) +func mapLess(m1, m2 map[uint32]struct{}) bool { + for k := range m2 { + if _, ok := m1[k]; !ok { + return true + } + } + return false +} + +func mapMore(m1, m2 map[uint32]struct{}) bool { + for k := range m1 { + if _, ok := m2[k]; !ok { + return true + } + } + return false +} + type pinoloChecker struct { c *testCase } @@ -41,7 +60,7 @@ func (n *pinoloChecker) check(sql string, isReduce bool) (ok bool, err error) { } //println(fmt.Sprintf("%s;", querySQL)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -67,7 +86,7 @@ func (n *pinoloChecker) check(sql string, isReduce bool) (ok bool, err error) { if r.Err == nil { rs2, err := n.c.execQueryForCnt(r.Sql) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { //logutil.BgLogger().Warn("ignore error", zap.String("query", r.Sql), zap.Error(err)) return false, nil } else { diff --git a/framework/run.go b/framework/run.go index 2638b51..c4d4e33 100644 --- a/framework/run.go +++ b/framework/run.go @@ -5,8 +5,6 @@ import ( "fmt" "os" "os/signal" - "regexp" - "strings" "syscall" "time" @@ -16,7 +14,6 @@ import ( var GlobalOutPut = "" var EnableTransactionTest = false -var RCIsolation = false var Prepare = false var CheckDDLExtraTimeout = 0 * time.Second var EnableApproximateQuerySynthesis = false @@ -35,7 +32,41 @@ func OpenDB(dsn string, maxIdleConns int) (*sql.DB, error) { return db, nil } -func Run(dbAddr string, dbName string, concurrency int, tablesToCreate int, mysqlCompatible bool, testTp DDLTestType, testTime time.Duration) { +func createDBs(dbDSN string) []*sql.DB { + dbs := make([]*sql.DB, 0, 2) + // Parallel send DDL request need more connection to send DDL request concurrently + db0, err := OpenDB(dbDSN, 20) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + db1, err := OpenDB(dbDSN, 1) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + dbs = append(dbs, db0) + dbs = append(dbs, db1) + return dbs +} + +func createDBsForPrepare(dbAddr, dbName string) []*sql.DB { + dbs := make([]*sql.DB, 0, 2) + db0, err := OpenDB(fmt.Sprintf("root:@tcp(%s)/%s", dbAddr, dbName), 20) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + dbcache, err := OpenDB(fmt.Sprintf("root:@tcp(%s)/%s", dbAddr, "testcache"), 20) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + dbcache.SetMaxOpenConns(32) + dbcache.SetMaxIdleConns(32) + dbcache.SetConnMaxLifetime(time.Hour) + dbs = append(dbs, db0) + dbs = append(dbs, dbcache) + return dbs +} + +func Run(cfg CaseConfig, testTime time.Duration) { wrapCtx := context.WithCancel if testTime > 0 { wrapCtx = func(ctx context.Context) (context.Context, context.CancelFunc) { @@ -43,28 +74,6 @@ func Run(dbAddr string, dbName string, concurrency int, tablesToCreate int, mysq } } ctx, cancel := wrapCtx(context.Background()) - dbss := make([][]*sql.DB, 0, concurrency) - dbDSN := fmt.Sprintf("root:@tcp(%s)/%s", dbAddr, dbName) - for i := 0; i < concurrency; i++ { - dbs := make([]*sql.DB, 0, 2) - // Parallel send DDL request need more connection to send DDL request concurrently - db0, err := OpenDB(dbDSN, 20) - if err != nil { - log.Fatalf("[ddl] create db client error %v", err) - } - db1, err := OpenDB(dbDSN, 1) - if err != nil { - log.Fatalf("[ddl] create db client error %v", err) - } - dbs = append(dbs, db0) - dbs = append(dbs, db1) - dbss = append(dbss, dbs) - } - globalDbs, err := OpenDB(dbDSN, 20) - if err != nil { - log.Fatalf("[ddl] create db client error %v", err) - } - sc := make(chan os.Signal, 1) signal.Notify(sc, syscall.SIGHUP, @@ -78,260 +87,22 @@ func Run(dbAddr string, dbName string, concurrency int, tablesToCreate int, mysq os.Exit(0) }() - cfg := CaseConfig{ - Concurrency: concurrency, - TablesToCreate: tablesToCreate, - MySQLCompatible: mysqlCompatible, - TestTp: testTp, - dbAddr: dbAddr, + dbss := make([][]*sql.DB, 0, cfg.Concurrency) + dbDSN := fmt.Sprintf("root:@tcp(%s)/%s", cfg.DBAddr, cfg.DBName) + for i := 0; i < cfg.Concurrency; i++ { + // Currently, we only use one testCase for plan cache test. + if cfg.TestPrepare { + dbss = append(dbss, createDBsForPrepare(cfg.DBAddr, cfg.DBName)) + } else { + dbss = append(dbss, createDBs(dbDSN)) + } } + ddl := NewDDLCase(&cfg) - if RCIsolation { - globalDbs.Exec("set global transaction_isolation='read-committed'") - } - _, err = globalDbs.Exec("set global tidb_enable_global_index=true") - if err != nil { - log.Fatalf("[ddl] set global tidb_enable_global_index=true error %v", err) - } - if err := ddl.Initialize(ctx, dbss, dbName); err != nil { + if err := ddl.Initialize(ctx, dbss, cfg.DBName); err != nil { log.Fatalf("[ddl] initialze error %v", err) } - if err := ddl.Execute(ctx, dbss); err != nil { + if err := ddl.Execute(ctx); err != nil { log.Fatalf("[ddl] execute error %v", err) } } - -var dmlIgnoreList = []string{ - "Table has no partition", - "can't have a default value", - "Invalid JSON bytes", - "Invalid JSON data provided to function", - "Invalid JSON value for CAST", - "Invalid JSON text", - "Data too long", - "character string", - - // bug - "slice bounds out of range", - "index out of range", - "writing inconsistent data in table", - "should ensure all columns have the same length", - "expected integer", - "invalid memory address or nil pointer dereference", - "encoding failed", - "invalid input value", - "region not found for key", - "Unsupported expression type MysqlBit", - "Unexpected missing column", - "strconv.Atoi", - //"other error for mpp stream", - - "Can't find a proper physical plan for this query", - "Your query has been cancelled due to exceeding the allowed memory limit", - "Cant peek from empty bytes", - - "maximum statement execution time exceeded", - "please skip this plan", - - "Subquery returns more", - "Some rows were cut by", - "invalid data type: Illegal Json text", - "Data Truncated", - - // unknown - "context canceled", - "cannot be pushed down", -} - -var ddlIgnoreList = []string{ - "already exists", - "A PRIMARY must include all columns", - "has an expression index dependency and cannot", - "Multiple definition of same constant", - "VALUES LESS THAN value must be strictly increasing for each partition", - "please split table instead", - "should less than the upper value", - "A primary key index cannot be invisible", - "Unsupported modify change collate", - "Failed to split region ranges: the region size is too small", - "Can't find dropped/truncated table", - "Can't find localTemporary/dropped/truncated", - "can't be flashback repeatedly", - "Invalid gbk character string", - "secondary index", - "cannot be used in key specification", - "Adding clustered primary key", - "Invalid use of NULL value", - "can not get 'tikv_gc_safe_point'", - "maximum statement execution time exceeded", - "Illegal mix of collations for operation", - "Unsupported Global Index", - "settings for table contains gbk charset", -} - -func dmlIgnoreError(err error) bool { - if err == nil { - return true - } - errStr := err.Error() - for _, ignore := range dmlIgnoreList { - if strings.Contains(errStr, ignore) { - return true - } - } - if strings.Contains(errStr, "Information schema is changed") && !RCIsolation { - return true - } - if strings.Contains(errStr, "try again later") { - return true - } - // Sometimes, there might be duplicated entry error caused by concurrent. - // So we ignore here. - if strings.Contains(errStr, "Duplicate entry") { - return true - } - // Sometimes, a insert to a table might generate an error caused by exceeding maximum auto increment id, - // we ignore this error here. - if strings.Contains(errStr, "Failed to read auto-increment value from storage engine") { - return true - } - if strings.Contains(errStr, "invalid connection") { - return true - } - if strings.Contains(errStr, "doesn't exist") || - strings.Contains(errStr, "column is deleted") || strings.Contains(errStr, "Can't find column") || - strings.Contains(errStr, "converting driver.Value type") || strings.Contains(errStr, "column specified twice") || - strings.Contains(errStr, "Out of range value for column") || strings.Contains(errStr, "Unknown column") || - strings.Contains(errStr, "column has index reference") || strings.Contains(errStr, "Data too long for column") || - strings.Contains(errStr, "Data truncated") || strings.Contains(errStr, "no rows in result set") || - strings.Contains(errStr, "Truncated incorrect") || strings.Contains(errStr, "Data truncated for column") || - // eg: For Incorrect tinyint value, Incorrect data value... - strings.Contains(errStr, "Incorrect") || - // eg: For constant 20030522161944 overflows tinyint - strings.Contains(errStr, "overflows") || - strings.Contains(errStr, "Bad Number") || - strings.Contains(errStr, "invalid year") || - strings.Contains(errStr, "value is out of range in") || - strings.Contains(errStr, "Data Too Long") || - strings.Contains(errStr, "doesn't have a default value") || - strings.Contains(errStr, "specified twice") || - strings.Contains(errStr, "cannot convert datum from") || - strings.Contains(errStr, "sql_mode=only_full_group_by") || - strings.Contains(errStr, "cannot be null") || - strings.Contains(errStr, "Column count doesn't match value count") || - strings.Contains(errStr, "Percentage value") || - strings.Contains(errStr, "Index column") || - strings.Contains(errStr, "Illegal mix of collations") || - strings.Contains(errStr, "Cannot convert string") || - strings.Contains(errStr, "interface conversion") || - strings.Contains(errStr, "connection is already closed") || - strings.Contains(errStr, "should contain a UNION") || - strings.Contains(errStr, "have different column counts") || - strings.Contains(errStr, "followed by one or more recursive ones") || - strings.Contains(errStr, "Not unique table/alias") || - strings.Contains(errStr, "have a different number of columns") || - strings.Contains(errStr, "Split table region lower value count") || - strings.Contains(errStr, "Out Of Memory") || - strings.Contains(errStr, "invalid syntax") || - strings.Contains(errStr, "newer than query schema version") || - strings.Contains(errStr, "PD server timeout") || - strings.Contains(errStr, "Information schema is out of date") || - strings.Contains(errStr, "Your query has been cancelled due to exceeding the allowed memory limit for a single SQL query") || - strings.Contains(errStr, "Value is out of range") { - return true - } - if strings.Contains(errStr, "Unsupported multi schema change") { - return true - } - if !RCIsolation && strings.Contains(errStr, "public column") { - return true - } - return false -} - -func ddlIgnoreError(err error) bool { - if err == nil { - return true - } - errStr := err.Error() - log.Warnf("check DDL err:%s", errStr) - for _, ignore := range ddlIgnoreList { - if strings.Contains(errStr, ignore) { - return true - } - } - if strings.Contains(errStr, "Information schema is changed") { - return true - } - // Sometimes, set shard row id bits to a large value might cause global auto ID overflow error. - // We ignore this error here. - if match, _ := regexp.MatchString(`cause next global auto ID( \d+ | )overflow`, errStr); match { - return true - } - if strings.Contains(errStr, "invalid connection") { - return true - } - if strings.Contains(errStr, "Unsupported shard_row_id_bits for table with primary key as row id") { - return true - } - // Ignore Column Type Change error. - if strings.Contains(errStr, "Unsupported modify column") || - strings.Contains(errStr, "Cancelled DDL job") || - strings.Contains(errStr, "Truncated incorrect") || - strings.Contains(errStr, "overflows") || - strings.Contains(errStr, "Invalid year value") || - strings.Contains(errStr, "Incorrect time value") || - strings.Contains(errStr, "Incorrect datetime value") || - strings.Contains(errStr, "Incorrect timestamp value") || - strings.Contains(errStr, "All parts of a PRIMARY KEY must be NOT NULL") || - strings.Contains(errStr, "value is out of range") || - strings.Contains(errStr, "Unsupported modify charset from") || - strings.Contains(errStr, "Unsupported modifying collation of column") || - strings.Contains(errStr, "Data truncated") || - strings.Contains(errStr, "Bad Number") || - strings.Contains(errStr, "cannot convert") || - strings.Contains(errStr, "Data Too Long") || - // eg: For v"BLOB/TEXT column '319de167-6d2e-4778-966c-60b95103a02c' used in key specification without a key length" - strings.Contains(errStr, "used in key specification without a key length") || - strings.Contains(errStr, "Specified key was too long; max key length is ") || - strings.Contains(errStr, "should be less than the total tiflash server count") || - strings.Contains(errStr, "Unsupported ALTER TiFlash settings") { - fmt.Println(errStr) - return true - } - if strings.Contains(errStr, "table doesn't exist") || - strings.Contains(errStr, "doesn't have a default value") || - strings.Contains(errStr, "with composite index covered or Primary Key covered now") || - strings.Contains(errStr, "does not exist, this column may have been updated by other DDL") || - strings.Contains(errStr, "is not exists") || strings.Contains(errStr, "column does not exist") || - strings.Contains(errStr, "doesn't exist") || strings.Contains(errStr, "Unknown table") || - strings.Contains(errStr, "admin show ddl jobs len != len(tasks)") || - strings.Contains(errStr, "check that column/key exists") || - strings.Contains(errStr, "Invalid default value") || - strings.Contains(errStr, "Duplicate column name") || - strings.Contains(errStr, "can't drop only column") || - strings.Contains(errStr, "doesn't exist") || strings.Contains(errStr, "not found") || - strings.Contains(errStr, "column is deleted") || strings.Contains(errStr, "Can't find column") || - strings.Contains(errStr, "converting driver.Value type") || strings.Contains(errStr, "column specified twice") || - strings.Contains(errStr, "Out of range value for column") || strings.Contains(errStr, "Unknown column") || - strings.Contains(errStr, "column has index reference") || strings.Contains(errStr, "Data too long for column") || - strings.Contains(errStr, "Data truncated") || strings.Contains(errStr, "no rows in result set") || - strings.Contains(errStr, "with tidb_enable_change_multi_schema is disable") || - strings.Contains(errStr, "not allowed type for this type of partitioning") || - strings.Contains(errStr, "A PRIMARY KEY must include all columns in the table's partitioning function") || - strings.Contains(errStr, "A UNIQUE INDEX must include all columns in the table's partitioning function") || - strings.Contains(errStr, "cannot convert datum") || - strings.Contains(errStr, "Duplicate entry") || - strings.Contains(errStr, "has a partitioning function dependency and cannot be dropped or renamed") || - strings.Contains(errStr, "A CLUSTERED INDEX must include all columns in the table's partitioning function") || - strings.Contains(errStr, "PD server timeout") || - strings.Contains(errStr, "Information schema is out of date") || - strings.Contains(errStr, "Invalid JSON data provided") || - strings.Contains(errStr, "Invalid JSON value for CAST") || - strings.Contains(errStr, "Invalid JSON text") || - strings.Contains(errStr, "since the unique index is not including all partitioning columns, and GLOBAL is not given as IndexOption") || - strings.Contains(errStr, "doesn't yet support") { - return true - } - return false -} diff --git a/framework/testcase.go b/framework/testcase.go new file mode 100644 index 0000000..54abd91 --- /dev/null +++ b/framework/testcase.go @@ -0,0 +1,719 @@ +package framework + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + crc322 "hash/crc32" + "math/rand" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/PingCAP-QE/schrddl/dump" + "github.com/PingCAP-QE/schrddl/norec" + "github.com/PingCAP-QE/schrddl/reduce" + "github.com/PingCAP-QE/schrddl/sqlgenerator" + "github.com/PingCAP-QE/schrddl/util" + "github.com/ngaut/log" + "github.com/pingcap/errors" + "github.com/pingcap/tidb/pkg/parser" + "github.com/pingcap/tidb/pkg/parser/model" + "github.com/pingcap/tidb/pkg/util/logutil" + "go.uber.org/zap" +) + +type CaseType int + +const ( + CaseTypeNormal CaseType = iota + CaseTypePlanCache +) + +type testCase struct { + cfg *CaseConfig + caseType CaseType + + dbname string + dbs []*sql.DB + caseIndex int + tables map[string]*ddlTestTable + schemas map[string]*ddlTestSchema + views map[string]*ddlTestView + tablesLock sync.RWMutex + schemasLock sync.Mutex + stop int32 + lastDDLID int + charsets []string + charsetsCollates map[string][]string + + tableMap map[string]*sqlgenerator.Table + outputWriter *os.File + + tidbParser *parser.Parser + + // stat info + queryPlanMap map[string]string + originalSQL string + reduceChangedSQL string + reduceSQL string + aggregationAsInnerSideOfIndexJoin int + planUseMvIndex int + cntOfOldOriginal int + cntOfNewOriginal int + cntOfOldReduce int + cntOfNewReduce int + + // cert + oldEstCntOriginal float64 + newEstCntOriginal float64 + oldEstCntReduce float64 + newEstCntReduce float64 + checkCERTCnt int + + // tlp + cntOfP int + cntOfN int + cntOfNull int + cntOfAll int + nQuery string + nullQuery string + allQuery string +} + +func (c *testCase) InitializeDB(dbname, dbDSN string) error { + // Parallel send DDL request need more connection to send DDL request concurrently + db0, err := OpenDB(dbDSN, 50) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + db1, err := OpenDB(dbDSN, 50) + if err != nil { + log.Fatalf("[ddl] create db client error %v", err) + } + + c.dbs = []*sql.DB{db0, db1} + c.tidbParser = parser.New() + c.dbname = dbname + + for _, db := range c.dbs { + _, err := db.Exec("set @@max_execution_time=800000") + if err != nil { + return err + } + _, err = db.Exec("set @@group_concat_max_len=10240000") + if err != nil { + return err + } + } + + return nil +} + +func (c *testCase) DisableKVGC() { + for _, db := range c.dbs { + disableTiKVGC(db) + } +} + +// initialize generates possible DDL and DML operations for one `testCase`. +// Different `testCase`s will be run in parallel according to the concurrent configuration. +func (c *testCase) initialize(dbs []*sql.DB) error { + //var err error + c.dbs = dbs + c.tidbParser = parser.New() + return nil +} + +// setCharsetsAndCollates sets the allowable character sets and associated collates for this testCase. +func (c *testCase) setCharsetsAndCollates(charsets []string, charsetsCollates map[string][]string) { + c.charsets = charsets + c.charsetsCollates = charsetsCollates +} + +func (c *testCase) checkError(err error) error { + if err != nil { + if c.cfg.MySQLCompatible { + if strings.Contains(err.Error(), "Duplicate entry") { + return nil + } + } + return errors.Trace(err) + } + return nil +} + +func (c *testCase) execQueryForPlanEstCnt(sql string) (float64, error) { + sql = "explain format='brief' " + sql + rows, err := c.dbs[0].Query(sql) + if err != nil { + return 0, err + } + defer func() { + rows.Close() + }() + var id string + var estRows float64 + var task string + var accessObject string + var operatorInfo string + rows.Next() + err = rows.Scan(&id, &estRows, &task, &accessObject, &operatorInfo) + return estRows, err +} + +func (c *testCase) execQueryForCnt(sql string) (int, error) { + rows, err := c.dbs[0].Query(sql) + if err != nil { + return 0, err + } + defer func() { + rows.Close() + }() + rs := 0 + for rows.Next() { + rs++ + } + if rows.Err() != nil { + return 0, rows.Err() + } + return rs, nil +} + +func (c *testCase) execQueryAndCheckEmpty(sql string) bool { + rows, err := c.dbs[0].Query(sql) + if err != nil { + logutil.BgLogger().Error("unexpected error", zap.Error(err)) + return false + } + defer func() { + rows.Close() + }() + + result := !rows.Next() + + cnt2 := 0 + if !result { + cols, _ := rows.Columns() + rawResult := make([][]byte, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range rawResult { + dest[i] = &rawResult[i] + } + + err := rows.Scan(dest...) + if err != nil { + panic(err) + } + + for rows.Next() { + cnt2++ + } + + for _, r := range rawResult { + c.outputWriter.WriteString(fmt.Sprintf("%s\n", string(r))) + } + + cnt, _ := c.execQueryForCnt(sql) + + rs, _ := c.execQuery(sql) + + c.outputWriter.WriteString(fmt.Sprintf("sql: %s, cnt: %d, cnt2: %d, cnt3: %d \n", sql, cnt, cnt2+1, len(rs))) + if rows.Err() != nil { + c.outputWriter.WriteString(fmt.Sprintf("err %s \n", rows.Err().Error())) + } + } + return result +} + +func (c *testCase) execQueryForCRC32(sql string) (map[uint32]struct{}, error) { + rows, err := c.dbs[0].Query(sql) + if err != nil { + return nil, err + } + defer func() { + rows.Close() + }() + + // Read all rows. + crc32s := make(map[uint32]struct{}, 0) + + for rows.Next() { + cols, err1 := rows.Columns() + if err1 != nil { + return nil, err + } + + //log.Infof("[ddl] [instance %d] rows.Columns():%v, len(cols):%v", c.caseIndex, cols, len(cols)) + + // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go + rawResult := make([][]byte, len(cols)) + result := make([]string, len(cols)) + dest := make([]interface{}, len(cols)) + ct, _ := rows.ColumnTypes() + for i := range rawResult { + dest[i] = &rawResult[i] + } + + err1 = rows.Scan(dest...) + if err1 != nil { + return nil, err + } + + for i, raw := range rawResult { + if raw == nil { + result[i] = ddlTestValueNull + } else { + //logutil.BgLogger().Warn("type to debug", zap.String("type", ct[i].DatabaseTypeName())) + if strings.EqualFold(ct[i].DatabaseTypeName(), "double") { + result[i] = fmt.Sprintf("'%s'", RoundToOneDecimals(string(raw))) + } else { + result[i] = fmt.Sprintf("'%s'", strings.ToLower(string(raw))) + } + //if typeNeedQuota(metaCols[i].k) { + // result[i] = fmt.Sprintf("'%s'", string(raw)) + //} + //result[i] = string(raw) + } + } + + crc32 := crc322.NewIEEE() + for _, r := range result { + crc32.Write([]byte(r)) + } + + sum := crc32.Sum32() + //logutil.BgLogger().Info("crc32", zap.Uint32("crc32", sum), zap.String("data", fmt.Sprintf("%v", result))) + + crc32s[sum] = struct{}{} + } + if rows.Err() != nil { + return nil, rows.Err() + } + + return crc32s, nil +} + +func (c *testCase) execQuery(sql string) ([][]string, error) { + return util.FetchRowsWithDB(c.dbs[0], sql) +} + +func (c *testCase) CheckData(tables []string) error { + for _, table := range tables { + same, err := util.CheckTableData(c.dbname, table, "testcache", table, c.dbs[0]) + if err != nil { + log.Fatalf("Error check result") + } + if !same { + return errors.Errorf("Table %s have different data", table) + } + } + + return nil +} + +func (c *testCase) dumpPrepare(prepare *sqlgenerator.Prepare) error { + dir, err := c.dumpErrorTables(prepare.SQLNoCache) + if err != nil { + return errors.Trace(err) + } + err = prepare.RecordError(dir) + return errors.Trace(err) +} + +func (c *testCase) dropData(tables []string) error { + for _, tbl := range tables { + for _, db := range c.dbs { + if _, err := db.Exec(fmt.Sprintf("truncate table %s", tbl)); err != nil { + return errors.Trace(err) + } + } + } + return nil +} + +// A special function to test instance plan cache +func (c *testCase) testPlanCache(ctx context.Context) error { + state := sqlgenerator.NewState() + + dbNoCache, dbWithCache := c.dbs[0], c.dbs[1] + + prepareStmtCnt := 100 + for i := 0; i < prepareStmtCnt; i++ { + startSQL, err := sqlgenerator.PlanCacheDataGen.Eval(state) + if err != nil { + return errors.Trace(err) + } + + _, err1 := dbNoCache.Exec(startSQL) + _, err2 := dbWithCache.Exec(startSQL) + if err1 != nil && err2 != nil { + continue + } + if err := sqlgenerator.CheckError(err1, err2); err != nil { + return errors.Trace(err) + } + } + + tableMetas := c.fetchTableInfo(state) + tblNames := make([]string, 0, len(tableMetas)) + for _, m := range tableMetas { + tblNames = append(tblNames, m.Name.L) + } + + stateDDL := sqlgenerator.NewState() + stateDDL.Tables = state.Tables + ddlFailed := false + go func() { + for { + select { + case <-ctx.Done(): + return + default: + time.Sleep(time.Second) + // if err := c.runDDL(stateDDL); err != nil { + // ddlFailed = true + // return + // } + } + } + }() + + noCache, withCache := 0, 0 + for i := 0; i < 50000; i++ { + if ddlFailed { + break + } + + if i%1000 == 0 { + if err := c.CheckData(tblNames); err != nil { + return errors.Trace(err) + } + log.Info("Check table data passed") + } + + useQuery := rand.Intn(10) > 6 + + prepare, err := sqlgenerator.GeneratePrepare(state, useQuery) + if err != nil { + log.Warn("Generate prepare statement failed, err = %s", err.Error()) + continue + } + useCache, err := prepare.UsePlanCache(dbWithCache) + if err != nil { + log.Warn("Check use cache failed, err = %s", err.Error()) + continue + } + + if !useCache { + noCache++ + continue + } + + withCache++ + if useQuery { + err = prepare.CheckQuery(dbWithCache, dbNoCache) + } else { + err = prepare.CheckExec(dbWithCache, dbNoCache) + + affectedTbls, _ := dump.ExtraFromSQL(prepare.SQLNoCache) + if err2 := c.CheckData(affectedTbls); err2 != nil { + c.dumpPrepare(prepare) + if err3 := c.dropData(affectedTbls); err != nil { + return errors.Trace(err3) + } + log.Warnf("Table %s data inconsistent after DML", affectedTbls[0]) + } + } + + if err != nil { + if err2 := c.dumpPrepare(prepare); err2 != nil { + return errors.Trace(err2) + } + if strings.Contains(err.Error(), "different error") { + continue + } + } + } + + log.Infof("WithCache: %d, NoCache: %d\n", withCache, noCache) + return nil +} + +func (c *testCase) fetchTableInfo(state *sqlgenerator.State) []*model.TableInfo { + tableMetas := make([]*model.TableInfo, 0) + for i := 0; i < len(state.Tables); i++ { + addr := c.cfg.DBAddr + stateAddr := strings.Replace(addr, "4000", "10080", -1) + path := fmt.Sprintf("%s/schema/%s/%s", stateAddr, c.dbname, state.Tables[i].Name) + rawMeta, err := exec.Command("curl", path).Output() + if err != nil { + log.Infof("curl error %s", err.Error()) + continue + } + var meta model.TableInfo + err = json.Unmarshal(rawMeta, &meta) + if err != nil { + logutil.BgLogger().Warn("unmarshal error", zap.Error(err), zap.String("table name", state.Tables[i].Name)) + state.Tables = append(state.Tables[:i], state.Tables[i+1:]...) + i-- + continue + } + tableMetas = append(tableMetas, &meta) + } + for _, table := range tableMetas { + log.Infof("table %s", table.Name.O) + } + log.Infof("tableMetas %d", len(tableMetas)) + return tableMetas +} + +// Dump related table data to local. +func (c *testCase) dumpErrorTables(sql string) (string, error) { + tblNames, err := dump.ExtraFromSQL(sql) + if err != nil { + return "", err + } + pwd := os.Getenv("PWD") + if GlobalOutPut != "" { + pwd = filepath.Dir(GlobalOutPut) + } + bugNum := globalBugSeqNum.Add(1) + dir := fmt.Sprintf("local://%s/bug-%s-%d", pwd, time.Now().Format("2006-01-02-15-04-05"), bugNum) + return dir, dump.DumpToFile("test", tblNames, dir, c.cfg.DBAddr) +} + +// Run random DDL, currently we only support add/drop index +func (c *testCase) runDDL(state *sqlgenerator.State) error { + sql, err := sqlgenerator.AddOrDropIndex.Eval(state) + if err != nil { + return err + } + + if err = util.ExecSQLWithDB(c.dbs[0], sql); err != nil { + return err + } + + return nil +} + +// execute iterates over two list of operations concurrently, one is +// ddl operations, one is dml operations. +// When one list completes, it starts over from the beginning again. +// When both of them ONCE complete, it exits. +func (c *testCase) execute(ctx context.Context) error { + state := sqlgenerator.NewState() + state.SetWeight(sqlgenerator.RenameColumn, 0) + state.SetWeight(sqlgenerator.WindowFunction, 0) + state.SetWeight(sqlgenerator.WindowClause, 0) + state.SetWeight(sqlgenerator.WindowFunctionOverW, 0) + state.SetWeight(sqlgenerator.WhereClause, 1) + //state.SetWeight(sqlgenerator.Limit, 0) + state.SetWeight(sqlgenerator.UnionSelect, 0) + //state.SetWeight(sqlgenerator.PartitionDefinitionHash, 1000) + state.SetWeight(sqlgenerator.PartitionDefinitionKey, 0) + //state.SetWeight(sqlgenerator.PartitionDefinitionRange, 1000) + + //state.SetWeight(sqlgenerator.AggSelect, 0) + + // Sub query is hard for NoREC + //state.SetWeight(sqlgenerator.SubSelect, 0) + + if !EnableApproximateQuerySynthesis { + state.SetWeight(sqlgenerator.ScalarSubQuery, 0) + state.SetWeight(sqlgenerator.SubSelect, 0) + } + + // bug + state.SetWeight(sqlgenerator.ColumnDefinitionTypesEnum, 0) + state.SetWeight(sqlgenerator.ColumnDefinitionTypesSet, 0) + state.SetWeight(sqlgenerator.ColumnDefinitionTypesYear, 0) + state.SetWeight(sqlgenerator.ColumnDefinitionTypesBit, 0) + + //state.Hook().Append(sqlgenerator.NewFnHookDebug()) + + //state.SetWeight(sqlgenerator.ColumnDefinitionTypesJSON, 0) + //state.SetWeight(sqlgenerator.JSONPredicate, 0) + + db := c.dbs[0] + + prepareStmtCnt := 50 + for i := 0; i < prepareStmtCnt; i++ { + startSQL, err := sqlgenerator.Start.Eval(state) + if err != nil { + return err + } + err = util.ExecSQLWithDB(db, startSQL) + //println(fmt.Sprintf("%s;", startSQL)) + if err != nil { + return err + } + } + + _, err := c.dbs[0].Exec("set @@max_execution_time=800000") + if err != nil { + return err + } + _, err = c.dbs[0].Exec("set @@group_concat_max_len=10240000") + if err != nil { + return err + } + + tableMetas := c.fetchTableInfo(state) + state.SetTableMeta(tableMetas) + sqlgenerator.PrepareIndexJoinColumns(state) + defer sqlgenerator.RemoveIndexJoinColumns(state) + + for cnt := 0; cnt < 20000; cnt++ { + if cnt%10000 == 0 { + err := c.readDataFromTiDB() + if err != nil { + if !util.DMLIgnoreError(err) { + return errors.Trace(err) + } + } + } + + // NoREC + rewriter := &norec.NoRecRewriter{} + var sb strings.Builder + + doDML := rand.Intn(2) == 0 + if doDML { + dmlSQL, err := sqlgenerator.DMLStmt.Eval(state) + //println(fmt.Sprintf("%s;", dmlSQL)) + if err != nil { + return err + } + if rand.Intn(100) == 0 { + dmlSQL, err = sqlgenerator.SetTiFlashReplica.Eval(state) + if err != nil { + return err + } + } else if rand.Intn(15) == 0 { + dmlSQL, err = sqlgenerator.SetVariable.Eval(state) + if err != nil { + return err + } + } + if rand.Intn(100) == 0 { + dmlSQL, err = sqlgenerator.AnalyzeTable.Eval(state) + if err != nil { + return err + } + } + err = util.ExecSQLWithDB(db, dmlSQL) + if err != nil { + return err + } + } else { + var ck checker + if EnableApproximateQuerySynthesis { + ck = &pinoloChecker{c: c} + } else if EnableCERT { + ck = &certChecker{c: c} + } else if EnableTLP { + ck = &tlpChecker{c: c} + } else { + ck = &norecChecker{c: c, rewriter: *rewriter, sb: sb} + } + + fn := sqlgenerator.QueryOrCTE + if EnableCERT { + fn = sqlgenerator.Query + } + querySQL, err := fn.Eval(state) + if err != nil { + return errors.Trace(err) + } + + //println(fmt.Sprintf("%s;", querySQL)) + + found, err := ck.check(querySQL, false) + if err != nil { + if !util.DMLIgnoreError(err) { + if strings.Contains(err.Error(), "plan not match") { + _, err = c.dbs[0].Exec(querySQL) + if err == nil { + continue + } + } + return errors.Trace(err) + } else { + continue + } + } + if found { + reduceSQL := reduce.ReduceSQL(ck.check, querySQL) + if EnableTLP { + _, err = c.outputWriter.WriteString( + fmt.Sprintf("old count of predicate reduce SQL:%d, count of non-predicate reduce SQL:%d,\ncount of null-predicate reduce SQL:%d, count of all-predicate reduce SQL:%d\n"+ + "query of orginal SQL: %s\n"+ + "predicate reduce query: %s\n"+ + "negative predicate reduce query: %s\n"+ + "isnull predicate reduce query: %s\n"+ + "all predicate reduce query: %s\n"+ + "\n\n", + c.cntOfP, c.cntOfN, c.cntOfNull, c.cntOfAll, c.originalSQL, c.reduceSQL, c.nQuery, c.nullQuery, c.allQuery)) + } else if EnableCERT { + _, err = c.outputWriter.WriteString( + fmt.Sprintf("old count of orginal SQL:%f, new count of orginal SQL:%f,\nold count of reduce SQL:%f, new count of reduce SQL:%f,\nold query of orginal SQL: %s\nnew query of reduce SQL: %s\nreduce query: %s\n\n\n", + c.oldEstCntOriginal, c.newEstCntOriginal, c.oldEstCntReduce, c.newEstCntReduce, c.originalSQL, c.reduceChangedSQL, c.reduceSQL)) + } else { + _, err = c.outputWriter.WriteString( + fmt.Sprintf("old count of orginal SQL:%d, new count of orginal SQL:%d,\nold count of reduce SQL:%d, new count of reduce SQL:%d,\nold query of orginal SQL: %s\nnew query of reduce SQL: %s\nreduce query: %s\n\n\n", + c.cntOfOldOriginal, c.cntOfNewOriginal, c.cntOfOldReduce, c.cntOfNewReduce, c.originalSQL, c.reduceChangedSQL, c.reduceSQL)) + } + + if _, err := c.dumpErrorTables(reduceSQL); err != nil { + return err + } + TestFail = true + break + } + } + } + + log.Infof("[ddl] [instance %d] Round completed", c.caseIndex) + log.Infof("[ddl] [instance %d] Executing post round operations...", c.caseIndex) + + if !c.cfg.MySQLCompatible { + if err := c.executeAdminCheck(state); err != nil { + return errors.Trace(err) + } + if err := c.readDataFromTiDB(); err != nil { + return errors.Trace(err) + } + } + + return nil +} + +func (c *testCase) readDataFromTiDB() error { + for _, table := range c.tables { + sql := fmt.Sprintf("select * from `%s`", table.name) + actualRows, err := util.FetchRowsWithDB(c.dbs[0], sql) + if err != nil { + return errors.Trace(err) + } + c.tableMap[table.name].Values = actualRows + } + + return nil +} + +func (c *testCase) executeAdminCheck(state *sqlgenerator.State) error { + for _, table := range state.Tables { + sql := fmt.Sprintf("ADMIN CHECK TABLE `%s`", table.Name) + log.Infof("[ddl] [instance %d] %s", c.caseIndex, sql) + if _, err := c.dbs[0].Exec(sql); err != nil { + return errors.Annotatef(err, "Error when executing SQL: %s", sql) + } + } + + return nil +} diff --git a/framework/tlpchecker.go b/framework/tlpchecker.go index 5d8f523..27e68fe 100644 --- a/framework/tlpchecker.go +++ b/framework/tlpchecker.go @@ -1,6 +1,8 @@ package framework import ( + "strings" + "github.com/PingCAP-QE/schrddl/tlp" "github.com/PingCAP-QE/schrddl/util" "github.com/pingcap/errors" @@ -8,7 +10,6 @@ import ( "github.com/pingcap/tidb/pkg/util/logutil" "go.uber.org/zap" "golang.org/x/exp/slices" - "strings" ) type tlpChecker struct { @@ -60,7 +61,7 @@ func (t *tlpChecker) check(sql string, isReduce bool) (ok bool, err error) { rs1, err := t.c.execQueryForCRC32(querySQL) //println(fmt.Sprintf("%s;", querySQL)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -98,7 +99,7 @@ func (t *tlpChecker) check(sql string, isReduce bool) (ok bool, err error) { rs2, err := t.c.execQueryForCRC32(negativeQuery) //println(fmt.Sprintf("%s;", newQuery)) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -120,7 +121,7 @@ func (t *tlpChecker) check(sql string, isReduce bool) (ok bool, err error) { isNullQuery = t.sb.String() rs3, err := t.c.execQueryForCRC32(isNullQuery) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) @@ -143,7 +144,7 @@ func (t *tlpChecker) check(sql string, isReduce bool) (ok bool, err error) { allQuery = t.sb.String() rsAll, err := t.c.execQueryForCRC32(allQuery) if err != nil { - if dmlIgnoreError(err) { + if util.DMLIgnoreError(err) { return false, nil } else { logutil.BgLogger().Error("unexpected error", zap.String("query", querySQL), zap.Error(err)) diff --git a/main.go b/main.go index 2c7c9d6..e8d11e8 100644 --- a/main.go +++ b/main.go @@ -23,6 +23,7 @@ import ( "time" . "github.com/PingCAP-QE/schrddl/framework" + "github.com/PingCAP-QE/schrddl/util" "github.com/go-sql-driver/mysql" "github.com/ngaut/log" ) @@ -51,10 +52,14 @@ func prepareEnv() { if err != nil { log.Fatalf("Can't open database, err: %s", err.Error()) } + defer tiDb.Close() + tidbC, err := tiDb.Conn(context.Background()) if err != nil { log.Fatalf("Can't connect to database, err: %s", err.Error()) } + defer tidbC.Close() + if _, err = tidbC.ExecContext(context.Background(), fmt.Sprintf("set global time_zone='%s'", Local.String())); err != nil { if _, err = tidbC.ExecContext(context.Background(), fmt.Sprintf("set global time_zone='%s'", time.Local.String())); err != nil { if _, err = tidbC.ExecContext(context.Background(), "set global time_zone='+8:00'"); err != nil { @@ -62,10 +67,23 @@ func prepareEnv() { } } } - // Enable index join on aggregation - tidbC.ExecContext(context.Background(), "set GLOBAL tidb_enable_inl_join_inner_multi_pattern='ON'") - tidbC.Close() + initSQLs := []string{ + "create database if not exists testcache", + "set GLOBAL tidb_enable_inl_join_inner_multi_pattern='ON'", + "set GLOBAL tidb_enable_instance_plan_cache=1", + "set global tidb_enable_global_index=true", + } + if util.RCIsolation { + initSQLs = append(initSQLs, "set global transaction_isolation='read-committed'") + } + + for _, sql := range initSQLs { + _, err = tidbC.ExecContext(context.Background(), sql) + if err != nil { + log.Fatalf("[DDL] %s failed", sql) + } + } mysql.SetLogger(log.Logger()) } @@ -79,7 +97,7 @@ func main() { EnableTransactionTest = true } if *rc { - RCIsolation = true + util.RCIsolation = true } if *prepare { Prepare = true @@ -111,7 +129,17 @@ func main() { http.ListenAndServe("127.0.0.1:6060", nil) }() - Run(*dbAddr, *dbName, *concurrency, *tablesToCreate, *mysqlCompatible, testType, *testTime) + cfg := CaseConfig{ + Concurrency: *concurrency, + TablesToCreate: *tablesToCreate, + MySQLCompatible: *mysqlCompatible, + DBName: *dbName, + DBAddr: *dbAddr, + TestTp: testType, + TestPrepare: *prepare, + } + + Run(cfg, *testTime) if TestFail { log.Fatalf("test failed") } diff --git a/sqlgenerator/db_constant.go b/sqlgenerator/db_constant.go index e0e7619..139bb2c 100644 --- a/sqlgenerator/db_constant.go +++ b/sqlgenerator/db_constant.go @@ -2,50 +2,13 @@ package sqlgenerator import ( "fmt" + "log" + "math" + "math/rand" "github.com/cznic/mathutil" ) -func (c *Column) EstimateSizeInBytes() int { - const bytesPerChar = 4 - switch c.Tp { - case ColumnTypeInt: - return 4 - case ColumnTypeBoolean, ColumnTypeTinyInt, ColumnTypeYear: - return 1 - case ColumnTypeSmallInt: - return 2 - case ColumnTypeMediumInt: - return 3 - case ColumnTypeBigInt: - return 8 - case ColumnTypeFloat: - return 4 - case ColumnTypeDouble, ColumnTypeDecimal: - return 8 - case ColumnTypeBit: - return mathutil.Max(c.Arg1, 1) - case ColumnTypeChar, ColumnTypeVarchar, ColumnTypeText, ColumnTypeBlob: - return bytesPerChar * c.Arg1 - case ColumnTypeBinary, ColumnTypeVarBinary: - return c.Arg1 - case ColumnTypeEnum: - return 2 - case ColumnTypeSet: - return 8 - case ColumnTypeDate, ColumnTypeTime: - return 3 - case ColumnTypeDatetime: - return 8 - case ColumnTypeTimestamp: - return 4 - case ColumnTypeJSON: - return c.Arg1 - } - panic(fmt.Sprintf("unknown column type %d", c.Tp)) - return 0 -} - type ColumnType int64 const ( @@ -101,12 +64,8 @@ func (tps ColumnTypes) Filter(pred func(tp ColumnType) bool) ColumnTypes { func (tps ColumnTypes) Concat(other ColumnTypes) ColumnTypes { ret := make(ColumnTypes, 0, len(tps)+len(other)) - for _, tp := range tps { - ret = append(ret, tp) - } - for _, tp := range other { - ret = append(ret, tp) - } + ret = append(ret, tps...) + ret = append(ret, other...) return ret } @@ -166,41 +125,6 @@ var ColumnTypeTimeTypes = ColumnTypes{ ColumnTypeDate, ColumnTypeTime, ColumnTypeDatetime, ColumnTypeTimestamp, } -type Collation struct { - ID int - CharsetName string - CollationName string - IsDefault bool -} - -type CollationType int64 - -const ( - CollationBinary CollationType = iota - CollationUtf8Bin - CollationUtf8mb4Bin - CollationUtf8GeneralCI - CollationUtf8mb4GeneralCI - CollationUtf8UnicodeCI - CollationUtf8mb4UnicodeCI - CollationGBKBin - CollationGBKChineseCI - - CollationTypeMax -) - -var Collations = map[CollationType]*Collation{ - CollationGBKChineseCI: {28, "gbk", "gbk_chinese_ci", true}, - CollationUtf8GeneralCI: {33, "utf8", "utf8_general_ci", false}, - CollationUtf8mb4GeneralCI: {45, "utf8mb4", "utf8mb4_general_ci", false}, - CollationUtf8mb4Bin: {46, "utf8mb4", "utf8mb4_bin", true}, - CollationBinary: {63, "binary", "binary", true}, - CollationUtf8Bin: {83, "utf8", "utf8_bin", true}, - CollationGBKBin: {87, "gbk", "gbk_bin", false}, - CollationUtf8UnicodeCI: {192, "utf8", "utf8_unicode_ci", false}, - CollationUtf8mb4UnicodeCI: {224, "utf8mb4", "utf8mb4_unicode_ci", false}, -} - func (c ColumnType) IsStringType() bool { switch c { case ColumnTypeChar, ColumnTypeVarchar, ColumnTypeText, @@ -330,6 +254,60 @@ func (c ColumnType) String() string { } } +func (c ColumnType) RandomMismatch() string { + if c.IsIntegerType() || c.IsFloatingType() || c == ColumnTypeBit { + if rand.Intn(2) == 0 { + return fmt.Sprintf("'%s'", randomStringRunes(8, false)) + } + return Num(math.MaxInt64) + } + if c.IsStringType() || c == ColumnTypeBoolean { + return RandomNum(10000, 1000000) + } + return fmt.Sprintf("'%s'", randomStringRunes(16, false)) + +} + +func (c *Column) EstimateSizeInBytes() int { + const bytesPerChar = 4 + switch c.Tp { + case ColumnTypeInt: + return 4 + case ColumnTypeBoolean, ColumnTypeTinyInt, ColumnTypeYear: + return 1 + case ColumnTypeSmallInt: + return 2 + case ColumnTypeMediumInt: + return 3 + case ColumnTypeBigInt: + return 8 + case ColumnTypeFloat: + return 4 + case ColumnTypeDouble, ColumnTypeDecimal: + return 8 + case ColumnTypeBit: + return mathutil.Max(c.Arg1, 1) + case ColumnTypeChar, ColumnTypeVarchar, ColumnTypeText, ColumnTypeBlob: + return bytesPerChar * c.Arg1 + case ColumnTypeBinary, ColumnTypeVarBinary: + return c.Arg1 + case ColumnTypeEnum: + return 2 + case ColumnTypeSet: + return 8 + case ColumnTypeDate, ColumnTypeTime: + return 3 + case ColumnTypeDatetime: + return 8 + case ColumnTypeTimestamp: + return 4 + case ColumnTypeJSON: + return c.Arg1 + } + log.Fatalf("unknown column type %d", c.Tp) + return 0 +} + type IndexType int64 const ( @@ -368,3 +346,38 @@ const ( QueryAggregation = "agg" ChosenSelection = "Selection" ) + +type Collation struct { + ID int + CharsetName string + CollationName string + IsDefault bool +} + +type CollationType int64 + +const ( + CollationBinary CollationType = iota + CollationUtf8Bin + CollationUtf8mb4Bin + CollationUtf8GeneralCI + CollationUtf8mb4GeneralCI + CollationUtf8UnicodeCI + CollationUtf8mb4UnicodeCI + CollationGBKBin + CollationGBKChineseCI + + CollationTypeMax +) + +var Collations = map[CollationType]*Collation{ + CollationGBKChineseCI: {28, "gbk", "gbk_chinese_ci", true}, + CollationUtf8GeneralCI: {33, "utf8", "utf8_general_ci", false}, + CollationUtf8mb4GeneralCI: {45, "utf8mb4", "utf8mb4_general_ci", false}, + CollationUtf8mb4Bin: {46, "utf8mb4", "utf8mb4_bin", true}, + CollationBinary: {63, "binary", "binary", true}, + CollationUtf8Bin: {83, "utf8", "utf8_bin", true}, + CollationGBKBin: {87, "gbk", "gbk_bin", false}, + CollationUtf8UnicodeCI: {192, "utf8", "utf8_unicode_ci", false}, + CollationUtf8mb4UnicodeCI: {224, "utf8mb4", "utf8mb4_unicode_ci", false}, +} diff --git a/sqlgenerator/db_generator.go b/sqlgenerator/db_generator.go index 9306523..23d8165 100644 --- a/sqlgenerator/db_generator.go +++ b/sqlgenerator/db_generator.go @@ -3,7 +3,6 @@ package sqlgenerator import ( "bytes" "fmt" - "github.com/twinj/uuid" "log" "math/rand" "sort" @@ -11,6 +10,8 @@ import ( "strings" "time" + "github.com/twinj/uuid" + "github.com/cznic/mathutil" "gonum.org/v1/gonum/stat/distuv" ) @@ -113,14 +114,6 @@ func LimitIndexColumnSize(cols []*Column, sizeLimit int) []*Column { return cols[:maxIdx] } -func GenNewPrepare(id int) *Prepare { - return &Prepare{ - ID: id, - Name: fmt.Sprintf("prepare_%d", id), - Args: nil, - } -} - func (t *Table) GenRandValues(cols []*Column) []string { if len(cols) == 0 { cols = t.Columns @@ -176,14 +169,6 @@ func (t *Table) GenMultipleRowsAscForIndexCols(count int, idx *Index) [][]string return rows } -func (p *Prepare) GenAssignments() []string { - todoSQLs := make([]string, len(p.Args)) - for i := 0; i < len(todoSQLs); i++ { - todoSQLs[i] = fmt.Sprintf("set @i%d = %s", i, p.Args[i]()) - } - return todoSQLs -} - func (c *Column) ZeroValue() string { switch c.Tp { case ColumnTypeTinyInt, ColumnTypeSmallInt, ColumnTypeMediumInt, ColumnTypeInt, ColumnTypeBigInt, ColumnTypeBoolean, ColumnTypeYear: @@ -265,7 +250,7 @@ func (c *Column) RandomValuesAsc(count int) []string { } else if length > 20 { length = 20 } - return RandStrings(length, count, c.Collation.CharsetName == "gbk") + return RandStrings(length, count, false) case ColumnTypeText, ColumnTypeBlob: length := c.Arg1 if length == 0 { @@ -273,7 +258,7 @@ func (c *Column) RandomValuesAsc(count int) []string { } else if length > 20 { length = 20 } - return RandStrings(length, count, c.Collation.CharsetName == "gbk") + return RandStrings(length, count, false) case ColumnTypeEnum, ColumnTypeSet: return RandEnums(c.Args, count) case ColumnTypeDate, ColumnTypeDatetime, ColumnTypeTimestamp: @@ -424,19 +409,6 @@ func RandJsons(count int) []string { return res } -var asciiRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789~!@#$%^&*()_+=-") - -func RandStringRunes(n int, mixCNChar bool) string { - b := make([]rune, n) - for i := range b { - b[i] = asciiRunes[rand.Intn(len(asciiRunes))] - if mixCNChar && rand.Intn(3) == 0 { - b[i] = rune(int('\u4e00') + rand.Intn(int('\u9fff')-int('\u4e00'))) - } - } - return string(b) -} - func RandGBKStringRunes(n int) string { b := make([]rune, n) for i := range b { @@ -461,7 +433,7 @@ func RandNumRunes(n int) string { func RandStrings(strLen int, count int, mixCNChar bool) []string { result := make([]string, count) for i := 0; i < count; i++ { - result[i] = fmt.Sprintf("'%s'", RandStringRunes(rand.Intn(strLen), mixCNChar)) + result[i] = fmt.Sprintf("'%s'", randomStringRunes(rand.Intn(strLen), mixCNChar)) } sort.Slice(result, func(i, j int) bool { return result[i] < result[j] diff --git a/sqlgenerator/db_mutator.go b/sqlgenerator/db_mutator.go index 43ba043..00e5d06 100644 --- a/sqlgenerator/db_mutator.go +++ b/sqlgenerator/db_mutator.go @@ -100,21 +100,6 @@ func (s *State) ParentCTE() *Table { return ctes[len(ctes)-1] } -func (s *State) AppendPrepare(pre *Prepare) { - s.prepareStmts = append(s.prepareStmts, pre) -} - -func (s *State) RemovePrepare(p *Prepare) { - var pos int - for i := range s.prepareStmts { - if s.prepareStmts[i].ID == p.ID { - pos = i - break - } - } - s.prepareStmts = append(s.prepareStmts[:pos], s.prepareStmts[pos+1:]...) -} - func (t *Table) AppendColumn(c *Column) { t.Columns = append(t.Columns, c) for i := range t.Values { @@ -284,11 +269,3 @@ func (i *Index) AppendColumnIfNotExists(cols ...*Column) { i.ColumnPrefix = append(i.ColumnPrefix, 0) } } - -func (p *Prepare) AppendColumns(cols ...*Column) { - for _, c := range cols { - p.Args = append(p.Args, func() string { - return c.RandomValue() - }) - } -} diff --git a/sqlgenerator/db_retriever.go b/sqlgenerator/db_retriever.go index 52a261e..698064b 100644 --- a/sqlgenerator/db_retriever.go +++ b/sqlgenerator/db_retriever.go @@ -1,7 +1,6 @@ package sqlgenerator import ( - "fmt" "math" "math/rand" "strings" @@ -100,10 +99,6 @@ func (s *State) PushSubQuery(sq *Table) { s.subQuery[len(s.subQuery)-1] = append(s.subQuery[len(s.subQuery)-1], sq) } -func (s *State) GetRandPrepare() *Prepare { - return s.prepareStmts[rand.Intn(len(s.prepareStmts))] -} - func (ts Tables) Copy() Tables { newTables := make(Tables, len(ts)) for i := range ts { @@ -354,11 +349,3 @@ func (i *Index) HasColumn(c *Column) bool { } return false } - -func (p *Prepare) UserVars() []string { - userVars := make([]string, len(p.Args)) - for i := 0; i < len(p.Args); i++ { - userVars[i] = fmt.Sprintf("@i%d", i) - } - return userVars -} diff --git a/sqlgenerator/db_type.go b/sqlgenerator/db_type.go index fb4c14b..7c38d9e 100644 --- a/sqlgenerator/db_type.go +++ b/sqlgenerator/db_type.go @@ -21,7 +21,7 @@ type State struct { env *Env - prepareStmts []*Prepare + prepareStmt *Prepare tableMeta []*model.TableInfo @@ -78,12 +78,6 @@ type Index struct { ColumnPrefix []int } -type Prepare struct { - ID int - Name string - Args []func() string -} - func NewState() *State { s := &State{ hooks: &Hooks{}, diff --git a/sqlgenerator/generator_lib.go b/sqlgenerator/generator_lib.go index 3cf9610..eb8dbd4 100644 --- a/sqlgenerator/generator_lib.go +++ b/sqlgenerator/generator_lib.go @@ -60,12 +60,14 @@ func Or(fns ...Fn) Fn { return NoneBecauseOf(fmt.Errorf("or exhausted")).Eval(state) } chosenFn := fns[chosenFnIdx] + l := state.RecordStack() rs, err := chosenFn.Eval(state) if err != nil { fnNames = append(fnNames, chosenFn.Info) errs = append(errs, err) fns[len(fns)-1], fns[chosenFnIdx] = fns[chosenFnIdx], fns[len(fns)-1] fns = fns[:len(fns)-1] + state.PopStack(l) continue } return rs, nil diff --git a/sqlgenerator/json.go b/sqlgenerator/json.go index e3fee78..581df32 100644 --- a/sqlgenerator/json.go +++ b/sqlgenerator/json.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "math/rand" - "time" ) // supported data types @@ -23,7 +22,7 @@ func wrapString(s string) string { return fmt.Sprintf("\"%s\"", s) } -func randomArrayJSONSubValue(dataType string) string { +func RandomValueWithType(dataType string) string { var val interface{} switch dataType { case dtSignedInt: @@ -50,8 +49,6 @@ func randomArrayJSONSubValue(dataType string) string { // randomArrayJSON returns a JSON array with random values of the given data types. func randomArrayJSON(size int, dataType string) (string, error) { - rand.Seed(time.Now().UnixNano()) - data := make([]interface{}, size) for i := range data { @@ -108,30 +105,3 @@ func randomBinary(length int) []byte { return b } - -// randomDate returns a random date string in yyyy-MM-dd format. -func randomDate() string { - min := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC).Unix() - max := time.Now().Unix() - randTime := time.Unix(rand.Int63n(max-min)+min, 0) - - return randTime.Format("2006-01-02") -} - -// randomDateTime returns a random datetime string in yyyy-MM-dd HH:mm:ss format. -func randomDateTime() string { - min := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC).Unix() - max := time.Now().Unix() - randTime := time.Unix(rand.Int63n(max-min)+min, 0) - - return randTime.Format("2006-01-02 15:04:05") -} - -// randomTime returns a random time string in HH:mm:ss format. -func randomTime() string { - min := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC).Unix() - max := time.Now().Unix() - randTime := time.Unix(rand.Int63n(max-min)+min, 0) - - return randTime.Format("15:04:05") -} diff --git a/sqlgenerator/json_test.go b/sqlgenerator/json_test.go index f91cc1f..0576f39 100644 --- a/sqlgenerator/json_test.go +++ b/sqlgenerator/json_test.go @@ -1,9 +1,10 @@ package sqlgenerator import ( + "testing" + "github.com/stretchr/testify/require" "golang.org/x/exp/rand" - "testing" ) func TestRandJsons(t *testing.T) { @@ -26,7 +27,7 @@ func TestRandJsons(t *testing.T) { }) t.Run("test rand json sub", func(t *testing.T) { for i := 0; i < 20; i++ { - str := randomArrayJSONSubValue(randArrayTp[rand.Intn(len(randArrayTp))]) + str := RandomValueWithType(randArrayTp[rand.Intn(len(randArrayTp))]) t.Log(str) } }) diff --git a/sqlgenerator/rule.go b/sqlgenerator/rule.go index b1c3a9f..9fdb749 100644 --- a/sqlgenerator/rule.go +++ b/sqlgenerator/rule.go @@ -411,6 +411,16 @@ var DropIndex = NewFn(func(state *State) Fn { return Strs("drop index", idx.Name) }) +var AddOrDropIndex = NewFn(func(state *State) Fn { + tbl := state.Tables.Rand() + state.env.Table = tbl + return And(Str("alter table"), Str(tbl.Name), + Or( + AddIndex, + DropIndex, + )) +}) + var AddColumn = NewFn(func(state *State) Fn { tbl := state.env.Table newCol := &Column{ID: state.alloc.AllocColumnID()} diff --git a/sqlgenerator/rule_prepare.go b/sqlgenerator/rule_prepare.go new file mode 100644 index 0000000..5dff049 --- /dev/null +++ b/sqlgenerator/rule_prepare.go @@ -0,0 +1,550 @@ +package sqlgenerator + +import ( + "context" + "database/sql" + "fmt" + "math" + "math/rand" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/PingCAP-QE/schrddl/util" + sqlutil "github.com/PingCAP-QE/schrddl/util" + "github.com/juju/errors" + "github.com/ngaut/log" +) + +func isIgnoredError(s string) bool { + return s == "" || + strings.Contains(s, "cannot be pushed down") || + strings.Contains(s, "Truncated incorrect") || + strings.Contains(s, "Duplicate entry") || + strings.Contains(s, "Incorrect datetime value") +} + +func CheckError(err1, err2 error) error { + if err1 == nil && err2 == nil { + return nil + } + + errStr1, errStr2 := "", "" + if err1 != nil { + errStr1 = err1.Error() + } + if err2 != nil { + errStr2 = err2.Error() + } + + // Filter some errors + if isIgnoredError(errStr1) && isIgnoredError(errStr2) { + return nil + } + + // Since the SQL is randomly generated, it may inherently fail to execute. + // In this case, we assume both sessions will return the same error. + if err1 == nil || err2 == nil { + log.Warn("Two sessions returns different error") + log.Warnf("Error1: %v", err1) + log.Warnf("Error2: %v", err2) + return errors.Errorf("Two sessions return different error") + } + + // TODO(joechenrh): pass dbname here + errStr2 = strings.ReplaceAll(errStr2, "testcache", "test") + if errStr1 != errStr2 { + log.Warn("Two sessions returns different error") + log.Warnf("Error1: %v", err1) + log.Warnf("Error2: %v", err2) + return errors.Errorf("Two sessions return different error") + } + + // Two sessions return same error, check if it can be ignroed. + if !util.DMLIgnoreError(err1) && !util.DDLIgnoreError(err1) { + log.Warnf("Two sessions get same error %v", err1) + return errors.Trace(err1) + } + return nil +} + +func checkLastUseCache(conn *sql.Conn) bool { + rows, err := util.FetchRowsWithConn(conn, "select @@last_plan_from_cache") + if err != nil { + log.Fatalf("Error fetch rows %v", err) + } + return rows[0][0] == "'1'" +} + +const Placeholder = "?" + +type ValueGenerator interface { + GenMismatch() string + GenMatch() string +} + +// ColummGenerator is used to generate random value based on given table and column. +type ColumnGenerator struct { + column *Column + table *Table +} + +func (g *ColumnGenerator) useTable() bool { + return len(g.table.Values) > 0 && rand.Intn(3) == 0 +} + +func (g *ColumnGenerator) GenMatch() string { + if g.useTable() { + if v := g.table.GetRandRowVal(g.column); len(v) != 0 { + return v + } + } + return g.column.RandomValue() +} + +func (g *ColumnGenerator) GenMismatch() (v string) { + return g.column.Tp.RandomMismatch() +} + +// ArrayValueGenerator is used to generate random value in column type. +type ArrayValueGenerator struct { + column *Column + table *Table +} + +func (g *ArrayValueGenerator) useTable() bool { + return len(g.table.Values) > 0 && rand.Intn(3) == 0 +} + +func (g *ArrayValueGenerator) GenMatch() string { + if g.useTable() { + if v := g.table.GetRandRowVal(g.column); len(v) != 0 { + return v + } + } + return RandomValueWithType(g.column.SubType) +} + +func (g *ArrayValueGenerator) GenMismatch() (v string) { + // We don't generate mismatch type of value in JSON + return g.GenMatch() +} + +// Currently, SimpleGenerator is only used in limit N. +type SimpleGenerator struct { + gen func() string +} + +func (g *SimpleGenerator) GenMatch() string { + return g.gen() +} + +func (g *SimpleGenerator) GenMismatch() string { + return g.gen() +} + +// RunAndCheckPlanCache checks whether this statement can be used in plan cache. +// +// Return: +// 1. whether this statement can use plan cache +// 2. error +func RunAndCheckPlanCache(sql string, db *sql.DB) (bool, error) { + // The generated statement may already have errors, check it first. + if _, err := db.Exec(sql); err != nil { + if sqlutil.DMLIgnoreError(err) { + return true, nil + } + return false, errors.Trace(err) + } + + rows, err := util.FetchRowsWithDB(db, "SHOW WARNINGS") + if err != nil { + return false, errors.Trace(err) + } + for _, row := range rows { + if strings.Contains(row[2], "skip plan-cache") { + return true, nil + } + } + + return false, nil +} + +type Prepare struct { + name string + generators []ValueGenerator + originalSQL string + prepareSQL string + executeSQL string + setSQL string + SQLNoCache string + setSQLs []string + + // error for no-cache execution and with-cache execution + err1 error + err2 error +} + +func NewPrepare() *Prepare { + return &Prepare{ + name: randomStringRunes(16, false), + } +} + +func (p *Prepare) RecordStack() int { + return len(p.generators) +} + +func (p *Prepare) PopStack(l int) { + if l < len(p.generators) { + p.generators = p.generators[:l] + } +} + +func (p *Prepare) RecordError(dir string) error { + dir = dir[8:] + filePath := filepath.Join(dir, "prepare.sql") + + content := fmt.Sprintf("SQLs:\n\n%s\n%s\n%s\nerr1:%v\nerr2:%v\n", + p.prepareSQL, p.executeSQL, p.SQLNoCache, p.err1, p.err2) + content += "\nParameters:\n\n" + for _, sql := range p.setSQLs { + content += sql + content += "\n" + } + + err := os.WriteFile(filePath, []byte(content), 0644) + if err != nil { + return fmt.Errorf("failed to write to file %s: %v", filePath, err) + } + + return nil +} + +func (p *Prepare) generateParams(genMatch bool) { + replacementsPairs := make([]string, 0, len(p.generators)) + setSQLs := make([]string, len(p.generators)) + for i, gen := range p.generators { + v := gen.GenMatch() + if !genMatch { + v = gen.GenMismatch() + } + setSQLs[i] = fmt.Sprintf("@i%d = %s", i, v) + replacementsPairs = append(replacementsPairs, v) + } + p.setSQL = fmt.Sprintf("set %s", strings.Join(setSQLs, ",")) + p.setSQLs = append(p.setSQLs, p.setSQL) + + var sb strings.Builder + replacementIndex := 0 + for _, char := range p.originalSQL { + if char == '?' { + sb.WriteString(replacementsPairs[replacementIndex]) + replacementIndex++ + } else { + sb.WriteRune(char) + } + } + p.SQLNoCache = sb.String() +} + +func getConn(db *sql.DB) (*sql.Conn, error) { + var ( + conn *sql.Conn + err error + ) + + conn, err = db.Conn(context.Background()) + if err != nil { + return nil, errors.Trace(err) + } + + maxRetries := 5 + for i := 0; i < maxRetries; i++ { + if err = conn.PingContext(context.Background()); err != nil { + conn.Close() + time.Sleep(10 * time.Millisecond) + conn, err = db.Conn(context.Background()) + if err != nil { + return nil, errors.Trace(err) + } + } else { + break + } + } + + return conn, errors.Trace(err) +} + +// Run the same query in two sessions, one with plan cache and one not. +// Return an error if two sessions return different errors. +func (p *Prepare) execAndCompare(dbWithCache, dbNoCache *sql.DB, genMatch bool, times int) error { + conn, err := getConn(dbWithCache) + if err != nil { + return errors.Trace(err) + } + defer conn.Close() + + // Prepare statement + err = util.ExecSQLWithConn(conn, p.prepareSQL) + if err != nil { + return err + } + + for i := 0; i < times; i++ { + p.generateParams(genMatch) + + // No cache + _, p.err1 = dbNoCache.Exec(p.SQLNoCache) + + // With cache + err := util.ExecSQLWithConn(conn, p.setSQL) + if err != nil { + return errors.Trace(err) + } + + _, p.err2 = conn.ExecContext(context.Background(), p.executeSQL) + if p.err2 != nil && (strings.Contains(p.err2.Error(), "invalid memory")) { + return errors.Trace(p.err2) + } + + lastExecuteUseCache := checkLastUseCache(conn) + + if err := CheckError(p.err1, p.err2); err != nil { + return errors.Trace(err) + } + + // Check cache used if no error. + if i > 0 && !lastExecuteUseCache { + skipCacheInExecute++ + } + } + + return nil +} + +var skipCacheInExecute = 0 + +// Run the same query in two sessions, one with plan cache and one not. +// Return an error if either: +// 1. two sessions return different errors. +// 2. two sessions return different results. +func (p *Prepare) queryAndCompare(dbWithCache, dbNoCache *sql.DB, genMatch bool, times int) error { + conn, err := getConn(dbWithCache) + if err != nil { + return errors.Trace(err) + } + defer conn.Close() + + // Prepare statement + err = util.ExecSQLWithConn(conn, p.prepareSQL) + if err != nil { + return err + } + + for i := 0; i < times; i++ { + p.generateParams(genMatch) + + // No cache + resWithoutCache, err1 := util.FetchRowsWithDB(dbNoCache, p.SQLNoCache) + p.err1 = err1 + + // With cache + err := util.ExecSQLWithConn(conn, p.setSQL) + if err != nil { + return errors.Trace(err) + } + + resWithCache, err2 := util.FetchRowsWithConn(conn, p.executeSQL) + if err2 != nil && (strings.Contains(err2.Error(), "invalid memory")) { + return errors.Trace(p.err2) + } + + lastExecuteUseCache := checkLastUseCache(conn) + p.err2 = err2 + + // If both executed successfully, check whether the results are same. + if err1 == nil && err2 == nil { + if i > 0 && !lastExecuteUseCache { + skipCacheInExecute++ + } + + same, err := util.CheckResults(resWithoutCache, resWithCache) + if err != nil { + return errors.Trace(err) + } + if !same { + return errors.Errorf("Two sessions' result mismatch") + } + return nil + } + + // Two sessions return different result + if err := CheckError(err1, err2); err != nil { + return errors.Trace(err) + } + + if i > 0 && !lastExecuteUseCache { + skipCacheInExecute++ + } + } + + return nil +} + +// Check whether this generated SQL can utilize plan cache +func (p *Prepare) UsePlanCache(dbWithCache *sql.DB) (bool, error) { + if len(p.generators) == 0 { + return false, nil + } + + skipped, err := RunAndCheckPlanCache(p.prepareSQL, dbWithCache) + return !skipped, err +} + +func (p *Prepare) CheckQuery(dbWithCache, dbNoCache *sql.DB) error { + err := p.queryAndCompare(dbWithCache, dbNoCache, true, 1) + if err != nil { + return errors.Trace(err) + } + + err = p.queryAndCompare(dbWithCache, dbNoCache, true, 10) + if err != nil { + return errors.Trace(err) + } + + return nil +} + +func (p *Prepare) CheckExec(dbWithCache, dbNoCache *sql.DB) error { + err := p.execAndCompare(dbWithCache, dbNoCache, true, 1) + if err != nil { + return errors.Trace(err) + } + + err = p.execAndCompare(dbWithCache, dbNoCache, true, 3) + if err != nil { + return errors.Trace(err) + } + + return nil +} + +// Use probability to determine whether to use placeholder +func (p *Prepare) CheckAdd() bool { + initialProb := 1.0 + decayFactor := 0.75 + + prob := initialProb * math.Pow(decayFactor, float64(len(p.generators))) + return rand.Float64() < prob +} + +func (p *Prepare) Add(gen ValueGenerator) { + p.generators = append(p.generators, gen) +} + +func (p *Prepare) SetPrepareSQL(sql string) { + p.originalSQL = sql + p.prepareSQL = fmt.Sprintf("prepare `%s` from %s", p.name, strconv.Quote(sql)) + vars := make([]string, len(p.generators)) + for i := range p.generators { + vars[i] = fmt.Sprintf("@i%d", i) + } + p.executeSQL = fmt.Sprintf("execute `%s` using %s", p.name, strings.Join(vars, ",")) +} + +func (s *State) EnablePrepare() { + s.prepareStmt = NewPrepare() +} + +func (s *State) PopPrepare() *Prepare { + p := s.prepareStmt + s.prepareStmt = nil + return p +} + +// Record current number of generators. +// It's used to pop generators after failure. +func (s *State) RecordStack() int { + if s.prepareStmt == nil { + return 0 + } + return s.prepareStmt.RecordStack() +} + +func (s *State) PopStack(l int) { + if s.prepareStmt == nil { + return + } + s.prepareStmt.PopStack(l) +} + +// GetValueFn is used to determine the return string. +// When we are generating a prepared statement, it will stores the generator and return a placeholder for later use. +// Otherwise, it directly gets a value from the generator and returns it. +func (s *State) GetValueFn(gen ValueGenerator) Fn { + if s.prepareStmt != nil && s.prepareStmt.CheckAdd() { + s.prepareStmt.Add(gen) + return Str(Placeholder) + } + + return Str(gen.GenMatch()) +} + +// Generate a random prepare statement +func GeneratePrepare(state *State, query bool) (*Prepare, error) { + state.EnablePrepare() + + var planCacheDML Fn + + if query { + planCacheDML = NewFn(func(state *State) Fn { + return Or( + SingleSelect.W(4), + MultiSelect.W(4), + UnionSelect.W(4), + MultiSelectWithSubQuery.W(4), + ) + }) + } else { + planCacheDML = NewFn(func(state *State) Fn { + return Or( + CommonInsertOrReplace.W(10), + CommonUpdate.W(10), + //CommonDelete.W(2), + ) + }) + } + + rawSQL, err := planCacheDML.Eval(state) + if err != nil { + return nil, err + } + + p := state.PopPrepare() + if strings.Count(rawSQL, "?") != len(p.generators) { + return nil, errors.Errorf("Generate prepare statement failed") + } + p.SetPrepareSQL(rawSQL) + + return p, nil +} + +var PlanCacheDataGen = NewFn(func(state *State) Fn { + return Or( + // Data preparation + CreateTable.W(20).P(NoTooMuchTables), + CommonInsertOrReplace.W(10).P(HasTables), + CommonUpdate.W(5).P(HasTables), + CommonDelete.W(5).P(HasTables), + // DDL + AlterTable.W(5).P(HasTables), + SplitRegion.W(1).P(HasTables), + SetTiFlashReplica.W(0).P(HasTables), + // Check + AdminCheck.W(1).P(HasTables), + AnalyzeTable.W(0).P(HasTables), + ) +}) diff --git a/sqlgenerator/rule_query.go b/sqlgenerator/rule_query.go index 17afefe..c728922 100644 --- a/sqlgenerator/rule_query.go +++ b/sqlgenerator/rule_query.go @@ -63,11 +63,22 @@ var UnionSelect = NewFn(func(state *State) Fn { if err != nil { return NoneBecauseOf(err) } + + orderString := func(n int) string { + numbers := make([]string, n) + for i := 1; i <= n; i++ { + numbers[i-1] = fmt.Sprintf("%d", i) + } + return strings.Join(numbers, ",") + }(fieldNum) + orderString = fmt.Sprintf("order by %s", orderString) + return Strs( "(", firstSelect, ")", setOpr, "(", secondSelect, ")", - "order by 1 limit", RandomNum(1, 1000), + orderString, + "limit", RandomNum(1, 1000), ) }) @@ -580,19 +591,26 @@ var Predicate = NewFn(func(state *State) Fn { ) }) +var JSONContainVal = NewFn(func(state *State) Fn { + v, err := ArrayRandVal.Eval(state) + if err != nil { + return NoneBecauseOf(err) + } + // Don't add quote to placeholder + if v == Placeholder { + return Str(v) + } + return Str(fmt.Sprintf("'%s'", strings.Trim(v, "'"))) +}) + var JSONPredicate = NewFn(func(state *State) Fn { tbl := state.env.Table randCol := state.env.Column colName := fmt.Sprintf("%s.%s", tbl.Name, randCol.Name) - arv, err := ArrayRandVal.Eval(state) - if err != nil { - return NoneBecauseOf(err) - } - jsContainVal := "'" + strings.Trim(arv, "'") + "'" pre := Or( - And(Str(arv), Str("MEMBER OF"), Str("("), Str(colName), Str(")")), - And(Str("JSON_CONTAINS("), Str(colName), Str(","), Str(jsContainVal), Str(")")), + And(ArrayRandVal, Str("MEMBER OF"), Str("("), Str(colName), Str(")")), + And(Str("JSON_CONTAINS("), Str(colName), Str(","), JSONContainVal, Str(")")), //And(Str("JSON_CONTAINS("), ArrayRandVal, Str(","), Str(colName), Str(")")), And(Str("JSON_OVERLAPS("), Str(colName), Str(","), RandVal, Str(")")), //And(Str("JSON_OVERLAPS("), RandVal, Str(","), Str(colName), Str(")")), @@ -619,33 +637,13 @@ var RandColVals = NewFn(func(state *State) Fn { }) var ArrayRandVal = NewFn(func(state *State) Fn { - tbl := state.env.Table - randCol := state.env.Column - var v string - if len(tbl.Values) == 0 || rand.Intn(3) == 0 { - v = randomArrayJSONSubValue(randCol.SubType) - } else { - v = tbl.GetRandArraySubVal(randCol) - } - if len(v) == 0 { - v = randomArrayJSONSubValue(randCol.SubType) - } - return Str(v) + gen := &ArrayValueGenerator{table: state.env.Table, column: state.env.Column} + return state.GetValueFn(gen) }) var RandVal = NewFn(func(state *State) Fn { - tbl := state.env.Table - randCol := state.env.Column - var v string - if len(tbl.Values) == 0 || rand.Intn(3) == 0 { - v = randCol.RandomValue() - } else { - v = tbl.GetRandRowVal(randCol) - } - if len(v) == 0 { - v = randCol.RandomValue() - } - return Str(v) + gen := &ColumnGenerator{table: state.env.Table, column: state.env.Column} + return state.GetValueFn(gen) }) var SubSelect = NewFn(func(state *State) Fn { @@ -734,9 +732,10 @@ var OrderBy = NewFn(func(state *State) Fn { }) var Limit = NewFn(func(state *State) Fn { - return Strs("limit", RandomNum(100000000, 1000000000)) - - //return Strs("limit", RandomNum(1000000, 2147483646)) + gen := &SimpleGenerator{gen: func() string { + return Num(rand.Intn(900000000) + 100000000) + }} + return And(Strs("limit"), state.GetValueFn(gen)) }) var Query2 Fn diff --git a/sqlgenerator/value_util.go b/sqlgenerator/value_util.go new file mode 100644 index 0000000..5a07eb6 --- /dev/null +++ b/sqlgenerator/value_util.go @@ -0,0 +1,79 @@ +package sqlgenerator + +import ( + "fmt" + "math/rand" + "time" +) + +var asciiRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789~!@#$%^&*()_+=-") + +// random choose from slice +func choose[T any](slice []T) T { + return slice[rand.Intn(len(slice))] +} + +func randomCNChar() rune { + lower, upper := int('\u4e00'), int('\u9fff') + return rune(lower + rand.Intn(upper-lower)) +} + +func randomStringRunes(n int, mixCNChar bool) string { + b := make([]rune, n) + for i := range b { + b[i] = choose(asciiRunes) + if mixCNChar && rand.Intn(3) == 0 { + b[i] = randomCNChar() + } + } + return string(b) +} + +const ( + yearFormat = "2006" + dateFormat = "2006-01-01" + timeFormat = "11:11:11.00" + dateTimeFormat = "2006-01-02 15:04:05" +) + +func randomGoTime() time.Time { + min := time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC).Unix() + max := time.Now().Unix() + return time.Unix(rand.Int63n(max-min)+min, 0) +} + +func randomTimes(count int, format string) []string { + result := make([]string, count) + for i := 0; i < count; i++ { + result[i] = fmt.Sprintf("'%s'", randomGoTime().Format(format)) + } + return result +} + +func randomYear() string { + return randomGoTime().Format(yearFormat) +} + +func RandomYears(count int) []string { + return randomTimes(count, yearFormat) +} + +func randomDate() string { + return randomGoTime().Format(dateFormat) +} + +func RandomDates(count int) []string { + return randomTimes(count, dateFormat) +} + +func randomTime() string { + return randomGoTime().Format(timeFormat) +} + +func RandomTimes(count int) []string { + return randomTimes(count, timeFormat) +} + +func randomDateTime() string { + return randomGoTime().Format(dateTimeFormat) +} diff --git a/util/errors.go b/util/errors.go new file mode 100644 index 0000000..9ef51fb --- /dev/null +++ b/util/errors.go @@ -0,0 +1,262 @@ +package util + +import ( + "fmt" + "regexp" + "strings" + + "github.com/ngaut/log" +) + +var dmlIgnoreList = []string{ + "Table has no partition", + "can't have a default value", + "Invalid JSON bytes", + "Invalid JSON data provided to function", + "Invalid JSON value for CAST", + "Invalid JSON text", + "Data too long", + "character string", + + // bug + "slice bounds out of range", + "index out of range", + "writing inconsistent data in table", + "should ensure all columns have the same length", + "expected integer", + // "invalid memory address or nil pointer dereference", + "encoding failed", + "invalid input value", + "region not found for key", + "Unsupported expression type MysqlBit", + "Unexpected missing column", + "strconv.Atoi", + //"other error for mpp stream", + + "Can't find a proper physical plan for this query", + "Your query has been cancelled due to exceeding the allowed memory limit", + "Cant peek from empty bytes", + + "maximum statement execution time exceeded", + "please skip this plan", + + "Subquery returns more", + "Some rows were cut by", + "invalid data type: Illegal Json text", + "Data Truncated", + + // unknown + "context canceled", + "cannot be pushed down", +} + +var ddlIgnoreList = []string{ + "already exists", + "A PRIMARY must include all columns", + "has an expression index dependency and cannot", + "Multiple definition of same constant", + "VALUES LESS THAN value must be strictly increasing for each partition", + "please split table instead", + "should less than the upper value", + "A primary key index cannot be invisible", + "Unsupported modify change collate", + "Failed to split region ranges: the region size is too small", + "Can't find dropped/truncated table", + "Can't find localTemporary/dropped/truncated", + "can't be flashback repeatedly", + "Invalid gbk character string", + "secondary index", + "cannot be used in key specification", + "Adding clustered primary key", + "Invalid use of NULL value", + "can not get 'tikv_gc_safe_point'", + "maximum statement execution time exceeded", + "Illegal mix of collations for operation", + "Unsupported Global Index", + "settings for table contains gbk charset", +} + +var RCIsolation = false + +func SetRCIsolation(isolation bool) { + RCIsolation = isolation +} + +func GetRCIsolation() bool { + return RCIsolation +} + +func DMLIgnoreError(err error) bool { + if err == nil { + return true + } + errStr := err.Error() + for _, ignore := range dmlIgnoreList { + if strings.Contains(errStr, ignore) { + return true + } + } + if strings.Contains(errStr, "Information schema is changed") && !RCIsolation { + return true + } + if strings.Contains(errStr, "try again later") { + return true + } + // Sometimes, there might be duplicated entry error caused by concurrent. + // So we ignore here. + if strings.Contains(errStr, "Duplicate entry") { + return true + } + // Sometimes, a insert to a table might generate an error caused by exceeding maximum auto increment id, + // we ignore this error here. + if strings.Contains(errStr, "Failed to read auto-increment value from storage engine") { + return true + } + if strings.Contains(errStr, "invalid connection") { + return true + } + if strings.Contains(errStr, "doesn't exist") || + strings.Contains(errStr, "column is deleted") || strings.Contains(errStr, "Can't find column") || + strings.Contains(errStr, "converting driver.Value type") || strings.Contains(errStr, "column specified twice") || + strings.Contains(errStr, "Out of range value for column") || strings.Contains(errStr, "Unknown column") || + strings.Contains(errStr, "column has index reference") || strings.Contains(errStr, "Data too long for column") || + strings.Contains(errStr, "Data truncated") || strings.Contains(errStr, "no rows in result set") || + strings.Contains(errStr, "Truncated incorrect") || strings.Contains(errStr, "Data truncated for column") || + // eg: For Incorrect tinyint value, Incorrect data value... + strings.Contains(errStr, "Incorrect") || + // eg: For constant 20030522161944 overflows tinyint + strings.Contains(errStr, "overflows") || + strings.Contains(errStr, "Bad Number") || + strings.Contains(errStr, "invalid year") || + strings.Contains(errStr, "value is out of range in") || + strings.Contains(errStr, "Data Too Long") || + strings.Contains(errStr, "doesn't have a default value") || + strings.Contains(errStr, "specified twice") || + strings.Contains(errStr, "cannot convert datum from") || + strings.Contains(errStr, "sql_mode=only_full_group_by") || + strings.Contains(errStr, "cannot be null") || + strings.Contains(errStr, "Column count doesn't match value count") || + strings.Contains(errStr, "Percentage value") || + strings.Contains(errStr, "Index column") || + strings.Contains(errStr, "Illegal mix of collations") || + strings.Contains(errStr, "Cannot convert string") || + strings.Contains(errStr, "interface conversion") || + strings.Contains(errStr, "connection is already closed") || + strings.Contains(errStr, "should contain a UNION") || + strings.Contains(errStr, "have different column counts") || + strings.Contains(errStr, "followed by one or more recursive ones") || + strings.Contains(errStr, "Not unique table/alias") || + strings.Contains(errStr, "have a different number of columns") || + strings.Contains(errStr, "Split table region lower value count") || + strings.Contains(errStr, "Out Of Memory") || + strings.Contains(errStr, "invalid syntax") || + strings.Contains(errStr, "newer than query schema version") || + strings.Contains(errStr, "PD server timeout") || + strings.Contains(errStr, "Information schema is out of date") || + strings.Contains(errStr, "Your query has been cancelled due to exceeding the allowed memory limit for a single SQL query") || + strings.Contains(errStr, "Value is out of range") { + return true + } + // PlanCache related errors + if strings.Contains(errStr, "Illegal mix of collations") || + strings.Contains(errStr, "SQL syntax") || + strings.Contains(errStr, "Unsupported type: Set") || + strings.Contains(errStr, "Invalid data type for JSON data in argument 2 to function json_contains") || + strings.Contains(errStr, "Prepared statement not found") { + return true + } + if strings.Contains(errStr, "Unsupported multi schema change") { + return true + } + if !RCIsolation && strings.Contains(errStr, "public column") { + return true + } + return false +} + +func DDLIgnoreError(err error) bool { + if err == nil { + return true + } + errStr := err.Error() + log.Warnf("check DDL err:%s", errStr) + for _, ignore := range ddlIgnoreList { + if strings.Contains(errStr, ignore) { + return true + } + } + if strings.Contains(errStr, "Information schema is changed") { + return true + } + // Sometimes, set shard row id bits to a large value might cause global auto ID overflow error. + // We ignore this error here. + if match, _ := regexp.MatchString(`cause next global auto ID( \d+ | )overflow`, errStr); match { + return true + } + if strings.Contains(errStr, "invalid connection") { + return true + } + if strings.Contains(errStr, "Unsupported shard_row_id_bits for table with primary key as row id") { + return true + } + // Ignore Column Type Change error. + if strings.Contains(errStr, "Unsupported modify column") || + strings.Contains(errStr, "Cancelled DDL job") || + strings.Contains(errStr, "Truncated incorrect") || + strings.Contains(errStr, "overflows") || + strings.Contains(errStr, "Invalid year value") || + strings.Contains(errStr, "Incorrect time value") || + strings.Contains(errStr, "Incorrect datetime value") || + strings.Contains(errStr, "Incorrect timestamp value") || + strings.Contains(errStr, "All parts of a PRIMARY KEY must be NOT NULL") || + strings.Contains(errStr, "value is out of range") || + strings.Contains(errStr, "Unsupported modify charset from") || + strings.Contains(errStr, "Unsupported modifying collation of column") || + strings.Contains(errStr, "Data truncated") || + strings.Contains(errStr, "Bad Number") || + strings.Contains(errStr, "cannot convert") || + strings.Contains(errStr, "Data Too Long") || + // eg: For v"BLOB/TEXT column '319de167-6d2e-4778-966c-60b95103a02c' used in key specification without a key length" + strings.Contains(errStr, "used in key specification without a key length") || + strings.Contains(errStr, "Specified key was too long; max key length is ") || + strings.Contains(errStr, "should be less than the total tiflash server count") || + strings.Contains(errStr, "Unsupported ALTER TiFlash settings") { + fmt.Println(errStr) + return true + } + if strings.Contains(errStr, "table doesn't exist") || + strings.Contains(errStr, "doesn't have a default value") || + strings.Contains(errStr, "with composite index covered or Primary Key covered now") || + strings.Contains(errStr, "does not exist, this column may have been updated by other DDL") || + strings.Contains(errStr, "is not exists") || strings.Contains(errStr, "column does not exist") || + strings.Contains(errStr, "doesn't exist") || strings.Contains(errStr, "Unknown table") || + strings.Contains(errStr, "admin show ddl jobs len != len(tasks)") || + strings.Contains(errStr, "check that column/key exists") || + strings.Contains(errStr, "Invalid default value") || + strings.Contains(errStr, "Duplicate column name") || + strings.Contains(errStr, "can't drop only column") || + strings.Contains(errStr, "doesn't exist") || strings.Contains(errStr, "not found") || + strings.Contains(errStr, "column is deleted") || strings.Contains(errStr, "Can't find column") || + strings.Contains(errStr, "converting driver.Value type") || strings.Contains(errStr, "column specified twice") || + strings.Contains(errStr, "Out of range value for column") || strings.Contains(errStr, "Unknown column") || + strings.Contains(errStr, "column has index reference") || strings.Contains(errStr, "Data too long for column") || + strings.Contains(errStr, "Data truncated") || strings.Contains(errStr, "no rows in result set") || + strings.Contains(errStr, "with tidb_enable_change_multi_schema is disable") || + strings.Contains(errStr, "not allowed type for this type of partitioning") || + strings.Contains(errStr, "A PRIMARY KEY must include all columns in the table's partitioning function") || + strings.Contains(errStr, "A UNIQUE INDEX must include all columns in the table's partitioning function") || + strings.Contains(errStr, "cannot convert datum") || + strings.Contains(errStr, "Duplicate entry") || + strings.Contains(errStr, "has a partitioning function dependency and cannot be dropped or renamed") || + strings.Contains(errStr, "A CLUSTERED INDEX must include all columns in the table's partitioning function") || + strings.Contains(errStr, "PD server timeout") || + strings.Contains(errStr, "Information schema is out of date") || + strings.Contains(errStr, "Invalid JSON data provided") || + strings.Contains(errStr, "Invalid JSON value for CAST") || + strings.Contains(errStr, "Invalid JSON text") || + strings.Contains(errStr, "since the unique index is not including all partitioning columns, and GLOBAL is not given as IndexOption") || + strings.Contains(errStr, "doesn't yet support") { + return true + } + return false +} diff --git a/framework/run_test.go b/util/errors_test.go similarity index 64% rename from framework/run_test.go rename to util/errors_test.go index 87c439e..95a1a25 100644 --- a/framework/run_test.go +++ b/util/errors_test.go @@ -1,4 +1,4 @@ -package framework +package util import ( "testing" @@ -9,9 +9,9 @@ import ( func TestDDLIgnoreError(t *testing.T) { err := errors.New("shard_row_id_bits 6 will cause next global auto ID 1648238210354062187 overflow random") - assert.True(t, ddlIgnoreError(err)) + assert.True(t, DDLIgnoreError(err)) err = errors.New("shard_row_id_bits 6 will cause next global auto ID 164823821035406218d overflow") - assert.False(t, ddlIgnoreError(err)) - assert.True(t, ddlIgnoreError(errors.New("cause next global auto ID 92738 overflow error"))) - assert.True(t, ddlIgnoreError(errors.New("cause next global auto ID overflow error"))) + assert.False(t, DDLIgnoreError(err)) + assert.True(t, DDLIgnoreError(errors.New("cause next global auto ID 92738 overflow error"))) + assert.True(t, DDLIgnoreError(errors.New("cause next global auto ID overflow error"))) } diff --git a/util/sql_util.go b/util/sql_util.go new file mode 100644 index 0000000..be36a63 --- /dev/null +++ b/util/sql_util.go @@ -0,0 +1,192 @@ +package util + +import ( + "context" + "database/sql" + "fmt" + "sort" + "strings" + + "github.com/juju/errors" + "github.com/ngaut/log" +) + +// Execute SQL with no returning rows. +// This function will check if the error can be ignored. +func ExecSQLWithConn(conn *sql.Conn, sql string) error { + _, err := conn.ExecContext(context.Background(), sql) + if err != nil && DMLIgnoreError(err) || DDLIgnoreError(err) { + return nil + } + if strings.Contains(err.Error(), "plan not match") { + _, err = conn.ExecContext(context.Background(), sql) + return err + } + return errors.Trace(err) +} + +// Execute SQL with no returning rows. +// This function will check if the error can be ignored. +func ExecSQLWithDB(db *sql.DB, sql string) error { + _, err := db.Exec(sql) + if err != nil && DMLIgnoreError(err) || DDLIgnoreError(err) { + return nil + } + if strings.Contains(err.Error(), "plan not match") { + _, err = db.Exec(sql) + return err + } + return errors.Trace(err) +} + +// Execute SQL and return all the rows. +func FetchRowsWithConn(conn *sql.Conn, sql string) ([][]string, error) { + rows, err := conn.QueryContext(context.Background(), sql) + if err != nil { + return nil, err + } + + defer func() { + rows.Close() + }() + + // Read all rows. + var actualRows [][]string + for rows.Next() { + cols, err := rows.Columns() + if err != nil { + return nil, err + } + + // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go + rawResult := make([][]byte, len(cols)) + result := make([]string, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range rawResult { + dest[i] = &rawResult[i] + } + + err = rows.Scan(dest...) + if err != nil { + return nil, err + } + + for i, raw := range rawResult { + if raw == nil { + result[i] = "NULL" + } else { + result[i] = fmt.Sprintf("'%s'", string(raw)) + } + } + + actualRows = append(actualRows, result) + } + if rows.Err() != nil { + return nil, rows.Err() + } + + return actualRows, nil +} + +// Execute SQL and return all the rows. +func FetchRowsWithDB(db *sql.DB, sql string) ([][]string, error) { + rows, err := db.Query(sql) + if err != nil { + return nil, err + } + + defer func() { + rows.Close() + }() + + // Read all rows. + var actualRows [][]string + for rows.Next() { + cols, err := rows.Columns() + if err != nil { + return nil, err + } + + // See https://stackoverflow.com/questions/14477941/read-select-columns-into-string-in-go + rawResult := make([][]byte, len(cols)) + result := make([]string, len(cols)) + dest := make([]interface{}, len(cols)) + for i := range rawResult { + dest[i] = &rawResult[i] + } + + err = rows.Scan(dest...) + if err != nil { + return nil, err + } + + for i, raw := range rawResult { + if raw == nil { + result[i] = "NULL" + } else { + result[i] = fmt.Sprintf("'%s'", string(raw)) + } + } + + actualRows = append(actualRows, result) + } + if rows.Err() != nil { + return nil, rows.Err() + } + + return actualRows, nil +} + +// CheckResults checks whether two result sets are same. +func CheckResults(res1, res2 [][]string) (bool, error) { + if len(res1) != len(res2) { + return false, nil + } + sort.Slice(res1, func(i, j int) bool { + for c := range res1[i] { + if res1[i][c] != res1[j][c] { + return res1[i][c] < res1[j][c] + } + } + return true + }) + sort.Slice(res2, func(i, j int) bool { + for c := range res2[i] { + if res2[i][c] != res2[j][c] { + return res2[i][c] < res2[j][c] + } + } + return true + }) + for i, row1 := range res1 { + row2 := res2[i] + if len(row1) != len(row2) { + return false, nil + } + for j, e := range row1 { + if e != row2[j] { + return false, nil + } + } + } + return true, nil +} + +func CheckTableData(db1, tb1, db2, tb2 string, db *sql.DB) (bool, error) { + sql := fmt.Sprintf("select * from `%s`.`%s`", db1, tb1) + rows1, err := FetchRowsWithDB(db, sql) + if err != nil { + return false, errors.Trace(err) + } + sql = fmt.Sprintf("select * from `%s`.`%s`", db2, tb2) + rows2, err := FetchRowsWithDB(db, sql) + if err != nil { + return false, errors.Trace(err) + } + same, err := CheckResults(rows1, rows2) + if err != nil { + log.Fatalf("Error check result") + } + + return same, nil +}