*: support create global binding

bindinfo/bind.go

@@ -0,0 +1,61 @@
+// Copyright 2019 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 bindinfo
+
+import (
+	"github.com/pingcap/tidb/sessionctx"
+)
+
+var _ Manager = (*BindManager)(nil)
+
+// BindManager implements Manager inferface.
+type BindManager struct {
+	GlobalAccessor GlobalBindAccessor
+}
+
+type keyType int
+
+func (k keyType) String() string {
+	return "bind-key"
+}
+
+// Manager is the interface for providing bind related operations.
+type Manager interface {
+	AddGlobalBind(originSQL, bindSQL, defaultDB, charset, collation string) error
+}
+
+const key keyType = 0
+
+// BindBinderManager binds Manager to context.
+func BindBinderManager(ctx sessionctx.Context, pc Manager) {
+	ctx.SetValue(key, pc)
+}
+
+// GetBindManager gets Checker from context.
+func GetBindManager(ctx sessionctx.Context) Manager {
+	if v, ok := ctx.Value(key).(Manager); ok {
+		return v
+	}
+	return nil
+}
+
+//AddGlobalBind implements Manager's AddGlobalBind interface.
+func (b *BindManager) AddGlobalBind(originSQL, bindSQL, defaultDB, charset, collation string) error {
+	return b.GlobalAccessor.AddGlobalBind(originSQL, bindSQL, defaultDB, charset, collation)
+}
+
+// GlobalBindAccessor is the interface for accessing global bind info.
+type GlobalBindAccessor interface {
+	AddGlobalBind(originSQL string, bindSQL string, defaultDB string, charset string, collation string) error
+}

bindinfo/bind_test.go

@@ -115,7 +115,7 @@ func (s *testSuite) TestBindParse(c *C) {
 	sql := fmt.Sprintf(`INSERT INTO mysql.bind_info(original_sql,bind_sql,default_db,status,create_time,update_time,charset,collation) VALUES ('%s', '%s', '%s', '%s', NOW(), NOW(),'%s', '%s')`,
 		originSQL, bindSQL, defaultDb, status, charset, collation)
 	tk.MustExec(sql)
-	bindHandle := bindinfo.NewHandle()
+	bindHandle := bindinfo.NewHandle(tk.Se)
 	bindCacheUpdater := bindinfo.NewBindCacheUpdater(tk.Se, bindHandle, s.Parser)
 	err := bindCacheUpdater.Update(true)
 	c.Check(err, IsNil)
@@ -134,3 +134,44 @@ func (s *testSuite) TestBindParse(c *C) {
 	c.Check(bindData[0].CreateTime, NotNil)
 	c.Check(bindData[0].UpdateTime, NotNil)
 }
+
+func (s *testSuite) TestGlobalBinding(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	s.cleanBindingEnv(tk)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("create table t(i int, s varchar(20))")
+	tk.MustExec("create table t1(i int, s varchar(20))")
+	tk.MustExec("create index index_t on t(i,s)")
+
+	_, err := tk.Exec("create global binding for select * from t where i>100 using select * from t use index(index_t) where i>100")
+	c.Assert(err, IsNil, Commentf("err %v", err))
+	_, err = tk.Exec("create global binding for select * from t where i>99 using select * from t use index(index_t) where i>99")
+	c.Assert(err, NotNil)
+
+	bindHandle := bindinfo.NewHandle(tk.Se)
+	bindCacheUpdater := bindinfo.NewBindCacheUpdater(tk.Se, bindHandle, s.Parser)
+	err = bindCacheUpdater.Update(true)
+	c.Check(err, IsNil)
+	c.Check(len(bindHandle.Get()), Equals, 1)
+
+	hash := parser.DigestHash("select * from t where i>100")
+	bindData := bindHandle.Get()[hash]
+	c.Check(bindData, NotNil)
+	c.Check(len(bindData), Equals, 1)
+	c.Check(bindData[0].OriginalSQL, Equals, "select * from t where i > ?")
+	c.Check(bindData[0].BindSQL, Equals, "select * from t use index(index_t) where i>100")
+	c.Check(bindData[0].Db, Equals, "test")
+	c.Check(bindData[0].Status, Equals, "using")
+	c.Check(bindData[0].Charset, NotNil)
+	c.Check(bindData[0].Collation, NotNil)
+	c.Check(bindData[0].CreateTime, NotNil)
+	c.Check(bindData[0].UpdateTime, NotNil)
+
+	_, err = tk.Exec("delete from mysql.bind_info")
+	c.Assert(err, IsNil)
+
+	_, err = tk.Exec("create global binding for select * from t using select * from t1 use index for join(index_t)")
+	c.Assert(err, NotNil, Commentf("err %v", err))
+}

bindinfo/cache.go

@@ -14,23 +14,28 @@
 package bindinfo
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"sync"
 	"sync/atomic"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/sessionctx"
+	"github.com/pingcap/tidb/store/tikv/oracle"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/chunk"
 	"github.com/pingcap/tidb/util/sqlexec"
 )
 
 const (
-	using   = "using"
-	deleted = "deleted"
+	// BindUsing is the bind info's in use status.
+	BindUsing = "using"
+	// BindDeleted is the bind info's deleted status.
+	BindDeleted = "deleted"
 )
 
 // bindMeta stores the basic bind info and bindSql astNode.
