Merge remote-tracking branch 'upstream/master' into move-column

.github/licenserc.yml

@@ -10,6 +10,7 @@ header:
     - '.golangci.yml'
     - '.golangci_br.yml'
     - 'LICENSES/'
+    - '**/*.key'
     - '**/*.md'
     - '**/*.json'
     - '**/*.pem'
@@ -28,4 +29,6 @@ header:
     - '.github/'
     - 'parser/'
     - 'dumpling/'
+    - 'tidb-binlog/driver/example'
+    - 'tidb-binlog/proto/go-binlog/secondary_binlog.pb.go'
   comment: on-failure

.github/workflows/br_compatible_test.yml

@@ -46,7 +46,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v2
       with:
-        go-version: 1.16
+        go-version: 1.18
 
     - name: Generate compatibility test backup data
       timeout-minutes: 15

.github/workflows/compile_br.yaml

@@ -48,7 +48,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
 
       - name: Run build
         run: make build_tools
@@ -73,7 +73,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
 
       - name: Run build
         run: make build_tools
@@ -88,7 +88,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
 
       - name: Compile for FreeBSD
         run: GOOS=freebsd make build_tools

.github/workflows/dumpling_integration_test.yml

@@ -50,10 +50,10 @@ jobs:
       - uses: actions/checkout@v2
       - name: Shutdown Ubuntu MySQL (SUDO)
         run: sudo service mysql stop # Shutdown the Default MySQL, "sudo" is necessary, please not remove it
-      - name: Set up Go 1.16
+      - name: Set up Go 1.18
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
       - uses: actions/cache@v2
         with:
           path: ~/go/pkg/mod
@@ -87,10 +87,10 @@ jobs:
       - uses: actions/checkout@v2
       - name: Shutdown Ubuntu MySQL (SUDO)
         run: sudo service mysql stop # Shutdown the Default MySQL, "sudo" is necessary, please not remove it
-      - name: Set up Go 1.16
+      - name: Set up Go 1.18
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
       - uses: actions/cache@v2
         with:
           path: ~/go/pkg/mod
@@ -124,10 +124,10 @@ jobs:
       - uses: actions/checkout@v2
       - name: Shutdown Ubuntu MySQL (SUDO)
         run: sudo service mysql stop # Shutdown the Default MySQL, "sudo" is necessary, please not remove it
-      - name: Set up Go 1.16
+      - name: Set up Go 1.18
         uses: actions/setup-go@v2
         with:
-          go-version: 1.16
+          go-version: 1.18
       - uses: actions/cache@v2
         with:
           path: ~/go/pkg/mod

Makefile

@@ -28,7 +28,7 @@ all: dev server benchkv
 parser:
 	@echo "remove this command later, when our CI script doesn't call it"
 
-dev: checklist check explaintest gogenerate br_unit_test test_part_parser_dev
+dev: checklist check explaintest gogenerate br_unit_test test_part_parser_dev ut
 	@>&2 echo "Great, all tests passed."
 
 # Install the check tools.
@@ -47,7 +47,7 @@ check-static: tools/bin/golangci-lint
 	GO111MODULE=on CGO_ENABLED=0 tools/bin/golangci-lint run -v $$($(PACKAGE_DIRECTORIES)) --config .golangci.yml
 
 unconvert:tools/bin/unconvert
-	@echo "unconvert check(skip check the genenrated or copied code in lightning)"
+	@echo "unconvert check(skip check the generated or copied code in lightning)"
 	@GO111MODULE=on tools/bin/unconvert $(UNCONVERT_PACKAGES)
 
 gogenerate:
@@ -245,7 +245,11 @@ tools/bin/errdoc-gen: tools/check/go.mod
 	$(GO) build -o ../bin/errdoc-gen github.com/pingcap/errors/errdoc-gen
 
 tools/bin/golangci-lint:
-	curl -sfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s -- -b ./tools/bin v1.41.1
+	cd tools/check; \
+	$(GO) build -o ../bin/golangci-lint github.com/golangci/golangci-lint/cmd/golangci-lint
+	# Build from source is not recommand. See https://golangci-lint.run/usage/install/
+	# But the following script from their website doesn't work with Go1.18:
+	# curl -sfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s -- -b ./tools/bin v1.44.2
 
 tools/bin/vfsgendev: tools/check/go.mod
 	cd tools/check; \

Makefile.common

@@ -54,7 +54,7 @@ PACKAGE_DIRECTORIES_TIDB_TESTS := $(PACKAGE_LIST_TIDB_TESTS) | sed 's|github.com
 FILES     := $$(find $$($(PACKAGE_DIRECTORIES)) -name "*.go")
 FILES_TIDB_TESTS := $$(find $$($(PACKAGE_DIRECTORIES_TIDB_TESTS)) -name "*.go")
 
