Repository: cloudspannerecosystem/memefish
Pull request #58 (2023-08-19 07:29:32)

support CHANGE STREAM
This PR adds support for [CHANGE STREAM statements](https://cloud.google.com/spanner/docs/reference/standard-sql/data-definition-language#change_stream_statements)

nktks: @makenowjust 
Thanks for your review quickly.

> Duplication of Watch and WatchAll

I investigated that the below two statements are different.(Although this behavior isn't docummented cleary.)
`CREATE CHANGE STREAM change_stream_name`: Just create change stream resource, not watching everything.
`CREATE CHANGE STREAM change_stream_name For ALL`: create change stream resource watching all.

So I added WatchAll to distinguish them.

nktks: I commited
 https://github.com/cloudspannerecosystem/memefish/pull/58/commits/116521d369d6ad720e9057bb41d9edef05ef0dec
so that:
- devide watch, options  ast
- fix `End()` implementation to save pos needed from `End()` functions.

nktks: I can delete `WatchAll` ( We can set`ALL` case when WatchTable length is 0) for now, I'll fix it.
makenowjust: @nktks Could you have any responses?
nktks: Thanks for your comment and sorry for delay...

According to your comment, I'll try to fix below when I have time.

- modify `CreateChangeStream`, `AlterChangeStream` struct to match [Google defined statements](https://cloud.google.com/spanner/docs/reference/standard-sql/data-definition-language#change_stream_statements)
- fix comment to match definition written in https://github.com/cloudspannerecosystem/memefish/blob/1b120de904be1345085ee862d3d061476dc2db2f/ast/ast.go#L1-L31

makenowjust: The overall design seems good to me. Please correct some coding styles.

Patch:
diff --git a/ast/ast.go b/ast/ast.go
index 2ac1148..39b6857 100644
--- a/ast/ast.go
+++ b/ast/ast.go
@@ -50,22 +50,25 @@ type Statement interface {
 	isStatement()
 }
 
-func (QueryStatement) isStatement() {}
-func (CreateDatabase) isStatement() {}
-func (CreateTable) isStatement()    {}
-func (CreateSequence) isStatement() {}
-func (CreateView) isStatement()     {}
-func (CreateIndex) isStatement()    {}
-func (CreateRole) isStatement()     {}
-func (AlterTable) isStatement()     {}
-func (DropTable) isStatement()      {}
-func (DropIndex) isStatement()      {}
-func (DropRole) isStatement()       {}
-func (Insert) isStatement()         {}
-func (Delete) isStatement()         {}
-func (Update) isStatement()         {}
-func (Grant) isStatement()          {}
-func (Revoke) isStatement()         {}
+func (QueryStatement) isStatement()     {}
+func (CreateDatabase) isStatement()     {}
+func (CreateTable) isStatement()        {}
+func (CreateSequence) isStatement()     {}
+func (CreateView) isStatement()         {}
+func (CreateIndex) isStatement()        {}
+func (CreateRole) isStatement()         {}
+func (AlterTable) isStatement()         {}
+func (DropTable) isStatement()          {}
+func (DropIndex) isStatement()          {}
+func (DropRole) isStatement()           {}
+func (Insert) isStatement()             {}
+func (Delete) isStatement()             {}
+func (Update) isStatement()             {}
+func (Grant) isStatement()              {}
+func (Revoke) isStatement()             {}
+func (CreateChangeStream) isStatement() {}
+func (AlterChangeStream) isStatement()  {}
+func (DropChangeStream) isStatement()   {}
 
 // QueryExpr represents set operator operands.
 type QueryExpr interface {
@@ -215,18 +218,21 @@ type DDL interface {
 	isDDL()
 }
 
-func (CreateDatabase) isDDL() {}
-func (CreateTable) isDDL()    {}
-func (CreateView) isDDL()     {}
-func (CreateSequence) isDDL() {}
-func (AlterTable) isDDL()     {}
-func (DropTable) isDDL()      {}
-func (CreateIndex) isDDL()    {}
-func (DropIndex) isDDL()      {}
-func (CreateRole) isDDL()     {}
-func (DropRole) isDDL()       {}
-func (Grant) isDDL()          {}
-func (Revoke) isDDL()         {}
+func (CreateDatabase) isDDL()     {}
+func (CreateTable) isDDL()        {}
+func (CreateView) isDDL()         {}
+func (CreateSequence) isDDL()     {}
+func (AlterTable) isDDL()         {}
+func (DropTable) isDDL()          {}
+func (CreateIndex) isDDL()        {}
+func (DropIndex) isDDL()          {}
+func (CreateRole) isDDL()         {}
+func (DropRole) isDDL()           {}
+func (Grant) isDDL()              {}
+func (Revoke) isDDL()             {}
+func (CreateChangeStream) isDDL() {}
+func (AlterChangeStream) isDDL()  {}
+func (DropChangeStream) isDDL()   {}
 
 // Constraint represents table constraint of CONSTARINT clause.
 type Constraint interface {
@@ -307,6 +313,25 @@ type InsertInput interface {
 func (ValuesInput) isInsertInput()   {}
 func (SubQueryInput) isInsertInput() {}
 
+// ChangeStreamFor represents FOR clause in CREATE/ALTER CHANGE STREAM statement.
+type ChangeStreamFor interface {
+	Node
+	isChangeStreamFor()
+}
+
+func (ChangeStreamForAll) isChangeStreamFor()    {}
+func (ChangeStreamForTables) isChangeStreamFor() {}
+
+// ChangeStreamAlternation represents ALTER CHANGE STREAM action.
+type ChangeStreamAlternation interface {
+	Node
+	isChangeStreamAlternation()
+}
+
+func (ChangeStreamSetFor) isChangeStreamAlternation()     {}
+func (ChangeStreamDropForAll) isChangeStreamAlternation() {}
+func (ChangeStreamSetOptions) isChangeStreamAlternation() {}
+
 // ================================================================================
 //
 // SELECT
@@ -1600,9 +1625,22 @@ type AlterTable struct {
 	TableAlternation TableAlternation
 }
 
+// AlterChangeStream is ALTER CHANGE STREAM statement node.
+//
+//	ALTER CHANGE STREAM {{.Name | sql}} {{.ChangeStreamAlternation | sql}}
+type AlterChangeStream struct {
+	// pos = Alter
+	// end = ChangeStreamAlternation.end
+
+	Alter token.Pos // position of "ALTER" keyword
+
+	Name                    *Ident
+	ChangeStreamAlternation ChangeStreamAlternation
+}
+
 // AddColumn is ADD COLUMN clause in ALTER TABLE.
 //
-//	ADD COLUMN {{.IfNotExists}}IF NOT EXISTS{{end}} {{.Column | sql}}
+//	ADD COLUMN {{if .IfNotExists}}IF NOT EXISTS{{end}} {{.Column | sql}}
 type AddColumn struct {
 	// pos = Add
 	// end = Column.end
@@ -1766,6 +1804,115 @@ type CreateIndex struct {
 	InterleaveIn *InterleaveIn // optional
 }
 
+// CreateChangeStream is CREATE CHANGE STREAM statement node.
+//
+//	CREATE CHANGE STREAM {{.Name | sql}} {{.For | sqlOpt}} {{.Options | sqlOpt}}
+type CreateChangeStream struct {
+	// pos = Create
+	// end = (Options ?? For ?? Name).end
+
+	Create token.Pos // position of "CREATE" keyword
+
+	Name    *Ident
+	For     ChangeStreamFor      // optional
+	Options *ChangeStreamOptions // optional
+}
+
+// ChangeStreamForAll is FOR ALL node in CREATE CHANGE STREAM
+//
+//	FOR ALL
+type ChangeStreamForAll struct {
+	// pos = For
+	// end = All
+
+	For token.Pos // position of "FOR" keyword
+	All token.Pos // position of "ALL" keyword
+}
+
+// ChangeStreamForTables is FOR tables node in CREATE CHANGE STREAM
+//
+//	FOR {{.Tables | sqlJoin ","}}
+type ChangeStreamForTables struct {
+	// pos = For
+	// end = Tables[$].end
+
+	For token.Pos // position of "FOR" keyword
+
+	Tables []*ChangeStreamForTable
+}
+
+// ChangeStreamForTable table node in CREATE CHANGE STREAM SET FOR
+//
+//	{{.TableName | sql}}{{if .Columns}}({{.Columns | sqlJoin ","}}){{end}}
+type ChangeStreamForTable struct {
+	// pos = TableName.pos
+	// end = TableName.end || Rparen + 1
+
+	Rparen token.Pos // position of ")"
+
+	TableName *Ident
+	Columns   []*Ident
+}
+
+// ChangeStreamOptions is OPTIONS clause node in CREATE CHANGE STREAM.
+//
+//	OPTIONS ({{.Records | sqlJoin ","}})
+type ChangeStreamOptions struct {
+	// pos = Options
+	// end = Rparen + 1
+
+	Options token.Pos // position of "OPTIONS" keyword
+	Rparen  token.Pos // position of ")"
+
+	Records []*ChangeStreamOptionsRecord // len(Records) > 0
+}
+
+// ChangeStreamOptionsRecord is OPTIONS record node.
+//
+//	{{.Key | sql}}={{.Expr | sql}}
+type ChangeStreamOptionsRecord struct {
+	// pos = Key.pos
+	// end = Value.end
+
+	Key   *Ident
+	Value Expr
+}
+
+// ChangeStreamSetFor is SET FOR tables node in ALTER CHANGE STREAM
+//
+//	SET FOR {{.For | sql}}
+type ChangeStreamSetFor struct {
+	// pos = Set
+	// end = For.end
+
+	Set token.Pos // position of "SET" keyword
+
+	For ChangeStreamFor
+}
+
+// ChangeStreamDropForAll is DROP FOR ALL node in ALTER CHANGE STREAM
+//
+//	DROP FOR ALL
+type ChangeStreamDropForAll struct {
+	// pos = Drop
+	// end = All + 3
+
+	Drop token.Pos // position of "DROP" keyword
+	All  token.Pos // position of "ALL" keyword
+}
+
+// ChangeStreamSetOptions is SET OPTIONS node in ALTER CHANGE STREAM
+//
+//	SET {{.Options | sql}}
+type ChangeStreamSetOptions struct {
+	// pos = Set
+	// end = Options.Rparen + 1
+
+	Set token.Pos // position of "SET" keyword
+
+	Options *ChangeStreamOptions
+}
+
 // Storing is STORING clause in CREATE INDEX.
 //
 //	STORING ({{.Columns | sqlJoin ","}})
@@ -1793,7 +1940,7 @@ type InterleaveIn struct {
 
 // DropIndex is DROP INDEX statement node.
 //
-//	DROP INDEX {{.IfExists}}IF EXISTS{{end}} {{.Name | sql}}
+//	DROP INDEX {{if .IfExists}}IF EXISTS{{end}} {{.Name | sql}}
 type DropIndex struct {
 	// pos = Drop
 	// end = Name.end
@@ -1828,6 +1975,18 @@ type DropRole struct {
 	Name *Ident
 }
 
+// DropChangeStream is DROP CHANGE STREAM  statement node.
+//
+//	DROP CHANGE STREAM {{.Name | sql}}
+type DropChangeStream struct {
+	// pos = Drop
+	// end = Name.end
+
+	Drop token.Pos // position of "DROP" keyword
+
+	Name *Ident
+}
+
 // Grant is GRANT statement node.
 //
 //	GRANT {{.Privilege | sql}} TO ROLE {{.Roles | sqlJoin ","}}
diff --git a/ast/pos.go b/ast/pos.go
index 5103424..c0d14c8 100644
--- a/ast/pos.go
+++ b/ast/pos.go
@@ -664,6 +664,28 @@ func (c *CreateIndex) End() token.Pos {
 	return c.Rparen + 1
 }
 
+func (c *CreateChangeStream) Pos() token.Pos {
+	return c.Create
+}
+
+func (c *CreateChangeStream) End() token.Pos {
+	if c.Options != nil {
+		return c.Options.End()
+	}
+	if c.For != nil {
+		return c.For.End()
+	}
+	return c.Name.End()
+}
+
+func (c *ChangeStreamOptions) Pos() token.Pos { return c.Options }
+func (c *ChangeStreamOptions) End() token.Pos { return c.Rparen + 1 }
+
+func (c *ChangeStreamForAll) Pos() token.Pos    { return c.For }
+func (c *ChangeStreamForAll) End() token.Pos    { return c.All }
+func (c *ChangeStreamForTables) Pos() token.Pos { return c.For }
+func (c *ChangeStreamForTables) End() token.Pos { return c.Tables[len(c.Tables)-1].End() }
+
 func (s *Storing) Pos() token.Pos { return s.Storing }
 func (s *Storing) End() token.Pos { return s.Rparen + 1 }
 
@@ -679,6 +701,21 @@ func (c *CreateRole) End() token.Pos { return c.Name.End() }
 func (d *DropRole) Pos() token.Pos { return d.Drop }
 func (d *DropRole) End() token.Pos { return d.Name.End() }
 
+func (d *DropChangeStream) Pos() token.Pos { return d.Drop }
+func (d *DropChangeStream) End() token.Pos { return d.Name.End() }
+
+func (a *AlterChangeStream) Pos() token.Pos      { return a.Alter }
+func (a *AlterChangeStream) End() token.Pos      { return a.ChangeStreamAlternation.End() }
+
+func (a *ChangeStreamSetFor) Pos() token.Pos     { return a.Set }
+func (a *ChangeStreamSetFor) End() token.Pos     { return a.For.End() }
+
+func (a *ChangeStreamDropForAll) Pos() token.Pos { return a.Drop }
+func (a *ChangeStreamDropForAll) End() token.Pos { return a.All + 3 }
+
+func (a *ChangeStreamSetOptions) Pos() token.Pos { return a.Set }
+func (a *ChangeStreamSetOptions) End() token.Pos { return a.Options.Rparen + 1 }
+
 func (g *Grant) Pos() token.Pos { return g.Grant }
 func (g *Grant) End() token.Pos { return g.Roles[len(g.Roles)-1].End() }
 
@@ -748,6 +785,13 @@ func (s *SizedSchemaType) End() token.Pos { return s.Rparen + 1 }
 func (a *ArraySchemaType) Pos() token.Pos { return a.Array }
 func (a *ArraySchemaType) End() token.Pos { return a.Gt + 1 }
 
+func (c *ChangeStreamForTable) End() token.Pos {
+	if len(c.Columns) == 0 {
+		return c.TableName.End()
+	}
+	return c.Rparen + 1
+}
+
 // ================================================================================
 //
 // DML
diff --git a/ast/sql.go b/ast/sql.go
index 690cf00..1750522 100644
--- a/ast/sql.go
+++ b/ast/sql.go
@@ -968,6 +968,79 @@ func (c *CreateIndex) SQL() string {
 	return sql
 }
 
+func (c *CreateChangeStream) SQL() string {
+	sql := "CREATE CHANGE STREAM " + c.Name.SQL()
+	if c.For != nil {
+		sql += " " + c.For.SQL()
+	}
+	if c.Options != nil {
+		sql += c.Options.SQL()
+	}
+	return sql
+}
+
+func (c *ChangeStreamForAll) SQL() string {
+	return "FOR ALL"
+}
+
+func (c *ChangeStreamForTables) SQL() string {
+	sql := "FOR "
+	for i, table := range c.Tables {
+		if i > 0 {
+			sql += ", "
+		}
+		sql += table.SQL()
+	}
+	return sql
+}
+
+func (c *ChangeStreamOptions) SQL() string {
+	sql := " OPTIONS ("
+	for i, record := range c.Records {
+		if i > 0 {
+			sql += ", "
+		}
+		sql += record.Key.SQL() + "=" + record.Value.SQL()
+	}
+	sql += ")"
+	return sql
+}
+
+func (a *AlterChangeStream) SQL() string {
+	return "ALTER CHANGE STREAM " + a.Name.SQL() + " " + a.ChangeStreamAlternation.SQL()
+}
+
+func (a ChangeStreamSetFor) SQL() string {
+	return "SET " + a.For.SQL()
+}
+
+func (a ChangeStreamDropForAll) SQL() string {
+	return "DROP FOR ALL"
+}
+
+func (a ChangeStreamSetOptions) SQL() string {
+	return "SET " + a.Options.SQL()
+}
+
+func (c *ChangeStreamForTable) SQL() string {
+	sql := c.TableName.SQL()
+	if len(c.Columns) > 0 {
+		sql += "("
+		for i, id := range c.Columns {
+			if i > 0 {
+				sql += ", "
+			}
+			sql += id.SQL()
+		}
+		sql += ")"
+	}
+	return sql
+}
+
+func (d *DropChangeStream) SQL() string {
+	return "DROP CHANGE STREAM " + d.Name.SQL()
+}
+
 func (s *Storing) SQL() string {
 	sql := "STORING ("
 	for i, c := range s.Columns {
diff --git a/parser.go b/parser.go
index ffa04a7..62e19c8 100644
--- a/parser.go
+++ b/parser.go
@@ -1988,11 +1988,19 @@ func (p *Parser) parseDDL() ast.DDL {
 			return p.parseCreateIndex(pos)
 		case p.Token.IsKeywordLike("ROLE"):
 			return p.parseCreateRole(pos)
+		case p.Token.IsKeywordLike("CHANGE"):
+			return p.parseCreateChangeStream(pos)
 		}
-		p.panicfAtToken(&p.Token, "expected pseudo keyword: DATABASE, TABLE, INDEX, UNIQUE, NULL_FILTERED, but: %s", p.Token.AsString)
+		p.panicfAtToken(&p.Token, "expected pseudo keyword: DATABASE, TABLE, INDEX, UNIQUE, NULL_FILTERED, ROLE, CHANGE but: %s", p.Token.AsString)
 	case p.Token.IsKeywordLike("ALTER"):
 		p.nextToken()
-		return p.parseAlterTable(pos)
+		switch {
+		case p.Token.IsKeywordLike("TABLE"):
+			return p.parseAlterTable(pos)
+		case p.Token.IsKeywordLike("CHANGE"):
+			return p.parseAlterChangeStream(pos)
+		}
+		p.panicfAtToken(&p.Token, "expected pseudo keyword: TABLE, CHANGE, but: %s", p.Token.AsString)
 	case p.Token.IsKeywordLike("DROP"):
 		p.nextToken()
 		switch {
@@ -2002,8 +2010,10 @@ func (p *Parser) parseDDL() ast.DDL {
 			return p.parseDropIndex(pos)
 		case p.Token.IsKeywordLike("ROLE"):
 			return p.parseDropRole(pos)
+		case p.Token.IsKeywordLike("CHANGE"):
+			return p.parseDropChangeStream(pos)
 		}
-		p.panicfAtToken(&p.Token, "expected pseudo keyword: TABLE, INDEX, but: %s", p.Token.AsString)
+		p.panicfAtToken(&p.Token, "expected pseudo keyword: TABLE, INDEX, ROLE, CHANGE, but: %s", p.Token.AsString)
 	case p.Token.IsKeywordLike("GRANT"):
 		p.nextToken()
 		return p.parseGrant(pos)
@@ -2482,6 +2492,143 @@ func (p *Parser) parseCreateIndex(pos token.Pos) *ast.CreateIndex {
 	}
 }
 
+func (p *Parser) parseCreateChangeStream(pos token.Pos) *ast.CreateChangeStream {
+	p.expectKeywordLike("CHANGE")
+	p.expectKeywordLike("STREAM")
+	name := p.parseIdent()
+	cs := &ast.CreateChangeStream{
+		Create: pos,
+		Name:   name,
+	}
+	if p.Token.Kind == "FOR" {
+		cs.For = p.parseChangeStreamFor()
+	}
+	if p.Token.IsKeywordLike("OPTIONS") {
+		cs.Options = p.parseChangeStreamOptions()
+	}
+	return cs
+
+}
+
+func (p *Parser) parseAlterChangeStream(pos token.Pos) *ast.AlterChangeStream {
+	p.expectKeywordLike("CHANGE")
+	p.expectKeywordLike("STREAM")
+	name := p.parseIdent()
+	cs := &ast.AlterChangeStream{
+		Alter: pos,
+		Name:  name,
+	}
+	if p.Token.Kind == "SET" {
+		setpos := p.Token.Pos
+		p.nextToken()
+		if p.Token.Kind == "FOR" {
+			cs.ChangeStreamAlternation = &ast.ChangeStreamSetFor{
+				Set: setpos,
+				For: p.parseChangeStreamFor(),
+			}
+			return cs
+		} else if p.Token.IsKeywordLike("OPTIONS") {
+			cs.ChangeStreamAlternation = &ast.ChangeStreamSetOptions{
+				Set:     setpos,
+				Options: p.parseChangeStreamOptions(),
+			}
+			return cs
+		}
+	} else if p.Token.IsKeywordLike("DROP") {
+		droppos := p.Token.Pos
+		p.nextToken()
+		p.expect("FOR")
+		allpos := p.expect("ALL").Pos
+		cs.ChangeStreamAlternation = &ast.ChangeStreamDropForAll{
+			Drop: droppos,
+			All:  allpos,
+		}
+		return cs
+	} else {
+		p.panicfAtToken(&p.Token, "expected SET FOR or DROP FOR ALL or SET OPTIONS")
+	}
+	return cs
+}
+
+func (p *Parser) parseDropChangeStream(pos token.Pos) *ast.DropChangeStream {
+	p.expectKeywordLike("CHANGE")
+	p.expectKeywordLike("STREAM")
+	name := p.parseIdent()
+	return &ast.DropChangeStream{
+		Drop: pos,
+		Name: name,
+	}
+
+}
+
+func (p *Parser) parseChangeStreamFor() ast.ChangeStreamFor {
+	pos := p.expect("FOR").Pos
+	if p.Token.Kind == "ALL" {
+		p.nextToken()
+		return &ast.ChangeStreamForAll{
+			For: pos,
+			All: p.Token.Pos,
+		}
+
+	}
+	cswt := &ast.ChangeStreamForTables{
+		For: pos,
+	}
+	for {
+		tname := p.parseIdent()
+		forTable := ast.ChangeStreamForTable{
+			TableName: tname,
+		}
+
+		if p.Token.Kind == "(" {
+			p.nextToken()
+			forTable.Columns = []*ast.Ident{p.parseIdent()}
+			for p.Token.Kind == "," {
+				p.nextToken()
+				forTable.Columns = append(forTable.Columns, p.parseIdent())
+			}
+			forTable.Rparen = p.expect(")").Pos
+		}
+
+		cswt.Tables = append(cswt.Tables, &forTable)
+		if p.Token.Kind == "," {
+			p.nextToken()
+			continue
+		}
+		break
+	}
+	return cswt
+
+}
+
+// Parse CHANGE STREAM OPTIONS.
+// We parse any expressions in OPTIONS. even if tokens includes unsupported expressions.
+// This is for the key, value which will supported in the future.
+// We don't need to modify this code for them.
+func (p *Parser) parseChangeStreamOptions() *ast.ChangeStreamOptions {
+	pos := p.expectKeywordLike("OPTIONS").Pos
+	p.expect("(")
+	cso := &ast.ChangeStreamOptions{Options: pos}
+	for {
+		key := p.parseIdent()
+		p.expect("=")
+		value := p.parseExpr()
+		cso.Records = append(cso.Records, &ast.ChangeStreamOptionsRecord{Key: key, Value: value})
+		if p.Token.Kind == "," {
+			p.nextToken()
+			continue
+		}
+		if p.Token.Kind == ")" {
+			cso.Rparen = p.Token.Pos
+			p.nextToken()
+			break
+		}
+		p.panicfAtToken(&p.Token, "expected expr or , or ), but: %s", p.Token.AsString)
+	}
+
+	return cso
+}
+
 func (p *Parser) tryParseStoring() *ast.Storing {
 	if !p.Token.IsKeywordLike("STORING") {
 		return nil


Test patch:
diff --git a/testdata/input/ddl/alter_change_stream_options.sql b/testdata/input/ddl/alter_change_stream_options.sql
new file mode 100644
index 0000000..5bd4777
--- /dev/null
+++ b/testdata/input/ddl/alter_change_stream_options.sql
@@ -0,0 +1,1 @@
+ALTER CHANGE STREAM change_stream_name SET OPTIONS (retention_period = '1d', value_capture_type = 'OLD_AND_NEW_VALUES')
diff --git a/testdata/input/ddl/alter_change_stream_set_drop_for_all.sql b/testdata/input/ddl/alter_change_stream_set_drop_for_all.sql
new file mode 100644
index 0000000..e6c9e0b
--- /dev/null
+++ b/testdata/input/ddl/alter_change_stream_set_drop_for_all.sql
@@ -0,0 +1,1 @@
+ALTER CHANGE STREAM change_stream_name DROP FOR ALL
diff --git a/testdata/input/ddl/alter_change_stream_set_for_all.sql b/testdata/input/ddl/alter_change_stream_set_for_all.sql
new file mode 100644
index 0000000..22cfb78
--- /dev/null
+++ b/testdata/input/ddl/alter_change_stream_set_for_all.sql
@@ -0,0 +1,1 @@
+ALTER CHANGE STREAM change_stream_name SET FOR ALL
diff --git a/testdata/input/ddl/alter_change_stream_set_for_table_columns.sql b/testdata/input/ddl/alter_change_stream_set_for_table_columns.sql
new file mode 100644
index 0000000..e9883ae
--- /dev/null
+++ b/testdata/input/ddl/alter_change_stream_set_for_table_columns.sql
@@ -0,0 +1,1 @@
+ALTER CHANGE STREAM change_stream_name SET FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/input/ddl/create_change_stream.sql b/testdata/input/ddl/create_change_stream.sql
new file mode 100644
index 0000000..4ea16d5
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name
diff --git a/testdata/input/ddl/create_change_stream_for_all.sql b/testdata/input/ddl/create_change_stream_for_all.sql
new file mode 100644
index 0000000..9426be6
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_all.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR ALL
diff --git a/testdata/input/ddl/create_change_stream_for_multiple_table_columns.sql b/testdata/input/ddl/create_change_stream_for_multiple_table_columns.sql
new file mode 100644
index 0000000..cf74479
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_multiple_table_columns.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/input/ddl/create_change_stream_for_table.sql b/testdata/input/ddl/create_change_stream_for_table.sql
new file mode 100644
index 0000000..c353153
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_table.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name
diff --git a/testdata/input/ddl/create_change_stream_for_table_column.sql b/testdata/input/ddl/create_change_stream_for_table_column.sql
new file mode 100644
index 0000000..44c02d8
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_table_column.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name(column)
diff --git a/testdata/input/ddl/create_change_stream_for_table_columns.sql b/testdata/input/ddl/create_change_stream_for_table_columns.sql
new file mode 100644
index 0000000..80b51ad
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_table_columns.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name(column1, column2)
diff --git a/testdata/input/ddl/create_change_stream_for_tables.sql b/testdata/input/ddl/create_change_stream_for_tables.sql
new file mode 100644
index 0000000..7459e14
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_tables.sql
@@ -0,0 +1,1 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name1, table_name2
diff --git a/testdata/input/ddl/create_change_stream_for_tables_with_option.sql b/testdata/input/ddl/create_change_stream_for_tables_with_option.sql
new file mode 100644
index 0000000..76d0054
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_tables_with_option.sql
@@ -0,0 +1,2 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d')
diff --git a/testdata/input/ddl/create_change_stream_for_tables_with_option_null.sql b/testdata/input/ddl/create_change_stream_for_tables_with_option_null.sql
new file mode 100644
index 0000000..7ff03c8
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_tables_with_option_null.sql
@@ -0,0 +1,2 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = null)
diff --git a/testdata/input/ddl/create_change_stream_for_tables_with_options.sql b/testdata/input/ddl/create_change_stream_for_tables_with_options.sql
new file mode 100644
index 0000000..3385ac9
--- /dev/null
+++ b/testdata/input/ddl/create_change_stream_for_tables_with_options.sql
@@ -0,0 +1,2 @@
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d', value_capture_type = 'NEW_ROW')
diff --git a/testdata/input/ddl/drop_change_stream.sql b/testdata/input/ddl/drop_change_stream.sql
new file mode 100644
index 0000000..719c320
--- /dev/null
+++ b/testdata/input/ddl/drop_change_stream.sql
@@ -0,0 +1,1 @@
+DROP CHANGE STREAM change_stream_name
diff --git a/testdata/result/ddl/alter_change_stream_options.sql.txt b/testdata/result/ddl/alter_change_stream_options.sql.txt
new file mode 100644
index 0000000..9d5a091
--- /dev/null
+++ b/testdata/result/ddl/alter_change_stream_options.sql.txt
@@ -0,0 +1,48 @@
+--- alter_change_stream_options.sql
+ALTER CHANGE STREAM change_stream_name SET OPTIONS (retention_period = '1d', value_capture_type = 'OLD_AND_NEW_VALUES')
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetOptions{
+    Set:     39,
+    Options: &ast.ChangeStreamOptions{
+      Options: 43,
+      Rparen:  118,
+      Records: []*ast.ChangeStreamOptionsRecord{
+        &ast.ChangeStreamOptionsRecord{
+          Key: &ast.Ident{
+            NamePos: 52,
+            NameEnd: 68,
+            Name:    "retention_period",
+          },
+          Value: &ast.StringLiteral{
+            ValuePos: 71,
+            ValueEnd: 75,
+            Value:    "1d",
+          },
+        },
+        &ast.ChangeStreamOptionsRecord{
+          Key: &ast.Ident{
+            NamePos: 77,
+            NameEnd: 95,
+            Name:    "value_capture_type",
+          },
+          Value: &ast.StringLiteral{
+            ValuePos: 98,
+            ValueEnd: 118,
+            Value:    "OLD_AND_NEW_VALUES",
+          },
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET  OPTIONS (retention_period="1d", value_capture_type="OLD_AND_NEW_VALUES")
diff --git a/testdata/result/ddl/alter_change_stream_set_drop_for_all.sql.txt b/testdata/result/ddl/alter_change_stream_set_drop_for_all.sql.txt
new file mode 100644
index 0000000..b8c4b40
--- /dev/null
+++ b/testdata/result/ddl/alter_change_stream_set_drop_for_all.sql.txt
@@ -0,0 +1,19 @@
+--- alter_change_stream_set_drop_for_all.sql
+ALTER CHANGE STREAM change_stream_name DROP FOR ALL
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamDropForAll{
+    Drop: 39,
+    All:  48,
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name DROP FOR ALL
diff --git a/testdata/result/ddl/alter_change_stream_set_for_all.sql.txt b/testdata/result/ddl/alter_change_stream_set_for_all.sql.txt
new file mode 100644
index 0000000..e6285a9
--- /dev/null
+++ b/testdata/result/ddl/alter_change_stream_set_for_all.sql.txt
@@ -0,0 +1,22 @@
+--- alter_change_stream_set_for_all.sql
+ALTER CHANGE STREAM change_stream_name SET FOR ALL
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetFor{
+    Set: 39,
+    For: &ast.ChangeStreamForAll{
+      For: 43,
+      All: 51,
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET FOR ALL
diff --git a/testdata/result/ddl/alter_change_stream_set_for_table_columns.sql.txt b/testdata/result/ddl/alter_change_stream_set_for_table_columns.sql.txt
new file mode 100644
index 0000000..4d7bfc3
--- /dev/null
+++ b/testdata/result/ddl/alter_change_stream_set_for_table_columns.sql.txt
@@ -0,0 +1,63 @@
+--- alter_change_stream_set_for_table_columns.sql
+ALTER CHANGE STREAM change_stream_name SET FOR table_name1(column1, column2), table_name2(column1, column2)
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetFor{
+    Set: 39,
+    For: &ast.ChangeStreamForTables{
+      For:    43,
+      Tables: []*ast.ChangeStreamForTable{
+        &ast.ChangeStreamForTable{
+          Rparen:    75,
+          TableName: &ast.Ident{
+            NamePos: 47,
+            NameEnd: 58,
+            Name:    "table_name1",
+          },
+          Columns: []*ast.Ident{
+            &ast.Ident{
+              NamePos: 59,
+              NameEnd: 66,
+              Name:    "column1",
+            },
+            &ast.Ident{
+              NamePos: 68,
+              NameEnd: 75,
+              Name:    "column2",
+            },
+          },
+        },
+        &ast.ChangeStreamForTable{
+          Rparen:    106,
+          TableName: &ast.Ident{
+            NamePos: 78,
+            NameEnd: 89,
+            Name:    "table_name2",
+          },
+          Columns: []*ast.Ident{
+            &ast.Ident{
+              NamePos: 90,
+              NameEnd: 97,
+              Name:    "column1",
+            },
+            &ast.Ident{
+              NamePos: 99,
+              NameEnd: 106,
+              Name:    "column2",
+            },
+          },
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/result/ddl/create_change_stream.sql.txt b/testdata/result/ddl/create_change_stream.sql.txt
new file mode 100644
index 0000000..08efb8b
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream.sql.txt
@@ -0,0 +1,17 @@
+--- create_change_stream.sql
+CREATE CHANGE STREAM change_stream_name
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For:     nil,
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name
diff --git a/testdata/result/ddl/create_change_stream_for_all.sql.txt b/testdata/result/ddl/create_change_stream_for_all.sql.txt
new file mode 100644
index 0000000..3bc320c
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_all.sql.txt
@@ -0,0 +1,20 @@
+--- create_change_stream_for_all.sql
+CREATE CHANGE STREAM change_stream_name FOR ALL
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForAll{
+    For: 40,
+    All: 48,
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR ALL
diff --git a/testdata/result/ddl/create_change_stream_for_multiple_table_columns.sql.txt b/testdata/result/ddl/create_change_stream_for_multiple_table_columns.sql.txt
new file mode 100644
index 0000000..508c30b
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_multiple_table_columns.sql.txt
@@ -0,0 +1,61 @@
+--- create_change_stream_for_multiple_table_columns.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/result/ddl/create_change_stream_for_table.sql.txt b/testdata/result/ddl/create_change_stream_for_table.sql.txt
new file mode 100644
index 0000000..2b27ab7
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_table.sql.txt
@@ -0,0 +1,30 @@
+--- create_change_stream_for_table.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name
diff --git a/testdata/result/ddl/create_change_stream_for_table_column.sql.txt b/testdata/result/ddl/create_change_stream_for_table_column.sql.txt
new file mode 100644
index 0000000..895d322
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_table_column.sql.txt
@@ -0,0 +1,36 @@
+--- create_change_stream_for_table_column.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name(column)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    61,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 55,
+            NameEnd: 61,
+            Name:    "column",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name(column)
diff --git a/testdata/result/ddl/create_change_stream_for_table_columns.sql.txt b/testdata/result/ddl/create_change_stream_for_table_columns.sql.txt
new file mode 100644
index 0000000..40f9f7f
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_table_columns.sql.txt
@@ -0,0 +1,41 @@
+--- create_change_stream_for_table_columns.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name(column1, column2)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    71,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 55,
+            NameEnd: 62,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 64,
+            NameEnd: 71,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name(column1, column2)
diff --git a/testdata/result/ddl/create_change_stream_for_tables.sql.txt b/testdata/result/ddl/create_change_stream_for_tables.sql.txt
new file mode 100644
index 0000000..b877bac
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_tables.sql.txt
@@ -0,0 +1,39 @@
+--- create_change_stream_for_tables.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1, table_name2
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 57,
+          NameEnd: 68,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1, table_name2
diff --git a/testdata/result/ddl/create_change_stream_for_tables_with_option.sql.txt b/testdata/result/ddl/create_change_stream_for_tables_with_option.sql.txt
new file mode 100644
index 0000000..f8cb97c
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_tables_with_option.sql.txt
@@ -0,0 +1,79 @@
+--- create_change_stream_for_tables_with_option.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d')
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  136,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 132,
+          ValueEnd: 136,
+          Value:    "1d",
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period="1d")
diff --git a/testdata/result/ddl/create_change_stream_for_tables_with_option_null.sql.txt b/testdata/result/ddl/create_change_stream_for_tables_with_option_null.sql.txt
new file mode 100644
index 0000000..7bb6e07
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_tables_with_option_null.sql.txt
@@ -0,0 +1,77 @@
+--- create_change_stream_for_tables_with_option_null.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = null)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  136,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.NullLiteral{
+          Null: 132,
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period=NULL)
diff --git a/testdata/result/ddl/create_change_stream_for_tables_with_options.sql.txt b/testdata/result/ddl/create_change_stream_for_tables_with_options.sql.txt
new file mode 100644
index 0000000..94e6612
--- /dev/null
+++ b/testdata/result/ddl/create_change_stream_for_tables_with_options.sql.txt
@@ -0,0 +1,91 @@
+--- create_change_stream_for_tables_with_options.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d', value_capture_type = 'NEW_ROW')
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  168,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 132,
+          ValueEnd: 136,
+          Value:    "1d",
+        },
+      },
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 138,
+          NameEnd: 156,
+          Name:    "value_capture_type",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 159,
+          ValueEnd: 168,
+          Value:    "NEW_ROW",
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period="1d", value_capture_type="NEW_ROW")
diff --git a/testdata/result/ddl/drop_change_stream.sql.txt b/testdata/result/ddl/drop_change_stream.sql.txt
new file mode 100644
index 0000000..ecbe86f
--- /dev/null
+++ b/testdata/result/ddl/drop_change_stream.sql.txt
@@ -0,0 +1,15 @@
+--- drop_change_stream.sql
+DROP CHANGE STREAM change_stream_name
+
+--- AST
+&ast.DropChangeStream{
+  Drop: 0,
+  Name: &ast.Ident{
+    NamePos: 19,
+    NameEnd: 37,
+    Name:    "change_stream_name",
+  },
+}
+
+--- SQL
+DROP CHANGE STREAM change_stream_name
diff --git a/testdata/result/statement/alter_change_stream_options.sql.txt b/testdata/result/statement/alter_change_stream_options.sql.txt
new file mode 100644
index 0000000..9d5a091
--- /dev/null
+++ b/testdata/result/statement/alter_change_stream_options.sql.txt
@@ -0,0 +1,48 @@
+--- alter_change_stream_options.sql
+ALTER CHANGE STREAM change_stream_name SET OPTIONS (retention_period = '1d', value_capture_type = 'OLD_AND_NEW_VALUES')
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetOptions{
+    Set:     39,
+    Options: &ast.ChangeStreamOptions{
+      Options: 43,
+      Rparen:  118,
+      Records: []*ast.ChangeStreamOptionsRecord{
+        &ast.ChangeStreamOptionsRecord{
+          Key: &ast.Ident{
+            NamePos: 52,
+            NameEnd: 68,
+            Name:    "retention_period",
+          },
+          Value: &ast.StringLiteral{
+            ValuePos: 71,
+            ValueEnd: 75,
+            Value:    "1d",
+          },
+        },
+        &ast.ChangeStreamOptionsRecord{
+          Key: &ast.Ident{
+            NamePos: 77,
+            NameEnd: 95,
+            Name:    "value_capture_type",
+          },
+          Value: &ast.StringLiteral{
+            ValuePos: 98,
+            ValueEnd: 118,
+            Value:    "OLD_AND_NEW_VALUES",
+          },
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET  OPTIONS (retention_period="1d", value_capture_type="OLD_AND_NEW_VALUES")
diff --git a/testdata/result/statement/alter_change_stream_set_drop_for_all.sql.txt b/testdata/result/statement/alter_change_stream_set_drop_for_all.sql.txt
new file mode 100644
index 0000000..b8c4b40
--- /dev/null
+++ b/testdata/result/statement/alter_change_stream_set_drop_for_all.sql.txt
@@ -0,0 +1,19 @@
+--- alter_change_stream_set_drop_for_all.sql
+ALTER CHANGE STREAM change_stream_name DROP FOR ALL
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamDropForAll{
+    Drop: 39,
+    All:  48,
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name DROP FOR ALL
diff --git a/testdata/result/statement/alter_change_stream_set_for_all.sql.txt b/testdata/result/statement/alter_change_stream_set_for_all.sql.txt
new file mode 100644
index 0000000..e6285a9
--- /dev/null
+++ b/testdata/result/statement/alter_change_stream_set_for_all.sql.txt
@@ -0,0 +1,22 @@
+--- alter_change_stream_set_for_all.sql
+ALTER CHANGE STREAM change_stream_name SET FOR ALL
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetFor{
+    Set: 39,
+    For: &ast.ChangeStreamForAll{
+      For: 43,
+      All: 51,
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET FOR ALL
diff --git a/testdata/result/statement/alter_change_stream_set_for_table_columns.sql.txt b/testdata/result/statement/alter_change_stream_set_for_table_columns.sql.txt
new file mode 100644
index 0000000..4d7bfc3
--- /dev/null
+++ b/testdata/result/statement/alter_change_stream_set_for_table_columns.sql.txt
@@ -0,0 +1,63 @@
+--- alter_change_stream_set_for_table_columns.sql
+ALTER CHANGE STREAM change_stream_name SET FOR table_name1(column1, column2), table_name2(column1, column2)
+
+--- AST
+&ast.AlterChangeStream{
+  Alter: 0,
+  Name:  &ast.Ident{
+    NamePos: 20,
+    NameEnd: 38,
+    Name:    "change_stream_name",
+  },
+  ChangeStreamAlternation: &ast.ChangeStreamSetFor{
+    Set: 39,
+    For: &ast.ChangeStreamForTables{
+      For:    43,
+      Tables: []*ast.ChangeStreamForTable{
+        &ast.ChangeStreamForTable{
+          Rparen:    75,
+          TableName: &ast.Ident{
+            NamePos: 47,
+            NameEnd: 58,
+            Name:    "table_name1",
+          },
+          Columns: []*ast.Ident{
+            &ast.Ident{
+              NamePos: 59,
+              NameEnd: 66,
+              Name:    "column1",
+            },
+            &ast.Ident{
+              NamePos: 68,
+              NameEnd: 75,
+              Name:    "column2",
+            },
+          },
+        },
+        &ast.ChangeStreamForTable{
+          Rparen:    106,
+          TableName: &ast.Ident{
+            NamePos: 78,
+            NameEnd: 89,
+            Name:    "table_name2",
+          },
+          Columns: []*ast.Ident{
+            &ast.Ident{
+              NamePos: 90,
+              NameEnd: 97,
+              Name:    "column1",
+            },
+            &ast.Ident{
+              NamePos: 99,
+              NameEnd: 106,
+              Name:    "column2",
+            },
+          },
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+ALTER CHANGE STREAM change_stream_name SET FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/result/statement/create_change_stream.sql.txt b/testdata/result/statement/create_change_stream.sql.txt
new file mode 100644
index 0000000..08efb8b
--- /dev/null
+++ b/testdata/result/statement/create_change_stream.sql.txt
@@ -0,0 +1,17 @@
+--- create_change_stream.sql
+CREATE CHANGE STREAM change_stream_name
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For:     nil,
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name
diff --git a/testdata/result/statement/create_change_stream_for_all.sql.txt b/testdata/result/statement/create_change_stream_for_all.sql.txt
new file mode 100644
index 0000000..3bc320c
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_all.sql.txt
@@ -0,0 +1,20 @@
+--- create_change_stream_for_all.sql
+CREATE CHANGE STREAM change_stream_name FOR ALL
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForAll{
+    For: 40,
+    All: 48,
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR ALL
diff --git a/testdata/result/statement/create_change_stream_for_multiple_table_columns.sql.txt b/testdata/result/statement/create_change_stream_for_multiple_table_columns.sql.txt
new file mode 100644
index 0000000..508c30b
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_multiple_table_columns.sql.txt
@@ -0,0 +1,61 @@
+--- create_change_stream_for_multiple_table_columns.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
diff --git a/testdata/result/statement/create_change_stream_for_table.sql.txt b/testdata/result/statement/create_change_stream_for_table.sql.txt
new file mode 100644
index 0000000..2b27ab7
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_table.sql.txt
@@ -0,0 +1,30 @@
+--- create_change_stream_for_table.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name
diff --git a/testdata/result/statement/create_change_stream_for_table_column.sql.txt b/testdata/result/statement/create_change_stream_for_table_column.sql.txt
new file mode 100644
index 0000000..895d322
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_table_column.sql.txt
@@ -0,0 +1,36 @@
+--- create_change_stream_for_table_column.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name(column)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    61,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 55,
+            NameEnd: 61,
+            Name:    "column",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name(column)
diff --git a/testdata/result/statement/create_change_stream_for_table_columns.sql.txt b/testdata/result/statement/create_change_stream_for_table_columns.sql.txt
new file mode 100644
index 0000000..40f9f7f
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_table_columns.sql.txt
@@ -0,0 +1,41 @@
+--- create_change_stream_for_table_columns.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name(column1, column2)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    71,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 54,
+          Name:    "table_name",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 55,
+            NameEnd: 62,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 64,
+            NameEnd: 71,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name(column1, column2)
diff --git a/testdata/result/statement/create_change_stream_for_tables.sql.txt b/testdata/result/statement/create_change_stream_for_tables.sql.txt
new file mode 100644
index 0000000..b877bac
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_tables.sql.txt
@@ -0,0 +1,39 @@
+--- create_change_stream_for_tables.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1, table_name2
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    0,
+        TableName: &ast.Ident{
+          NamePos: 57,
+          NameEnd: 68,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident(nil),
+      },
+    },
+  },
+  Options: (*ast.ChangeStreamOptions)(nil),
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1, table_name2
diff --git a/testdata/result/statement/create_change_stream_for_tables_with_option.sql.txt b/testdata/result/statement/create_change_stream_for_tables_with_option.sql.txt
new file mode 100644
index 0000000..f8cb97c
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_tables_with_option.sql.txt
@@ -0,0 +1,79 @@
+--- create_change_stream_for_tables_with_option.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d')
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  136,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 132,
+          ValueEnd: 136,
+          Value:    "1d",
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period="1d")
diff --git a/testdata/result/statement/create_change_stream_for_tables_with_option_null.sql.txt b/testdata/result/statement/create_change_stream_for_tables_with_option_null.sql.txt
new file mode 100644
index 0000000..7bb6e07
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_tables_with_option_null.sql.txt
@@ -0,0 +1,77 @@
+--- create_change_stream_for_tables_with_option_null.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = null)
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  136,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.NullLiteral{
+          Null: 132,
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period=NULL)
diff --git a/testdata/result/statement/create_change_stream_for_tables_with_options.sql.txt b/testdata/result/statement/create_change_stream_for_tables_with_options.sql.txt
new file mode 100644
index 0000000..94e6612
--- /dev/null
+++ b/testdata/result/statement/create_change_stream_for_tables_with_options.sql.txt
@@ -0,0 +1,91 @@
+--- create_change_stream_for_tables_with_options.sql
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2)
+OPTIONS(retention_period = '1d', value_capture_type = 'NEW_ROW')
+
+--- AST
+&ast.CreateChangeStream{
+  Create: 0,
+  Name:   &ast.Ident{
+    NamePos: 21,
+    NameEnd: 39,
+    Name:    "change_stream_name",
+  },
+  For: &ast.ChangeStreamForTables{
+    For:    40,
+    Tables: []*ast.ChangeStreamForTable{
+      &ast.ChangeStreamForTable{
+        Rparen:    72,
+        TableName: &ast.Ident{
+          NamePos: 44,
+          NameEnd: 55,
+          Name:    "table_name1",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 56,
+            NameEnd: 63,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 65,
+            NameEnd: 72,
+            Name:    "column2",
+          },
+        },
+      },
+      &ast.ChangeStreamForTable{
+        Rparen:    103,
+        TableName: &ast.Ident{
+          NamePos: 75,
+          NameEnd: 86,
+          Name:    "table_name2",
+        },
+        Columns: []*ast.Ident{
+          &ast.Ident{
+            NamePos: 87,
+            NameEnd: 94,
+            Name:    "column1",
+          },
+          &ast.Ident{
+            NamePos: 96,
+            NameEnd: 103,
+            Name:    "column2",
+          },
+        },
+      },
+    },
+  },
+  Options: &ast.ChangeStreamOptions{
+    Options: 105,
+    Rparen:  168,
+    Records: []*ast.ChangeStreamOptionsRecord{
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 113,
+          NameEnd: 129,
+          Name:    "retention_period",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 132,
+          ValueEnd: 136,
+          Value:    "1d",
+        },
+      },
+      &ast.ChangeStreamOptionsRecord{
+        Key: &ast.Ident{
+          NamePos: 138,
+          NameEnd: 156,
+          Name:    "value_capture_type",
+        },
+        Value: &ast.StringLiteral{
+          ValuePos: 159,
+          ValueEnd: 168,
+          Value:    "NEW_ROW",
+        },
+      },
+    },
+  },
+}
+
+--- SQL
+CREATE CHANGE STREAM change_stream_name FOR table_name1(column1, column2), table_name2(column1, column2) OPTIONS (retention_period="1d", value_capture_type="NEW_ROW")
diff --git a/testdata/result/statement/drop_change_stream.sql.txt b/testdata/result/statement/drop_change_stream.sql.txt
new file mode 100644
index 0000000..ecbe86f
--- /dev/null
+++ b/testdata/result/statement/drop_change_stream.sql.txt
@@ -0,0 +1,15 @@
+--- drop_change_stream.sql
+DROP CHANGE STREAM change_stream_name
+
+--- AST
+&ast.DropChangeStream{
+  Drop: 0,
+  Name: &ast.Ident{
+    NamePos: 19,
+    NameEnd: 37,
+    Name:    "change_stream_name",
+  },
+}
+
+--- SQL
+DROP CHANGE STREAM change_stream_name