Support update pump or drainer status

ast/misc.go

@@ -721,6 +721,74 @@ func (n *SetPwdStmt) Accept(v Visitor) (Node, bool) {
 	return v.Leave(n)
 }
 
+type ChangePumpStmt struct {
+	stmtNode
+
+	StateName string
+	State     string
+	IpAndPort string
+}
+
+// Restore implements Node interface.
+func (n *ChangePumpStmt) Restore(ctx *RestoreCtx) error {
+	ctx.WriteKeyWord("CHANGE PUMP TO")
+	ctx.WriteString("PUMP_STATE")
+	ctx.WritePlain("=")
+	ctx.WriteString(n.State)
+	ctx.WriteKeyWord("FOR NodeID")
+	ctx.WriteKeyWord(n.IpAndPort)
+	return nil
+}
+
+// SecureText implements SensitiveStatement interface.
+func (n *ChangePumpStmt) SecureText() string {
+	return fmt.Sprintf("change pump to pump_state='paused' for NodeID '%s'", n.IpAndPort)
+}
+
+// Accept implements Node Accept interface.
+func (n *ChangePumpStmt) Accept(v Visitor) (Node, bool) {
+	newNode, skipChildren := v.Enter(n)
+	if skipChildren {
+		return v.Leave(newNode)
+	}
+	n = newNode.(*ChangePumpStmt)
+	return v.Leave(n)
+}
+
+type ChangeDrainerStmt struct {
+	stmtNode
+
+	StateName string
+	State     string
+	IpAndPort string
+}
+
+// Restore implements Node interface.
+func (n *ChangeDrainerStmt) Restore(ctx *RestoreCtx) error {
+	ctx.WriteKeyWord("CHANGE DRAINER TO")
+	ctx.WriteString("PUMP_STATE")
+	ctx.WritePlain("=")
+	ctx.WriteString(n.State)
+	ctx.WriteKeyWord("FOR NodeID")
+	ctx.WriteKeyWord(n.IpAndPort)
+	return nil
+}
+
+// SecureText implements SensitiveStatement interface.
+func (n *ChangeDrainerStmt) SecureText() string {
+	return fmt.Sprintf("change drainer to drainer_state='paused' for NodeID '%s'", n.IpAndPort)
+}
+
+// Accept implements Node Accept interface.
+func (n *ChangeDrainerStmt) Accept(v Visitor) (Node, bool) {
+	newNode, skipChildren := v.Enter(n)
+	if skipChildren {
+		return v.Leave(newNode)
+	}
+	n = newNode.(*ChangeDrainerStmt)
+	return v.Leave(n)
+}
+
 // UserSpec is used for parsing create user statement.
 type UserSpec struct {
 	User    *auth.UserIdentity

ast/misc_test.go

@@ -126,6 +126,17 @@ load data infile '/tmp/t.csv' into table t fields terminated by 'ab' enclosed by
 		stmt.Accept(visitor1{})
 	}
 }
+func (ts *testMiscSuite) TestChangeStmt(c *C) {
+	sql := `change pump to pump_state='paused' for NodeID "127.0.0.1:8249";`
+
+	p := parser.New()
+	stmts, _, err := p.Parse(sql, "", "")
+	c.Assert(err, IsNil)
+	for _, stmt := range stmts {
+		stmt.Accept(visitor{})
+		stmt.Accept(visitor1{})
+	}
+}
 
 func (ts *testMiscSuite) TestSensitiveStatement(c *C) {
 	positive := []StmtNode{

parser.y

@@ -112,6 +112,7 @@ import (
 	distinctRow		"DISTINCTROW"
 	div 			"DIV"
 	doubleType		"DOUBLE"
+	drainer_state		"DRAINER_STATE"
 	drop			"DROP"
 	dual 			"DUAL"
 	elseKwd			"ELSE"
@@ -180,6 +181,7 @@ import (
 	minuteMicrosecond	"MINUTE_MICROSECOND"
 	minuteSecond 		"MINUTE_SECOND"
 	mod 			"MOD"
+	NodeID			"NODEID"
 	not			"NOT"
 	noWriteToBinLog 	"NO_WRITE_TO_BINLOG"
 	nthValue		"NTH_VALUE"
@@ -200,6 +202,7 @@ import (
 	precisionType		"PRECISION"
 	primary			"PRIMARY"
 	procedure		"PROCEDURE"
+	PUMP_STATE		"PUMP_STATE"
 	shardRowIDBits		"SHARD_ROW_ID_BITS"
 	rangeKwd		"RANGE"
 	rank			"RANK"
@@ -621,7 +624,8 @@ import (
 	RevokeStmt			"Revoke statement"
 	RevokeRoleStmt      "Revoke role statement"
 	RollbackStmt			"ROLLBACK statement"
-	SetStmt				"Set variable statement"
+	SetStmt				"Update Pump or Drainer's status"
+	ChangeStmt				"Change statement"
 	SetRoleStmt				"Set active role statement"
 	SetDefaultRoleStmt			"Set default statement for some user"
 	ShowStmt			"Show engines/databases/tables/user/columns/warnings/status statement"
@@ -5513,6 +5517,22 @@ UnionSelect:
 UnionOpt:
 DefaultTrueDistinctOpt
 
+/********************Change Statement*******************************/
+ChangeStmt:
+|	"CHANGE" "PUMP" to "PUMP_STATE" eq stringLit forKwd "NODEID" stringLit
+	{
+		$$ = &ast.ChangePumpStmt{
+			State: $6.(string),
+			IpAndPort: $9.(string),
+		}
+	}
+|	"CHANGE" "DRAINER" to "DRAINER_STATE" eq stringLit forKwd "NODEID" stringLit
+	{
+		$$ = &ast.ChangeDrainerStmt{
+			State: $6.(string),
+			IpAndPort: $9.(string),
+		}
+	}
 
 /********************Set Statement*******************************/
 SetStmt:
@@ -6472,6 +6492,7 @@ Statement:
 |	DeleteFromStmt
 |	ExecuteStmt
 |	ExplainStmt
+|	ChangeStmt
 |	CreateDatabaseStmt
 |	CreateIndexStmt
 |	CreateTableStmt