refactor(func): sharing array func (#2124)

Let the array func and aggregate func share the implementation if possible.

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

docs/en_US/sqls/functions/array_functions.md

@@ -90,8 +90,7 @@ Returns a union of the two arrays, with all duplicates removed.
 array_max(array)
 ```
 
-Returns an element which is greater than or equal to all other elements of the array. If an element of the array is
-null, it returns null.
+Returns an element which is greater than or equal to all other elements of the array. The null element will be ignored.
 
 ## ARRAY_MIN
 
@@ -99,8 +98,7 @@ null, it returns null.
 array_min(array)
 ```
 
-Returns an element which is less than or equal to all other elements of the array. If an element of the array is null,
-it returns null.
+Returns an element which is less than or equal to all other elements of the array. The null element will be ignored.
 
 ## ARRAY_EXCEPT
 

docs/zh_CN/sqls/functions/array_functions.md

@@ -88,7 +88,7 @@ array_union(array1, array2)
 array_max(array)
 ```
 
-返回数组中的最大值, 若数组元素中存在 null，则返回 null。
+返回数组中的最大值, 数组元素中的 null 值将被忽略。
 
 ## ARRAY_MIN
 
@@ -96,7 +96,7 @@ array_max(array)
 array_min(array)
 ```
 
-返回数组中的最小值, 若数组元素中存在 null，则返回 null。
+返回数组中的最小值, 数组元素中的 null 值将被忽略。
 
 ## ARRAY_EXCEPT
 

internal/binder/function/common_array_funcs.go

@@ -0,0 +1,272 @@
+// Copyright 2023 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.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package function
+
+import (
+	"fmt"
+
+	"github.com/lf-edge/ekuiper/pkg/cast"
+)
+
+// The functions here are used to implement the array functions to be referred in
+// 1. Aggregate function
+// 2. Array function
+
+func max(arr []interface{}) (interface{}, bool) {
+	if len(arr) > 0 {
+		v := getFirstValidArg(arr)
+		switch t := v.(type) {
+		case int:
+			if r, err := sliceIntMax(arr, int64(t)); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case int64:
+			if r, err := sliceIntMax(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case float64:
+			if r, err := sliceFloatMax(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case string:
+			if r, err := sliceStringMax(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case nil:
+			return nil, true
+		default:
+			return fmt.Errorf("found invalid arg %[1]T(%[1]v)", v), false
+		}
+	}
+	return nil, true
+}
+
+func min(arr []interface{}) (interface{}, bool) {
+	if len(arr) > 0 {
+		v := getFirstValidArg(arr)
+		switch t := v.(type) {
+		case int:
+			if r, err := sliceIntMin(arr, int64(t)); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case int64:
+			if r, err := sliceIntMin(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case float64:
+			if r, err := sliceFloatMin(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case string:
+			if r, err := sliceStringMin(arr, t); err != nil {
+				return err, false
+			} else {
+				return r, true
+			}
+		case nil:
+			return nil, true
+		default:
+			return fmt.Errorf("found invalid arg %[1]T(%[1]v)", v), false
+		}
+	}
+	return nil, true
+}
+
+func getCount(s []interface{}) int {
+	c := 0
+	for _, v := range s {
+		if v != nil {
+			c++
+		}
+	}
+	return c
+}
+
+func getFirstValidArg(s []interface{}) interface{} {
+	for _, v := range s {
+		if v != nil {
+			return v
+		}
+	}
+	return nil
+}
+
+func sliceIntTotal(s []interface{}) (int64, error) {
+	var total int64
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
+			total += vi
+		} else if v != nil {
+			return 0, fmt.Errorf("requires int but found %[1]T(%[1]v)", v)
+		}
+	}
+	return total, nil
+}
+
+func sliceFloatTotal(s []interface{}) (float64, error) {
+	var total float64
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		if vf, ok := v.(float64); ok {
+			total += vf
+		} else if v != nil {
+			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
+		}
+	}
+	return total, nil
+}
+
+func sliceIntMax(s []interface{}, max int64) (int64, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
+			if vi > max {
+				max = vi
+			}
+		} else if v != nil {
+			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
+		}
+	}
+	return max, nil
+}
+
+func sliceFloatMax(s []interface{}, max float64) (float64, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		if vf, ok := v.(float64); ok {
+			if max < vf {
+				max = vf
+			}
+		} else if v != nil {
+			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
+		}
+	}
+	return max, nil
+}
+
+func sliceStringMax(s []interface{}, max string) (string, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		if vs, ok := v.(string); ok {
+			if max < vs {
+				max = vs
+			}
+		} else if v != nil {
+			return "", fmt.Errorf("requires string but found %[1]T(%[1]v)", v)
+		}
+	}
+	return max, nil
+}
+
+func sliceIntMin(s []interface{}, min int64) (int64, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
+		if err == nil {
+			if vi < min {
+				min = vi
+			}
+		} else if v != nil {
+			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
+		}
+	}
+	return min, nil
+}
+
+func sliceFloatMin(s []interface{}, min float64) (float64, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		if vf, ok := v.(float64); ok {
+			if min > vf {
+				min = vf
+			}
+		} else if v != nil {
+			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
+		}
+	}
+	return min, nil
+}
+
+func sliceStringMin(s []interface{}, min string) (string, error) {
+	for _, v := range s {
+		if v == nil {
+			continue
+		}
+		if vs, ok := v.(string); ok {
+			if vs < min {
+				min = vs
+			}
+		} else if v != nil {
+			return "", fmt.Errorf("requires string but found %[1]T(%[1]v)", v)
+		}
+	}
+	return min, nil
+}
+
+func dedup(r []interface{}, col []interface{}, all bool) (interface{}, error) {
+	keyset := make(map[string]bool)
+	result := make([]interface{}, 0)
+	for i, m := range col {
+		key := fmt.Sprintf("%v", m)
+		if _, ok := keyset[key]; !ok {
+			if all {
+				result = append(result, r[i])
+			} else if i == len(col)-1 {
+				result = append(result, r[i])
+			}
+			keyset[key] = true
+		}
+	}
+	if !all {
+		if len(result) == 0 {
+			return nil, nil
+		} else {
+			return result[0], nil
+		}
+	} else {
+		return result, nil
+	}
+}

