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

schema cache: cache schema version by timestamp #40768

Merged
merged 8 commits into from
Feb 6, 2023
Merged

schema cache: cache schema version by timestamp #40768

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5982d7c
cache schema version by timestamp
hihihuhu Jan 23, 2023
b01a24b
fix test
hihihuhu Jan 25, 2023
611f7e7
avoid recursive locking
hihihuhu Feb 1, 2023
4da7ca7
address comment
hihihuhu Feb 2, 2023
1710996
optimize the code
hihihuhu Feb 2, 2023
3f9ce17
Merge branch 'master' into master
ti-chi-bot Feb 6, 2023
7145ef8
Merge branch 'master' into master
ti-chi-bot Feb 6, 2023
f804d62
Merge branch 'master' into master
ti-chi-bot Feb 6, 2023
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
78 changes: 65 additions & 13 deletions infoschema/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
package infoschema

import (
"fmt"
"sort"
"sync"

Expand All @@ -38,20 +39,29 @@ 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
// schema versions sorted by tiemstamp
sortedTS []versionAndTimestamp
}

type versionAndTimestamp struct {
version int64
timestamp int64
}

// NewCache creates a new InfoCache.
func NewCache(capacity int) *InfoCache {
return &InfoCache{cache: make([]InfoSchema, 0, capacity)}
return &InfoCache{
cache: make([]InfoSchema, 0, capacity),
sortedTS: make([]versionAndTimestamp, 0, capacity),
}
}

// Reset resets the cache.
func (h *InfoCache) Reset(capacity int) {
h.mu.Lock()
defer h.mu.Unlock()
h.cache = make([]InfoSchema, 0, capacity)
h.sortedTS = make([]versionAndTimestamp, 0, capacity)
}

// GetLatest gets the newest information schema.
Expand All @@ -66,10 +76,32 @@ func (h *InfoCache) GetLatest() InfoSchema {
return nil
}

// GetVersionByTimestamp returns the schema version used at the specific timestamp
func (h *InfoCache) GetVersionByTimestamp(ts uint64) (int64, error) {
h.mu.RLock()
defer h.mu.RUnlock()
return h.getVersionByTimestampNoLock(ts)
}

func (h *InfoCache) getVersionByTimestampNoLock(ts uint64) (int64, error) {
i := sort.Search(len(h.sortedTS), func(i int) bool {
return uint64(h.sortedTS[i].timestamp) <= ts
})
if i < len(h.sortedTS) {
return h.sortedTS[i].version, nil
}

return 0, fmt.Errorf("no schema cached for timestamp %d", ts)
}

// GetByVersion gets the information schema based on schemaVersion. Returns nil if it is not loaded.
func (h *InfoCache) GetByVersion(version int64) InfoSchema {
h.mu.RLock()
defer h.mu.RUnlock()
return h.getByVersionNoLock(version)
}

