Add ARRAY_REVERSE, ARRAY_MAX and ARRAY_MIN SQL functions

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java

@@ -58,6 +58,9 @@
 import org.apache.geaflow.dsl.udf.table.array.ArrayAppend;
 import org.apache.geaflow.dsl.udf.table.array.ArrayContains;
 import org.apache.geaflow.dsl.udf.table.array.ArrayDistinct;
+import org.apache.geaflow.dsl.udf.table.array.ArrayMax;
+import org.apache.geaflow.dsl.udf.table.array.ArrayMin;
+import org.apache.geaflow.dsl.udf.table.array.ArrayReverse;
 import org.apache.geaflow.dsl.udf.table.array.ArrayUnion;
 import org.apache.geaflow.dsl.udf.table.date.AddMonths;
 import org.apache.geaflow.dsl.udf.table.date.DateAdd;
@@ -166,6 +169,9 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
             .add(GeaFlowFunction.of(ArrayAppend.class))
             .add(GeaFlowFunction.of(ArrayContains.class))
             .add(GeaFlowFunction.of(ArrayDistinct.class))
+            .add(GeaFlowFunction.of(ArrayMax.class))
+            .add(GeaFlowFunction.of(ArrayMin.class))
+            .add(GeaFlowFunction.of(ArrayReverse.class))
             .add(GeaFlowFunction.of(ArrayUnion.class))
 
             // udf.table.math

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMax.java

@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+package org.apache.geaflow.dsl.udf.table.array;
+
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDF;
+
+@Description(name = "array_max", description = "Return the maximum element of input array.")
+public class ArrayMax extends UDF {
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public Object eval(Object[] input) {
+        if (input == null || input.length == 0) {
+            return null;
+        }
+        Comparable max = null;
+        for (Object element : input) {
+            if (element == null) {
+                continue;
+            }
+            if (!(element instanceof Comparable)) {
+                throw new IllegalArgumentException(
+                    "array_max only supports comparable element types");
+            }
+            Comparable comparable = (Comparable) element;
+            if (max == null || comparable.compareTo(max) > 0) {
+                max = comparable;
+            }
+        }
+        return max;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMin.java

@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+
+package org.apache.geaflow.dsl.udf.table.array;
+
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDF;
+
+@Description(name = "array_min", description = "Return the minimum element of input array.")
+public class ArrayMin extends UDF {
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public Object eval(Object[] input) {
+        if (input == null || input.length == 0) {
+            return null;
+        }
+        Comparable min = null;
+        for (Object element : input) {
+            if (element == null) {
+                continue;
+            }
+            if (!(element instanceof Comparable)) {
+                throw new IllegalArgumentException(
+                    "array_min only supports comparable element types");
+            }
+            Comparable comparable = (Comparable) element;
+            if (min == null || comparable.compareTo(min) < 0) {
+                min = comparable;
+            }
+        }
+        return min;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayReverse.java

@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+package org.apache.geaflow.dsl.udf.table.array;
+
+import org.apache.geaflow.dsl.common.function.Description;
+import org.apache.geaflow.dsl.common.function.UDF;
+
+@Description(name = "array_reverse", description = "Reverse input array.")
+public class ArrayReverse extends UDF {
+
+    public Object[] eval(Object[] input) {
+        if (input == null) {
+            return null;
+        }
+        Object[] result = new Object[input.length];
+        for (int i = 0; i < input.length; i++) {
+            result[i] = input[input.length - 1 - i];
+        }
+        return result;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/array/ArrayTest.java

@@ -22,6 +22,9 @@
 import org.apache.geaflow.dsl.udf.table.array.ArrayAppend;
 import org.apache.geaflow.dsl.udf.table.array.ArrayContains;
 import org.apache.geaflow.dsl.udf.table.array.ArrayDistinct;
+import org.apache.geaflow.dsl.udf.table.array.ArrayMax;
+import org.apache.geaflow.dsl.udf.table.array.ArrayMin;
+import org.apache.geaflow.dsl.udf.table.array.ArrayReverse;
 import org.apache.geaflow.dsl.udf.table.array.ArrayUnion;
 import org.testng.Assert;
 import org.testng.annotations.Test;
@@ -60,4 +63,47 @@ public void testArrayUnion() throws Exception {
         Object[] res = udf.eval(input1, input2);
         Assert.assertEquals(res.length, 8);
     }
+
+    @Test
+    public void testArrayReverse() throws Exception {
+        ArrayReverse udf = new ArrayReverse();
+        Assert.assertEquals(udf.eval(new Object[]{1, 2, 4, -1}), new Object[]{-1, 4, 2, 1});
+        Assert.assertEquals(udf.eval(new Object[]{1, null, 3}), new Object[]{3, null, 1});
+        Assert.assertEquals(udf.eval(new Object[]{}), new Object[]{});
+        Assert.assertNull(udf.eval(null));
+    }
+
+    @Test
+    public void testArrayMax() throws Exception {
+        ArrayMax udf = new ArrayMax();
+        Assert.assertEquals(udf.eval(new Object[]{1, 2, 4, -1}), 4);
+        Assert.assertEquals(udf.eval(new Object[]{"a", "c", "b"}), "c");
+        Assert.assertEquals(udf.eval(new Object[]{1, null, 3}), 3);
+        Assert.assertNull(udf.eval(new Object[]{}));
+        Assert.assertNull(udf.eval(new Object[]{null, null}));
+        Assert.assertNull(udf.eval(null));
+    }
+
+    @Test(expectedExceptions = IllegalArgumentException.class)
+    public void testArrayMaxWithUnsupportedElementType() throws Exception {
+        ArrayMax udf = new ArrayMax();
+        udf.eval(new Object[]{new Object()});
+    }
+
+    @Test
+    public void testArrayMin() throws Exception {
+        ArrayMin udf = new ArrayMin();
+        Assert.assertEquals(udf.eval(new Object[]{1, 2, 4, -1}), -1);
+        Assert.assertEquals(udf.eval(new Object[]{"a", "c", "b"}), "a");
+        Assert.assertEquals(udf.eval(new Object[]{1, null, 3}), 1);
+        Assert.assertNull(udf.eval(new Object[]{}));
+        Assert.assertNull(udf.eval(new Object[]{null, null}));
+        Assert.assertNull(udf.eval(null));
+    }
+
+    @Test(expectedExceptions = IllegalArgumentException.class)
+    public void testArrayMinWithUnsupportedElementType() throws Exception {
+        ArrayMin udf = new ArrayMin();
+        udf.eval(new Object[]{new Object()});
+    }
 }