fix(planner): avoid schema null panics in slice mode

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

Author: ngjaying

internal/topo/operator/project_operator.go

@@ -55,6 +55,8 @@ type ProjectOp struct {
 	// compiledExprs caches pre-compiled accessors for ExprFields.
 	// For simple FieldRef fields, isDirect=true and we skip the AST walk.
 	compiledExprs []compiledField
+	aliasIndices  []int
+	fieldIndices  []int
 }
 
 // compiledField caches whether an ExprField is a simple direct lookup
@@ -178,6 +180,14 @@ func (pp *ProjectOp) getVE(tuple xsql.RawRow, agg xsql.AggregateData, wr *xsql.W
 			}
 			pp.compiledExprs = append(pp.compiledExprs, cf)
 		}
+		pp.aliasIndices = make([]int, len(pp.AliasFields))
+		for i, f := range pp.AliasFields {
+			pp.aliasIndices[i] = getExprIndex(f.Expr)
+		}
+		pp.fieldIndices = make([]int, len(pp.Fields))
+		for i, f := range pp.Fields {
+			pp.fieldIndices[i] = getExprIndex(f.Expr)
+		}
 	}
 
 	pp.wv.Data = tuple
@@ -217,33 +227,32 @@ func (pp *ProjectOp) getRowVE(tuple xsql.Row, wr *xsql.WindowRange, fv *xsql.Fun
 func (pp *ProjectOp) project(row xsql.RawRow, ve *xsql.ValuerEval) error {
 	switch rt := row.(type) {
 	case *xsql.SliceTuple:
-		for _, f := range pp.AliasFields {
+		for i, f := range pp.AliasFields {
 			vi := ve.Eval(f.Expr)
 			if e, ok := vi.(error); ok {
 				return fmt.Errorf("expr: %s meet error, err:%v", f.Expr.String(), e)
 			}
 			if pp.SendNil && vi == nil {
-				// set it to a typed nil to distinguish from nil
-				// so that the encoder can treat it differently from nil
 				vi = cast.TNil
 			}
-			fr := f.Expr.(*ast.FieldRef)
-			rt.SetByIndex(fr.Index, vi)
+			index := pp.aliasIndices[i]
+			if index >= 0 {
+				rt.SetByIndex(index, vi)
+			}
 		}
-		for _, f := range pp.Fields {
+		for i, f := range pp.Fields {
 			if f.AName == "" {
 				vi := ve.Eval(f.Expr)
 				if e, ok := vi.(error); ok {
 					return fmt.Errorf("expr: %s meet error, err:%v", f.Expr.String(), e)
 				}
-				// TODO deal with other types
 				if pp.SendNil && vi == nil {
-					// set it to a typed nil to distinguish from nil
-					// so that the encoder can treat it differently from nil
 					vi = cast.TNil
 				}
-				fr := f.Expr.(*ast.FieldRef)
-				rt.SetByIndex(fr.Index, vi)
+				index := pp.fieldIndices[i]
+				if index >= 0 {
+					rt.SetByIndex(index, vi)
+				}
 			}
 		}
 		rt.Compact(pp.FieldLen)
@@ -305,3 +314,18 @@ func (pp *ProjectOp) project(row xsql.RawRow, ve *xsql.ValuerEval) error {
 	}
 	return nil
 }
+
+func getExprIndex(expr ast.Expr) int {
+	if fr, ok := expr.(*ast.FieldRef); ok {
+		return fr.Index
+	}
+	var targetIdx int = -1
+	ast.WalkFunc(expr, func(n ast.Node) bool {
+		if fr, ok := n.(*ast.FieldRef); ok {
+			targetIdx = fr.Index
+			return false // stop walking
+		}
+		return true
+	})
+	return targetIdx
+}

internal/topo/operator/project_test.go

@@ -3358,3 +3358,36 @@ func TestSliceNilField(t *testing.T) {
 		})
 	}
 }
+
+
+func TestProjectOp_SliceTuple_ComplexExpression(t *testing.T) {
+	sql := "SELECT abs(a) FROM test"
+	stmt, err := xsql.NewParser(strings.NewReader(sql)).Parse()
+	require.NoError(t, err)
+
+	pp := &ProjectOp{}
+	parseStmtWithSlice(pp, stmt.Fields, true)
+
+	// In current parseStmtWithSlice, inner ast.FieldRef elements of ast.Call 
+	// don't get HasIndex=true automatically. Let's fix that for this test.
+	ast.WalkFunc(stmt.Fields[0].Expr, func(n ast.Node) bool {
+		if fr, ok := n.(*ast.FieldRef); ok {
+			fr.HasIndex = true
+			fr.SourceIndex = constSourceIndex[fr.Name]
+		}
+		return true
+	})
+
+	fv, afv := xsql.NewFunctionValuersForOp(nil)
+	
+	sliceTuple := &xsql.SliceTuple{
+		SourceContent: make([]interface{}, 5),
+	}
+	sliceTuple.SourceContent[constSourceIndex["a"]] = int64(-42)
+	
+	ve := pp.getRowVE(sliceTuple, nil, fv, afv)
+	err = pp.project(sliceTuple, ve)
+	require.NoError(t, err)
+
+	require.Equal(t, int64(42), sliceTuple.SourceContent[0])
+}

internal/topo/planner/planner.go

@@ -108,6 +108,12 @@ func updateFieldIndex(ctx api.StreamContext, stmt *ast.SelectStatement, af []*as
 			case *ast.FieldRef:
 				if nf.IsColumn() {
 					sc := schema.GetStreamSchemaIndex(string(nf.StreamName))
+					if sc == nil {
+						streamsFromStmt := xsql.GetStreams(stmt)
+						if len(streamsFromStmt) == 1 {
+							sc = schema.GetStreamSchemaIndex(streamsFromStmt[0])
+						}
+					}
 					if sc != nil {
 						if si, ok := sc[nf.Name]; ok {
 							nf.SourceIndex = si
@@ -148,10 +154,10 @@ func updateFieldIndex(ctx api.StreamContext, stmt *ast.SelectStatement, af []*as
 	}
 	index := len(stmt.Fields)
 	for _, fieldExpr := range fieldExprs {
-		index = doUpdateIndex(ctx, fieldExpr, index, aliasIndex)
+		index = doUpdateIndex(ctx, stmt, fieldExpr, index, aliasIndex)
 	}
 	// Add sink index for other non-select fields
-	index = doUpdateIndex(ctx, stmt, index, aliasIndex)
+	index = doUpdateIndex(ctx, stmt, stmt, index, aliasIndex)
 	ctx.GetLogger().Infof("assign %d field index", index)
 	// Set temp index for analytic funcs
 	index = 0
@@ -166,13 +172,19 @@ func updateFieldIndex(ctx api.StreamContext, stmt *ast.SelectStatement, af []*as
 	ctx.GetLogger().Infof("assign %d temp index", index)
 }
 
-func doUpdateIndex(ctx api.StreamContext, root ast.Node, index int, aliasIndex map[string]int) int {
+func doUpdateIndex(ctx api.StreamContext, stmt *ast.SelectStatement, root ast.Node, index int, aliasIndex map[string]int) int {
 	ast.WalkFunc(root, func(n ast.Node) bool {
 		switch nf := n.(type) {
 		case *ast.FieldRef:
 			nf.HasIndex = true
 			if nf.IsColumn() {
 				sc := schema.GetStreamSchemaIndex(string(nf.StreamName))
+				if sc == nil {
+					streamsFromStmt := xsql.GetStreams(stmt)
+					if len(streamsFromStmt) == 1 {
+						sc = schema.GetStreamSchemaIndex(streamsFromStmt[0])
+					}
+				}
 				if sc != nil {
 					if si, ok := sc[nf.Name]; ok {
 						nf.SourceIndex = si