Fix demo case (#380)

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java

@@ -55,6 +55,7 @@
 import com.antgroup.geaflow.dsl.udf.table.other.EdgeTargetId;
 import com.antgroup.geaflow.dsl.udf.table.other.EdgeTimestamp;
 import com.antgroup.geaflow.dsl.udf.table.other.If;
+import com.antgroup.geaflow.dsl.udf.table.other.IsDecimal;
 import com.antgroup.geaflow.dsl.udf.table.other.Label;
 import com.antgroup.geaflow.dsl.udf.table.other.VertexId;
 import com.antgroup.geaflow.dsl.udf.table.string.Ascii2String;
@@ -171,6 +172,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
             .add(GeaFlowFunction.of(EdgeSrcId.class))
             .add(GeaFlowFunction.of(EdgeTargetId.class))
             .add(GeaFlowFunction.of(EdgeTimestamp.class))
+            .add(GeaFlowFunction.of(IsDecimal.class))
             // UDGA
             .add(GeaFlowFunction.of(SingleSourceShortestPath.class))
             .add(GeaFlowFunction.of(PageRank.class))

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/udf/table/other/IsDecimal.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2023 AntGroup 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.
+ */
+
+package com.antgroup.geaflow.dsl.udf.table.other;
+
+import com.antgroup.geaflow.dsl.common.function.Description;
+import com.antgroup.geaflow.dsl.common.function.UDF;
+
+@Description(name = "is_decimal", description = "Returns true if only contains digits and is "
+    + "non-null, otherwise return false.")
+public class IsDecimal extends UDF {
+
+    public boolean eval(String s) {
+        if (s == null) {
+            return false;
+        }
+        if (isInteger(s) || isLong(s) || isDouble(s)) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    private boolean isInteger(String s) {
+        boolean flag = true;
+        try {
+            Integer.parseInt(s);
+        } catch (NumberFormatException e) {
+            flag = false;
+        }
+        return flag;
+    }
+
+    private boolean isLong(String s) {
+        boolean flag = true;
+        try {
+            Long.parseLong(s);
+        } catch (NumberFormatException e) {
+            flag = false;
+        }
+        return flag;
+    }
+
+    private boolean isDouble(String s) {
+        boolean flag = true;
+        try {
+            Double.parseDouble(s);
+        } catch (NumberFormatException e) {
+            flag = false;
+        }
+        return flag;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/engine/GeaFlowRuntimeTable.java

@@ -81,6 +81,8 @@
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class GeaFlowRuntimeTable implements RuntimeTable {
 
@@ -286,6 +288,8 @@ public Row map(Row row) {
 
     private static class TableFilterFunction implements FilterFunction<Row> {
 
+        private static final Logger LOGGER = LoggerFactory.getLogger(TableFilterFunction.class);
+
         private final WhereFunction whereFunction;
 
         public TableFilterFunction(WhereFunction whereFunction) {

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/com/antgroup/geaflow/dsl/runtime/query/GQLAlgorithmTest.java

@@ -83,6 +83,15 @@ public void testAlgorithm_006() throws Exception {
             .checkSinkResult();
     }
 
+    @Test
+    public void testAlgorithm_008() throws Exception {
+        QueryTester
+            .build()
+            .withQueryPath("/query/find_loop.sql")
+            .execute()
+            .checkSinkResult();
+    }
+
     @Test
     public void testAlgorithmKHop() throws Exception {
         QueryTester

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/data/loop.txt

@@ -0,0 +1,15 @@
+. 1,jim
+. 2,kate
+. 3,lily
+. 4,lucy
+. 5,brown
+. 6,jack
+. 7,jackson
+- 1,2,0.2
+- 2,3,0.3
+- 3,4,0.2
+- 4,1,0.1
+- 4,5,0.1
+- 5,1,0.2
+- 5,6,0.1
+- 6,7,0.1

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/find_loop.txt

@@ -0,0 +1,4 @@
+2,3,4,1,2
+4,1,2,3,4
+3,4,1,2,3
+1,2,3,4,1

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/find_loop.sql

@@ -0,0 +1,76 @@
+set geaflow.dsl.window.size = 1;
+set geaflow.dsl.ignore.exception = true;
+
+CREATE GRAPH IF NOT EXISTS dy_modern (
+  Vertex person (
+    id bigint ID,
+    name varchar
+  ),
+  Edge knows (
+    srcId bigint SOURCE ID,
+    targetId bigint DESTINATION ID,
+    weight double
+  )
+) WITH (
+  storeType='rocksdb',
+  shardCount = 1
+);
+
+CREATE TABLE IF NOT EXISTS tbl_source (
+  text varchar
+) WITH (
+  type='file',
+  `geaflow.dsl.column.separator` = '#',
+  geaflow.dsl.file.path = 'resource:///data/loop.txt'
+);
+
+CREATE TABLE IF NOT EXISTS tbl_result (
+  a_id bigint,
+  b_id bigint,
+  c_id bigint,
+  d_id bigint,
+  a1_id bigint
+) WITH (
+  type='file',
+  geaflow.dsl.file.path = '${target}'
+);
+
+USE GRAPH dy_modern;
+
+INSERT INTO dy_modern.person(id, name)
+SELECT
+cast(trim(split_ex(t1, ',', 0)) as bigint),
+split_ex(trim(t1), ',', 1)
+FROM (
+  Select trim(substr(text, 2)) as t1
+  FROM tbl_source
+  WHERE substr(text, 1, 1) = '.'
+  AND is_decimal(trim(split_ex(trim(substr(text, 2)), ',', 0)))
+);
+
+INSERT INTO dy_modern.knows
+SELECT
+ cast(split_ex(t1, ',', 0) as bigint),
+ cast(split_ex(t1, ',', 1) as bigint),
+ cast(split_ex(t1, ',', 2) as double)
+FROM (
+  Select trim(substr(text, 2)) as t1
+  FROM tbl_source
+  WHERE substr(text, 1, 1) = '-'
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 0))
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 1))
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 2))
+);
+
+INSERT INTO tbl_result
+SELECT DISTINCT
+  a_id,
+  b_id,
+  c_id,
+  d_id,
+  a1_id
+FROM (
+  MATCH (a:person) -[:knows]->(b:person) -[:knows]-> (c:person)
+   -[:knows]-> (d:person) -> (a:person)
+  RETURN a.id as a_id, b.id as b_id, c.id as c_id, d.id as d_id, a.id as a1_id
+);

geaflow/geaflow-examples/gql/loop_detection.sql

@@ -48,6 +48,7 @@ FROM (
   Select trim(substr(text, 2)) as t1
   FROM tbl_source
   WHERE substr(text, 1, 1) = '.'
+  AND is_decimal(trim(split_ex(trim(substr(text, 2)), ',', 0)))
 );
 
 INSERT INTO dy_modern.knows
@@ -59,6 +60,9 @@ FROM (
   Select trim(substr(text, 2)) as t1
   FROM tbl_source
   WHERE substr(text, 1, 1) = '-'
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 0))
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 1))
+  AND is_decimal(split_ex(trim(substr(text, 2)), ',', 2))
 );
 
 INSERT INTO tbl_result