-UNCONVERT_PACKAGES_LIST := go list ./...| grep -vE "lightning\/checkpoints|lightning\/manual|lightning\/common"
+UNCONVERT_PACKAGES_LIST := go list ./...| grep -vE "lightning\/checkpoints|lightning\/manual|lightning\/common|tidb-binlog\/proto\/go-binlog"
 UNCONVERT_PACKAGES := $$($(UNCONVERT_PACKAGES_LIST))
 
 FAILPOINT_ENABLE  := find $$PWD/ -type d | grep -vE "(\.git|tools)" | xargs tools/bin/failpoint-ctl enable

bindinfo/bind_cache.go

@@ -15,6 +15,7 @@
 package bindinfo
 
 import (
+	"errors"
 	"sync"
 
 	"github.com/cznic/mathutil"
@@ -54,7 +55,7 @@ func newBindCache() *bindCache {
 	cache := kvcache.NewSimpleLRUCache(mathutil.MaxUint, 0, 0)
 	c := bindCache{
 		cache:       cache,
-		memCapacity: variable.MemQuotaBindCache.Load(),
+		memCapacity: variable.MemQuotaBindingCache.Load(),
 		memTracker:  memory.NewTracker(memory.LabelForBindCache, -1),
 	}
 	return &c
@@ -73,28 +74,48 @@ func (c *bindCache) get(key bindCacheKey) []*BindRecord {
 	return typedValue
 }
 
+// getCopiedVal gets a copied cache item according to cache key.
+// The return value can be modified.
+// If you want to modify the return value, use the 'getCopiedVal' function rather than 'get' function.
+// We use the copy on write way to operate the bindRecord in cache for safety and accuracy of memory usage.
+func (c *bindCache) getCopiedVal(key bindCacheKey) []*BindRecord {
+	bindRecords := c.get(key)
+	if bindRecords != nil {
+		copiedRecords := make([]*BindRecord, len(bindRecords))
+		for i, bindRecord := range bindRecords {
+			copiedRecords[i] = bindRecord.shallowCopy()
+		}
+		return copiedRecords
+	}
+	return bindRecords
+}
+
 // set inserts an item to the cache. It's not thread-safe.
 // Only other functions of the bindCache can use this function.
-func (c *bindCache) set(key bindCacheKey, value []*BindRecord) bool {
+// The set operation will return error message when the memory usage of binding_cache exceeds its capacity.
+func (c *bindCache) set(key bindCacheKey, value []*BindRecord) (ok bool, err error) {
 	mem := calcBindCacheKVMem(key, value)
 	if mem > c.memCapacity { // ignore this kv pair if its size is too large
-		return false
+		err = errors.New("The memory usage of all available bindings exceeds the cache's mem quota. As a result, all available bindings cannot be held on the cache. Please increase the value of the system variable 'tidb_mem_quota_binding_cache' and execute 'admin reload bindings' to ensure that all bindings exist in the cache and can be used normally")
+		return
 	}
 	bindRecords := c.get(key)
 	if bindRecords != nil {
 		// Remove the origin key-value pair.
 		mem -= calcBindCacheKVMem(key, bindRecords)
 	}
 	for mem+c.memTracker.BytesConsumed() > c.memCapacity {
+		err = errors.New("The memory usage of all available bindings exceeds the cache's mem quota. As a result, all available bindings cannot be held on the cache. Please increase the value of the system variable 'tidb_mem_quota_binding_cache' and execute 'admin reload bindings' to ensure that all bindings exist in the cache and can be used normally")
 		evictedKey, evictedValue, evicted := c.cache.RemoveOldest()
 		if !evicted {
-			return false
+			return
 		}
 		c.memTracker.Consume(-calcBindCacheKVMem(evictedKey.(bindCacheKey), evictedValue.([]*BindRecord)))
 	}
 	c.memTracker.Consume(mem)
 	c.cache.Put(key, value)
-	return true
+	ok = true
+	return
 }
 
 // delete remove an item from the cache. It's not thread-safe.
@@ -105,8 +126,9 @@ func (c *bindCache) delete(key bindCacheKey) bool {
 		mem := calcBindCacheKVMem(key, bindRecords)
 		c.cache.Delete(key)
 		c.memTracker.Consume(-mem)
+		return true
 	}
-	return true
+	return false
 }
 
 // GetBindRecord gets the BindRecord from the cache.
@@ -140,25 +162,26 @@ func (c *bindCache) GetAllBindRecords() []*BindRecord {
 
 // SetBindRecord sets the BindRecord to the cache.
 // The function is thread-safe.
-func (c *bindCache) SetBindRecord(hash string, meta *BindRecord) {
+func (c *bindCache) SetBindRecord(hash string, meta *BindRecord) (err error) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	cacheKey := bindCacheKey(hash)
-	metas := c.get(cacheKey)
+	metas := c.getCopiedVal(cacheKey)
 	for i := range metas {
 		if metas[i].OriginalSQL == meta.OriginalSQL {
 			metas[i] = meta
 		}
 	}
-	c.set(cacheKey, []*BindRecord{meta})
+	_, err = c.set(cacheKey, []*BindRecord{meta})
+	return
 }
 
 // RemoveBindRecord removes the BindRecord which has same originSQL with specified BindRecord.
 // The function is thread-safe.
 func (c *bindCache) RemoveBindRecord(hash string, meta *BindRecord) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	metas := c.get(bindCacheKey(hash))
+	metas := c.getCopiedVal(bindCacheKey(hash))
 	if metas == nil {
 		return
 	}
@@ -175,7 +198,9 @@ func (c *bindCache) RemoveBindRecord(hash string, meta *BindRecord) {
 			}
 		}
 	}
-	c.set(bindCacheKey(hash), metas)
+	// This function can guarantee the memory usage for the cache will never grow up.
+	// So we don't need to handle the return value here.
+	_, _ = c.set(bindCacheKey(hash), metas)
 }
 
 // SetMemCapacity sets the memory capacity for the cache.
