planner, executor: support Change and ChangeExec

executor/builder.go

@@ -71,6 +71,8 @@ func (b *executorBuilder) build(p plannercore.Plan) Executor {
 	switch v := p.(type) {
 	case nil:
 		return nil
+	case *plannercore.Change:
+		return b.buildChange(v)
 	case *plannercore.CheckTable:
 		return b.buildCheckTable(v)
 	case *plannercore.CheckIndex:
@@ -195,6 +197,12 @@ func (b *executorBuilder) buildCancelDDLJobs(v *plannercore.CancelDDLJobs) Execu
 	}
 	return e
 }
+func (b *executorBuilder) buildChange(v *plannercore.Change) Executor {
+	e := &ChangeExec{
+		Statement: v.Statement,
+	}
+	return e
+}
 
 func (b *executorBuilder) buildShowNextRowID(v *plannercore.ShowNextRowID) Executor {
 	e := &ShowNextRowIDExec{

executor/change.go

@@ -0,0 +1,70 @@
+// 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"
+	"fmt"
+	"github.com/pingcap/errors"
+	"github.com/pingcap/parser/ast"
+	"github.com/pingcap/tidb-tools/tidb-binlog/node"
+	"github.com/pingcap/tidb/config"
+	"github.com/pingcap/tidb/util/chunk"
+)
+
+// ChangeExec represents a change executor.
+type ChangeExec struct {
+	baseExecutor
+
+	Statement *ast.ChangeStmt
+}
+
+// Next implements the Executor Next interface.
+func (e *ChangeExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
+
+	stmt := e.Statement
+	fmt.Println(stmt.NodeID)
+	cfg := config.GetGlobalConfig()
+	return updateNodeState(cfg.Path, stmt.NodeType, stmt.NodeID, stmt.State)
+}
+
+// updateNodeState update pump or drainer's state.
+func updateNodeState(urls, kind, nodeID, state string) error {
+	/*
+		node's state can be online, pausing, paused, closing and offline.
+		if the state is one of them, will update the node's state saved in etcd directly.
+	*/
+	registry, err := createRegistry(urls)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	nodes, _, err := registry.Nodes(context.Background(), node.NodePrefix[kind])
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	for _, n := range nodes {
+		if n.NodeID != nodeID {
+			continue
+		}
+		switch state {
+		case node.Online, node.Pausing, node.Paused, node.Closing, node.Offline:
+			n.State = state
+			return registry.UpdateNode(context.Background(), node.NodePrefix[kind], n)
+		default:
+			return errors.Errorf("state %s is illegal", state)
+		}
+	}
+
+	return errors.NotFoundf("node %s, id %s from etcd %s", kind, nodeID, urls)
+}

go.mod

@@ -49,7 +49,7 @@ require (
 	github.com/pingcap/goleveldb v0.0.0-20171020122428-b9ff6c35079e
 	github.com/pingcap/kvproto v0.0.0-20190215154024-7f2fc73ef562
 	github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596
-	github.com/pingcap/parser v0.0.0-20190312024907-3f6280b08c8b
+	github.com/pingcap/parser v0.0.0-20190318135537-2575f2d0ba4b
 	github.com/pingcap/pd v2.1.0-rc.4+incompatible
 	github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible
 	github.com/pingcap/tipb v0.0.0-20190107072121-abbec73437b7

go.sum

@@ -3,6 +3,8 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/StackExchange/wmi v0.0.0-20180725035823-b12b22c5341f h1:5ZfJxyXo8KyX8DgGXC5B7ILL8y51fci/qYz2B4j8iLY=
 github.com/StackExchange/wmi v0.0.0-20180725035823-b12b22c5341f/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
+github.com/aliiohs/parser v0.0.0-20190318091245-2946eb6d5d02 h1:L0NeP/soQ+LDqqmhtFWseqt4+n2uva+bau/ojaBlHo0=
+github.com/aliiohs/parser v0.0.0-20190318091245-2946eb6d5d02/go.mod h1:WzwSl32vdxLYkpa8qu1hC3AGxGROygU9lQiy1JAyKYk=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/blacktear23/go-proxyprotocol v0.0.0-20180807104634-af7a81e8dd0d h1:rQlvB2AYWme2bIB18r/SipGiMEVJYE9U0z+MGoU/LtQ=
@@ -117,6 +119,8 @@ github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596 h1:t2OQTpPJnrPDGlvA+3F
 github.com/pingcap/log v0.0.0-20190307075452-bd41d9273596/go.mod h1:WpHUKhNZ18v116SvGrmjkA9CBhYmuUTKL+p8JC9ANEw=
 github.com/pingcap/parser v0.0.0-20190312024907-3f6280b08c8b h1:NlvTrxqezIJh6CD5Leky12IZ8E/GtpEEmzgNNb34wbw=
 github.com/pingcap/parser v0.0.0-20190312024907-3f6280b08c8b/go.mod h1:1FNvfp9+J0wvc4kl8eGNh7Rqrxveg15jJoWo/a0uHwA=
+github.com/pingcap/parser v0.0.0-20190318135537-2575f2d0ba4b h1:cfmy3QgCQkjjsHzw3HgWLZYg62MBLXSRXHN6TcsKhSA=
+github.com/pingcap/parser v0.0.0-20190318135537-2575f2d0ba4b/go.mod h1:1FNvfp9+J0wvc4kl8eGNh7Rqrxveg15jJoWo/a0uHwA=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible h1:/buwGk04aHO5odk/+O8ZOXGs4qkUjYTJ2UpCJXna8NE=
 github.com/pingcap/pd v2.1.0-rc.4+incompatible/go.mod h1:nD3+EoYes4+aNNODO99ES59V83MZSI+dFbhyr667a0E=
 github.com/pingcap/tidb-tools v2.1.3-0.20190116051332-34c808eef588+incompatible h1:e9Gi/LP9181HT3gBfSOeSBA+5JfemuE4aEAhqNgoE4k=

planner/core/common_plans.go

@@ -133,6 +133,13 @@ type CancelDDLJobs struct {
 	JobIDs []int64
 }
 
+// Change represents deallocate plan.
+type Change struct {
+	baseSchemaProducer
+
+	Statement *ast.ChangeStmt
+}
+
 // Prepare represents prepare plan.
 type Prepare struct {
 	baseSchemaProducer

planner/core/planbuilder.go

@@ -217,9 +217,15 @@ func (b *PlanBuilder) Build(node ast.Node) (Plan, error) {
 		return b.buildSimple(node.(ast.StmtNode))
 	case ast.DDLNode:
 		return b.buildDDL(x)
+	case *ast.ChangeStmt:
+		return b.buildChange(x)
 	}
 	return nil, ErrUnsupportedType.GenWithStack("Unsupported type %T", node)
 }
+func (b *PlanBuilder) buildChange(v *ast.ChangeStmt) (Plan, error) {
+	exe := &Change{Statement: v}
+	return exe, nil
+}
 
 func (b *PlanBuilder) buildExecute(v *ast.ExecuteStmt) (Plan, error) {
 	vars := make([]expression.Expression, 0, len(v.UsingVars))