Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

infoschema: maintain the snapshotTS in the infocache #25322

Merged
merged 4 commits into from
Jun 11, 2021
Merged

infoschema: maintain the snapshotTS in the infocache #25322

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5b41099
infoschema: maintain the snapshotTS in the infocache
nolouch Jun 10, 2021
d827fe3
address comments
nolouch Jun 10, 2021
fd16fb0
address comments
nolouch Jun 11, 2021
7a15b31
Merge branch 'master' into improve-schema-cache
ti-chi-bot Jun 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ddl/ddl_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ func TestT(t *testing.T) {
func testNewDDLAndStart(ctx context.Context, c *C, options ...Option) *ddl {
// init infoCache and a stub infoSchema
ic := infoschema.NewCache(2)
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
options = append(options, WithInfoCache(ic))
d := newDDL(ctx, options...)
err := d.Start(nil)
Expand Down
4 changes: 2 additions & 2 deletions ddl/util/syncer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ func TestSyncerSimple(t *testing.T) {
cli := clus.RandClient()
ctx := goctx.Background()
ic := infoschema.NewCache(2)
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d := NewDDL(
ctx,
WithEtcdClient(cli),
Expand Down Expand Up @@ -115,7 +115,7 @@ func TestSyncerSimple(t *testing.T) {
}

ic2 := infoschema.NewCache(2)
ic2.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic2.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d1 := NewDDL(
ctx,
WithEtcdClient(cli),
Expand Down
8 changes: 6 additions & 2 deletions domain/domain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ func (do *Domain) loadInfoSchema(startTS uint64) (infoschema.InfoSchema, bool, i
if currentSchemaVersion != 0 && neededSchemaVersion > currentSchemaVersion && neededSchemaVersion-currentSchemaVersion < 100 {
is, relatedChanges, err := do.tryLoadSchemaDiffs(m, currentSchemaVersion, neededSchemaVersion)
if err == nil {
do.infoCache.Insert(is)
do.infoCache.Insert(is, startTS)
logutil.BgLogger().Info("diff load InfoSchema success",
zap.Int64("currentSchemaVersion", currentSchemaVersion),
zap.Int64("neededSchemaVersion", neededSchemaVersion),
Expand Down Expand Up @@ -158,7 +158,7 @@ func (do *Domain) loadInfoSchema(startTS uint64) (infoschema.InfoSchema, bool, i
zap.Duration("start time", time.Since(startTime)))

is := newISBuilder.Build()
do.infoCache.Insert(is)
do.infoCache.Insert(is, startTS)
return is, false, currentSchemaVersion, nil, nil
}

Expand Down Expand Up @@ -290,6 +290,10 @@ func (do *Domain) InfoSchema() infoschema.InfoSchema {

// GetSnapshotInfoSchema gets a snapshot information schema.
func (do *Domain) GetSnapshotInfoSchema(snapshotTS uint64) (infoschema.InfoSchema, error) {
// if the snapshotTS is new enough, we can get infoschema directly through sanpshotTS.
if is := do.infoCache.GetBySnapshotTS(snapshotTS); is != nil {
return is, nil
}
Comment on lines 292 to +296
Copy link
Contributor

@Yisaer Yisaer Jun 11, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this may cause the infoschema inconsistency in following case.

1. infoschema cache holding `t1` as maxUpdatedSnapshotTS and infoschema for `schema version` v1
2. ddl worker running ddl job with `t2` in txn and increased `schema version` to v2 while infoschema cache have not been awared of this changing yet.
3. load infoschema with snapshotTS t3 (t3 > t2 > t1) and using infoschema with v1 schema version.

Copy link
Contributor

@xhebox xhebox Jun 11, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Infocache is used everywhere, it is promised to sync the latest infoschema for the current tidb instance(not necessarily the latest for all tidb instances, actually ddl owner tidb can not immediately get the latest, too).

And of course, there are edge cases. It is indeed not a 100% consistent method. But it is also impossible to avoid kvget by promising 100% consistency.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In my example, I think we don't need to guarantee that the query must use the schema version v2's infoschema if the changing is not related to the tables we visit.

But I think the following checking is ncessary:

  1. check whether there exists any schema version changed bewtween t1(maxUpdatedSnapshotTS) and t3(snapshotTS)
  2. If exists, check whether the changing is related to the table we vistied.
  3. If related, reloading the infoschema with t3 snapshotTS to use the right infoschema.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In my example, I think we don't need to guarantee that the query must use the schema version v2's infoschema if the changing is not related to the tables we visit.

But I think the following checking is necessary:

  1. check whether there exists any schema version changed bewtween t1(maxUpdatedSnapshotTS) and t3(snapshotTS)
  2. If exists, check whether the changing is related to the table we vistied.
  3. If related, reloading the infoschema with t3 snapshotTS to use the right infoschema.

is, _, _, _, err := do.loadInfoSchema(snapshotTS)
return is, err
}
Expand Down
33 changes: 28 additions & 5 deletions infoschema/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@ type InfoCache struct {
mu sync.RWMutex
// cache is sorted by SchemaVersion in descending order
cache []InfoSchema
// record SnapshotTS of the latest schema Insert.
maxUpdatedSnapshotTS uint64
}

// NewCache creates a new InfoCache.
Expand All @@ -38,9 +40,9 @@ func NewCache(capcity int) *InfoCache {
func (h *InfoCache) GetLatest() InfoSchema {
h.mu.RLock()
defer h.mu.RUnlock()
metrics.InfoCacheCounters.WithLabelValues("get").Inc()
metrics.InfoCacheCounters.WithLabelValues("get", "latest").Inc()
if len(h.cache) > 0 {
metrics.InfoCacheCounters.WithLabelValues("hit").Inc()
metrics.InfoCacheCounters.WithLabelValues("hit", "latest").Inc()
return h.cache[0]
}
return nil
Expand All @@ -50,21 +52,38 @@ func (h *InfoCache) GetLatest() InfoSchema {
func (h *InfoCache) GetByVersion(version int64) InfoSchema {
h.mu.RLock()
defer h.mu.RUnlock()
metrics.InfoCacheCounters.WithLabelValues("get").Inc()
metrics.InfoCacheCounters.WithLabelValues("get", "version").Inc()
i := sort.Search(len(h.cache), func(i int) bool {
return h.cache[i].SchemaMetaVersion() <= version
})
if i < len(h.cache) && h.cache[i].SchemaMetaVersion() == version {
metrics.InfoCacheCounters.WithLabelValues("hit").Inc()
metrics.InfoCacheCounters.WithLabelValues("hit", "version").Inc()
return h.cache[i]
}
return nil
}

// GetBySnapshotTS gets the information schema based on snapshotTS.
// If the snapshotTS is new than maxUpdatedSnapshotTS, that's mean it can directly use
// the latest infoschema. otherwise, will return nil.
func (h *InfoCache) GetBySnapshotTS(snapshotTS uint64) InfoSchema {
h.mu.RLock()
defer h.mu.RUnlock()

metrics.InfoCacheCounters.WithLabelValues("get", "ts").Inc()
if snapshotTS >= h.maxUpdatedSnapshotTS {
if len(h.cache) > 0 {
metrics.InfoCacheCounters.WithLabelValues("hit", "ts").Inc()
return h.cache[0]
}
}
return nil
}

// Insert will **TRY** to insert the infoschema into the cache.
// It only promised to cache the newest infoschema.
// It returns 'true' if it is cached, 'false' otherwise.
func (h *InfoCache) Insert(is InfoSchema) bool {
func (h *InfoCache) Insert(is InfoSchema, snapshotTS uint64) bool {
h.mu.Lock()
defer h.mu.Unlock()

Expand All @@ -73,6 +92,10 @@ func (h *InfoCache) Insert(is InfoSchema) bool {
return h.cache[i].SchemaMetaVersion() <= version
})

if h.maxUpdatedSnapshotTS < snapshotTS {
h.maxUpdatedSnapshotTS = snapshotTS
}

// cached entry
if i < len(h.cache) && h.cache[i].SchemaMetaVersion() == version {
return true
Expand Down
75 changes: 45 additions & 30 deletions infoschema/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,63 +33,78 @@ func (s *testInfoCacheSuite) TestInsert(c *C) {
c.Assert(ic, NotNil)

is2 := infoschema.MockInfoSchemaWithSchemaVer(nil, 2)
ic.Insert(is2)
c.Assert(ic.GetByVersion(2), NotNil)
ic.Insert(is2, 2)
c.Assert(ic.GetByVersion(2), DeepEquals, is2)
xhebox marked this conversation as resolved.
Show resolved Hide resolved
c.Assert(ic.GetBySnapshotTS(2), DeepEquals, is2)
nolouch marked this conversation as resolved.
Show resolved Hide resolved
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is2)
c.Assert(ic.GetBySnapshotTS(0), IsNil)

// newer
is5 := infoschema.MockInfoSchemaWithSchemaVer(nil, 5)
ic.Insert(is5)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(2), NotNil)
ic.Insert(is5, 5)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(2), DeepEquals, is2)
c.Assert(ic.GetBySnapshotTS(2), IsNil)
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is5)

// older
is0 := infoschema.MockInfoSchemaWithSchemaVer(nil, 0)
ic.Insert(is0)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(2), NotNil)
c.Assert(ic.GetByVersion(0), NotNil)
ic.Insert(is0, 0)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(2), DeepEquals, is2)
c.Assert(ic.GetByVersion(0), DeepEquals, is0)

// replace 5, drop 0
is6 := infoschema.MockInfoSchemaWithSchemaVer(nil, 6)
ic.Insert(is6)
c.Assert(ic.GetByVersion(6), NotNil)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(2), NotNil)
ic.Insert(is6, 6)
c.Assert(ic.GetByVersion(6), DeepEquals, is6)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(2), DeepEquals, is2)
c.Assert(ic.GetByVersion(0), IsNil)
c.Assert(ic.GetBySnapshotTS(2), IsNil)
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is6)

// replace 2, drop 2
is3 := infoschema.MockInfoSchemaWithSchemaVer(nil, 3)
ic.Insert(is3)
c.Assert(ic.GetByVersion(6), NotNil)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(3), NotNil)
ic.Insert(is3, 3)
c.Assert(ic.GetByVersion(6), DeepEquals, is6)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(3), DeepEquals, is3)
c.Assert(ic.GetByVersion(2), IsNil)
c.Assert(ic.GetByVersion(0), IsNil)
c.Assert(ic.GetBySnapshotTS(2), IsNil)
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is6)

// insert 2, but failed silently
ic.Insert(is2)
c.Assert(ic.GetByVersion(6), NotNil)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(3), NotNil)
ic.Insert(is2, 2)
c.Assert(ic.GetByVersion(6), DeepEquals, is6)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(3), DeepEquals, is3)
c.Assert(ic.GetByVersion(2), IsNil)
c.Assert(ic.GetByVersion(0), IsNil)
c.Assert(ic.GetBySnapshotTS(2), IsNil)
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is6)