func (h *InfoCache) getByVersionNoLock(version int64) InfoSchema {
getVersionCounter.Inc()
i := sort.Search(len(h.cache), func(i int) bool {
return h.cache[i].SchemaMetaVersion() <= version
Expand Down Expand Up @@ -108,11 +140,9 @@ func (h *InfoCache) GetBySnapshotTS(snapshotTS uint64) InfoSchema {
defer h.mu.RUnlock()

getTSCounter.Inc()
if snapshotTS >= h.maxUpdatedSnapshotTS {
if len(h.cache) > 0 {
hitTSCounter.Inc()
return h.cache[0]
}
if version, err := h.getVersionByTimestampNoLock(snapshotTS); err == nil {
hitTSCounter.Inc()
return h.getByVersionNoLock(version)
}
return nil
}
Expand All @@ -125,14 +155,36 @@ func (h *InfoCache) Insert(is InfoSchema, snapshotTS uint64) bool {
defer h.mu.Unlock()

version := is.SchemaMetaVersion()
i := sort.Search(len(h.cache), func(i int) bool {
return h.cache[i].SchemaMetaVersion() <= version
})

if h.maxUpdatedSnapshotTS < snapshotTS {
h.maxUpdatedSnapshotTS = snapshotTS
i := sort.Search(len(h.sortedTS), func(i int) bool {
return h.sortedTS[i].version <= version
})
xhebox marked this conversation as resolved.
Show resolved Hide resolved
if i < len(h.sortedTS) && h.sortedTS[i].version == version {
// find the version, update the schema timestamp if the new timestamp is earlier
if snapshotTS < uint64(h.sortedTS[i].timestamp) {
h.sortedTS[i].timestamp = int64(snapshotTS)
}
} else {
if len(h.sortedTS) < cap(h.sortedTS) {
h.sortedTS = h.sortedTS[:len(h.sortedTS)+1]
copy(h.sortedTS[i+1:], h.sortedTS[i:])
h.sortedTS[i] = versionAndTimestamp{
version: version,
timestamp: int64(snapshotTS),
}
} else if i < len(h.sortedTS) {
copy(h.sortedTS[i+1:], h.sortedTS[i:])
h.sortedTS[i] = versionAndTimestamp{
version: version,
timestamp: int64(snapshotTS),
}
}
}

i = sort.Search(len(h.cache), func(i int) bool {
return h.cache[i].SchemaMetaVersion() <= version
})

// cached entry
if i < len(h.cache) && h.cache[i].SchemaMetaVersion() == version {
return true
Expand Down
64 changes: 61 additions & 3 deletions infoschema/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func TestInsert(t *testing.T) {
ic.Insert(is5, 5)
require.Equal(t, is5, ic.GetByVersion(5))
require.Equal(t, is2, ic.GetByVersion(2))
require.Nil(t, ic.GetBySnapshotTS(2))
require.Equal(t, is2, ic.GetBySnapshotTS(2))
require.Equal(t, is5, ic.GetBySnapshotTS(10))

// older
Expand All @@ -59,7 +59,7 @@ func TestInsert(t *testing.T) {
require.Equal(t, is5, ic.GetByVersion(5))
require.Equal(t, is2, ic.GetByVersion(2))
require.Nil(t, ic.GetByVersion(0))
require.Nil(t, ic.GetBySnapshotTS(2))
require.Equal(t, is2, ic.GetBySnapshotTS(2))
require.Equal(t, is6, ic.GetBySnapshotTS(10))

// replace 2, drop 2
Expand Down Expand Up @@ -91,7 +91,7 @@ func TestInsert(t *testing.T) {
require.Nil(t, ic.GetByVersion(2))
require.Nil(t, ic.GetByVersion(0))
require.Nil(t, ic.GetBySnapshotTS(2))
require.Nil(t, ic.GetBySnapshotTS(5))
require.Equal(t, is5, ic.GetBySnapshotTS(5))
require.Equal(t, is6, ic.GetBySnapshotTS(10))
}

Expand Down Expand Up @@ -129,3 +129,61 @@ func TestGetLatest(t *testing.T) {
ic.Insert(is0, 0)
require.Equal(t, is2, ic.GetLatest())
}

func TestGetByTimestamp(t *testing.T) {
ic := infoschema.NewCache(16)
require.NotNil(t, ic)
require.Nil(t, ic.GetLatest())

is1 := infoschema.MockInfoSchemaWithSchemaVer(nil, 1)
ic.Insert(is1, 1)
require.Equal(t, is1, ic.GetLatest())
_, err := ic.GetVersionByTimestamp(0)
require.NotNil(t, err)
ver, err := ic.GetVersionByTimestamp(1)
require.Nil(t, err)
require.Equal(t, int64(1), ver)
require.Equal(t, is1, ic.GetBySnapshotTS(1))
ver, err = ic.GetVersionByTimestamp(2)
require.Nil(t, err)
require.Equal(t, int64(1), ver)
require.Equal(t, is1, ic.GetBySnapshotTS(2))

is2 := infoschema.MockInfoSchemaWithSchemaVer(nil, 2)
ic.Insert(is2, 2)
require.Equal(t, is2, ic.GetLatest())
_, err = ic.GetVersionByTimestamp(0)
require.NotNil(t, err)
ver, err = ic.GetVersionByTimestamp(1)
require.Nil(t, err)
require.Equal(t, int64(1), ver)
require.Equal(t, is1, ic.GetBySnapshotTS(1))
ver, err = ic.GetVersionByTimestamp(2)
require.Nil(t, err)
require.Equal(t, int64(2), ver)
require.Equal(t, is2, ic.GetBySnapshotTS(2))
ver, err = ic.GetVersionByTimestamp(3)
require.Nil(t, err)
require.Equal(t, int64(2), ver)
require.Equal(t, is2, ic.GetBySnapshotTS(3))

is0 := infoschema.MockInfoSchemaWithSchemaVer(nil, 0)
ic.Insert(is0, 0)
require.Equal(t, is2, ic.GetLatest())
ver, err = ic.GetVersionByTimestamp(0)
require.Nil(t, err)
require.Equal(t, int64(0), ver)
require.Equal(t, is0, ic.GetBySnapshotTS(0))
ver, err = ic.GetVersionByTimestamp(1)
require.Nil(t, err)
require.Equal(t, int64(1), ver)
require.Equal(t, is1, ic.GetBySnapshotTS(1))
ver, err = ic.GetVersionByTimestamp(2)
require.Nil(t, err)
require.Equal(t, int64(2), ver)
require.Equal(t, is2, ic.GetBySnapshotTS(2))
ver, err = ic.GetVersionByTimestamp(3)
require.Nil(t, err)
require.Equal(t, int64(2), ver)
require.Equal(t, is2, ic.GetBySnapshotTS(3))
}