Parse and analyze database-scoped names for views

Signed-off-by: Alejandro García Montoro <alejandro.garciamontoro@gmail.com>

Author: agarciamontoro

sql/parse/util.go

@@ -127,6 +127,27 @@ func readLetter(r *bufio.Reader, buf *bytes.Buffer) error {
 	return nil
 }
 
+func readLetterOrPoint(r *bufio.Reader, buf *bytes.Buffer) error {
+	ru, _, err := r.ReadRune()
+	if err != nil {
+		if err == io.EOF {
+			return nil
+		}
+
+		return err
+	}
+
+	if !unicode.IsLetter(ru) && ru != '.' {
+		if err := r.UnreadRune(); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	buf.WriteRune(ru)
+	return nil
+}
+
 func readValidIdentRune(r *bufio.Reader, buf *bytes.Buffer) error {
 	ru, _, err := r.ReadRune()
 	if err != nil {
@@ -144,6 +165,23 @@ func readValidIdentRune(r *bufio.Reader, buf *bytes.Buffer) error {
 	return nil
 }
 
+func readValidScopedIdentRune(r *bufio.Reader, separator rune, buf *bytes.Buffer) error {
+	ru, _, err := r.ReadRune()
+	if err != nil {
+		return err
+	}
+
+	if !unicode.IsLetter(ru) && !unicode.IsDigit(ru) && ru != '_' && ru != separator {
+		if err := r.UnreadRune(); err != nil {
+			return err
+		}
+		return io.EOF
+	}
+
+	buf.WriteRune(ru)
+	return nil
+}
+
 func readValidQuotedIdentRune(r *bufio.Reader, buf *bytes.Buffer) error {
 	bs, err := r.Peek(2)
 	if err != nil {
@@ -199,6 +237,29 @@ func readIdent(ident *string) parseFunc {
 	}
 }
 
+func readScopedIdent(separator rune, idents *[]string) parseFunc {
+	return func(r *bufio.Reader) error {
+		var buf bytes.Buffer
+		if err := readLetter(r, &buf); err != nil {
+			return err
+		}
+
+		for {
+			if err := readValidScopedIdentRune(r, separator, &buf); err == io.EOF {
+				break
+			} else if err != nil {
+				return err
+			}
+		}
+
+		*idents = append(
+			*idents,
+			strings.Split(strings.ToLower(buf.String()), string(separator))...,
+		)
+		return nil
+	}
+}
+
 func readQuotedIdent(ident *string) parseFunc {
 	return func(r *bufio.Reader) error {
 		var buf bytes.Buffer

sql/parse/views.go

@@ -11,18 +11,21 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
+var ErrMalformedViewName = errors.NewKind("the view name '%s' is not correct")
 var ErrMalformedCreateView = errors.NewKind("view definition %#v is not a SELECT query")
 
 // Parses
-// CREATE [OR REPLACE] VIEW view_name [(col1, col2, ...)] AS select_statement
+// CREATE [OR REPLACE] VIEW [db_name.]view_name [(col1, col2, ...)] AS select_statement
 // and returns a NewCreateView node in case of success
 func parseCreateView(ctx *sql.Context, s string) (sql.Node, error) {
 	r := bufio.NewReader(strings.NewReader(s))
 
 	var (
-		viewName, subquery string
-		columns            []string
-		isReplace          bool
+		databaseName, viewName string
+		scopedName             []string
+		subquery               string
+		columns                []string
+		isReplace              bool
 	)
 
 	err := parseFuncs{
@@ -32,7 +35,7 @@ func parseCreateView(ctx *sql.Context, s string) (sql.Node, error) {
 		skipSpaces,
 		expect("view"),
 		skipSpaces,
-		readIdent(&viewName),
+		readScopedIdent('.', &scopedName),
 		skipSpaces,
 		maybeList('(', ',', ')', &columns),
 		skipSpaces,
@@ -46,6 +49,18 @@ func parseCreateView(ctx *sql.Context, s string) (sql.Node, error) {
 		return nil, err
 	}
 
+	if len(scopedName) < 1 || len(scopedName) > 2 {
+		return nil, ErrMalformedViewName.New(strings.Join(scopedName, "."))
+	}
+
+	if len(scopedName) == 1 {
+		viewName = scopedName[0]
+	}
+	if len(scopedName) == 2 {
+		databaseName = scopedName[0]
+		viewName = scopedName[1]
+	}
+
 	subqueryStatement, err := sqlparser.Parse(subquery)
 	if err != nil {
 		return nil, err
@@ -64,6 +79,6 @@ func parseCreateView(ctx *sql.Context, s string) (sql.Node, error) {
 	subqueryAlias := plan.NewSubqueryAlias(viewName, subqueryNode)
 
 	return plan.NewCreateView(
-		sql.UnresolvedDatabase(""), viewName, columns, subqueryAlias,
+		sql.UnresolvedDatabase(databaseName), viewName, columns, subqueryAlias,
 	), nil
 }

sql/plan/create_view.go

@@ -12,7 +12,7 @@ type CreateView struct {
 	database sql.Database
 	Name     string
 	Columns  []string
-	Catalog *sql.Catalog
+	Catalog  *sql.Catalog
 }
 
 func NewCreateView(
@@ -71,3 +71,13 @@ func (create *CreateView) WithChildren(children ...sql.Node) (sql.Node, error) {
 	newCreate.Child = children[0]
 	return newCreate, nil
 }
+
+func (create *CreateView) Database() sql.Database {
+	return create.database
+}
+
+func (create *CreateView) WithDatabase(database sql.Database) (sql.Node, error) {
+	newCreate := *create
+	newCreate.database = database
+	return &newCreate, nil
+}