statistics: use PrevLastVersion to avoid error. (#3060)

Currently, we use startTS of a txn as the version of a stats table. There may be an error like this:
txn1 with version 2 write firstly.
TiDB update it, and the latest udpate version is 2.
then txn2 with version 1 write.
TiDB will never get it with the version 2.

So we choose to get the table stats which version is greator than the version before last two lease.

statistics/statscache.go

@@ -30,9 +30,15 @@ type statsCache map[int64]*Table
 
 // Handle can update stats info periodically.
 type Handle struct {
-	ctx         context.Context
-	lastVersion uint64
-	statsCache  atomic.Value
+	ctx context.Context
+	// LastVersion is the latest update version before last lease. Exported for test.
+	LastVersion uint64
+	// PrevLastVersion is the latest update version before two lease. Exported for test.
+	// We need this because for two tables, the smaller version may write later than the one with larger version.
+	// We can read the version with lastTwoVersion if the diff between commit time and version is less than one lease.
+	// PrevLastVersion will be assigned by LastVersion every time Update is called.
+	PrevLastVersion uint64
+	statsCache      atomic.Value
 	// ddlEventCh is a channel to notify a ddl operation has happened. It is sent only by owner and read by stats handle.
 	ddlEventCh chan *ddl.Event
 
@@ -45,7 +51,8 @@ type Handle struct {
 // Clear the statsCache, only for test.
 func (h *Handle) Clear() {
 	h.statsCache.Store(statsCache{})
-	h.lastVersion = 0
+	h.LastVersion = 0
+	h.PrevLastVersion = 0
 }
 
 // NewHandle creates a Handle for update stats.
@@ -62,11 +69,12 @@ func NewHandle(ctx context.Context) *Handle {
 
 // Update reads stats meta from store and updates the stats map.
 func (h *Handle) Update(is infoschema.InfoSchema) error {
-	sql := fmt.Sprintf("SELECT version, table_id, count from mysql.stats_meta where version > %d order by version", h.lastVersion)
+	sql := fmt.Sprintf("SELECT version, table_id, count from mysql.stats_meta where version > %d order by version", h.PrevLastVersion)
 	rows, _, err := h.ctx.(sqlexec.RestrictedSQLExecutor).ExecRestrictedSQL(h.ctx, sql)
 	if err != nil {
 		return errors.Trace(err)
 	}
+	h.PrevLastVersion = h.LastVersion
 	tables := make([]*Table, 0, len(rows))
 	for _, row := range rows {
 		version, tableID, count := row.Data[0].GetUint64(), row.Data[1].GetInt64(), row.Data[2].GetInt64()
@@ -83,7 +91,7 @@ func (h *Handle) Update(is infoschema.InfoSchema) error {
 			continue
 		}
 		tables = append(tables, tbl)
-		h.lastVersion = version
+		h.LastVersion = version
 	}
 	h.updateTableStats(tables)
 	return nil

statistics/statscache_test.go

@@ -242,10 +242,75 @@ func (s *testStatsCacheSuite) TestDDL(c *C) {
 	err = h.HandleDDLEvent(<-h.DDLEventCh())
 	c.Assert(err, IsNil)
 	h.Update(is)
-	statsTbl := do.StatsHandle().GetTableStats(tableInfo)
+	statsTbl := h.GetTableStats(tableInfo)
 	c.Assert(statsTbl.Pseudo, IsFalse)
 }
 
+func (s *testStatsCacheSuite) TestVersion(c *C) {
+	store, do, err := newStoreWithBootstrap()
+	c.Assert(err, IsNil)
+	defer store.Close()
+	testKit := testkit.NewTestKit(c, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("create table t1 (c1 int, c2 int)")
+	testKit.MustExec("analyze table t1")
+	is := do.InfoSchema()
+	tbl1, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	c.Assert(err, IsNil)
+	tableInfo1 := tbl1.Meta()
+	h := statistics.NewHandle(testKit.Se)
+	testKit.MustExec("update mysql.stats_meta set version = 2 where table_id = ?", tableInfo1.ID)
+
+	h.Update(is)
+	c.Assert(h.LastVersion, Equals, uint64(2))
+	c.Assert(h.PrevLastVersion, Equals, uint64(0))
+	statsTbl1 := h.GetTableStats(tableInfo1)
+	c.Assert(statsTbl1.Pseudo, IsFalse)
+
+	testKit.MustExec("create table t2 (c1 int, c2 int)")
+	testKit.MustExec("analyze table t2")
+	is = do.InfoSchema()
+	tbl2, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("t2"))
+	c.Assert(err, IsNil)
+	tableInfo2 := tbl2.Meta()
+	// A smaller version write, and we can still read it.
+	testKit.MustExec("update mysql.stats_meta set version = 1 where table_id = ?", tableInfo2.ID)
+	h.Update(is)
+	c.Assert(h.LastVersion, Equals, uint64(2))
+	c.Assert(h.PrevLastVersion, Equals, uint64(2))
+	statsTbl2 := h.GetTableStats(tableInfo2)
+	c.Assert(statsTbl2.Pseudo, IsFalse)
+
+	testKit.MustExec("insert t1 values(1,2)")
+	testKit.MustExec("analyze table t1")
+	testKit.MustExec("update mysql.stats_meta set version = 4 where table_id = ?", tableInfo1.ID)
+	h.Update(is)
+	c.Assert(h.LastVersion, Equals, uint64(4))
+	c.Assert(h.PrevLastVersion, Equals, uint64(2))
+	statsTbl1 = h.GetTableStats(tableInfo1)
+	c.Assert(statsTbl1.Count, Equals, int64(1))
+
+	testKit.MustExec("insert t2 values(1,2)")
+	testKit.MustExec("analyze table t2")
+	// A smaller version write, and we can still read it.
+	testKit.MustExec("update mysql.stats_meta set version = 3 where table_id = ?", tableInfo2.ID)
+	h.Update(is)
+	c.Assert(h.LastVersion, Equals, uint64(4))
+	c.Assert(h.PrevLastVersion, Equals, uint64(4))
+	statsTbl2 = h.GetTableStats(tableInfo2)
+	c.Assert(statsTbl2.Count, Equals, int64(1))
+
+	testKit.MustExec("insert t2 values(1,2)")
+	testKit.MustExec("analyze table t2")
+	// A smaller version write, and we cannot read it. Because at this time, lastTwo Version is 4.
+	testKit.MustExec("update mysql.stats_meta set version = 3 where table_id = ?", tableInfo2.ID)
+	h.Update(is)
+	c.Assert(h.LastVersion, Equals, uint64(4))
+	c.Assert(h.PrevLastVersion, Equals, uint64(4))
+	statsTbl2 = h.GetTableStats(tableInfo2)
+	c.Assert(statsTbl2.Count, Equals, int64(1))
+}
+
 func (s *testStatsCacheSuite) TestLoadHist(c *C) {
 	store, do, err := newStoreWithBootstrap()
 	c.Assert(err, IsNil)