Add basic VIEWS support.

engine.go

@@ -120,7 +120,7 @@ func (e *Engine) Query(
 	case *plan.CreateIndex:
 		typ = sql.CreateIndexProcess
 		perm = auth.ReadPerm | auth.WritePerm
-	case *plan.InsertInto, *plan.DeleteFrom, *plan.Update, *plan.DropIndex, *plan.UnlockTables, *plan.LockTables:
+	case *plan.InsertInto, *plan.DeleteFrom, *plan.Update, *plan.DropIndex, *plan.UnlockTables, *plan.LockTables, *plan.CreateView, *plan.DropView:
 		perm = auth.ReadPerm | auth.WritePerm
 	}
 

engine_test.go

@@ -3661,17 +3661,19 @@ func TestReadOnly(t *testing.T) {
 	_, _, err := e.Query(newCtx(), `SELECT i FROM mytable`)
 	require.NoError(err)
 
-	_, _, err = e.Query(newCtx(), `CREATE INDEX foo ON mytable USING pilosa (i, s)`)
-	require.Error(err)
-	require.True(auth.ErrNotAuthorized.Is(err))
-
-	_, _, err = e.Query(newCtx(), `DROP INDEX foo ON mytable`)
-	require.Error(err)
-	require.True(auth.ErrNotAuthorized.Is(err))
+	writingQueries := []string{
+		`CREATE INDEX foo ON mytable USING pilosa (i, s)`,
+		`DROP INDEX foo ON mytable`,
+		`INSERT INTO mytable (i, s) VALUES(42, 'yolo')`,
+		`CREATE VIEW myview AS SELECT i FROM mytable`,
+		`DROP VIEW myview`,
+	}
 
-	_, _, err = e.Query(newCtx(), `INSERT INTO mytable (i, s) VALUES(42, 'yolo')`)
-	require.Error(err)
-	require.True(auth.ErrNotAuthorized.Is(err))
+	for _, query := range writingQueries {
+		_, _, err = e.Query(newCtx(), query)
+		require.Error(err)
+		require.True(auth.ErrNotAuthorized.Is(err))
+	}
 }
 
 func TestSessionVariables(t *testing.T) {

sql/analyzer/assign_catalog.go

@@ -60,6 +60,14 @@ func assignCatalog(ctx *sql.Context, a *Analyzer, n sql.Node) (sql.Node, error)
 			nc := *node
 			nc.Catalog = a.Catalog
 			return &nc, nil
+		case *plan.CreateView:
+			nc := *node
+			nc.Catalog = a.Catalog
+			return &nc, nil
+		case *plan.DropView:
+			nc := *node
+			nc.Catalog = a.Catalog
+			return &nc, nil
 		default:
 			return n, nil
 		}

sql/analyzer/assign_catalog_test.go

@@ -73,4 +73,18 @@ func TestAssignCatalog(t *testing.T) {
 	ut, ok := node.(*plan.UnlockTables)
 	require.True(ok)
 	require.Equal(c, ut.Catalog)
+
+	mockSubquery := plan.NewSubqueryAlias("mock", plan.NewResolvedTable(tbl))
+	mockView := plan.NewCreateView(db, "", nil, mockSubquery, false)
+	node, err = f.Apply(sql.NewEmptyContext(), a, mockView)
+	require.NoError(err)
+	cv, ok := node.(*plan.CreateView)
+	require.True(ok)
+	require.Equal(c, cv.Catalog)
+
+	node, err = f.Apply(sql.NewEmptyContext(), a, plan.NewDropView(nil, false))
+	require.NoError(err)
+	dv, ok := node.(*plan.DropView)
+	require.True(ok)
+	require.Equal(c, dv.Catalog)
 }

sql/analyzer/resolve_tables.go

@@ -39,17 +39,26 @@ func resolveTables(ctx *sql.Context, a *Analyzer, n sql.Node) (sql.Node, error)
 		}
 
 		rt, err := a.Catalog.Table(db, name)
-		if err != nil {
-			if sql.ErrTableNotFound.Is(err) && name == dualTableName {
+		if err == nil {
+			a.Log("table resolved: %q", t.Name())
+			return plan.NewResolvedTable(rt), nil
+		}
+
+		if sql.ErrTableNotFound.Is(err) {
+			if name == dualTableName {
 				rt = dualTable
 				name = dualTableName
-			} else {
-				return nil, err
+
+				a.Log("table resolved: %q", t.Name())
+				return plan.NewResolvedTable(rt), nil
 			}
-		}
 
-		a.Log("table resolved: %q", t.Name())
+			if view, err := a.Catalog.ViewRegistry.View(db, name); err == nil {
+				a.Log("table %q is a view: replacing plans", t.Name())
+				return view.Definition(), nil
+			}
+		}
 
-		return plan.NewResolvedTable(rt), nil
+		return nil, err
 	})
 }

sql/analyzer/resolve_tables_test.go