@@ -187,6 +212,14 @@ func (c *bindCache) SetMemCapacity(capacity int64) {
 	c.memCapacity = capacity
 }
 
+// GetMemUsage get the memory Usage for the cache.
+// The function is thread-safe.
+func (c *bindCache) GetMemUsage() int64 {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	return c.memTracker.BytesConsumed()
+}
+
 // GetMemCapacity get the memory capacity for the cache.
 // The function is thread-safe.
 func (c *bindCache) GetMemCapacity() int64 {
@@ -197,17 +230,22 @@ func (c *bindCache) GetMemCapacity() int64 {
 
 // Copy copies a new bindCache from the origin cache.
 // The function is thread-safe.
-func (c *bindCache) Copy() *bindCache {
+func (c *bindCache) Copy() (newCache *bindCache, err error) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
-	newCache := newBindCache()
+	newCache = newBindCache()
+	if c.memTracker.BytesConsumed() > newCache.GetMemCapacity() {
+		err = errors.New("The memory usage of all available bindings exceeds the cache's mem quota. As a result, all available bindings cannot be held on the cache. Please increase the value of the system variable 'tidb_mem_quota_binding_cache' and execute 'admin reload bindings' to ensure that all bindings exist in the cache and can be used normally")
+	}
 	keys := c.cache.Keys()
 	for _, key := range keys {
 		cacheKey := key.(bindCacheKey)
 		v := c.get(cacheKey)
 		bindRecords := make([]*BindRecord, len(v))
 		copy(bindRecords, v)
-		newCache.set(cacheKey, bindRecords)
+		// The memory usage of cache has been handled at the beginning of this function.
+		// So we don't need to handle the return value here.
+		_, _ = newCache.set(cacheKey, bindRecords)
 	}
-	return newCache
+	return newCache, err
 }

bindinfo/bind_cache_test.go

@@ -25,39 +25,55 @@ import (
 )
 
 func TestBindCache(t *testing.T) {
-	variable.MemQuotaBindCache.Store(100)
+	variable.MemQuotaBindingCache.Store(200)
 	bindCache := newBindCache()
 
 	value := make([][]*BindRecord, 3)
 	key := make([]bindCacheKey, 3)
+	var bigKey string
 	for i := 0; i < 3; i++ {
 		cacheKey := strings.Repeat(strconv.Itoa(i), 50)
 		key[i] = bindCacheKey(hack.Slice(cacheKey))
 		record := &BindRecord{OriginalSQL: cacheKey, Db: ""}
 		value[i] = []*BindRecord{record}
+		bigKey += cacheKey
 
 		require.Equal(t, int64(100), calcBindCacheKVMem(key[i], value[i]))
 	}
 
-	ok := bindCache.set(key[0], value[0])
+	ok, err := bindCache.set(key[0], value[0])
 	require.True(t, ok)
+	require.Nil(t, err)
 	result := bindCache.get(key[0])
 	require.NotNil(t, result)
 
-	ok = bindCache.set(key[1], value[1])
+	ok, err = bindCache.set(key[1], value[1])
 	require.True(t, ok)
+	require.Nil(t, err)
 	result = bindCache.get(key[1])
 	require.NotNil(t, result)
 
-	ok = bindCache.set(key[2], value[2])
+	ok, err = bindCache.set(key[2], value[2])
 	require.True(t, ok)
+	require.NotNil(t, err)
 	result = bindCache.get(key[2])
 	require.NotNil(t, result)
 
-	// Both key[0] and key[1] are not in the cache
+	// key[0] is not in the cache
 	result = bindCache.get(key[0])
 	require.Nil(t, result)
 
+	// key[1] is still in the cache
 	result = bindCache.get(key[1])
+	require.NotNil(t, result)
+
+	bigBindCacheKey := bindCacheKey(hack.Slice(bigKey))
+	bigRecord := &BindRecord{OriginalSQL: bigKey, Db: ""}
+	bigBindCacheValue := []*BindRecord{bigRecord}
+	require.Equal(t, int64(300), calcBindCacheKVMem(bigBindCacheKey, bigBindCacheValue))
+	ok, err = bindCache.set(bigBindCacheKey, bigBindCacheValue)
+	require.False(t, ok)
+	require.NotNil(t, err)
+	result = bindCache.get(bigBindCacheKey)
 	require.Nil(t, result)
 }