Skip to content

Commit

Permalink
Merge branch 'master' into fk-plan2
Browse files Browse the repository at this point in the history
  • Loading branch information
crazycs520 authored Dec 2, 2022
2 parents fc7bc74 + 40d1ddb commit f34af39
Show file tree
Hide file tree
Showing 59 changed files with 10,862 additions and 9,609 deletions.
9 changes: 9 additions & 0 deletions bindinfo/bind_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1428,14 +1428,23 @@ func TestCreateBindingFromHistory(t *testing.T) {
tk.MustExec(bindSQL)
planDigest := tk.MustQuery(fmt.Sprintf("select plan_digest from information_schema.statements_summary where query_sample_text = '%s'", bindSQL)).Rows()
tk.MustExec(fmt.Sprintf("create session binding from history using plan digest '%s'", planDigest[0][0]))
showRes := tk.MustQuery("show bindings").Rows()
require.Equal(t, len(showRes), 1)
require.Equal(t, planDigest[0][0], showRes[0][10])
for _, sql := range testCase.sqls {
tk.MustExec(fmt.Sprintf(sql, ""))
tk.MustQuery("select @@last_plan_from_binding").Check(testkit.Rows("1"))
}
}
showRes := tk.MustQuery("show bindings").Rows()
require.Equal(t, len(showRes), 1)
tk.MustExec(fmt.Sprintf("drop binding for sql digest '%s'", showRes[0][9]))
}

// exception cases
tk.MustGetErrMsg(fmt.Sprintf("create binding from history using plan digest '%s'", "1"), "can't find any plans for '1'")
tk.MustGetErrMsg(fmt.Sprintf("create binding from history using plan digest '%s'", ""), "plan digest is empty")
tk.MustExec("create binding for select * from t1, t2 where t1.id = t2.id using select /*+ merge_join(t1, t2) */ * from t1, t2 where t1.id = t2.id")
showRes := tk.MustQuery("show bindings").Rows()
require.Equal(t, showRes[0][10], "") // plan digest should be nil by create for
}
8 changes: 4 additions & 4 deletions br/pkg/storage/s3.go
Original file line number Diff line number Diff line change
Expand Up @@ -399,7 +399,7 @@ func NewS3Storage(backend *backuppb.S3, opts *ExternalStorageOptions) (obj *S3St
options: &qs,
}
if opts.CheckS3ObjectLockOptions {
backend.ObjectLockEnabled = s3Storage.isObjectLockEnabled()
backend.ObjectLockEnabled = s3Storage.IsObjectLockEnabled()
}
return s3Storage, nil
}
Expand Down Expand Up @@ -446,7 +446,7 @@ func getObject(svc *s3.S3, qs *backuppb.S3) error {
return nil
}