@@ -91,3 +91,45 @@ func TestResolveTablesNested(t *testing.T) {
 	)
 	require.Equal(expected, analyzed)
 }
+
+func TestResolveViews(t *testing.T) {
+	require := require.New(t)
+
+	f := getRule("resolve_tables")
+
+	table := memory.NewTable("mytable", sql.Schema{{Name: "i", Type: sql.Int32}})
+	db := memory.NewDatabase("mydb")
+	db.AddTable("mytable", table)
+
+	// Resolved plan that corresponds to query "SELECT i FROM mytable"
+	subquery := plan.NewProject(
+		[]sql.Expression{
+			expression.NewGetFieldWithTable(
+				1, sql.Int32, table.Name(), "i", true),
+		},
+		plan.NewResolvedTable(table),
+	)
+	subqueryAlias := plan.NewSubqueryAlias("myview", subquery)
+	view := sql.NewView("myview", subqueryAlias)
+
+	catalog := sql.NewCatalog()
+	catalog.AddDatabase(db)
+	err := catalog.ViewRegistry.Register(db.Name(), view)
+	require.NoError(err)
+
+	a := NewBuilder(catalog).AddPostAnalyzeRule(f.Name, f.Apply).Build()
+
+	var notAnalyzed sql.Node = plan.NewUnresolvedTable("myview", "")
+	analyzed, err := f.Apply(sql.NewEmptyContext(), a, notAnalyzed)
+	require.NoError(err)
+	require.Equal(subqueryAlias, analyzed)
+
+	notAnalyzed = plan.NewUnresolvedTable("MyVieW", "")
+	analyzed, err = f.Apply(sql.NewEmptyContext(), a, notAnalyzed)
+	require.NoError(err)
+	require.Equal(subqueryAlias, analyzed)
+
+	analyzed, err = f.Apply(sql.NewEmptyContext(), a, subqueryAlias)
+	require.NoError(err)
+	require.Equal(subqueryAlias, analyzed)
+}

sql/catalog.go