// insert 5, but it is already in
ic.Insert(is5)
c.Assert(ic.GetByVersion(6), NotNil)
c.Assert(ic.GetByVersion(5), NotNil)
c.Assert(ic.GetByVersion(3), NotNil)
ic.Insert(is5, 5)
c.Assert(ic.GetByVersion(6), DeepEquals, is6)
c.Assert(ic.GetByVersion(5), DeepEquals, is5)
c.Assert(ic.GetByVersion(3), DeepEquals, is3)
c.Assert(ic.GetByVersion(2), IsNil)
c.Assert(ic.GetByVersion(0), IsNil)
c.Assert(ic.GetBySnapshotTS(2), IsNil)
c.Assert(ic.GetBySnapshotTS(5), IsNil)
c.Assert(ic.GetBySnapshotTS(10), DeepEquals, is6)

}

func (s *testInfoCacheSuite) TestGetByVersion(c *C) {
ic := infoschema.NewCache(2)
c.Assert(ic, NotNil)
is1 := infoschema.MockInfoSchemaWithSchemaVer(nil, 1)
ic.Insert(is1)
ic.Insert(is1, 1)
is3 := infoschema.MockInfoSchemaWithSchemaVer(nil, 3)
ic.Insert(is3)
ic.Insert(is3, 3)

c.Assert(ic.GetByVersion(1), Equals, is1)
c.Assert(ic.GetByVersion(3), Equals, is3)
Expand All @@ -104,16 +119,16 @@ func (s *testInfoCacheSuite) TestGetLatest(c *C) {
c.Assert(ic.GetLatest(), IsNil)

is1 := infoschema.MockInfoSchemaWithSchemaVer(nil, 1)
ic.Insert(is1)
ic.Insert(is1, 1)
c.Assert(ic.GetLatest(), Equals, is1)

// newer change the newest
is2 := infoschema.MockInfoSchemaWithSchemaVer(nil, 2)
ic.Insert(is2)
ic.Insert(is2, 2)
c.Assert(ic.GetLatest(), Equals, is2)

// older schema doesn't change the newest
is0 := infoschema.MockInfoSchemaWithSchemaVer(nil, 0)
ic.Insert(is0)
ic.Insert(is0, 0)
c.Assert(ic.GetLatest(), Equals, is2)
}
2 changes: 1 addition & 1 deletion metrics/domain.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ var (
Subsystem: "domain",
Name: "infocache_counters",
Help: "Counters of infoCache: get/hit.",
}, []string{LblType})
}, []string{LblAction, LblType})
// InfoCacheCounterGet is the total number of getting entry.
InfoCacheCounterGet = "get"
// InfoCacheCounterHit is the cache hit numbers for get.
Expand Down
8 changes: 4 additions & 4 deletions owner/manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ func TestSingle(t *testing.T) {
cli := clus.RandClient()
ctx := goctx.Background()
ic := infoschema.NewCache(2)
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d := NewDDL(
ctx,
WithEtcdClient(cli),
Expand Down Expand Up @@ -147,7 +147,7 @@ func TestCluster(t *testing.T) {

cli := clus.Client(0)
ic := infoschema.NewCache(2)
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d := NewDDL(
goctx.Background(),
WithEtcdClient(cli),
Expand All @@ -165,7 +165,7 @@ func TestCluster(t *testing.T) {
}
cli1 := clus.Client(1)
ic2 := infoschema.NewCache(2)
ic2.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic2.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d1 := NewDDL(
goctx.Background(),
WithEtcdClient(cli1),
Expand Down Expand Up @@ -200,7 +200,7 @@ func TestCluster(t *testing.T) {
// d3 (not owner) stop
cli3 := clus.Client(3)
ic3 := infoschema.NewCache(2)
ic3.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0))
ic3.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 0), 0)
d3 := NewDDL(
goctx.Background(),
WithEtcdClient(cli3),
Expand Down