Support SELECT ON CHANGE STREAM privilege (#169)

* Implement SELECT ON CHANGE STREAM in GRANT/REVOKE

* Update testdadta

* Reorder

ast/ast.go

@@ -334,6 +334,7 @@ type Privilege interface {
 }
 
 func (PrivilegeOnTable) isPrivilege()                {}
+func (SelectPrivilegeOnChangeStream) isPrivilege()   {}
 func (SelectPrivilegeOnView) isPrivilege()           {}
 func (ExecutePrivilegeOnTableFunction) isPrivilege() {}
 func (RolePrivilege) isPrivilege()                   {}
@@ -2492,6 +2493,18 @@ type DeletePrivilege struct {
 	Delete token.Pos // position of "DELETE" keyword
 }
 
+// SelectPrivilegeOnChangeStream is SELECT ON CHANGE STREAM privilege node in GRANT and REVOKE.
+//
+//	SELECT ON CHANGE STREAM {{.Names | sqlJoin ", "}}
+type SelectPrivilegeOnChangeStream struct {
+	// pos = Select
+	// end = Names[$].end
+
+	Select token.Pos
+
+	Names []*Ident // len(Names) > 0
+}
+
 // SelectPrivilegeOnView is SELECT ON VIEW privilege node in GRANT and REVOKE.
 //
 //	SELECT ON VIEW {{.Names | sqlJoin ","}}

ast/pos.go

@@ -874,6 +874,9 @@ func (u *UpdatePrivilege) End() token.Pos {
 func (d *DeletePrivilege) Pos() token.Pos { return d.Delete }
 func (d *DeletePrivilege) End() token.Pos { return d.Delete + 6 }
 
+func (p *SelectPrivilegeOnChangeStream) Pos() token.Pos { return p.Select }
+func (p *SelectPrivilegeOnChangeStream) End() token.Pos { return lastEnd(p.Names) }
+
 func (s *SelectPrivilegeOnView) Pos() token.Pos { return s.Select }
 func (s *SelectPrivilegeOnView) End() token.Pos { return s.Names[len(s.Names)-1].End() }
 

ast/sql.go

@@ -1206,6 +1206,10 @@ func (d *DeletePrivilege) SQL() string {
 	return "DELETE"
 }
 
+func (p *SelectPrivilegeOnChangeStream) SQL() string {
+	return "SELECT ON CHANGE STREAM " + sqlJoin(p.Names, ", ")
+}
+
 func (s *SelectPrivilegeOnView) SQL() string {
 	sql := "SELECT ON VIEW " + s.Names[0].SQL()
 	for _, v := range s.Names[1:] {

parser.go

@@ -3237,9 +3237,12 @@ func (p *Parser) parsePrivilege() ast.Privilege {
 	if e := p.tryParseExecutePrivilegeOnTableFunction(); e != nil {
 		return e
 	}
-	if r := p.tryRolePrivilege(); r != nil {
+	if r := p.tryParseRolePrivilege(); r != nil {
 		return r
 	}
+	if c := p.tryParseSelectPrivilegeOnChangeStream(); c != nil {
+		return c
+	}
 	return p.parsePrivilegeOnTable()
 }
 
@@ -3281,7 +3284,7 @@ func (p *Parser) tryParseExecutePrivilegeOnTableFunction() *ast.ExecutePrivilege
 	}
 }
 
-func (p *Parser) tryRolePrivilege() *ast.RolePrivilege {
+func (p *Parser) tryParseRolePrivilege() *ast.RolePrivilege {
 	if !p.Token.IsKeywordLike("ROLE") {
 		return nil
 	}
@@ -3293,6 +3296,31 @@ func (p *Parser) tryRolePrivilege() *ast.RolePrivilege {
 	}
 }
 
+func (p *Parser) tryParseSelectPrivilegeOnChangeStream() *ast.SelectPrivilegeOnChangeStream {
+	if p.Token.Kind != "SELECT" {
+		return nil
+	}
+	lexer := p.Lexer.Clone()
+	pos := p.expect("SELECT").Pos
+	if p.Token.Kind != "ON" {
+		p.Lexer = lexer
+		return nil
+	}
+	p.expect("ON")
+	if !p.Token.IsKeywordLike("CHANGE") {
+		p.Lexer = lexer
+		return nil
+	}
+	p.expectKeywordLike("CHANGE")
+	p.expectKeywordLike("STREAM")
+	names := parseCommaSeparatedList(p, p.parseIdent)
+
+	return &ast.SelectPrivilegeOnChangeStream{
+		Select: pos,
+		Names:  names,
+	}
+}
+
 func (p *Parser) parsePrivilegeOnTable() *ast.PrivilegeOnTable {
 	privileges := parseCommaSeparatedList(p, p.parseTablePrivilege)
 	p.expect("ON")

testdata/input/ddl/grant_change_stream_privilege.sql

@@ -0,0 +1 @@
+GRANT SELECT ON CHANGE STREAM cs_name_one, cs_name_two TO ROLE hr_manager

testdata/input/ddl/revoke_change_stream_privilege.sql

@@ -0,0 +1 @@
+REVOKE SELECT ON CHANGE STREAM cs_name_one, cs_name_two FROM ROLE hr_manager

testdata/result/ddl/grant_change_stream_privilege.sql.txt

@@ -0,0 +1,31 @@
+--- grant_change_stream_privilege.sql
+GRANT SELECT ON CHANGE STREAM cs_name_one, cs_name_two TO ROLE hr_manager
+--- AST
+&ast.Grant{
+  Grant:     0,
+  Privilege: &ast.SelectPrivilegeOnChangeStream{
+    Select: 6,
+    Names:  []*ast.Ident{
+      &ast.Ident{
+        NamePos: 30,
+        NameEnd: 41,
+        Name:    "cs_name_one",
+      },
+      &ast.Ident{
+        NamePos: 43,
+        NameEnd: 54,
+        Name:    "cs_name_two",
+      },
+    },
+  },
+  Roles: []*ast.Ident{
+    &ast.Ident{
+      NamePos: 63,
+      NameEnd: 73,
+      Name:    "hr_manager",
+    },
+  },
+}
+
+--- SQL
+GRANT SELECT ON CHANGE STREAM cs_name_one, cs_name_two TO ROLE hr_manager

testdata/result/ddl/revoke_change_stream_privilege.sql.txt

@@ -0,0 +1,31 @@
+--- revoke_change_stream_privilege.sql
+REVOKE SELECT ON CHANGE STREAM cs_name_one, cs_name_two FROM ROLE hr_manager
+--- AST
+&ast.Revoke{
+  Revoke:    0,
+  Privilege: &ast.SelectPrivilegeOnChangeStream{
+    Select: 7,
+    Names:  []*ast.Ident{
+      &ast.Ident{
+        NamePos: 31,
+        NameEnd: 42,
+        Name:    "cs_name_one",
+      },
+      &ast.Ident{
+        NamePos: 44,
+        NameEnd: 55,
+        Name:    "cs_name_two",
+      },
+    },
+  },
+  Roles: []*ast.Ident{
+    &ast.Ident{
+      NamePos: 66,
+      NameEnd: 76,
+      Name:    "hr_manager",
+    },
+  },
+}
+
+--- SQL
+REVOKE SELECT ON CHANGE STREAM cs_name_one, cs_name_two FROM ROLE hr_manager

testdata/result/statement/grant_change_stream_privilege.sql.txt

@@ -0,0 +1,31 @@
+--- grant_change_stream_privilege.sql
+GRANT SELECT ON CHANGE STREAM cs_name_one, cs_name_two TO ROLE hr_manager
+--- AST
+&ast.Grant{
+  Grant:     0,
+  Privilege: &ast.SelectPrivilegeOnChangeStream{
+    Select: 6,
+    Names:  []*ast.Ident{
+      &ast.Ident{
+        NamePos: 30,
+        NameEnd: 41,
+        Name:    "cs_name_one",
+      },
+      &ast.Ident{
+        NamePos: 43,
+        NameEnd: 54,
+        Name:    "cs_name_two",
+      },
+    },
+  },
+  Roles: []*ast.Ident{
+    &ast.Ident{
+      NamePos: 63,
+      NameEnd: 73,
+      Name:    "hr_manager",
+    },
+  },
+}
+
+--- SQL
+GRANT SELECT ON CHANGE STREAM cs_name_one, cs_name_two TO ROLE hr_manager

testdata/result/statement/revoke_change_stream_privilege.sql.txt

@@ -0,0 +1,31 @@
+--- revoke_change_stream_privilege.sql
+REVOKE SELECT ON CHANGE STREAM cs_name_one, cs_name_two FROM ROLE hr_manager
+--- AST
+&ast.Revoke{
+  Revoke:    0,
+  Privilege: &ast.SelectPrivilegeOnChangeStream{
+    Select: 7,
+    Names:  []*ast.Ident{
+      &ast.Ident{
+        NamePos: 31,
+        NameEnd: 42,
+        Name:    "cs_name_one",
+      },
+      &ast.Ident{
+        NamePos: 44,
+        NameEnd: 55,
+        Name:    "cs_name_two",
+      },
+    },
+  },
+  Roles: []*ast.Ident{
+    &ast.Ident{
+      NamePos: 66,
+      NameEnd: 76,
+      Name:    "hr_manager",
+    },
+  },
+}
+
+--- SQL
+REVOKE SELECT ON CHANGE STREAM cs_name_one, cs_name_two FROM ROLE hr_manager