@@ -18,6 +18,7 @@ var ErrDatabaseNotFound = errors.NewKind("database not found: %s")
 type Catalog struct {
 	FunctionRegistry
 	*IndexRegistry
+	*ViewRegistry
 	*ProcessList
 	*MemoryManager
 
@@ -38,6 +39,7 @@ func NewCatalog() *Catalog {
 	return &Catalog{
 		FunctionRegistry: NewFunctionRegistry(),
 		IndexRegistry:    NewIndexRegistry(),
+		ViewRegistry:     NewViewRegistry(),
 		MemoryManager:    NewMemoryManager(ProcessMemory),
 		ProcessList:      NewProcessList(),
 		locks:            make(sessionLocks),

sql/index.go

@@ -589,7 +589,7 @@ func exprListsEqual(a, b []string) bool {
 // marked as creating, so nobody can't register two indexes with the same
 // expression or id while the other is still being created.
 // When something is sent through the returned channel, it means the index has
-// finished it's creation and will be marked as ready.
+// finished its creation and will be marked as ready.
 // Another channel is returned to notify the user when the index is ready.
 func (r *IndexRegistry) AddIndex(
 	idx Index,

sql/parse/parse.go

@@ -47,7 +47,6 @@ var (
 	unlockTablesRegex    = regexp.MustCompile(`^unlock\s+tables$`)
 	lockTablesRegex      = regexp.MustCompile(`^lock\s+tables\s`)
 	setRegex             = regexp.MustCompile(`^set\s+`)
-	createViewRegex      = regexp.MustCompile(`^create\s+view\s+`)
 )
 
 // These constants aren't exported from vitess for some reason. This could be removed if we changed this.
@@ -104,9 +103,6 @@ func Parse(ctx *sql.Context, query string) (sql.Node, error) {
 		return parseLockTables(ctx, s)
 	case setRegex.MatchString(lowerQuery):
 		s = fixSetQuery(s)
-	case createViewRegex.MatchString(lowerQuery):
-		// CREATE VIEW parses as a CREATE DDL statement with an empty table spec
-		return nil, ErrUnsupportedFeature.New("CREATE VIEW")
 	}
 
 	stmt, err := sqlparser.Parse(s)
@@ -163,7 +159,7 @@ func convert(ctx *sql.Context, stmt sqlparser.Statement, query string) (sql.Node
 		if err != nil {
 			return nil, err
 		}
-		return convertDDL(ddl.(*sqlparser.DDL))
+		return convertDDL(ctx, ddl.(*sqlparser.DDL))
 	case *sqlparser.Set:
 		return convertSet(ctx, n)
 	case *sqlparser.Use:
@@ -369,11 +365,17 @@ func convertSelect(ctx *sql.Context, s *sqlparser.Select) (sql.Node, error) {
 	return node, nil
 }
 
-func convertDDL(c *sqlparser.DDL) (sql.Node, error) {
+func convertDDL(ctx *sql.Context, c *sqlparser.DDL) (sql.Node, error) {
 	switch c.Action {
 	case sqlparser.CreateStr:
+		if !c.View.IsEmpty() {
+			return convertCreateView(ctx, c)
+		}
 		return convertCreateTable(c)
 	case sqlparser.DropStr:
+                if len(c.FromViews) != 0 {
+			return convertDropView(ctx, c)
+		}
 		return convertDropTable(c)
 	default:
 		return nil, ErrUnsupportedSyntax.New(c)
@@ -398,6 +400,31 @@ func convertCreateTable(c *sqlparser.DDL) (sql.Node, error) {
 		sql.UnresolvedDatabase(""), c.Table.Name.String(), schema), nil
 }
 
+func convertCreateView(ctx *sql.Context, c *sqlparser.DDL) (sql.Node, error) {
+	selectStatement, ok := c.ViewExpr.(*sqlparser.Select)
+	if !ok {
+		return nil, ErrUnsupportedSyntax.New(c.ViewExpr)
+	}
+
+        queryNode, err := convertSelect(ctx, selectStatement)
+        if err != nil {
+                return nil, err
+        }
+
+	queryAlias := plan.NewSubqueryAlias(c.View.Name.String(), queryNode)
+
+	return plan.NewCreateView(
+		sql.UnresolvedDatabase(""), c.View.Name.String(), []string{}, queryAlias, c.OrReplace), nil
+}
+
+func convertDropView(ctx *sql.Context, c *sqlparser.DDL) (sql.Node, error) {
+	plans := make([]sql.Node, len(c.FromViews))
+	for i, v := range c.FromViews {
+		plans[i] = plan.NewSingleDropView(sql.UnresolvedDatabase(""), v.Name.String())
+	}
+	return plan.NewDropView(plans, c.IfExists), nil
+}
+
 func convertInsert(ctx *sql.Context, i *sqlparser.Insert) (sql.Node, error) {
 	if len(i.OnDup) > 0 {
 		return nil, ErrUnsupportedFeature.New("ON DUPLICATE KEY")

sql/parse/parse_test.go

@@ -55,44 +55,44 @@ var fixtures = map[string]sql.Node{
 		sql.UnresolvedDatabase(""),
 		"t1",
 		sql.Schema{{
-			Name:     "a",
-			Type:     sql.Int32,
-			Nullable: false,
+			Name:       "a",
+			Type:       sql.Int32,
+			Nullable:   false,
 			PrimaryKey: true,
 		}, {
-			Name:     "b",
-			Type:     sql.Text,
-			Nullable: true,
+			Name:       "b",
+			Type:       sql.Text,
+			Nullable:   true,
 			PrimaryKey: false,
 		}},
 	),
 	`CREATE TABLE t1(a INTEGER, b TEXT, PRIMARY KEY (a))`: plan.NewCreateTable(
 		sql.UnresolvedDatabase(""),
 		"t1",
 		sql.Schema{{
-			Name:     "a",
-			Type:     sql.Int32,
-			Nullable: false,
+			Name:       "a",
+			Type:       sql.Int32,
+			Nullable:   false,
 			PrimaryKey: true,
 		}, {
-			Name:     "b",
-			Type:     sql.Text,
-			Nullable: true,
+			Name:       "b",
+			Type:       sql.Text,
+			Nullable:   true,
 			PrimaryKey: false,
 		}},
 	),
 	`CREATE TABLE t1(a INTEGER, b TEXT, PRIMARY KEY (a, b))`: plan.NewCreateTable(
 		sql.UnresolvedDatabase(""),
 		"t1",
 		sql.Schema{{
-			Name:     "a",
-			Type:     sql.Int32,
-			Nullable: false,
+			Name:       "a",
+			Type:       sql.Int32,
+			Nullable:   false,
 			PrimaryKey: true,
 		}, {
-			Name:     "b",
-			Type:     sql.Text,
-			Nullable: false,
+			Name:       "b",
+			Type:       sql.Text,
+			Nullable:   false,
 			PrimaryKey: true,
 		}},
 	),
@@ -1321,7 +1321,7 @@ var fixturesErrors = map[string]*errors.Kind{
 	`SELECT INTERVAL 1 DAY + INTERVAL 1 DAY`:                  ErrUnsupportedSyntax,
 	`SELECT '2018-05-01' + (INTERVAL 1 DAY + INTERVAL 1 DAY)`: ErrUnsupportedSyntax,
 	`SELECT AVG(DISTINCT foo) FROM b`:                         ErrUnsupportedSyntax,
-	`CREATE VIEW view1 AS SELECT x FROM t1 WHERE x>0`:         ErrUnsupportedFeature,
+	`CREATE VIEW myview AS SELECT AVG(DISTINCT foo) FROM b`:   ErrUnsupportedSyntax,
 }
 
 func TestParseErrors(t *testing.T) {