@@ -45,6 +50,9 @@ type cache map[string][]*bindMeta
 // Handle holds an atomic cache.
 type Handle struct {
 	atomic.Value
+	lock sync.Mutex
+	ctx  sessionctx.Context
+	exec sqlexec.SQLExecutor
 }
 
 // BindCacheUpdater is used to update the global cache.
@@ -83,8 +91,13 @@ func NewBindCacheUpdater(ctx sessionctx.Context, handle *Handle, parser *parser.
 }
 
 // NewHandle creates a Handle with a cache.
-func NewHandle() *Handle {
-	handle := &Handle{}
+func NewHandle(ctx sessionctx.Context) *Handle {
+	exec, _ := ctx.(sqlexec.SQLExecutor)
+	handle := &Handle{
+		ctx:  ctx,
+		exec: exec,
+	}
+
 	return handle
 }
 
@@ -177,7 +190,7 @@ func (b cache) appendNode(newBindRecord *bindRecord, sparser *parser.Parser) err
 			}
 		}
 	}
-	if newBindRecord.Status == deleted {
+	if newBindRecord.Status == BindDeleted {
 		return nil
 	}
 	stmtNodes, _, err := sparser.Parse(newBindRecord.BindSQL, newBindRecord.Charset, newBindRecord.Collation)
@@ -191,3 +204,77 @@ func (b cache) appendNode(newBindRecord *bindRecord, sparser *parser.Parser) err
 	b[hash] = append(b[hash], newNode)
 	return nil
 }
+
+// AddGlobalBind implements GlobalBindAccessor.AddGlobalBind interface.
+func (h *Handle) AddGlobalBind(originSQL, bindSQL, defaultDB, charset, collation string) error {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	ctx := context.TODO()
+	_, err := h.exec.Execute(ctx, "BEGIN")
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer func() {
+		if err == nil {
+			_, err = h.exec.Execute(ctx, "COMMIT")
+		} else {
+			_, rbErr := h.exec.Execute(ctx, "ROLLBACK")
+			terror.Log(errors.Trace(rbErr))
+		}
+	}()
+
+	sql := fmt.Sprintf("SELECT status FROM mysql.bind_info WHERE original_sql='%s' AND default_db='%s'",
+		originSQL, defaultDB)
+	rs, err := h.exec.Execute(ctx, sql)
+	if err != nil {
+		return err
+	}
+	if len(rs) == 1 {
+		chkBatch := rs[0].NewRecordBatch()
+		for {
+			err = rs[0].Next(context.TODO(), chkBatch)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			if chkBatch.NumRows() == 0 {
+				break
+			}
+			it := chunk.NewIterator4Chunk(chkBatch.Chunk)
+			for row := it.Begin(); row != it.End(); row = it.Next() {
+				status := row.GetString(0)
+				if status == BindUsing {
+					err = errors.New("origin sql already has binding sql")
+					return err
+				}
+				sql = fmt.Sprintf("DELETE FROM mysql.bind_info WHERE original_sql='%s' and default_db='%s'", originSQL, defaultDB)
+				_, err = h.exec.Execute(ctx, sql)
+				if err != nil {
+					return errors.Trace(err)
+				}
+			}
+		}
+	}
+
+	txn, err := h.ctx.Txn(true)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	ts := oracle.GetTimeFromTS(txn.StartTS())
+	bindSQL = getEscapeCharacter(bindSQL)
+	sql = fmt.Sprintf(`INSERT INTO mysql.bind_info(original_sql,bind_sql,default_db,status,create_time,update_time,charset,collation) VALUES ('%s', '%s', '%s', '%s', '%s', '%s','%s', '%s')`,
+		originSQL, bindSQL, defaultDB, BindUsing, ts, ts, charset, collation)
+	_, err = h.exec.Execute(ctx, sql)
+	return errors.Trace(err)
+}
+
+func getEscapeCharacter(str string) string {
+	var buffer bytes.Buffer
+	for _, v := range str {
+		if v == '\'' || v == '"' || v == '\\' {
+			buffer.WriteString("\\")
+		}
+		buffer.WriteString(string(v))
+	}
+	return buffer.String()
+}

