schemadiff: support views with CTE (#18893)

Signed-off-by: Shlomi Noach <[email protected]>

Author: shlomi-noach

go/vt/schemadiff/schema.go

@@ -134,21 +134,31 @@ func getForeignKeyParentTableNames(createTable *sqlparser.CreateTable) (names []
 }
 
 // getViewDependentTableNames analyzes a CREATE VIEW definition and extracts all tables/views read by this view
-func getViewDependentTableNames(createView *sqlparser.CreateView) (names []string) {
+func getViewDependentTableNames(createView *sqlparser.CreateView) (names []string, cteNames []string) {
+	cteMap := make(map[string]bool)
 	_ = sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 		switch node := node.(type) {
+		case *sqlparser.CommonTableExpr:
+			if !cteMap[node.ID.String()] {
+				cteNames = append(cteNames, node.ID.String())
+				cteMap[node.ID.String()] = true
+			}
 		case *sqlparser.TableName:
-			names = append(names, node.Name.String())
+			if _, isCte := cteMap[node.Name.String()]; !isCte {
+				names = append(names, node.Name.String())
+			}
 		case *sqlparser.AliasedTableExpr:
 			if tableName, ok := node.Expr.(sqlparser.TableName); ok {
-				names = append(names, tableName.Name.String())
+				if _, isCte := cteMap[tableName.Name.String()]; !isCte {
+					names = append(names, tableName.Name.String())
+				}
 			}
 			// or, this could be a more complex expression, like a derived table `(select * from v1) as derived`,
 			// in which case further Walk-ing will eventually find the "real" table name
 		}
 		return true, nil
 	}, createView)
-	return names
+	return names, cteNames
 }
 
 // normalize is called as part of Schema creation process. The user may only get a hold of normalized schema.
@@ -310,7 +320,7 @@ func (s *Schema) normalize(hints *DiffHints) error {
 				continue
 			}
 			// Not handled. Is this view dependent on already handled objects?
