diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java index 47addc84a..35d306abc 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java @@ -39,6 +39,7 @@ import org.apache.geaflow.dsl.udf.graph.ClusterCoefficient; import org.apache.geaflow.dsl.udf.graph.CommonNeighbors; import org.apache.geaflow.dsl.udf.graph.ConnectedComponents; +import org.apache.geaflow.dsl.udf.graph.Degree; import org.apache.geaflow.dsl.udf.graph.IncKHopAlgorithm; import org.apache.geaflow.dsl.udf.graph.IncMinimumSpanningTree; import org.apache.geaflow.dsl.udf.graph.IncWeakConnectedComponents; @@ -58,6 +59,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 +170,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 @@ -228,6 +235,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable { .add(GeaFlowFunction.of(PageRank.class)) .add(GeaFlowFunction.of(KHop.class)) .add(GeaFlowFunction.of(KCore.class)) + .add(GeaFlowFunction.of(Degree.class)) .add(GeaFlowFunction.of(IncrementalKCore.class)) .add(GeaFlowFunction.of(IncMinimumSpanningTree.class)) .add(GeaFlowFunction.of(ClosenessCentrality.class)) diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java new file mode 100644 index 000000000..f863a7d3c --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java @@ -0,0 +1,69 @@ +/* + * 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.graph; + +import java.util.Iterator; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.LongType; +import org.apache.geaflow.dsl.common.algo.AlgorithmRuntimeContext; +import org.apache.geaflow.dsl.common.algo.AlgorithmUserFunction; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.RowVertex; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.function.Description; +import org.apache.geaflow.dsl.common.types.GraphSchema; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.model.graph.edge.EdgeDirection; + +@Description(name = "degree", description = "built-in udga for Degree") +public class Degree implements AlgorithmUserFunction { + + private AlgorithmRuntimeContext context; + + @Override + public void init(AlgorithmRuntimeContext context, Object[] params) { + this.context = context; + if (params.length > 0) { + throw new IllegalArgumentException("Only support no arguments, usage: degree()"); + } + } + + @Override + public void process(RowVertex vertex, Optional updatedValues, Iterator messages) { + long inDegree = context.loadEdges(EdgeDirection.IN).size(); + long outDegree = context.loadEdges(EdgeDirection.OUT).size(); + context.take(ObjectRow.create(vertex.getId(), inDegree, outDegree)); + } + + @Override + public void finish(RowVertex graphVertex, Optional updatedValues) { + + } + + @Override + public StructType getOutputType(GraphSchema graphSchema) { + return new StructType( + new TableField("id", graphSchema.getIdType(), false), + new TableField("in_degree", LongType.INSTANCE, false), + new TableField("out_degree", LongType.INSTANCE, false) + ); + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMax.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMax.java new file mode 100644 index 000000000..9a142dbf9 --- /dev/null +++ b/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; + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMin.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayMin.java new file mode 100644 index 000000000..f99f7b5f5 --- /dev/null +++ b/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; + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayReverse.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/array/ArrayReverse.java new file mode 100644 index 000000000..0933c94c3 --- /dev/null +++ b/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; + } +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/array/ArrayTest.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/array/ArrayTest.java index 691eb7a87..503b84d06 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/array/ArrayTest.java +++ b/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()}); + } } diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java index 8c75d6e78..69d2d6f0c 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java @@ -115,6 +115,16 @@ public void testAlgorithmKCore() throws Exception { .checkSinkResult(); } + @Test + public void testAlgorithmDegree() throws Exception { + QueryTester + .build() + .withGraphDefine("/query/modern_graph.sql") + .withQueryPath("/query/gql_algorithm_degree.sql") + .execute() + .checkSinkResult(); + } + @Test public void testAlgorithmClosenessCentrality() throws Exception { QueryTester diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt new file mode 100644 index 000000000..20251d872 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt @@ -0,0 +1,6 @@ +1,0,3 +2,1,0 +3,3,0 +4,1,2 +5,1,0 +6,0,1 diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql new file mode 100644 index 000000000..4eb89e52e --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql @@ -0,0 +1,34 @@ +/* + * 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. + */ + +CREATE TABLE result_tb ( + id bigint, + in_degree bigint, + out_degree bigint +) WITH ( + type='file', + geaflow.dsl.file.path='${target}' +); + +USE GRAPH modern; + +INSERT INTO result_tb +CALL degree() YIELD (id, in_degree, out_degree) +RETURN cast(id as bigint), in_degree, out_degree +;