domain/domain.go

@@ -784,7 +784,7 @@ func (do *Domain) BindHandle() *bindinfo.Handle {
 // be called only once in BootstrapSession.
 func (do *Domain) LoadBindInfoLoop(ctx sessionctx.Context, parser *parser.Parser) error {
 	ctx.GetSessionVars().InRestrictedSQL = true
-	do.bindHandle = bindinfo.NewHandle()
+	do.bindHandle = bindinfo.NewHandle(ctx)
 
 	bindCacheUpdater := bindinfo.NewBindCacheUpdater(ctx, do.BindHandle(), parser)
 	err := bindCacheUpdater.Update(true)

executor/bind.go

@@ -0,0 +1,50 @@
+// Copyright 2019 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 executor
+
+import (
+	"context"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/ast"
+	"github.com/pingcap/tidb/bindinfo"
+	"github.com/pingcap/tidb/util/chunk"
+)
+
+// CreateBindExec represents a create bind executor.
+type CreateBindExec struct {
+	baseExecutor
+
+	originSQL string
+	bindSQL   string
+	defaultDB string
+	charset   string
+	collation string
+	isGlobal  bool
+	bindAst   ast.StmtNode
+}
+
+// Next implements the Executor Next interface.
+func (e *CreateBindExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
+	req.Reset()
+	bm := bindinfo.GetBindManager(e.ctx)
+	if bm == nil {
+		return errors.New("bind manager is nil")
+	}
+	var err error
+	if e.isGlobal {
+		err = bm.AddGlobalBind(e.originSQL, e.bindSQL, e.defaultDB, e.charset, e.collation)
+	}
+	return err
+}

executor/builder.go

@@ -180,6 +180,8 @@ func (b *executorBuilder) build(p plannercore.Plan) Executor {
 		return b.buildIndexLookUpReader(v)
 	case *plannercore.PhysicalWindow:
 		return b.buildWindow(v)
+	case *plannercore.CreateBindPlan:
+		return b.buildCreateBind(v)
 	default:
 		if mp, ok := p.(MockPhysicalPlan); ok {
 			return mp.GetExecutor()
@@ -1974,6 +1976,24 @@ func (b *executorBuilder) buildWindow(v *plannercore.PhysicalWindow) *WindowExec
 	}
 }
 
+func (b *executorBuilder) buildCreateBind(v *plannercore.CreateBindPlan) Executor {
+	base := newBaseExecutor(b.ctx, v.Schema(), v.ExplainID())
+	base.initCap = chunk.ZeroCapacity
+
+	charset, collation := b.ctx.GetSessionVars().GetCharsetInfo()
+	e := &CreateBindExec{
+		baseExecutor: base,
+		originSQL:    v.OriginSQL,
+		bindSQL:      v.BindSQL,
+		defaultDB:    v.DefaultDB,
+		charset:      charset,
+		collation:    collation,
+		isGlobal:     v.IsGlobal,
+		bindAst:      v.BindStmt,
+	}
+	return e
+}
+
 func getPhysicalTableID(t table.Table) int64 {
 	if p, ok := t.(table.PhysicalTable); ok {
 		return p.GetPhysicalID()