fix(planner): fix default stream for alias

Signed-off-by: Jiyong Huang <huangjy@emqx.io>

internal/topo/planner/analyzer.go

@@ -369,8 +369,6 @@ func (s *streamFieldMapSchemaless) bindRef(fr *ast.FieldRef) error {
 			for sk := range s.content {
 				fr.StreamName = sk
 			}
-		} else {
-			fr.StreamName = s.defaultStream
 		}
 	}
 	k := fr.StreamName
@@ -378,7 +376,8 @@ func (s *streamFieldMapSchemaless) bindRef(fr *ast.FieldRef) error {
 		switch l {
 		case 0: // must be a column because alias are fields and have been traversed
 			// reserve a hole and do nothing
-			s.content[k] = nil
+			fr.StreamName = s.defaultStream
+			s.content[s.defaultStream] = nil
 			return nil
 		case 1: // if alias or single col, return this
 			for sk, sv := range s.content {
@@ -393,10 +392,12 @@ func (s *streamFieldMapSchemaless) bindRef(fr *ast.FieldRef) error {
 				fr.StreamName = ast.AliasStream
 				return nil
 			} else {
-				return fmt.Errorf("ambiguous field ")
+				fr.StreamName = s.defaultStream
 			}
 		}
-	} else {
+	}
+
+	if fr.StreamName != ast.DefaultStream {
 		r, ok := s.content[k]
 		if !ok { // reserver a hole
 			s.content[k] = nil
@@ -405,4 +406,5 @@ func (s *streamFieldMapSchemaless) bindRef(fr *ast.FieldRef) error {
 		}
 		return nil
 	}
+	return fmt.Errorf("ambiguous field ")
 }

internal/topo/planner/analyzer_test.go

@@ -112,6 +112,10 @@ var tests = []struct {
 		sql: `SELECT sin(temp) as temp1, cos(temp1) FROM src1`,
 		r:   newErrorStructWithS("unknown field temp1", ""),
 	},
+	{ // 14
+		sql: `SELECT collect(*)[-1] as current FROM src1 GROUP BY COUNTWINDOW(2, 1) HAVING isNull(current->name) = false`,
+		r:   newErrorStruct(""),
+	},
 }
 
 func Test_validation(t *testing.T) {