-			dependentNames := getViewDependentTableNames(v.CreateView)
+			dependentNames, _ := getViewDependentTableNames(v.CreateView)
 			if allNamesFoundInLowerLevel(dependentNames, iterationLevel) {
 				s.sorted = append(s.sorted, v)
 				dependencyLevels[v.Name()] = iterationLevel
@@ -341,7 +351,7 @@ func (s *Schema) normalize(hints *DiffHints) error {
 			if _, ok := dependencyLevels[v.Name()]; !ok {
 				// We _know_ that in this iteration, at least one view is found unassigned a dependency level.
 				// We gather all the errors.
-				dependentNames := getViewDependentTableNames(v.CreateView)
+				dependentNames, _ := getViewDependentTableNames(v.CreateView)
 				missingReferencedEntities := []string{}
 				for _, name := range dependentNames {
 					if _, ok := dependencyLevels[name]; !ok {
@@ -974,12 +984,16 @@ func (s *Schema) SchemaDiff(other *Schema, hints *DiffHints) (*SchemaDiff, error
 	for _, diff := range schemaDiff.UnorderedDiffs() {
 		switch diff := diff.(type) {
 		case *CreateViewEntityDiff:
-			checkDependencies(diff, getViewDependentTableNames(diff.createView))
+			dependentNames, _ := getViewDependentTableNames(diff.createView)
+			checkDependencies(diff, dependentNames)
 		case *AlterViewEntityDiff:
-			checkDependencies(diff, getViewDependentTableNames(diff.from.CreateView))
-			checkDependencies(diff, getViewDependentTableNames(diff.to.CreateView))
+			fromDependentNames, _ := getViewDependentTableNames(diff.from.CreateView)
+			checkDependencies(diff, fromDependentNames)
+			toDependentNames, _ := getViewDependentTableNames(diff.to.CreateView)
+			checkDependencies(diff, toDependentNames)
 		case *DropViewEntityDiff:
-			checkDependencies(diff, getViewDependentTableNames(diff.from.CreateView))
+			dependentNames, _ := getViewDependentTableNames(diff.from.CreateView)
+			checkDependencies(diff, dependentNames)
 		case *CreateTableEntityDiff:
 			checkDependencies(diff, getForeignKeyParentTableNames(diff.CreateTable()))
 			_ = sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
@@ -1142,13 +1156,15 @@ func (s *Schema) getViewColumnNames(v *CreateViewEntity, schemaInformation *decl
 	for _, node := range v.Select.GetColumns() {
 		switch node := node.(type) {
 		case *sqlparser.StarExpr:
+			dependentNames, cteNames := getViewDependentTableNames(v.CreateView)
 			if tableName := node.TableName.Name.String(); tableName != "" {
-				for _, col := range schemaInformation.Tables[tableName].Columns {
-					name := sqlparser.Clone(col.Name)
-					columnNames = append(columnNames, &name)
+				if tbl, ok := schemaInformation.Tables[tableName]; ok {
+					for _, col := range tbl.Columns {
+						name := sqlparser.Clone(col.Name)
+						columnNames = append(columnNames, &name)
+					}
 				}
 			} else {
-				dependentNames := getViewDependentTableNames(v.CreateView)
 				// add all columns from all referenced tables and views
 				for _, entityName := range dependentNames {
 					if schemaInformation.Tables[entityName] != nil { // is nil for dual/DUAL
@@ -1159,7 +1175,10 @@ func (s *Schema) getViewColumnNames(v *CreateViewEntity, schemaInformation *decl
 					}
 				}
 			}
-			if len(columnNames) == 0 {
+			if len(columnNames) == 0 && len(cteNames) == 0 {
+				// *-expressions that do not resolve to any columns are invalid in views.
+				// For CTEs, schemadiff does not analyze the list of columns returned by the CTE (even if the CTE defines it).
+				// TODO(shlomi): analyze CTE columns as well.
 				return nil, &InvalidStarExprInViewError{View: v.Name()}
 			}
 		case *sqlparser.AliasedExpr:

go/vt/schemadiff/schema_test.go

@@ -176,6 +176,52 @@ func TestNewSchemaFromQueriesViewFromDualImplicit(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestNewSchemaFromQueriesViewWithCTEFail(t *testing.T) {
+	queries := []string{"create view v30 as with vcte as (select 1) select * from vcte2"}
+	_, err := NewSchemaFromQueries(NewTestEnv(), queries)
+	assert.Error(t, err)
+	assert.EqualError(t, err, (&ViewDependencyUnresolvedError{View: "v30", MissingReferencedEntities: []string{"dual", "vcte2"}}).Error())
+}
+
+func TestNewSchemaFromQueriesViewWithCTE(t *testing.T) {
+	tcases := []struct {
+		name    string
+		queries []string
+	}{
+		{
+			"no table",
+			[]string{"create view v20 as with vcte as (select 1) select * from vcte"},
+		},
+		{
+			"with table",
+			[]string{
+				"create table orders (id int primary key, info int not null)",
+				"create view v21 as with vcte as (select * from orders) select * from vcte",
+			},
+		},
+		{
+			"with table and column aliasing",
+			[]string{
+				"create table orders (id int primary key, info int not null)",
+				"create view v22 as with vcte as (select id, info as val from orders) select * from vcte",
+			},
+		},
+		{
+			"with table and select all from cte",
+			[]string{
+				"create table orders (id int primary key, info int not null)",
+				"create view v22 as with vcte as (select id, info as val from orders) select vcte.* from vcte",
+			},
+		},
+	}
+	for _, tc := range tcases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewSchemaFromQueries(NewTestEnv(), tc.queries)
+			assert.NoError(t, err)
+		})
+	}
+}
+
 func TestNewSchemaFromQueriesLoop(t *testing.T) {
 	// v7 and v8 depend on each other
 	queries := append(schemaTestCreateQueries,
@@ -213,6 +259,7 @@ func TestGetViewDependentTableNames(t *testing.T) {
 		name   string
 		view   string
 		tables []string
+		ctes   []string
 	}{
 		{
 			view:   "create view v6 as select * from v4",
@@ -242,6 +289,16 @@ func TestGetViewDependentTableNames(t *testing.T) {
 			view:   "create view v9 as select 1",
 			tables: []string{"dual"},
 		},
+		{
+			view:   "create view v20 as with vcte as (select 1) select * from vcte",
+			tables: []string{"dual"},
+			ctes:   []string{"vcte"},
+		},
+		{
+			view:   "create view v21 as with vcte as (select * from orders) select * from vcte",
+			tables: []string{"orders"},
+			ctes:   []string{"vcte"},
+		},
 	}
 	for _, ts := range tt {
 		t.Run(ts.view, func(t *testing.T) {
@@ -250,8 +307,9 @@ func TestGetViewDependentTableNames(t *testing.T) {
 			createView, ok := stmt.(*sqlparser.CreateView)
 			require.True(t, ok)
 
-			tables := getViewDependentTableNames(createView)
+			tables, ctes := getViewDependentTableNames(createView)
 			assert.Equal(t, ts.tables, tables)
+			assert.Equal(t, ts.ctes, ctes)
 		})
 	}
 }