You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
963 lines
33 KiB
963 lines
33 KiB
// Copyright 2015 PingCAP, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package autoid
|
|
|
|
import (
|
|
"context"
|
|
"math"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/cznic/mathutil"
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/failpoint"
|
|
"github.com/pingcap/parser/model"
|
|
"github.com/pingcap/parser/mysql"
|
|
"github.com/pingcap/tidb/kv"
|
|
"github.com/pingcap/tidb/meta"
|
|
"github.com/pingcap/tidb/metrics"
|
|
"github.com/pingcap/tidb/types"
|
|
"github.com/pingcap/tidb/util/logutil"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// Attention:
|
|
// For reading cluster TiDB memory tables, the system schema/table should be same.
|
|
// Once the system schema/table id been allocated, it can't be changed any more.
|
|
// Change the system schema/table id may have the compatibility problem.
|
|
const (
|
|
// SystemSchemaIDFlag is the system schema/table id flag, uses the highest bit position as system schema ID flag, it's exports for test.
|
|
SystemSchemaIDFlag = 1 << 62
|
|
// InformationSchemaDBID is the information_schema schema id, it's exports for test.
|
|
InformationSchemaDBID int64 = SystemSchemaIDFlag | 1
|
|
// PerformanceSchemaDBID is the performance_schema schema id, it's exports for test.
|
|
PerformanceSchemaDBID int64 = SystemSchemaIDFlag | 10000
|
|
// MetricSchemaDBID is the metrics_schema schema id, it's exported for test.
|
|
MetricSchemaDBID int64 = SystemSchemaIDFlag | 20000
|
|
)
|
|
|
|
const (
|
|
minStep = 30000
|
|
maxStep = 2000000
|
|
defaultConsumeTime = 10 * time.Second
|
|
minIncrement = 1
|
|
maxIncrement = 65535
|
|
)
|
|
|
|
// RowIDBitLength is the bit number of a row id in TiDB.
|
|
const RowIDBitLength = 64
|
|
|
|
// DefaultAutoRandomBits is the default value of auto sharding.
|
|
const DefaultAutoRandomBits = 5
|
|
|
|
// MaxAutoRandomBits is the max value of auto sharding.
|
|
const MaxAutoRandomBits = 15
|
|
|
|
// Test needs to change it, so it's a variable.
|
|
var step = int64(30000)
|
|
|
|
// AllocatorType is the type of allocator for generating auto-id. Different type of allocators use different key-value pairs.
|
|
type AllocatorType = uint8
|
|
|
|
const (
|
|
// RowIDAllocType indicates the allocator is used to allocate row id.
|
|
RowIDAllocType AllocatorType = iota
|
|
// AutoIncrementType indicates the allocator is used to allocate auto increment value.
|
|
AutoIncrementType
|
|
// AutoRandomType indicates the allocator is used to allocate auto-shard id.
|
|
AutoRandomType
|
|
// SequenceType indicates the allocator is used to allocate sequence value.
|
|
SequenceType
|
|
)
|
|
|
|
// CustomAutoIncCacheOption is one kind of AllocOption to customize the allocator step length.
|
|
type CustomAutoIncCacheOption int64
|
|
|
|
// ApplyOn is implement the AllocOption interface.
|
|
func (step CustomAutoIncCacheOption) ApplyOn(alloc *allocator) {
|
|
alloc.step = int64(step)
|
|
alloc.customStep = true
|
|
}
|
|
|
|
// AllocOption is a interface to define allocator custom options coming in future.
|
|
type AllocOption interface {
|
|
ApplyOn(*allocator)
|
|
}
|
|
|
|
// Allocator is an auto increment id generator.
|
|
// Just keep id unique actually.
|
|
type Allocator interface {
|
|
// Alloc allocs N consecutive autoID for table with tableID, returning (min, max] of the allocated autoID batch.
|
|
// It gets a batch of autoIDs at a time. So it does not need to access storage for each call.
|
|
// The consecutive feature is used to insert multiple rows in a statement.
|
|
// increment & offset is used to validate the start position (the allocator's base is not always the last allocated id).
|
|
// The returned range is (min, max]:
|
|
// case increment=1 & offset=1: you can derive the ids like min+1, min+2... max.
|
|
// case increment=x & offset=y: you firstly need to seek to firstID by `SeekToFirstAutoIDXXX`, then derive the IDs like firstID, firstID + increment * 2... in the caller.
|
|
Alloc(tableID int64, n uint64, increment, offset int64) (int64, int64, error)
|
|
|
|
// AllocSeqCache allocs sequence batch value cached in table level(rather than in alloc), the returned range covering
|
|
// the size of sequence cache with it's increment. The returned round indicates the sequence cycle times if it is with
|
|
// cycle option.
|
|
AllocSeqCache(sequenceID int64) (min int64, max int64, round int64, err error)
|
|
|
|
// Rebase rebases the autoID base for table with tableID and the new base value.
|
|
// If allocIDs is true, it will allocate some IDs and save to the cache.
|
|
// If allocIDs is false, it will not allocate IDs.
|
|
Rebase(tableID, newBase int64, allocIDs bool) error
|
|
|
|
// RebaseSeq rebases the sequence value in number axis with tableID and the new base value.
|
|
RebaseSeq(table, newBase int64) (int64, bool, error)
|
|
|
|
// Base return the current base of Allocator.
|
|
Base() int64
|
|
// End is only used for test.
|
|
End() int64
|
|
// NextGlobalAutoID returns the next global autoID.
|
|
NextGlobalAutoID(tableID int64) (int64, error)
|
|
GetType() AllocatorType
|
|
}
|
|
|
|
// Allocators represents a set of `Allocator`s.
|
|
type Allocators []Allocator
|
|
|
|
// NewAllocators packs multiple `Allocator`s into Allocators.
|
|
func NewAllocators(allocators ...Allocator) Allocators {
|
|
return allocators
|
|
}
|
|
|
|
// Get returns the Allocator according to the AllocatorType.
|
|
func (all Allocators) Get(allocType AllocatorType) Allocator {
|
|
for _, a := range all {
|
|
if a.GetType() == allocType {
|
|
return a
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type allocator struct {
|
|
mu sync.Mutex
|
|
base int64
|
|
end int64
|
|
store kv.Storage
|
|
// dbID is current database's ID.
|
|
dbID int64
|
|
isUnsigned bool
|
|
lastAllocTime time.Time
|
|
step int64
|
|
customStep bool
|
|
allocType AllocatorType
|
|
sequence *model.SequenceInfo
|
|
}
|
|
|
|
// GetStep is only used by tests
|
|
func GetStep() int64 {
|
|
return step
|
|
}
|
|
|
|
// SetStep is only used by tests
|
|
func SetStep(s int64) {
|
|
step = s
|
|
}
|
|
|
|
// Base implements autoid.Allocator Base interface.
|
|
func (alloc *allocator) Base() int64 {
|
|
return alloc.base
|
|
}
|
|
|
|
// End implements autoid.Allocator End interface.
|
|
func (alloc *allocator) End() int64 {
|
|
return alloc.end
|
|
}
|
|
|
|
// NextGlobalAutoID implements autoid.Allocator NextGlobalAutoID interface.
|
|
func (alloc *allocator) NextGlobalAutoID(tableID int64) (int64, error) {
|
|
var autoID int64
|
|
startTime := time.Now()
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
var err1 error
|
|
m := meta.NewMeta(txn)
|
|
autoID, err1 = getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return errors.Trace(err1)
|
|
}
|
|
return nil
|
|
})
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.GlobalAutoID, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if alloc.isUnsigned {
|
|
return int64(uint64(autoID) + 1), err
|
|
}
|
|
return autoID + 1, err
|
|
}
|
|
|
|
func (alloc *allocator) rebase4Unsigned(tableID int64, requiredBase uint64, allocIDs bool) error {
|
|
// Satisfied by alloc.base, nothing to do.
|
|
if requiredBase <= uint64(alloc.base) {
|
|
return nil
|
|
}
|
|
// Satisfied by alloc.end, need to update alloc.base.
|
|
if requiredBase <= uint64(alloc.end) {
|
|
alloc.base = int64(requiredBase)
|
|
return nil
|
|
}
|
|
var newBase, newEnd uint64
|
|
startTime := time.Now()
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
currentEnd, err1 := getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
uCurrentEnd := uint64(currentEnd)
|
|
if allocIDs {
|
|
newBase = mathutil.MaxUint64(uCurrentEnd, requiredBase)
|
|
newEnd = mathutil.MinUint64(math.MaxUint64-uint64(alloc.step), newBase) + uint64(alloc.step)
|
|
} else {
|
|
if uCurrentEnd >= requiredBase {
|
|
newBase = uCurrentEnd
|
|
newEnd = uCurrentEnd
|
|
// Required base satisfied, we don't need to update KV.
|
|
return nil
|
|
}
|
|
// If we don't want to allocate IDs, for example when creating a table with a given base value,
|
|
// We need to make sure when other TiDB server allocates ID for the first time, requiredBase + 1
|
|
// will be allocated, so we need to increase the end to exactly the requiredBase.
|
|
newBase = requiredBase
|
|
newEnd = requiredBase
|
|
}
|
|
_, err1 = generateAutoIDByAllocType(m, alloc.dbID, tableID, int64(newEnd-uCurrentEnd), alloc.allocType)
|
|
return err1
|
|
})
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDRebase, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
alloc.base, alloc.end = int64(newBase), int64(newEnd)
|
|
return nil
|
|
}
|
|
|
|
func (alloc *allocator) rebase4Signed(tableID, requiredBase int64, allocIDs bool) error {
|
|
// Satisfied by alloc.base, nothing to do.
|
|
if requiredBase <= alloc.base {
|
|
return nil
|
|
}
|
|
// Satisfied by alloc.end, need to update alloc.base.
|
|
if requiredBase <= alloc.end {
|
|
alloc.base = requiredBase
|
|
return nil
|
|
}
|
|
var newBase, newEnd int64
|
|
startTime := time.Now()
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
currentEnd, err1 := getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
if allocIDs {
|
|
newBase = mathutil.MaxInt64(currentEnd, requiredBase)
|
|
newEnd = mathutil.MinInt64(math.MaxInt64-alloc.step, newBase) + alloc.step
|
|
} else {
|
|
if currentEnd >= requiredBase {
|
|
newBase = currentEnd
|
|
newEnd = currentEnd
|
|
// Required base satisfied, we don't need to update KV.
|
|
return nil
|
|
}
|
|
// If we don't want to allocate IDs, for example when creating a table with a given base value,
|
|
// We need to make sure when other TiDB server allocates ID for the first time, requiredBase + 1
|
|
// will be allocated, so we need to increase the end to exactly the requiredBase.
|
|
newBase = requiredBase
|
|
newEnd = requiredBase
|
|
}
|
|
_, err1 = generateAutoIDByAllocType(m, alloc.dbID, tableID, newEnd-currentEnd, alloc.allocType)
|
|
return err1
|
|
})
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDRebase, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
alloc.base, alloc.end = newBase, newEnd
|
|
return nil
|
|
}
|
|
|
|
// rebase4Sequence won't alloc batch immediately, cause it won't cache value in allocator.
|
|
func (alloc *allocator) rebase4Sequence(tableID, requiredBase int64) (int64, bool, error) {
|
|
startTime := time.Now()
|
|
alreadySatisfied := false
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
currentEnd, err := getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if alloc.sequence.Increment > 0 {
|
|
if currentEnd >= requiredBase {
|
|
// Required base satisfied, we don't need to update KV.
|
|
alreadySatisfied = true
|
|
return nil
|
|
}
|
|
} else {
|
|
if currentEnd <= requiredBase {
|
|
// Required base satisfied, we don't need to update KV.
|
|
alreadySatisfied = true
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// If we don't want to allocate IDs, for example when creating a table with a given base value,
|
|
// We need to make sure when other TiDB server allocates ID for the first time, requiredBase + 1
|
|
// will be allocated, so we need to increase the end to exactly the requiredBase.
|
|
_, err = generateAutoIDByAllocType(m, alloc.dbID, tableID, requiredBase-currentEnd, alloc.allocType)
|
|
return err
|
|
})
|
|
// TODO: sequence metrics
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDRebase, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return 0, false, err
|
|
}
|
|
if alreadySatisfied {
|
|
return 0, true, nil
|
|
}
|
|
return requiredBase, false, err
|
|
}
|
|
|
|
// Rebase implements autoid.Allocator Rebase interface.
|
|
// The requiredBase is the minimum base value after Rebase.
|
|
// The real base may be greater than the required base.
|
|
func (alloc *allocator) Rebase(tableID, requiredBase int64, allocIDs bool) error {
|
|
if tableID == 0 {
|
|
return errInvalidTableID.GenWithStack("Invalid tableID")
|
|
}
|
|
|
|
alloc.mu.Lock()
|
|
defer alloc.mu.Unlock()
|
|
|
|
if alloc.isUnsigned {
|
|
return alloc.rebase4Unsigned(tableID, uint64(requiredBase), allocIDs)
|
|
}
|
|
return alloc.rebase4Signed(tableID, requiredBase, allocIDs)
|
|
}
|
|
|
|
// Rebase implements autoid.Allocator RebaseSeq interface.
|
|
// The return value is quite same as expression function, bool means whether it should be NULL,
|
|
// here it will be used in setval expression function (true meaning the set value has been satisfied, return NULL).
|
|
// case1:When requiredBase is satisfied with current value, it will return (0, true, nil),
|
|
// case2:When requiredBase is successfully set in, it will return (requiredBase, false, nil).
|
|
// If some error occurs in the process, return it immediately.
|
|
func (alloc *allocator) RebaseSeq(tableID, requiredBase int64) (int64, bool, error) {
|
|
if tableID == 0 {
|
|
return 0, false, errInvalidTableID.GenWithStack("Invalid tableID")
|
|
}
|
|
|
|
alloc.mu.Lock()
|
|
defer alloc.mu.Unlock()
|
|
return alloc.rebase4Sequence(tableID, requiredBase)
|
|
}
|
|
|
|
func (alloc *allocator) GetType() AllocatorType {
|
|
return alloc.allocType
|
|
}
|
|
|
|
// NextStep return new auto id step according to previous step and consuming time.
|
|
func NextStep(curStep int64, consumeDur time.Duration) int64 {
|
|
failpoint.Inject("mockAutoIDCustomize", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
failpoint.Return(3)
|
|
}
|
|
})
|
|
failpoint.Inject("mockAutoIDChange", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
failpoint.Return(step)
|
|
}
|
|
})
|
|
|
|
consumeRate := defaultConsumeTime.Seconds() / consumeDur.Seconds()
|
|
res := int64(float64(curStep) * consumeRate)
|
|
if res < minStep {
|
|
return minStep
|
|
} else if res > maxStep {
|
|
return maxStep
|
|
}
|
|
return res
|
|
}
|
|
|
|
// NewAllocator returns a new auto increment id generator on the store.
|
|
func NewAllocator(store kv.Storage, dbID int64, isUnsigned bool, allocType AllocatorType, opts ...AllocOption) Allocator {
|
|
alloc := &allocator{
|
|
store: store,
|
|
dbID: dbID,
|
|
isUnsigned: isUnsigned,
|
|
step: step,
|
|
lastAllocTime: time.Now(),
|
|
allocType: allocType,
|
|
}
|
|
for _, fn := range opts {
|
|
fn.ApplyOn(alloc)
|
|
}
|
|
return alloc
|
|
}
|
|
|
|
// NewSequenceAllocator returns a new sequence value generator on the store.
|
|
func NewSequenceAllocator(store kv.Storage, dbID int64, info *model.SequenceInfo) Allocator {
|
|
return &allocator{
|
|
store: store,
|
|
dbID: dbID,
|
|
// Sequence allocator is always signed.
|
|
isUnsigned: false,
|
|
lastAllocTime: time.Now(),
|
|
allocType: SequenceType,
|
|
sequence: info,
|
|
}
|
|
}
|
|
|
|
// NewAllocatorsFromTblInfo creates an array of allocators of different types with the information of model.TableInfo.
|
|
func NewAllocatorsFromTblInfo(store kv.Storage, schemaID int64, tblInfo *model.TableInfo) Allocators {
|
|
var allocs []Allocator
|
|
dbID := tblInfo.GetDBID(schemaID)
|
|
if tblInfo.AutoIdCache > 0 {
|
|
allocs = append(allocs, NewAllocator(store, dbID, tblInfo.IsAutoIncColUnsigned(), RowIDAllocType, CustomAutoIncCacheOption(tblInfo.AutoIdCache)))
|
|
} else {
|
|
allocs = append(allocs, NewAllocator(store, dbID, tblInfo.IsAutoIncColUnsigned(), RowIDAllocType))
|
|
}
|
|
if tblInfo.ContainsAutoRandomBits() {
|
|
allocs = append(allocs, NewAllocator(store, dbID, tblInfo.IsAutoRandomBitColUnsigned(), AutoRandomType))
|
|
}
|
|
if tblInfo.IsSequence() {
|
|
allocs = append(allocs, NewSequenceAllocator(store, dbID, tblInfo.Sequence))
|
|
}
|
|
return NewAllocators(allocs...)
|
|
}
|
|
|
|
// Alloc implements autoid.Allocator Alloc interface.
|
|
// For autoIncrement allocator, the increment and offset should always be positive in [1, 65535].
|
|
// Attention:
|
|
// When increment and offset is not the default value(1), the return range (min, max] need to
|
|
// calculate the correct start position rather than simply the add 1 to min. Then you can derive
|
|
// the successive autoID by adding increment * cnt to firstID for (n-1) times.
|
|
//
|
|
// Example:
|
|
// (6, 13] is returned, increment = 4, offset = 1, n = 2.
|
|
// 6 is the last allocated value for other autoID or handle, maybe with different increment and step,
|
|
// but actually we don't care about it, all we need is to calculate the new autoID corresponding to the
|
|
// increment and offset at this time now. To simplify the rule is like (ID - offset) % increment = 0,
|
|
// so the first autoID should be 9, then add increment to it to get 13.
|
|
func (alloc *allocator) Alloc(tableID int64, n uint64, increment, offset int64) (int64, int64, error) {
|
|
if tableID == 0 {
|
|
return 0, 0, errInvalidTableID.GenWithStackByArgs("Invalid tableID")
|
|
}
|
|
if n == 0 {
|
|
return 0, 0, nil
|
|
}
|
|
if alloc.allocType == AutoIncrementType || alloc.allocType == RowIDAllocType {
|
|
if !validIncrementAndOffset(increment, offset) {
|
|
return 0, 0, errInvalidIncrementAndOffset.GenWithStackByArgs(increment, offset)
|
|
}
|
|
}
|
|
alloc.mu.Lock()
|
|
defer alloc.mu.Unlock()
|
|
if alloc.isUnsigned {
|
|
return alloc.alloc4Unsigned(tableID, n, increment, offset)
|
|
}
|
|
return alloc.alloc4Signed(tableID, n, increment, offset)
|
|
}
|
|
|
|
func (alloc *allocator) AllocSeqCache(tableID int64) (int64, int64, int64, error) {
|
|
if tableID == 0 {
|
|
return 0, 0, 0, errInvalidTableID.GenWithStackByArgs("Invalid tableID")
|
|
}
|
|
alloc.mu.Lock()
|
|
defer alloc.mu.Unlock()
|
|
return alloc.alloc4Sequence(tableID)
|
|
}
|
|
|
|
func validIncrementAndOffset(increment, offset int64) bool {
|
|
return (increment >= minIncrement && increment <= maxIncrement) && (offset >= minIncrement && offset <= maxIncrement)
|
|
}
|
|
|
|
// CalcNeededBatchSize is used to calculate batch size for autoID allocation.
|
|
// It firstly seeks to the first valid position based on increment and offset,
|
|
// then plus the length remained, which could be (n-1) * increment.
|
|
func CalcNeededBatchSize(base, n, increment, offset int64, isUnsigned bool) int64 {
|
|
if increment == 1 {
|
|
return n
|
|
}
|
|
if isUnsigned {
|
|
// SeekToFirstAutoIDUnSigned seeks to the next unsigned valid position.
|
|
nr := SeekToFirstAutoIDUnSigned(uint64(base), uint64(increment), uint64(offset))
|
|
// Calculate the total batch size needed.
|
|
nr += (uint64(n) - 1) * uint64(increment)
|
|
return int64(nr - uint64(base))
|
|
}
|
|
nr := SeekToFirstAutoIDSigned(base, increment, offset)
|
|
// Calculate the total batch size needed.
|
|
nr += (n - 1) * increment
|
|
return nr - base
|
|
}
|
|
|
|
// CalcSequenceBatchSize calculate the next sequence batch size.
|
|
func CalcSequenceBatchSize(base, size, increment, offset, MIN, MAX int64) (int64, error) {
|
|
// The sequence is positive growth.
|
|
if increment > 0 {
|
|
if increment == 1 {
|
|
// Sequence is already allocated to the end.
|
|
if base >= MAX {
|
|
return 0, ErrAutoincReadFailed
|
|
}
|
|
// The rest of sequence < cache size, return the rest.
|
|
if MAX-base < size {
|
|
return MAX - base, nil
|
|
}
|
|
// The rest of sequence is adequate.
|
|
return size, nil
|
|
}
|
|
nr, ok := SeekToFirstSequenceValue(base, increment, offset, MIN, MAX)
|
|
if !ok {
|
|
return 0, ErrAutoincReadFailed
|
|
}
|
|
// The rest of sequence < cache size, return the rest.
|
|
if MAX-nr < (size-1)*increment {
|
|
return MAX - base, nil
|
|
}
|
|
return (nr - base) + (size-1)*increment, nil
|
|
}
|
|
// The sequence is negative growth.
|
|
if increment == -1 {
|
|
if base <= MIN {
|
|
return 0, ErrAutoincReadFailed
|
|
}
|
|
if base-MIN < size {
|
|
return base - MIN, nil
|
|
}
|
|
return size, nil
|
|
}
|
|
nr, ok := SeekToFirstSequenceValue(base, increment, offset, MIN, MAX)
|
|
if !ok {
|
|
return 0, ErrAutoincReadFailed
|
|
}
|
|
// The rest of sequence < cache size, return the rest.
|
|
if nr-MIN < (size-1)*(-increment) {
|
|
return base - MIN, nil
|
|
}
|
|
return (base - nr) + (size-1)*(-increment), nil
|
|
}
|
|
|
|
// SeekToFirstSequenceValue seeks to the next valid value (must be in range of [MIN, MAX]),
|
|
// the bool indicates whether the first value is got.
|
|
// The seeking formula is describe as below:
|
|
//
|
|
// nr := (base + increment - offset) / increment
|
|
//
|
|
// first := nr*increment + offset
|
|
// Because formula computation will overflow Int64, so we transfer it to uint64 for distance computation.
|
|
func SeekToFirstSequenceValue(base, increment, offset, MIN, MAX int64) (int64, bool) {
|
|
if increment > 0 {
|
|
// Sequence is already allocated to the end.
|
|
if base >= MAX {
|
|
return 0, false
|
|
}
|
|
uMax := EncodeIntToCmpUint(MAX)
|
|
uBase := EncodeIntToCmpUint(base)
|
|
uOffset := EncodeIntToCmpUint(offset)
|
|
uIncrement := uint64(increment)
|
|
if uMax-uBase < uIncrement {
|
|
// Enum the possible first value.
|
|
for i := uBase + 1; i <= uMax; i++ {
|
|
if (i-uOffset)%uIncrement == 0 {
|
|
return DecodeCmpUintToInt(i), true
|
|
}
|
|
}
|
|
return 0, false
|
|
}
|
|
nr := (uBase + uIncrement - uOffset) / uIncrement
|
|
nr = nr*uIncrement + uOffset
|
|
first := DecodeCmpUintToInt(nr)
|
|
return first, true
|
|
}
|
|
// Sequence is already allocated to the end.
|
|
if base <= MIN {
|
|
return 0, false
|
|
}
|
|
uMin := EncodeIntToCmpUint(MIN)
|
|
uBase := EncodeIntToCmpUint(base)
|
|
uOffset := EncodeIntToCmpUint(offset)
|
|
uIncrement := uint64(-increment)
|
|
if uBase-uMin < uIncrement {
|
|
// Enum the possible first value.
|
|
for i := uBase - 1; i >= uMin; i-- {
|
|
if (uOffset-i)%uIncrement == 0 {
|
|
return DecodeCmpUintToInt(i), true
|
|
}
|
|
}
|
|
return 0, false
|
|
}
|
|
nr := (uOffset - uBase + uIncrement) / uIncrement
|
|
nr = uOffset - nr*uIncrement
|
|
first := DecodeCmpUintToInt(nr)
|
|
return first, true
|
|
}
|
|
|
|
// SeekToFirstAutoIDSigned seeks to the next valid signed position.
|
|
func SeekToFirstAutoIDSigned(base, increment, offset int64) int64 {
|
|
nr := (base + increment - offset) / increment
|
|
nr = nr*increment + offset
|
|
return nr
|
|
}
|
|
|
|
// SeekToFirstAutoIDUnSigned seeks to the next valid unsigned position.
|
|
func SeekToFirstAutoIDUnSigned(base, increment, offset uint64) uint64 {
|
|
nr := (base + increment - offset) / increment
|
|
nr = nr*increment + offset
|
|
return nr
|
|
}
|
|
|
|
func (alloc *allocator) alloc4Signed(tableID int64, n uint64, increment, offset int64) (int64, int64, error) {
|
|
// Check offset rebase if necessary.
|
|
if offset-1 > alloc.base {
|
|
if err := alloc.rebase4Signed(tableID, offset-1, true); err != nil {
|
|
return 0, 0, err
|
|
}
|
|
}
|
|
// CalcNeededBatchSize calculates the total batch size needed.
|
|
n1 := CalcNeededBatchSize(alloc.base, int64(n), increment, offset, alloc.isUnsigned)
|
|
|
|
// Condition alloc.base+N1 > alloc.end will overflow when alloc.base + N1 > MaxInt64. So need this.
|
|
if math.MaxInt64-alloc.base <= n1 {
|
|
return 0, 0, ErrAutoincReadFailed
|
|
}
|
|
// The local rest is not enough for allocN, skip it.
|
|
if alloc.base+n1 > alloc.end {
|
|
var newBase, newEnd int64
|
|
startTime := time.Now()
|
|
nextStep := alloc.step
|
|
if !alloc.customStep {
|
|
// Although it may skip a segment here, we still think it is consumed.
|
|
consumeDur := startTime.Sub(alloc.lastAllocTime)
|
|
nextStep = NextStep(alloc.step, consumeDur)
|
|
}
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
var err1 error
|
|
newBase, err1 = getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
// CalcNeededBatchSize calculates the total batch size needed on global base.
|
|
n1 = CalcNeededBatchSize(newBase, int64(n), increment, offset, alloc.isUnsigned)
|
|
// Although the step is customized by user, we still need to make sure nextStep is big enough for insert batch.
|
|
if nextStep < n1 {
|
|
nextStep = n1
|
|
}
|
|
tmpStep := mathutil.MinInt64(math.MaxInt64-newBase, nextStep)
|
|
// The global rest is not enough for alloc.
|
|
if tmpStep < n1 {
|
|
return ErrAutoincReadFailed
|
|
}
|
|
newEnd, err1 = generateAutoIDByAllocType(m, alloc.dbID, tableID, tmpStep, alloc.allocType)
|
|
return err1
|
|
})
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDAlloc, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
// Store the step for non-customized-step allocator to calculate next dynamic step.
|
|
if !alloc.customStep {
|
|
alloc.step = nextStep
|
|
}
|
|
alloc.lastAllocTime = time.Now()
|
|
if newBase == math.MaxInt64 {
|
|
return 0, 0, ErrAutoincReadFailed
|
|
}
|
|
alloc.base, alloc.end = newBase, newEnd
|
|
}
|
|
logutil.Logger(context.TODO()).Debug("alloc N signed ID",
|
|
zap.Uint64("from ID", uint64(alloc.base)),
|
|
zap.Uint64("to ID", uint64(alloc.base+n1)),
|
|
zap.Int64("table ID", tableID),
|
|
zap.Int64("database ID", alloc.dbID))
|
|
min := alloc.base
|
|
alloc.base += n1
|
|
return min, alloc.base, nil
|
|
}
|
|
|
|
func (alloc *allocator) alloc4Unsigned(tableID int64, n uint64, increment, offset int64) (int64, int64, error) {
|
|
// Check offset rebase if necessary.
|
|
if uint64(offset-1) > uint64(alloc.base) {
|
|
if err := alloc.rebase4Unsigned(tableID, uint64(offset-1), true); err != nil {
|
|
return 0, 0, err
|
|
}
|
|
}
|
|
// CalcNeededBatchSize calculates the total batch size needed.
|
|
n1 := CalcNeededBatchSize(alloc.base, int64(n), increment, offset, alloc.isUnsigned)
|
|
|
|
// Condition alloc.base+n1 > alloc.end will overflow when alloc.base + n1 > MaxInt64. So need this.
|
|
if math.MaxUint64-uint64(alloc.base) <= uint64(n1) {
|
|
return 0, 0, ErrAutoincReadFailed
|
|
}
|
|
// The local rest is not enough for alloc, skip it.
|
|
if uint64(alloc.base)+uint64(n1) > uint64(alloc.end) {
|
|
var newBase, newEnd int64
|
|
startTime := time.Now()
|
|
nextStep := alloc.step
|
|
if !alloc.customStep {
|
|
// Although it may skip a segment here, we still treat it as consumed.
|
|
consumeDur := startTime.Sub(alloc.lastAllocTime)
|
|
nextStep = NextStep(alloc.step, consumeDur)
|
|
}
|
|
err := kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
var err1 error
|
|
newBase, err1 = getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
// CalcNeededBatchSize calculates the total batch size needed on new base.
|
|
n1 = CalcNeededBatchSize(newBase, int64(n), increment, offset, alloc.isUnsigned)
|
|
// Although the step is customized by user, we still need to make sure nextStep is big enough for insert batch.
|
|
if nextStep < n1 {
|
|
nextStep = n1
|
|
}
|
|
tmpStep := int64(mathutil.MinUint64(math.MaxUint64-uint64(newBase), uint64(nextStep)))
|
|
// The global rest is not enough for alloc.
|
|
if tmpStep < n1 {
|
|
return ErrAutoincReadFailed
|
|
}
|
|
newEnd, err1 = generateAutoIDByAllocType(m, alloc.dbID, tableID, tmpStep, alloc.allocType)
|
|
return err1
|
|
})
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDAlloc, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return 0, 0, err
|
|
}
|
|
// Store the step for non-customized-step allocator to calculate next dynamic step.
|
|
if !alloc.customStep {
|
|
alloc.step = nextStep
|
|
}
|
|
alloc.lastAllocTime = time.Now()
|
|
if uint64(newBase) == math.MaxUint64 {
|
|
return 0, 0, ErrAutoincReadFailed
|
|
}
|
|
alloc.base, alloc.end = newBase, newEnd
|
|
}
|
|
logutil.Logger(context.TODO()).Debug("alloc unsigned ID",
|
|
zap.Uint64(" from ID", uint64(alloc.base)),
|
|
zap.Uint64("to ID", uint64(alloc.base+n1)),
|
|
zap.Int64("table ID", tableID),
|
|
zap.Int64("database ID", alloc.dbID))
|
|
min := alloc.base
|
|
// Use uint64 n directly.
|
|
alloc.base = int64(uint64(alloc.base) + uint64(n1))
|
|
return min, alloc.base, nil
|
|
}
|
|
|
|
// alloc4Sequence is used to alloc value for sequence, there are several aspects different from autoid logic.
|
|
// 1: sequence allocation don't need check rebase.
|
|
// 2: sequence allocation don't need auto step.
|
|
// 3: sequence allocation may have negative growth.
|
|
// 4: sequence allocation batch length can be dissatisfied.
|
|
// 5: sequence batch allocation will be consumed immediately.
|
|
func (alloc *allocator) alloc4Sequence(tableID int64) (min int64, max int64, round int64, err error) {
|
|
increment := alloc.sequence.Increment
|
|
offset := alloc.sequence.Start
|
|
minValue := alloc.sequence.MinValue
|
|
maxValue := alloc.sequence.MaxValue
|
|
cacheSize := alloc.sequence.CacheValue
|
|
if !alloc.sequence.Cache {
|
|
cacheSize = 1
|
|
}
|
|
|
|
var newBase, newEnd int64
|
|
startTime := time.Now()
|
|
err = kv.RunInNewTxn(alloc.store, true, func(txn kv.Transaction) error {
|
|
m := meta.NewMeta(txn)
|
|
var (
|
|
err1 error
|
|
seqStep int64
|
|
)
|
|
// Get the real offset if the sequence is in cycle.
|
|
// round is used to count cycle times in sequence with cycle option.
|
|
if alloc.sequence.Cycle {
|
|
// GetSequenceCycle is used to get the flag `round`, which indicates whether the sequence is already in cycle.
|
|
round, err1 = m.GetSequenceCycle(alloc.dbID, tableID)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
if round > 0 {
|
|
if increment > 0 {
|
|
offset = alloc.sequence.MinValue
|
|
} else {
|
|
offset = alloc.sequence.MaxValue
|
|
}
|
|
}
|
|
}
|
|
|
|
// Get the global new base.
|
|
newBase, err1 = getAutoIDByAllocType(m, alloc.dbID, tableID, alloc.allocType)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
|
|
// CalcNeededBatchSize calculates the total batch size needed.
|
|
seqStep, err1 = CalcSequenceBatchSize(newBase, cacheSize, increment, offset, minValue, maxValue)
|
|
|
|
if err1 != nil && err1 == ErrAutoincReadFailed {
|
|
if !alloc.sequence.Cycle {
|
|
return err1
|
|
}
|
|
// Reset the sequence base and offset.
|
|
if alloc.sequence.Increment > 0 {
|
|
newBase = alloc.sequence.MinValue - 1
|
|
offset = alloc.sequence.MinValue
|
|
} else {
|
|
newBase = alloc.sequence.MaxValue + 1
|
|
offset = alloc.sequence.MaxValue
|
|
}
|
|
err1 = m.SetSequenceValue(alloc.dbID, tableID, newBase)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
|
|
// Reset sequence round state value.
|
|
round++
|
|
// SetSequenceCycle is used to store the flag `round` which indicates whether the sequence is already in cycle.
|
|
// round > 0 means the sequence is already in cycle, so the offset should be minvalue / maxvalue rather than sequence.start.
|
|
// TiDB is a stateless node, it should know whether the sequence is already in cycle when restart.
|
|
err1 = m.SetSequenceCycle(alloc.dbID, tableID, round)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
|
|
// Recompute the sequence next batch size.
|
|
seqStep, err1 = CalcSequenceBatchSize(newBase, cacheSize, increment, offset, minValue, maxValue)
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
}
|
|
var delta int64
|
|
if alloc.sequence.Increment > 0 {
|
|
delta = seqStep
|
|
} else {
|
|
delta = -seqStep
|
|
}
|
|
newEnd, err1 = generateAutoIDByAllocType(m, alloc.dbID, tableID, delta, alloc.allocType)
|
|
return err1
|
|
})
|
|
|
|
// TODO: sequence metrics
|
|
metrics.AutoIDHistogram.WithLabelValues(metrics.TableAutoIDAlloc, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
|
|
if err != nil {
|
|
return 0, 0, 0, err
|
|
}
|
|
logutil.Logger(context.TODO()).Debug("alloc sequence value",
|
|
zap.Uint64(" from value", uint64(newBase)),
|
|
zap.Uint64("to value", uint64(newEnd)),
|
|
zap.Int64("table ID", tableID),
|
|
zap.Int64("database ID", alloc.dbID))
|
|
return newBase, newEnd, round, nil
|
|
}
|
|
|
|
func getAutoIDByAllocType(m *meta.Meta, dbID, tableID int64, allocType AllocatorType) (int64, error) {
|
|
switch allocType {
|
|
// Currently, row id allocator and auto-increment value allocator shares the same key-value pair.
|
|
case RowIDAllocType, AutoIncrementType:
|
|
return m.GetAutoTableID(dbID, tableID)
|
|
case AutoRandomType:
|
|
return m.GetAutoRandomID(dbID, tableID)
|
|
case SequenceType:
|
|
return m.GetSequenceValue(dbID, tableID)
|
|
default:
|
|
return 0, ErrInvalidAllocatorType.GenWithStackByArgs()
|
|
}
|
|
}
|
|
|
|
func generateAutoIDByAllocType(m *meta.Meta, dbID, tableID, step int64, allocType AllocatorType) (int64, error) {
|
|
switch allocType {
|
|
case RowIDAllocType, AutoIncrementType:
|
|
return m.GenAutoTableID(dbID, tableID, step)
|
|
case AutoRandomType:
|
|
return m.GenAutoRandomID(dbID, tableID, step)
|
|
case SequenceType:
|
|
return m.GenSequenceValue(dbID, tableID, step)
|
|
default:
|
|
return 0, ErrInvalidAllocatorType.GenWithStackByArgs()
|
|
}
|
|
}
|
|
|
|
const signMask uint64 = 0x8000000000000000
|
|
|
|
// EncodeIntToCmpUint make int v to comparable uint type
|
|
func EncodeIntToCmpUint(v int64) uint64 {
|
|
return uint64(v) ^ signMask
|
|
}
|
|
|
|
// DecodeCmpUintToInt decodes the u that encoded by EncodeIntToCmpUint
|
|
func DecodeCmpUintToInt(u uint64) int64 {
|
|
return int64(u ^ signMask)
|
|
}
|
|
|
|
// TestModifyBaseAndEndInjection exported for testing modifying the base and end.
|
|
func TestModifyBaseAndEndInjection(alloc Allocator, base, end int64) {
|
|
alloc.(*allocator).mu.Lock()
|
|
alloc.(*allocator).base = base
|
|
alloc.(*allocator).end = end
|
|
alloc.(*allocator).mu.Unlock()
|
|
}
|
|
|
|
// AutoRandomIDLayout is used to calculate the bits length of different section in auto_random id.
|
|
// The primary key with auto_random can only be `bigint` column, the total layout length of auto random is 64 bits.
|
|
// These are two type of layout:
|
|
// 1. Signed bigint:
|
|
// | [sign_bit] | [shard_bits] | [incremental_bits] |
|
|
// sign_bit(1 fixed) + shard_bits(15 max) + incremental_bits(the rest) = total_layout_bits(64 fixed)
|
|
// 2. Unsigned bigint:
|
|
// | [shard_bits] | [incremental_bits] |
|
|
// shard_bits(15 max) + incremental_bits(the rest) = total_layout_bits(64 fixed)
|
|
//
|
|
// Please always use NewAutoRandomIDLayout() to instantiate.
|
|
type AutoRandomIDLayout struct {
|
|
FieldType *types.FieldType
|
|
ShardBits uint64
|
|
// Derived fields.
|
|
TypeBitsLength uint64
|
|
IncrementalBits uint64
|
|
HasSignBit bool
|
|
}
|
|
|
|
// NewAutoRandomIDLayout create an instance of AutoRandomIDLayout.
|
|
func NewAutoRandomIDLayout(fieldType *types.FieldType, shardBits uint64) *AutoRandomIDLayout {
|
|
typeBitsLength := uint64(mysql.DefaultLengthOfMysqlTypes[mysql.TypeLonglong] * 8)
|
|
incrementalBits := typeBitsLength - shardBits
|
|
hasSignBit := !mysql.HasUnsignedFlag(fieldType.Flag)
|
|
if hasSignBit {
|
|
incrementalBits -= 1
|
|
}
|
|
return &AutoRandomIDLayout{
|
|
FieldType: fieldType,
|
|
ShardBits: shardBits,
|
|
TypeBitsLength: typeBitsLength,
|
|
IncrementalBits: incrementalBits,
|
|
HasSignBit: hasSignBit,
|
|
}
|
|
}
|
|
|
|
// IncrementalBitsCapacity returns the max capacity of incremental section of the current layout.
|
|
func (l *AutoRandomIDLayout) IncrementalBitsCapacity() uint64 {
|
|
return uint64(math.Pow(2, float64(l.IncrementalBits))) - 1
|
|
}
|
|
|
|
// IncrementalMask returns 00..0[11..1], where [xxx] is the incremental section of the current layout.
|
|
func (l *AutoRandomIDLayout) IncrementalMask() int64 {
|
|
return (1 << l.IncrementalBits) - 1
|
|
}
|
|
|