perf(sql): optimize expression evaluation (#4075)

## Summary
- Reuse the project evaluator and its valuer chain across input rows.
- Normalize numeric operands with one type switch instead of repeated
type checks and cast dispatch.
- Compare common `changed_cols` value types directly before falling back
to `reflect.DeepEqual`.

## Why
- SQL expression evaluation runs for every input row, so small setup and
dispatch costs accumulate in projection, arithmetic, filtering, sorting,
and change detection.
- The project operator rebuilt the same evaluator scaffolding for every
row even though only the current tuple, window, and aggregate data
changed.
- Numeric normalization inspected each operand type multiple times,
while `changed_cols` used reflection even for primitive values.

## Changes In This PR
- Expose reusable internal multi-valuer implementations and reset their
per-row data from `ProjectOp` without changing expression indices or
SliceTuple output semantics.
- Replace `convertNum` helper dispatch with a single type switch that
preserves the existing integer and floating-point conversions.
- Add primitive fast paths for string, int64, float64, bool, and int
comparisons in `changed_cols`; complex values retain `reflect.DeepEqual`
behavior.
- Add focused regression tests and benchmarks for all three paths.
- Organize the changes as three independent commits for review.

## Notes
- Benchmark medians on linux/amd64, Intel i9-14900HX, seven runs:
- int64 `SimpleDataEval`: 10.82 ns/op to 8.08 ns/op (about 25% faster),
0 allocations before and after.
- SliceTuple projection: 158.6 ns/op to 54.9 ns/op (about 65% faster),
from 112 B / 4 allocations to 0 B / 0 allocations.
- stable `changed_cols`: 337.0 ns/op to 304.8 ns/op (about 10% faster),
with allocations unchanged.
- Passed `go test -race ./internal/xsql ./internal/topo/operator
-count=1` and focused race tests for the changed column functions.
- `git diff --check` passes.
- A full local binder/function run still hits the pre-existing
`TestIncAggFunctionErr` failure, reproduced unchanged on
`upstream/master`. `go vet` likewise reports the existing lock-copy
warning in untouched `internal/xsql/row.go:187`.

---------

Signed-off-by: Jiayin Ng <ngjaying@gmail.com>

Author: ngjaying

internal/binder/function/funcs_cols.go

@@ -1,4 +1,4 @@
-// Copyright 2022-2024 EMQ Technologies Co., Ltd.
+// Copyright 2022-2026 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -89,7 +89,7 @@ func changedFunc(ctx api.FunctionContext, args []interface{}, keys []string) (Re
 		if err != nil {
 			return nil, err
 		}
-		if !reflect.DeepEqual(v, lv) {
+		if !changedValueEqual(v, lv) {
 			if r == nil {
 				r = make(ResultCols)
 			}
@@ -102,3 +102,32 @@ func changedFunc(ctx api.FunctionContext, args []interface{}, keys []string) (Re
 	}
 	return r, nil
 }
+
+func changedValueEqual(v1, v2 interface{}) bool {
+	if v1 == nil || v2 == nil {
+		return v1 == v2
+	}
+	switch t1 := v1.(type) {
+	case string:
+		if t2, ok := v2.(string); ok {
+			return t1 == t2
+		}
+	case int64:
+		if t2, ok := v2.(int64); ok {
+			return t1 == t2
+		}
+	case float64:
+		if t2, ok := v2.(float64); ok {
+			return t1 == t2
+		}
+	case bool:
+		if t2, ok := v2.(bool); ok {
+			return t1 == t2
+		}
+	case int:
+		if t2, ok := v2.(int); ok {
+			return t1 == t2
+		}
+	}
+	return reflect.DeepEqual(v1, v2)
+}

internal/binder/function/funcs_cols_test.go

@@ -1,4 +1,4 @@
-// Copyright 2022 EMQ Technologies Co., Ltd.
+// Copyright 2022-2026 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -305,3 +305,53 @@ func TestExecIgnoreNull(t *testing.T) {
 		}
 	}
 }