internal/binder/function/funcs_agg.go

@@ -69,40 +69,7 @@ func registerAggFunc() {
 		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
-			if len(arg0) > 0 {
-				v := getFirstValidArg(arg0)
-				switch t := v.(type) {
-				case int:
-					if r, err := sliceIntMax(arg0, int64(t)); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case int64:
-					if r, err := sliceIntMax(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case float64:
-					if r, err := sliceFloatMax(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case string:
-					if r, err := sliceStringMax(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case nil:
-					return nil, true
-				default:
-					return fmt.Errorf("run max function error: found invalid arg %[1]T(%[1]v)", v), false
-				}
-			}
-			return nil, true
+			return max(arg0)
 		},
 		val:   ValidateOneNumberArg,
 		check: returnNilIfHasAnyNil,
@@ -111,40 +78,7 @@ func registerAggFunc() {
 		fType: ast.FuncTypeAgg,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
 			arg0 := args[0].([]interface{})
-			if len(arg0) > 0 {
-				v := getFirstValidArg(arg0)
-				switch t := v.(type) {
-				case int:
-					if r, err := sliceIntMin(arg0, int64(t)); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case int64:
-					if r, err := sliceIntMin(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case float64:
-					if r, err := sliceFloatMin(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case string:
-					if r, err := sliceStringMin(arg0, t); err != nil {
-						return err, false
-					} else {
-						return r, true
-					}
-				case nil:
-					return nil, true
-				default:
-					return fmt.Errorf("run min function error: found invalid arg %[1]T(%[1]v)", v), false
-				}
-			}
-			return nil, true
+			return min(arg0)
 		},
 		val:   ValidateOneNumberArg,
 		check: returnNilIfHasAnyNil,
@@ -443,176 +377,3 @@ func registerAggFunc() {
 		check: returnNilIfHasAnyNil,
 	}
 }
-
-func getCount(s []interface{}) int {
-	c := 0
-	for _, v := range s {
-		if v != nil {
-			c++
-		}
-	}
-	return c
-}
-
-func getFirstValidArg(s []interface{}) interface{} {
-	for _, v := range s {
-		if v != nil {
-			return v
-		}
-	}
-	return nil
-}
-
-func sliceIntTotal(s []interface{}) (int64, error) {
-	var total int64
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
-		if err == nil {
-			total += vi
-		} else if v != nil {
-			return 0, fmt.Errorf("requires int but found %[1]T(%[1]v)", v)
-		}
-	}
-	return total, nil
-}
-
-func sliceFloatTotal(s []interface{}) (float64, error) {
-	var total float64
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		if vf, ok := v.(float64); ok {
-			total += vf
-		} else if v != nil {
-			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
-		}
-	}
-	return total, nil
-}
-
-func sliceIntMax(s []interface{}, max int64) (int64, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
-		if err == nil {
-			if vi > max {
-				max = vi
-			}
-		} else if v != nil {
-			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
-		}
-	}
-	return max, nil
-}
-
-func sliceFloatMax(s []interface{}, max float64) (float64, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		if vf, ok := v.(float64); ok {
-			if max < vf {
-				max = vf
-			}
-		} else if v != nil {
-			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
-		}
-	}
-	return max, nil
-}
-
-func sliceStringMax(s []interface{}, max string) (string, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		if vs, ok := v.(string); ok {
-			if max < vs {
-				max = vs
-			}
-		} else if v != nil {
-			return "", fmt.Errorf("requires string but found %[1]T(%[1]v)", v)
-		}
-	}
-	return max, nil
-}
-
-func sliceIntMin(s []interface{}, min int64) (int64, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		vi, err := cast.ToInt64(v, cast.CONVERT_SAMEKIND)
-		if err == nil {
-			if vi < min {
-				min = vi
-			}
-		} else if v != nil {
-			return 0, fmt.Errorf("requires int64 but found %[1]T(%[1]v)", v)
-		}
-	}
-	return min, nil
-}
-
-func sliceFloatMin(s []interface{}, min float64) (float64, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		if vf, ok := v.(float64); ok {
-			if min > vf {
-				min = vf
-			}
-		} else if v != nil {
-			return 0, fmt.Errorf("requires float64 but found %[1]T(%[1]v)", v)
-		}
-	}
-	return min, nil
-}
-
-func sliceStringMin(s []interface{}, min string) (string, error) {
-	for _, v := range s {
-		if v == nil {
-			continue
-		}
-		if vs, ok := v.(string); ok {
-			if vs < min {
-				min = vs
-			}
-		} else if v != nil {
-			return "", fmt.Errorf("requires string but found %[1]T(%[1]v)", v)
-		}
-	}
-	return min, nil
-}
-
-func dedup(r []interface{}, col []interface{}, all bool) (interface{}, error) {
-	keyset := make(map[string]bool)
-	result := make([]interface{}, 0)
-	for i, m := range col {
-		key := fmt.Sprintf("%v", m)
-		if _, ok := keyset[key]; !ok {
-			if all {
-				result = append(result, r[i])
-			} else if i == len(col)-1 {
-				result = append(result, r[i])
-			}
-			keyset[key] = true
-		}
-	}
-	if !all {
-		if len(result) == 0 {
-			return nil, nil
-		} else {
-			return result[0], nil
-		}
-	} else {
-		return result, nil
-	}
-}

