fix: fix alias ref cycle check (#2121)

* fix test

Signed-off-by: yisaer <disxiaofei@163.com>

* fix test

Signed-off-by: yisaer <disxiaofei@163.com>

---------

Signed-off-by: yisaer <disxiaofei@163.com>

internal/topo/planner/analyzer.go

@@ -51,6 +51,9 @@ func decorateStmt(s *ast.SelectStatement, store kv.KeyValue) ([]*streamInfo, []*
 			isSchemaless = true
 		}
 	}
+	if checkAliasReferenceCycle(s) {
+		return nil, nil, fmt.Errorf("select fields have cycled alias")
+	}
 	if !isSchemaless {
 		aliasFieldTopoSort(s, streamStmts)
 	}
@@ -219,6 +222,51 @@ func (a aliasTopoDegrees) Swap(i, j int) {
 	a[i], a[j] = a[j], a[i]
 }
 
+// checkAliasReferenceCycle checks whether exists select a + 1 as b, b + 1 as a from demo;
+func checkAliasReferenceCycle(s *ast.SelectStatement) bool {
+	aliasRef := make(map[string]map[string]struct{})
+	for _, field := range s.Fields {
+		if len(field.AName) > 0 {
+			aliasRef[field.AName] = make(map[string]struct{})
+		}
+	}
+	if len(aliasRef) < 1 {
+		return false
+	}
+	hasCycleAlias := false
+	for _, field := range s.Fields {
+		if len(field.AName) > 0 {
+			ast.WalkFunc(&field, func(node ast.Node) bool {
+				switch f := node.(type) {
+				case *ast.FieldRef:
+					if len(f.Name) > 0 {
+						if f.Name == field.AName {
+							return true
+						}
+						_, ok := aliasRef[f.Name]
+						if ok {
+							aliasRef[field.AName][f.Name] = struct{}{}
+							v, ok1 := aliasRef[f.Name]
+							if ok1 {
+								_, ok2 := v[field.AName]
+								if ok2 {
+									hasCycleAlias = true
+									return false
+								}
+							}
+						}
+					}
+				}
+				return true
+			})
+			if hasCycleAlias {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func aliasFieldTopoSort(s *ast.SelectStatement, streamStmts []*streamInfo) {
 	nonAliasFields := make([]ast.Field, 0)
 	aliasDegreeMap := make(map[string]*aliasTopoDegree)

internal/topo/planner/analyzer_test.go

@@ -138,6 +138,10 @@ var tests = []struct {
 		sql: `SELECT * FROM src1 GROUP BY SlidingWindow(ss,5) Over (WHEN last_hit_time() > 1) HAVING last_agg_hit_count() < 3`,
 		r:   newErrorStruct(""),
 	},
+	{
+		sql: "select a + 1 as b, b + 1 as a from src1",
+		r:   newErrorStruct("select fields have cycled alias"),
+	},
 	//{ // 19 already captured in parser
 	//	sql: `SELECT * FROM src1 GROUP BY SlidingWindow(ss,5) Over (WHEN abs(sum(a)) > 1) HAVING last_agg_hit_count() < 3`,
 	//	r:   newErrorStruct("error compile sql: Not allowed to call aggregate functions in GROUP BY clause."),