+
+func TestChangedValueEqual(t *testing.T) {
+	tests := []struct {
+		name string
+		v1   any
+		v2   any
+		want bool
+	}{
+		{name: "nil", want: true},
+		{name: "one nil", v1: int64(1), want: false},
+		{name: "string", v1: "value", v2: "value", want: true},
+		{name: "different string", v1: "value", v2: "other", want: false},
+		{name: "int64", v1: int64(1), v2: int64(1), want: true},
+		{name: "different numeric types", v1: int64(1), v2: int(1), want: false},
+		{name: "float64", v1: float64(1.5), v2: float64(1.5), want: true},
+		{name: "bool", v1: true, v2: true, want: true},
+		{name: "int", v1: int(1), v2: int(1), want: true},
+		{name: "slice fallback", v1: []int{1, 2}, v2: []int{1, 2}, want: true},
+		{name: "map fallback", v1: map[string]any{"a": 1}, v2: map[string]any{"a": 2}, want: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := changedValueEqual(tt.v1, tt.v2); got != tt.want {
+				t.Errorf("changedValueEqual(%v, %v) = %v, want %v", tt.v1, tt.v2, got, tt.want)
+			}
+		})
+	}
+}
+
+func BenchmarkChangedColsStable(b *testing.B) {
+	contextLogger := conf.Log.WithField("rule", "BenchmarkChangedColsStable")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, err := state.CreateStore("BenchmarkChangedColsStable", def.AtMostOnce)
+	if err != nil {
+		b.Fatal(err)
+	}
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("BenchmarkChangedColsStable", "changed_cols", tempStore), 1)
+	args := []any{"p_", true, "same", int64(42), true}
+	keys := []string{"prefix", "ignore", "string", "integer", "boolean"}
+	if _, err := changedFunc(fctx, args, keys); err != nil {
+		b.Fatal(err)
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if _, err := changedFunc(fctx, args, keys); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

internal/topo/operator/project_operator.go

@@ -50,6 +50,12 @@ type ProjectOp struct {
 
 	kvs   []interface{}
 	alias []interface{}
+
+	ve  *xsql.ValuerEval
+	wv  *xsql.WildcardValuer
+	wrv *xsql.WindowRangeValuer
+	mvs xsql.MultiValuerList
+	mav *xsql.AggregateMultiValuer
 }
 
 // Apply
@@ -121,14 +127,41 @@ func (pp *ProjectOp) Apply(ctx api.StreamContext, data interface{}, fv *xsql.Fun
 
 func (pp *ProjectOp) getVE(tuple xsql.RawRow, agg xsql.AggregateData, wr *xsql.WindowRange, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) *xsql.ValuerEval {
 	afv.SetData(agg)
+	if pp.ve == nil {
+		pp.ve = &xsql.ValuerEval{}
+		pp.wv = &xsql.WildcardValuer{}
+		pp.wrv = &xsql.WindowRangeValuer{}
+		pp.mvs = make(xsql.MultiValuerList, 4)
+		if pp.IsAggregate {
+			pp.mav = &xsql.AggregateMultiValuer{MultiValuerList: pp.mvs}
+			pp.ve.Valuer = pp.mav
+		} else {
+			pp.ve.Valuer = &pp.mvs
+		}
+	}
+	pp.wv.Data = tuple
 	if pp.IsAggregate {
-		return &xsql.ValuerEval{Valuer: xsql.MultiAggregateValuer(agg, fv, tuple, fv, afv, &xsql.WildcardValuer{Data: tuple})}
+		pp.mvs[0] = tuple
+		pp.mvs[1] = fv
+		pp.mvs[2] = afv
+		pp.mvs[3] = pp.wv
+		pp.mav.Reset(agg, fv)
 	} else {
 		if wr != nil {
-			return &xsql.ValuerEval{Valuer: xsql.MultiValuer(tuple, &xsql.WindowRangeValuer{WindowRange: wr}, fv, &xsql.WildcardValuer{Data: tuple})}
+			pp.mvs = pp.mvs[:4]
+			pp.wrv.WindowRange = wr
+			pp.mvs[0] = tuple
+			pp.mvs[1] = pp.wrv
+			pp.mvs[2] = fv
+			pp.mvs[3] = pp.wv
+		} else {
+			pp.mvs = pp.mvs[:3]
+			pp.mvs[0] = tuple
+			pp.mvs[1] = fv
+			pp.mvs[2] = pp.wv
 		}
-		return &xsql.ValuerEval{Valuer: xsql.MultiValuer(tuple, fv, &xsql.WildcardValuer{Data: tuple})}
 	}
+	return pp.ve
 }
 
 func (pp *ProjectOp) getRowVE(tuple xsql.Row, wr *xsql.WindowRange, fv *xsql.FunctionValuer, afv *xsql.AggregateFunctionValuer) *xsql.ValuerEval {

internal/topo/operator/project_test.go

@@ -3630,3 +3630,48 @@ func TestProjectClearsCachedAliasesAfterError(t *testing.T) {
 	require.Same(t, succeeded, result)
 	require.Equal(t, map[string]any{"calculated": int64(4)}, succeeded.ToMap())
 }
+
+func TestProjectReusesValuer(t *testing.T) {
+	pp := &ProjectOp{}
+	fv, afv := xsql.NewFunctionValuersForOp(nil)
+	first := &xsql.Tuple{Emitter: "test", Message: xsql.Message{"a": int64(1)}}
+	second := &xsql.Tuple{Emitter: "test", Message: xsql.Message{"a": int64(2)}}
+	field := &ast.FieldRef{Name: "a"}
+
+	ve := pp.getRowVE(first, nil, fv, afv)
+	require.Equal(t, int64(1), ve.Eval(field))
+	reused := pp.getRowVE(second, nil, fv, afv)
+	require.Same(t, ve, reused)
+	require.Equal(t, int64(2), reused.Eval(field))
+
+	windowed := pp.getVE(second, nil, xsql.NewWindowRange(10, 20, 30), fv, afv)
+	require.Same(t, ve, windowed)
+	require.Equal(t, int64(10), windowed.Eval(&ast.Call{Name: "window_start"}))
+
+	withoutWindow := pp.getRowVE(first, nil, fv, afv)
+	require.Same(t, ve, withoutWindow)
+	require.Equal(t, int64(1), withoutWindow.Eval(field))
+}
+
+func BenchmarkSliceProjectEvaluation(b *testing.B) {
+	stmt, err := xsql.NewParser(strings.NewReader("SELECT a, b, c FROM test")).Parse()
+	if err != nil {
+		b.Fatal(err)
+	}
+	pp := &ProjectOp{}
+	parseStmtWithSlice(pp, stmt.Fields, true)
+	fv, afv := xsql.NewFunctionValuersForOp(nil)
+	source := model.SliceVal{int64(10), int64(20), int64(30)}
+	sink := make(model.SliceVal, 0, pp.FieldLen)
+	tuple := &xsql.SliceTuple{}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		tuple.SourceContent = source
+		tuple.SinkContent = sink[:0]
+		ve := pp.getRowVE(tuple, nil, fv, afv)
+		if err := pp.project(tuple, ve); err != nil {
+			b.Fatal(err)
+		}
+	}
+}