feat(func): support median function (#3755)

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

Author: Yisaer

docs/en_US/sqls/functions/aggregate_functions.md

@@ -177,6 +177,14 @@ Examples:
      SELECT deduplicate(a, false)->a as r1 FROM demo GROUP BY SlidingWindow(hh, 1)
      ```
 
+## MEDIAN
+
+```text
+median(col)
+```
+
+Returns the median value of expression in the group.
+
 ## STDDEV
 
 ```text

docs/zh_CN/sqls/functions/aggregate_functions.md

@@ -148,6 +148,14 @@ deduplicate(col, false)
      SELECT deduplicate(a, false)->a as r1 FROM demo GROUP BY SlidingWindow(hh, 1)
      ```
 
+## MEDIAN
+
+```text
+median(col)
+```
+
+返回组中所有值的中位数。
+
 ## STDDEV
 
 ```text

internal/binder/function/funcs_agg.go

@@ -16,6 +16,7 @@ package function
 
 import (
 	"fmt"
+	"sort"
 
 	"github.com/lf-edge/ekuiper/contract/v2/api"
 	"github.com/montanaflynn/stats"
@@ -25,6 +26,33 @@ import (
 )
 
 func registerAggFunc() {
+	builtins["median"] = builtinFunc{
+		fType: ast.FuncTypeAgg,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			arg0 := args[0].([]interface{})
+			if len(arg0) < 1 {
+				return int64(0), true
+			}
+			switch arg0[0].(type) {
+			case float64:
+				f64s, err := cast.ToFloat64Slice(arg0, cast.CONVERT_SAMEKIND, cast.IGNORE_NIL)
+				if err != nil {
+					return err, false
+				}
+				return median(f64s), true
+			case int64, int:
+				i64s, err := cast.ToInt64Slice(arg0, cast.CONVERT_SAMEKIND)
+				if err != nil {
+					return err, false
+				}
+				return median(i64s), true
+			default:
+				return fmt.Errorf("%v should be number", arg0[0]), false
+			}
+		},
+		val:   ValidateOneNumberArg,
+		check: returnNilIfHasAnyNil,
+	}
 	builtins["avg"] = builtinFunc{
 		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
@@ -379,3 +407,22 @@ func registerAggFunc() {
 		check: returnNilIfHasAnyNil,
 	}
 }
+
+type Number interface {
+	int64 | float64
+}
+
+func median[T Number](nums []T) interface{} {
+	sort.Slice(nums, func(i, j int) bool {
+		return nums[i] < nums[j]
+	})
+	n := len(nums)
+	if n == 0 {
+		return 0
+	}
+	if n%2 == 1 {
+		return nums[n/2]
+	} else {
+		return float64((nums[n/2-1])+(nums[n/2])) / 2
+	}
+}

internal/binder/function/funcs_agg_test.go

@@ -667,3 +667,106 @@ func TestLastValueValidation(t *testing.T) {
 		}
 	}
 }
+
+func TestMedian(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    interface{}
+		expected interface{}
+	}{
+		{
+			name:     "single int64 element",
+			input:    []int64{5},
+			expected: int64(5),
+		},
+		{
+			name:     "odd length int64 slice",
+			input:    []int64{1, 3, 5, 7, 9},
+			expected: int64(5),
+		},
+		{
+			name:     "even length int64 slice",
+			input:    []int64{1, 3, 5, 7, 9, 11},
+			expected: float64(6),
+		},
+		{
+			name:     "unsorted int64 slice",
+			input:    []int64{9, 1, 5, 3, 7},
+			expected: int64(5),
+		},
+		{
+			name:     "empty float64 slice",
+			input:    []float64{},
+			expected: 0,
+		},
+		{
+			name:     "single float64 element",
+			input:    []float64{5.5},
+			expected: 5.5,
+		},
+		{
+			name:     "odd length float64 slice",
+			input:    []float64{1.1, 3.3, 5.5, 7.7, 9.9},
+			expected: 5.5,
+		},
+		{
+			name:     "even length float64 slice",
+			input:    []float64{1.1, 3.3, 5.5, 7.7, 9.9, 11.1},
+			expected: 6.6,
+		},
+		{
+			name:     "unsorted float64 slice",
+			input:    []float64{9.9, 1.1, 5.5, 3.3, 7.7},
+			expected: 5.5,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var result interface{}
+			switch v := tt.input.(type) {
+			case []int64:
+				result = median(v)
+			case []float64:
+				result = median(v)
+			default:
+				t.Fatalf("unsupported input type: %T", v)
+			}
+			require.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestMedianFunc(t *testing.T) {
+	fm, ok := builtins["median"]
+	require.True(t, ok)
+	contextLogger := conf.Log.WithField("rule", "testExec")
+	ctx := kctx.WithValue(kctx.Background(), kctx.LoggerKey, contextLogger)
+	tempStore, _ := state.CreateStore("mockRule0", def.AtMostOnce)
+	fctx := kctx.NewDefaultFuncContext(ctx.WithMeta("mockRule0", "test", tempStore), 2)
+	tests := []struct {
+		args   []interface{}
+		expect interface{}
+	}{
+		{
+			args:   []interface{}{},
+			expect: int64(0),
+		},
+		{
+			args: []interface{}{
+				int64(5), int64(1), int64(2), int64(3), int64(4),
+			},
+			expect: int64(3),
+		},
+		{
+			args: []interface{}{
+				float64(5), float64(1), float64(2), float64(3), float64(4),
+			},
+			expect: float64(3),
+		},
+	}
+	for _, tt := range tests {
+		got, ok := fm.exec(fctx, []interface{}{tt.args})
+		require.True(t, ok)
+		require.Equal(t, tt.expect, got)
+	}
+}