[Fix][transform-v2]SQL transform support max/min function (apache#8625)

CosmosNi · web-flow · commit 9d201bb103d9 · 2025-02-10T21:52:10.000+08:00
diff --git a/docs/en/transform-v2/sql-functions.md b/docs/en/transform-v2/sql-functions.md
@@ -385,6 +385,26 @@ Example:
 
 ACOS(D)
 
+### ARRAY_MAX
+
+```ARRAY_MAX(ARRAY)```
+
+The MAX function returns the maximum value of the expression.
+
+Example:
+
+ARRAY_MAX(I)
+
+### ARRAY_MIN
+
+```ARRAY_MIN(ARRAY)```
+
+The MIN function returns the minimum value of the expression.
+
+Example:
+
+ARRAY_MIN(I)
+
 ### ASIN
 
 ```ASIN(numeric)```
diff --git a/docs/zh/transform-v2/sql-functions.md b/docs/zh/transform-v2/sql-functions.md
@@ -386,6 +386,27 @@ ABS(I)
 
 ACOS(D)
 
+### ARRAY_MAX
+
+```ARRAY_MAX(ARRAY)```
+
+MAX 函数返回表达式的最大值。
+
+示例:
+
+ARRAY_MAX(I)
+
+### ARRAY_MIN
+
+```ARRAY_MIN(ARRAY)```
+
+MIN 函数返回表达式的最小值。
+
+示例:
+
+ARRAY_MIN(I)
+
+
 ### ASIN
 
 ```ASIN(numeric)```
diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLIT.java b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/java/org/apache/seatunnel/e2e/transform/TestSQLIT.java
@@ -77,6 +77,10 @@ public void testSQLTransform(TestContainer container) throws IOException, Interr
 
         Container.ExecResult splitSql = container.executeJob("/sql_transform/func_split.conf");
         Assertions.assertEquals(0, splitSql.getExitCode());
+
+        Container.ExecResult maxMinSql =
+                container.executeJob("/sql_transform/func_array_max_min.conf");
+        Assertions.assertEquals(0, maxMinSql.getExitCode());
     }
 
     @TestTemplate
diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_array_max_min.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_array_max_min.conf
@@ -0,0 +1,113 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+
+env {
+   job.mode = "BATCH"
+   parallelism = 1
+ }
+
+source {
+  FakeSource {
+    plugin_output = "fake"
+    schema = {
+      fields {
+        c_string = string
+        c_num_array = "array<int>"
+        c_string_array = "array<string>"
+      }
+    }
+    rows = [
+      {
+        kind = INSERT
+        fields = ["c_string",[1,2,3], ["a","b","c"]]
+      }
+    ]
+  }
+}
+
+ transform {
+   Sql {
+     plugin_input = "fake"
+     plugin_output = "fake1"
+     query = """select c_string,
+                     ARRAY_MAX(c_num_array) as c_num_max_array,
+                     ARRAY_MIN(c_num_array) as c_num_min_array,
+                     ARRAY_MAX(c_string_array) as c_string_max_array,
+                     ARRAY_MIN(c_string_array) as c_string_min_array
+                     from fake1"""
+   }
+ }
+
+sink {
+  Assert {
+    plugin_input = "fake1"
+    rules =
+      {
+        row_rules = [
+          {
+            rule_type = MIN_ROW
+            rule_value = 1
+          },
+          {
+            rule_type = MAX_ROW
+            rule_value = 1
+          }
+        ],
+        field_rules = [
+          {
+            field_name = "c_string"
+            field_type = "string"
+            field_value = [
+              {equals_to = "c_string"}
+            ]
+          },
+         {
+           field_name = "c_num_max_array"
+           field_type = "int"
+           field_value = [
+             {equals_to = 3}
+           ]
+         },
+         {
+           field_name = "c_num_min_array"
+           field_type = "int"
+           field_value = [
+             {equals_to = 1}
+           ]
+         },
+         {
+           field_name = "c_string_max_array"
+           field_type = "string"
+           field_value = [
+             {equals_to = "c"}
+           ]
+         },
+         {
+           field_name = "c_string_min_array"
+           field_type = "string"
+           field_value = [
+             {equals_to = "a"}
+           ]
+         }
+        ]
+      }
+  }
+}
diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java
@@ -148,6 +148,8 @@ public class ZetaSQLFunction {
     public static final String SIGN = "SIGN";
     public static final String TRUNC = "TRUNC";
     public static final String TRUNCATE = "TRUNCATE";
+    public static final String ARRAY_MAX = "ARRAY_MAX";
+    public static final String ARRAY_MIN = "ARRAY_MIN";
 
     // -------------------------time and date functions----------------------------
     public static final String CURRENT_DATE = "CURRENT_DATE";
@@ -554,6 +556,10 @@ public Object executeFunctionExpr(String functionName, List<Object> args) {
                 return SystemFunction.nullif(args);
             case ARRAY:
                 return ArrayFunction.array(args);
+            case ARRAY_MAX:
+                return ArrayFunction.arrayMax(args);
+            case ARRAY_MIN:
+                return ArrayFunction.arrayMin(args);
             case UUID:
                 return randomUUID().toString();
             default:
diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java
@@ -450,6 +450,9 @@ private SeaTunnelDataType<?> getFunctionType(Function function) {
                 return BasicType.DOUBLE_TYPE;
             case ZetaSQLFunction.ARRAY:
                 return ArrayFunction.castArrayTypeMapping(function, inputRowType);
+            case ZetaSQLFunction.ARRAY_MAX:
+            case ZetaSQLFunction.ARRAY_MIN:
+                return ArrayFunction.getElementType(function, inputRowType);
             case ZetaSQLFunction.SPLIT:
                 return ArrayType.STRING_ARRAY_TYPE;
             case ZetaSQLFunction.NOW:
diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/ArrayFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/ArrayFunction.java
@@ -17,8 +17,11 @@
 package org.apache.seatunnel.transform.sql.zeta.functions;
 
 import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.exception.CommonErrorCode;
 import org.apache.seatunnel.common.utils.SeaTunnelException;
+import org.apache.seatunnel.transform.exception.TransformException;
 
 import net.sf.jsqlparser.expression.DoubleValue;
 import net.sf.jsqlparser.expression.Expression;
@@ -30,10 +33,60 @@
 import net.sf.jsqlparser.schema.Column;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
 
 public class ArrayFunction {
 
+    public static Object arrayMax(List<Object> args) {
+        if (args == null || args.isEmpty()) {
+            return null;
+        }
+        Object[] dataList = (Object[]) args.get(0);
+        if (dataList == null || dataList.length == 0) {
+            return null;
+        }
+        if (dataList[0] instanceof String) {
+            return Arrays.stream(dataList)
+                    .map(String.class::cast)
+                    .max(String::compareTo)
+                    .orElse(null);
+        } else if (dataList[0] instanceof Number) {
+            return Arrays.stream(dataList)
+                    .map(Number.class::cast)
+                    .max(Comparator.comparingDouble(Number::doubleValue))
+                    .orElse(null);
+        }
+        throw new TransformException(
+                CommonErrorCode.UNSUPPORTED_DATA_TYPE,
+                String.format("Unsupported function max() arguments: %s", args));
+    }
+
+    public static Object arrayMin(List<Object> args) {
+        if (args == null || args.isEmpty()) {
+            return null;
+        }
+        Object[] dataList = (Object[]) args.get(0);
+        if (dataList == null || dataList.length == 0) {
+            return null;
+        }
+        if (dataList[0] instanceof String) {
+            return Arrays.stream(dataList)
+                    .map(String.class::cast)
+                    .min(String::compareTo)
+                    .orElse(null);
+        } else if (dataList[0] instanceof Number) {
+            return Arrays.stream(dataList)
+                    .map(Number.class::cast)
+                    .min(Comparator.comparingDouble(Number::doubleValue))
+                    .orElse(null);
+        }
+        throw new TransformException(
+                CommonErrorCode.UNSUPPORTED_DATA_TYPE,
+                String.format("Unsupported function max() arguments: %s", args));
+    }
+
     public static Object[] array(List<Object> args) {
         if (args == null || args.isEmpty()) {
             return new Object[0];
@@ -150,6 +203,14 @@ private static Class<?> getDataClassType(List<Object> args) {
         return arrayType == null ? String.class : arrayType;
     }
 
+    public static SeaTunnelDataType<?> getElementType(
+            Function function, SeaTunnelRowType inputRowType) {
+        String columnName = function.getParameters().getExpressions().get(0).toString();
+        int columnIndex = inputRowType.indexOf(columnName);
+        ArrayType arrayType = (ArrayType) inputRowType.getFieldType(columnIndex);
+        return arrayType.getElementType();
+    }
+
     private static List<Class<?>> getFunctionArgs(
             Function function, SeaTunnelRowType inputRowType) {
         ExpressionList<Expression> expressionList =

Original file line number	Diff line number	Diff line change
`@@ -77,6 +77,10 @@ public void testSQLTransform(TestContainer container) throws IOException, Interr`
`77`	`77`
`78`	`78`	`Container.ExecResult splitSql = container.executeJob("/sql_transform/func_split.conf");`
`79`	`79`	`Assertions.assertEquals(0, splitSql.getExitCode());`
	`80`	`+`
	`81`	`+ Container.ExecResult maxMinSql =`
	`82`	`+ container.executeJob("/sql_transform/func_array_max_min.conf");`
	`83`	`+ Assertions.assertEquals(0, maxMinSql.getExitCode());`
`80`	`84`	`}`
`81`	`85`
`82`	`86`	`@TestTemplate`