internal/binder/function/funcs_array.go

@@ -16,7 +16,6 @@ package function
 
 import (
 	"fmt"
-	"math"
 	"math/rand"
 	"reflect"
 	"strings"
@@ -36,7 +35,6 @@ var (
 	errorArraySecondArgumentNotStringError = fmt.Errorf("second argument should be string")
 	errorArrayThirdArgumentNotIntError     = fmt.Errorf("third argument should be int")
 	errorArrayThirdArgumentNotStringError  = fmt.Errorf("third argument should be string")
-	errorArrayContainsNonNumOrBoolValError = fmt.Errorf("array contain elements that are not numeric or Boolean")
 	errorArrayNotArrayElementError         = fmt.Errorf("array elements should be array")
 	errorArrayNotStringElementError        = fmt.Errorf("array elements should be string")
 )
@@ -158,15 +156,38 @@ func registerArrayFunc() {
 		},
 		check: returnNilIfHasAnyNil,
 	}
+	builtins["array_position"] = builtinFunc{
+		fType: ast.FuncTypeScalar,
+		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			if args[0] == nil {
+				return -1, true
+			}
+			array, ok := args[0].([]interface{})
+			if !ok {
+				return errorArrayFirstArgumentNotArrayError, false
+			}
+			for i, item := range array {
+				if item == args[1] {
+					return i, true
+				}
+			}
+			return -1, true
+		},
+		val: func(ctx api.FunctionContext, args []ast.Expr) error {
+			return ValidateLen(2, len(args))
+		},
+	}
 
 	builtins["array_last_position"] = builtinFunc{
 		fType: ast.FuncTypeScalar,
 		exec: func(ctx api.FunctionContext, args []interface{}) (interface{}, bool) {
+			if args[0] == nil {
+				return -1, true
+			}
 			array, ok := args[0].([]interface{})
 			if !ok {
 				return errorArrayFirstArgumentNotArrayError, false
 			}
-
 			lastPos := -1
 			for i := len(array) - 1; i >= 0; i-- {
 				if array[i] == args[1] {
@@ -174,13 +195,11 @@ func registerArrayFunc() {
 					break
 				}
 			}
-
 			return lastPos, true
 		},
 		val: func(ctx api.FunctionContext, args []ast.Expr) error {
 			return ValidateLen(2, len(args))
 		},
-		check: returnFalseIfHasAnyNil,
 	}
 
 	builtins["array_contains_any"] = builtinFunc{
@@ -296,24 +315,7 @@ func registerArrayFunc() {
 			if !ok {
 				return errorArrayFirstArgumentNotArrayError, false
 			}
-			var res interface{}
-			var maxVal float64 = math.Inf(-1)
-
-			for _, val := range array {
-				if val == nil {
-					return nil, true
-				}
-				f, err := cast.ToFloat64(val, cast.CONVERT_ALL)
-				if err != nil {
-					return errorArrayContainsNonNumOrBoolValError, false
-				}
-
-				if f > maxVal {
-					maxVal = f
-					res = val
-				}
-			}
-			return res, true
+			return max(array)
 		},
 		val: func(ctx api.FunctionContext, args []ast.Expr) error {
 			return ValidateLen(1, len(args))
@@ -327,25 +329,7 @@ func registerArrayFunc() {
 			if !ok {
 				return errorArrayFirstArgumentNotArrayError, false
 			}
-			var res interface{}
-			var min float64 = math.Inf(1)
-
-			for _, val := range array {
-				if val == nil {
-					return nil, true
-				}
-
-				f, err := cast.ToFloat64(val, cast.CONVERT_ALL)
-				if err != nil {
-					return errorArrayContainsNonNumOrBoolValError, false
-				}
-
-				if f < min {
-					min = f
-					res = val
-				}
-			}
-			return res, true
+			return min(array)
 		},
 		val: func(ctx api.FunctionContext, args []ast.Expr) error {
 			return ValidateLen(1, len(args))
@@ -465,7 +449,7 @@ func registerArrayFunc() {
 			if !ok {
 				return errorArrayFirstArgumentNotArrayError, false
 			}
-			return len(array), true
+			return getCount(array), true
 		},
 		val: func(ctx api.FunctionContext, args []ast.Expr) error {
 			return ValidateLen(1, len(args))

internal/binder/function/funcs_array_test.go

@@ -15,10 +15,12 @@
 package function
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/lf-edge/ekuiper/internal/conf"
@@ -284,28 +286,28 @@ func TestArrayCommonFunctions(t *testing.T) {
 			args: []interface{}{
 				[]interface{}{1},
 			},
-			result: 1,
+			result: int64(1),
 		},
 		{
 			name: "array_max",
 			args: []interface{}{
 				[]interface{}{1, nil, 3},
 			},
-			result: nil,
+			result: int64(3),
 		},
 		{
 			name: "array_max",
 			args: []interface{}{
 				[]interface{}{1, "4", 3},
 			},
-			result: "4",
+			result: errors.New("requires int64 but found string(4)"),
 		},
 		{
 			name: "array_max",
 			args: []interface{}{
 				[]interface{}{1, "a4a", 3},
 			},
-			result: errorArrayContainsNonNumOrBoolValError,
+			result: errors.New("requires int64 but found string(a4a)"),
 		},
 		{
 			name: "array_max",
@@ -319,21 +321,21 @@ func TestArrayCommonFunctions(t *testing.T) {
 			args: []interface{}{
 				[]interface{}{1, 3.2, 4.1, 2},
 			},
-			result: 4.1,
+			result: int64(4),
 		},
 		{
 			name: "array_min",
 			args: []interface{}{
 				[]interface{}{1, nil, 3},
 			},
-			result: nil,
+			result: int64(1),
 		},
 		{
 			name: "array_min",
 			args: []interface{}{
 				[]interface{}{1, "0", 3},
 			},
-			result: "0",
+			result: errors.New("requires int64 but found string(0)"),
 		},
 		{
 			name: "array_min",
@@ -347,14 +349,14 @@ func TestArrayCommonFunctions(t *testing.T) {
 			args: []interface{}{
 				[]interface{}{1, "a4a", 3},
 			},
-			result: errorArrayContainsNonNumOrBoolValError,
+			result: errors.New("requires int64 but found string(a4a)"),
 		},
 		{
 			name: "array_min",
 			args: []interface{}{
 				[]interface{}{1, 3.2, 4.1, 2},
 			},
-			result: 1,
+			result: int64(1),
 		},
 		{
 			name: "array_except",
@@ -708,18 +710,16 @@ func TestArrayCommonFunctions(t *testing.T) {
 				[]interface{}{1},
 				nil,
 			},
-			result: errorArrayNotArrayElementError,
+			result: nil,
 		},
 	}
-	for i, tt := range tests {
-		f, ok := builtins[tt.name]
-		if !ok {
-			t.Fatal(fmt.Sprintf("builtin %v not found", tt.name))
-		}
-		result, _ := f.exec(fctx, tt.args)
-		if !reflect.DeepEqual(result, tt.result) {
-			t.Errorf("%d result mismatch,\ngot:\t%v \nwant:\t%v", i, result, tt.result)
-		}
+
+	fe := funcExecutor{}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, _ := fe.ExecWithName(tt.args, fctx, tt.name)
+			assert.Equal(t, tt.result, result)
+		})
 	}
 }
 
@@ -793,11 +793,11 @@ func TestArrayFuncNil(t *testing.T) {
 			r, b = mathFunc.exec(fctx, []interface{}{nil, 1})
 			require.True(t, b, fmt.Sprintf("%v failed", mathFuncName))
 			require.Equal(t, r, []interface{}{1}, fmt.Sprintf("%v failed", mathFuncName))
-		case "array_position":
+		case "array_position", "array_last_position":
 			r, b := mathFunc.exec(fctx, []interface{}{nil})
 			require.True(t, b, fmt.Sprintf("%v failed", mathFuncName))
 			require.Equal(t, r, -1, fmt.Sprintf("%v failed", mathFuncName))
-		case "array_contains", "array_last_position", "array_contains_any":
+		case "array_contains", "array_contains_any":
 			r, b := mathFunc.check([]interface{}{nil})
 			require.True(t, b, fmt.Sprintf("%v failed", mathFuncName))
 			require.False(t, r.(bool), fmt.Sprintf("%v failed", mathFuncName))