func (rs *S3Storage) isObjectLockEnabled() bool {
func (rs *S3Storage) IsObjectLockEnabled() bool {
input := &s3.GetObjectLockConfigurationInput{
Bucket: aws.String(rs.options.Bucket),
}
Expand All @@ -455,8 +455,8 @@ func (rs *S3Storage) isObjectLockEnabled() bool {
log.Warn("failed to check object lock for bucket", zap.String("bucket", rs.options.Bucket), zap.Error(err))
return false
}
if resp.ObjectLockConfiguration != nil {
if s3.ObjectLockEnabledEnabled == *resp.ObjectLockConfiguration.ObjectLockEnabled {
if resp != nil && resp.ObjectLockConfiguration != nil {
if s3.ObjectLockEnabledEnabled == aws.StringValue(resp.ObjectLockConfiguration.ObjectLockEnabled) {
return true
}
}
Expand Down
45 changes: 45 additions & 0 deletions br/pkg/storage/s3_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1141,3 +1141,48 @@ func TestSendCreds(t *testing.T) {
sentSecretAccessKey = backend.GetS3().SecretAccessKey
require.Equal(t, "", sentSecretAccessKey)
}

func TestObjectLock(t *testing.T) {
s := createS3Suite(t)
// resp is nil
s.s3.EXPECT().GetObjectLockConfiguration(gomock.Any()).Return(nil, nil)
require.Equal(t, false, s.storage.IsObjectLockEnabled())

// resp is not nil, but resp.ObjectLockConfiguration is nil
s.s3.EXPECT().GetObjectLockConfiguration(gomock.Any()).Return(
&s3.GetObjectLockConfigurationOutput{
ObjectLockConfiguration: nil,
}, nil,
)
require.Equal(t, false, s.storage.IsObjectLockEnabled())

// resp.ObjectLockConfiguration is not nil, but resp.ObjectLockConfiguration.ObjectLockEnabled is nil
s.s3.EXPECT().GetObjectLockConfiguration(gomock.Any()).Return(
&s3.GetObjectLockConfigurationOutput{
ObjectLockConfiguration: &s3.ObjectLockConfiguration{
ObjectLockEnabled: nil,
},
}, nil,
)
require.Equal(t, false, s.storage.IsObjectLockEnabled())

// resp.ObjectLockConfiguration.ObjectLockEnabled is illegal string
s.s3.EXPECT().GetObjectLockConfiguration(gomock.Any()).Return(
&s3.GetObjectLockConfigurationOutput{
ObjectLockConfiguration: &s3.ObjectLockConfiguration{
ObjectLockEnabled: aws.String("EnaBled"),
},
}, nil,
)
require.Equal(t, false, s.storage.IsObjectLockEnabled())

// resp.ObjectLockConfiguration.ObjectLockEnabled is enabled
s.s3.EXPECT().GetObjectLockConfiguration(gomock.Any()).Return(
&s3.GetObjectLockConfigurationOutput{
ObjectLockConfiguration: &s3.ObjectLockConfiguration{
ObjectLockEnabled: aws.String("Enabled"),
},
}, nil,
)
require.Equal(t, true, s.storage.IsObjectLockEnabled())
}
3 changes: 2 additions & 1 deletion config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,8 @@ const (
// DefTxnEntrySizeLimit is the default value of TxnEntrySizeLimit.
DefTxnEntrySizeLimit = 6 * 1024 * 1024
// DefTxnTotalSizeLimit is the default value of TxnTxnTotalSizeLimit.
DefTxnTotalSizeLimit = 100 * 1024 * 1024
DefTxnTotalSizeLimit = 100 * 1024 * 1024
SuperLargeTxnSize uint64 = 100 * 1024 * 1024 * 1024 * 1024 // 100T, we expect a txn can never be this large
// DefMaxIndexLength is the maximum index length(in bytes). This value is consistent with MySQL.
DefMaxIndexLength = 3072
// DefMaxOfMaxIndexLength is the maximum index length(in bytes) for TiDB v3.0.7 and previous version.
Expand Down
22 changes: 5 additions & 17 deletions ddl/db_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3036,21 +3036,6 @@ func TestAutoIncrementForceAutoIDCache(t *testing.T) {
return gid
}

// Rebase _tidb_row_id.
tk.MustExec("create table t (a int) AUTO_ID_CACHE 1")
tk.MustExec("alter table t force auto_increment = 2;")
tk.MustExec("insert into t values (1),(2);")
tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("1 1", "2 2"))
// Cannot set next global ID to 0.
tk.MustExec("alter table t force auto_increment = 0;")
tk.MustExec("alter table t force auto_increment = 1;")
require.Equal(t, uint64(3), getNextGlobalID())
// inserting new rows can overwrite the existing data.
tk.MustExec("insert into t values (3);")
tk.MustExec("insert into t values (3);")
tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("1 1", "2 2", "3 3", "3 4"))
tk.MustExec("drop table if exists t;")

// When AUTO_ID_CACHE is 1, row id and auto increment id use separate allocator, so the behaviour differs.
// "Alter table t force auto_increment" has no effect on row id.
tk.MustExec("create table t (a int) AUTO_ID_CACHE 1")
Expand All @@ -3060,11 +3045,14 @@ func TestAutoIncrementForceAutoIDCache(t *testing.T) {
// Cannot set next global ID to 0.
tk.MustExec("alter table t force auto_increment = 0;")
tk.MustExec("alter table t force auto_increment = 1;")
require.Equal(t, uint64(3), getNextGlobalID())
tk.MustQuery("show table t next_row_id").Check(testkit.Rows(
"auto_inc_force t _tidb_rowid 5001 _TIDB_ROWID",
))

// inserting new rows can overwrite the existing data.
tk.MustExec("insert into t values (3);")
tk.MustExec("insert into t values (3);")
tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("1 1", "2 2", "3 3", "3 4"))
tk.MustQuery("select a, _tidb_rowid from t;").Check(testkit.Rows("1 1", "2 2", "3 5001", "3 5002"))
tk.MustExec("drop table if exists t;")

// Rebase auto_increment.
Expand Down
1 change: 1 addition & 0 deletions errno/errcode.go
Original file line number Diff line number Diff line change
Expand Up @@ -892,6 +892,7 @@ const (
ErrIllegalPrivilegeLevel = 3619
ErrCTEMaxRecursionDepth = 3636
ErrNotHintUpdatable = 3637
ErrExistsInHistoryPassword = 3638
ErrForeignKeyCannotDropParent = 3730
ErrForeignKeyCannotUseVirtualColumn = 3733
ErrForeignKeyNoColumnInParent = 3734
Expand Down
1 change: 1 addition & 0 deletions errno/errname.go
Original file line number Diff line number Diff line change
Expand Up @@ -887,6 +887,7 @@ var MySQLErrName = map[uint16]*mysql.ErrMessage{
ErrMaxExecTimeExceeded: mysql.Message("Query execution was interrupted, max_execution_time exceeded.", nil),
ErrLockAcquireFailAndNoWaitSet: mysql.Message("Statement aborted because lock(s) could not be acquired immediately and NOWAIT is set.", nil),
ErrNotHintUpdatable: mysql.Message("Variable '%s' cannot be set using SET_VAR hint.", nil),
ErrExistsInHistoryPassword: mysql.Message("Cannot use these credentials for '%s@%s' because they contradict the password history policy.", nil),
ErrForeignKeyCannotDropParent: mysql.Message("Cannot drop table '%s' referenced by a foreign key constraint '%s' on table '%s'.", nil),
ErrForeignKeyCannotUseVirtualColumn: mysql.Message("Foreign key '%s' uses virtual column '%s' which is not supported.", nil),
ErrForeignKeyNoColumnInParent: mysql.Message("Failed to add the foreign key constraint. Missing column '%s' for constraint '%s' in the referenced table '%s'", nil),
Expand Down
5 changes: 5 additions & 0 deletions errors.toml
Original file line number Diff line number Diff line change
Expand Up @@ -1506,6 +1506,11 @@ error = '''
Recursive query aborted after %d iterations. Try increasing @@cte_max_recursion_depth to a larger value
'''

["executor:3638"]
error = '''
Cannot use these credentials for '%s@%s' because they contradict the password history policy.
'''

["executor:3929"]
error = '''
Dynamic privilege '%s' is not registered with the server.
Expand Down
7 changes: 7 additions & 0 deletions executor/analyze_col_v2.go
Original file line number Diff line number Diff line change
Expand Up @@ -595,6 +595,13 @@ func (e *AnalyzeColumnsExecV2) subMergeWorker(resultCh chan<- *samplingMergeResu
failpoint.Inject("mockAnalyzeSamplingMergeWorkerPanic", func() {
panic("failpoint triggered")
})
failpoint.Inject("mockAnalyzeMergeWorkerSlowConsume", func(val failpoint.Value) {
times := val.(int)
for i := 0; i < times; i++ {
e.memTracker.Consume(5 << 20)
time.Sleep(100 * time.Millisecond)
}
})
retCollector := statistics.NewRowSampleCollector(int(e.analyzePB.ColReq.SampleSize), e.analyzePB.ColReq.GetSampleRate(), l)
for i := 0; i < l; i++ {
retCollector.Base().FMSketches = append(retCollector.Base().FMSketches, statistics.NewFMSketch(maxSketchSize))
Expand Down
11 changes: 9 additions & 2 deletions executor/analyze_utils.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,15 @@ package executor
import (
"context"
"strconv"
"strings"
"sync"

"github.com/pingcap/errors"
"github.com/pingcap/tidb/config"
"github.com/pingcap/tidb/sessionctx"
"github.com/pingcap/tidb/sessionctx/variable"
"github.com/pingcap/tidb/statistics"
"github.com/pingcap/tidb/util/memory"
"go.uber.org/atomic"
)

Expand All @@ -45,8 +47,13 @@ func isAnalyzeWorkerPanic(err error) bool {
}

func getAnalyzePanicErr(r interface{}) error {
if msg, ok := r.(string); ok && msg == globalPanicAnalyzeMemoryExceed {
return errAnalyzeOOM
if msg, ok := r.(string); ok {
if msg == globalPanicAnalyzeMemoryExceed {
return errAnalyzeOOM
}
if strings.Contains(msg, memory.PanicMemoryExceed) {
return errors.Errorf(msg, errAnalyzeOOM)
}
}
if err, ok := r.(error); ok {
if err.Error() == globalPanicAnalyzeMemoryExceed {
Expand Down
18 changes: 18 additions & 0 deletions executor/autoidtest/autoid_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -749,3 +749,21 @@ func TestMockAutoIDServiceError(t *testing.T) {
// Cover a bug that the autoid client retry non-retryable errors forever cause dead loop.
tk.MustExecToErr("insert into t_mock_err values (),()") // mock error, instead of dead loop
}

func TestIssue39528(t *testing.T) {
// When AUTO_ID_CACHE is 1, it should not affect row id setting when autoid and rowid are separated.
store := testkit.CreateMockStore(t)
tk := testkit.NewTestKit(t, store)
tk.MustExec("use test;")
tk.MustExec("create table issue39528 (id int unsigned key nonclustered auto_increment) shard_row_id_bits=4 auto_id_cache 1;")
tk.MustExec("insert into issue39528 values ()")
tk.MustExec("insert into issue39528 values ()")

ctx := context.Background()
var codeRun bool
ctx = context.WithValue(ctx, "testIssue39528", &codeRun)
_, err := tk.ExecWithContext(ctx, "insert into issue39528 values ()")
require.NoError(t, err)
// Make sure the code does not visit tikv on allocate path.
require.False(t, codeRun)
}
5 changes: 3 additions & 2 deletions executor/benchmark_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -913,6 +913,7 @@ func prepare4HashJoin(testCase *hashJoinTestCase, innerExec, outerExec Executor)
e := &HashJoinExec{
baseExecutor: newBaseExecutor(testCase.ctx, joinSchema, 5, innerExec, outerExec),
hashJoinCtx: &hashJoinCtx{
sessCtx: testCase.ctx,
joinType: testCase.joinType, // 0 for InnerJoin, 1 for LeftOutersJoin, 2 for RightOuterJoin
isOuterJoin: false,
useOuterToBuild: testCase.useOuterToBuild,
Expand All @@ -936,13 +937,13 @@ func prepare4HashJoin(testCase *hashJoinTestCase, innerExec, outerExec Executor)
for i := uint(0); i < e.concurrency; i++ {
e.probeWorkers[i] = &probeWorker{
workerID: i,
sessCtx: e.ctx,
hashJoinCtx: e.hashJoinCtx,
joiner: newJoiner(testCase.ctx, e.joinType, true, defaultValues,
nil, lhsTypes, rhsTypes, childrenUsedSchema, false),
probeKeyColIdx: probeKeysColIdx,
}
}
e.buildWorker.hashJoinCtx = e.hashJoinCtx
memLimit := int64(-1)
if testCase.disk {
memLimit = 1
Expand Down Expand Up @@ -1200,7 +1201,7 @@ func benchmarkBuildHashTable(b *testing.B, casTest *hashJoinTestCase, dataSource
close(innerResultCh)

b.StartTimer()
if err := exec.buildHashTableForList(innerResultCh); err != nil {
if err := exec.buildWorker.buildHashTableForList(innerResultCh); err != nil {
b.Fatal(err)
}

Expand Down
13 changes: 7 additions & 6 deletions executor/bind.go
Original file line number Diff line number Diff line change
Expand Up @@ -128,12 +128,13 @@ func (e *SQLBindExec) createSQLBind() error {
}()

bindInfo := bindinfo.Binding{
BindSQL: e.bindSQL,
Charset: e.charset,
Collation: e.collation,
Status: bindinfo.Enabled,
Source: e.source,
SQLDigest: e.sqlDigest,
BindSQL: e.bindSQL,
Charset: e.charset,
Collation: e.collation,
Status: bindinfo.Enabled,
Source: e.source,
SQLDigest: e.sqlDigest,
PlanDigest: e.planDigest,
}
record := &bindinfo.BindRecord{
OriginalSQL: e.normdOrigSQL,
Expand Down
6 changes: 4 additions & 2 deletions executor/builder.go
Original file line number Diff line number Diff line change
Expand Up @@ -1417,12 +1417,14 @@ func (b *executorBuilder) buildHashJoin(v *plannercore.PhysicalHashJoin) Executo
probeWorkers: make([]*probeWorker, v.Concurrency),
buildWorker: &buildWorker{},
hashJoinCtx: &hashJoinCtx{
sessCtx: b.ctx,
isOuterJoin: v.JoinType.IsOuterJoin(),
useOuterToBuild: v.UseOuterToBuild,
joinType: v.JoinType,
concurrency: v.Concurrency,
},
}
e.hashJoinCtx.allocPool = e.AllocPool
defaultValues := v.DefaultValues
lhsTypes, rhsTypes := retTypes(leftExec), retTypes(rightExec)
if v.InnerChildIdx == 1 {
Expand Down Expand Up @@ -1494,13 +1496,12 @@ func (b *executorBuilder) buildHashJoin(v *plannercore.PhysicalHashJoin) Executo
e.probeWorkers[i] = &probeWorker{
hashJoinCtx: e.hashJoinCtx,
workerID: i,
sessCtx: e.ctx,
joiner: newJoiner(b.ctx, v.JoinType, v.InnerChildIdx == 0, defaultValues, v.OtherConditions, lhsTypes, rhsTypes, childrenUsedSchema, isNAJoin),
probeKeyColIdx: probeKeyColIdx,
probeNAKeyColIdx: probeNAKeColIdx,
}
}
e.buildWorker.buildKeyColIdx, e.buildWorker.buildNAKeyColIdx, e.buildWorker.buildSideExec = buildKeyColIdx, buildNAKeyColIdx, buildSideExec
e.buildWorker.buildKeyColIdx, e.buildWorker.buildNAKeyColIdx, e.buildWorker.buildSideExec, e.buildWorker.hashJoinCtx = buildKeyColIdx, buildNAKeyColIdx, buildSideExec, e.hashJoinCtx
e.hashJoinCtx.isNullAware = isNAJoin
executorCountHashJoinExec.Inc()

Expand Down Expand Up @@ -4817,6 +4818,7 @@ func (b *executorBuilder) buildSQLBindExec(v *plannercore.SQLBindPlan) Executor
newStatus: v.NewStatus,
source: v.Source,
sqlDigest: v.SQLDigest,
planDigest: v.PlanDigest,
}
return e
}
Expand Down
1 change: 1 addition & 0 deletions executor/errors.go
Original file line number Diff line number Diff line change
Expand Up @@ -73,4 +73,5 @@ var (
ErrWrongStringLength = dbterror.ClassDDL.NewStd(mysql.ErrWrongStringLength)
errUnsupportedFlashbackTmpTable = dbterror.ClassDDL.NewStdErr(mysql.ErrUnsupportedDDLOperation, parser_mysql.Message("Recover/flashback table is not supported on temporary tables", nil))
errTruncateWrongInsertValue = dbterror.ClassTable.NewStdErr(mysql.ErrTruncatedWrongValue, parser_mysql.Message("Incorrect %-.32s value: '%-.128s' for column '%.192s' at row %d", nil))
ErrExistsInHistoryPassword = dbterror.ClassExecutor.NewStd(mysql.ErrExistsInHistoryPassword)
)
27 changes: 14 additions & 13 deletions executor/executor.go
Original file line number Diff line number Diff line change
Expand Up @@ -1975,21 +1975,22 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
if _, ok := s.(*ast.AnalyzeTableStmt); ok {
sc.InitMemTracker(memory.LabelForAnalyzeMemory, -1)
vars.MemTracker.SetBytesLimit(-1)
vars.MemTracker.AttachTo(GlobalAnalyzeMemoryTracker)
} else {
sc.InitMemTracker(memory.LabelForSQLText, -1)
logOnQueryExceedMemQuota := domain.GetDomain(ctx).ExpensiveQueryHandle().LogOnQueryExceedMemQuota
switch variable.OOMAction.Load() {
case variable.OOMActionCancel:
action := &memory.PanicOnExceed{ConnID: vars.ConnectionID}
action.SetLogHook(logOnQueryExceedMemQuota)
vars.MemTracker.SetActionOnExceed(action)
case variable.OOMActionLog:
fallthrough
default:
action := &memory.LogOnExceed{ConnID: vars.ConnectionID}
action.SetLogHook(logOnQueryExceedMemQuota)
vars.MemTracker.SetActionOnExceed(action)
}
}
logOnQueryExceedMemQuota := domain.GetDomain(ctx).ExpensiveQueryHandle().LogOnQueryExceedMemQuota
switch variable.OOMAction.Load() {
case variable.OOMActionCancel:
action := &memory.PanicOnExceed{ConnID: vars.ConnectionID}
action.SetLogHook(logOnQueryExceedMemQuota)
vars.MemTracker.SetActionOnExceed(action)
case variable.OOMActionLog:
fallthrough
default:
action := &memory.LogOnExceed{ConnID: vars.ConnectionID}
action.SetLogHook(logOnQueryExceedMemQuota)
vars.MemTracker.SetActionOnExceed(action)
}
sc.MemTracker.SessionID = vars.ConnectionID
sc.MemTracker.AttachTo(vars.MemTracker)
Expand Down
Loading

0 comments on commit f34af39

Please sign in to comment.