Fix negative from index handling in INSTR

geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/table/string/Instr.java

@@ -41,14 +41,32 @@ public Long eval(String str, String target, Long from, Long nth) {
         if (nth <= 0) {
             return null;
         }
-        int fromIndex = from.intValue() - 1;
-        if (fromIndex < 0) {
+        if (from > 0) {
+            int fromIndex = from.intValue() - 1;
+            for (int i = 0; i < nth; ++i) {
+                fromIndex = str.indexOf(target, fromIndex) + 1;
+            }
+            return (long) fromIndex;
+        }
+        if (from == 0) {
             return null;
         }
+
+        long fromIndex = str.length() + from;
+        if (fromIndex < 0) {
+            return 0L;
+        }
+        int index = (int) fromIndex;
         for (int i = 0; i < nth; ++i) {
-            fromIndex = str.indexOf(target, fromIndex) + 1;
+            index = str.lastIndexOf(target, index);
+            if (index < 0) {
+                return 0L;
+            }
+            if (i < nth - 1) {
+                index--;
+            }
         }
-        return (long) fromIndex;
+        return (long) index + 1;
     }
 
     public Long eval(BinaryString str, BinaryString target) {
@@ -66,13 +84,48 @@ public Long eval(BinaryString str, BinaryString target, Long from, Long nth) {
         if (nth <= 0) {
             return null;
         }
-        int fromIndex = from.intValue() - 1;
-        if (fromIndex < 0) {
+        if (from > 0) {
+            int fromIndex = from.intValue() - 1;
+            for (int i = 0; i < nth; ++i) {
+                fromIndex = str.indexOf(target, fromIndex) + 1;
+            }
+            return (long) fromIndex;
+        }
+        if (from == 0) {
             return null;
         }
+
+        long fromIndex = str.getLength() + from;
+        if (fromIndex < 0) {
+            return 0L;
+        }
+        int index = (int) fromIndex;
         for (int i = 0; i < nth; ++i) {
-            fromIndex = str.indexOf(target, fromIndex) + 1;
+            index = lastIndexOf(str, target, index);
+            if (index < 0) {
+                return 0L;
+            }
+            if (i < nth - 1) {
+                index--;
+            }
+        }
+        return (long) index + 1;
+    }
+
+    private int lastIndexOf(BinaryString str, BinaryString target, int fromIndex) {
+        if (target.getLength() == 0) {
+            return fromIndex;
+        }
+        int index = -1;
+        int searchFrom = 0;
+        while (searchFrom <= fromIndex) {
+            int nextIndex = str.indexOf(target, searchFrom);
+            if (nextIndex < 0 || nextIndex > fromIndex) {
+                break;
+            }
+            index = nextIndex;
+            searchFrom = nextIndex + 1;
         }
-        return (long) fromIndex;
+        return index;
     }
 }

geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/org/apache/geaflow/dsl/udf/string/UDFInstrTest.java

@@ -50,6 +50,16 @@ public void test() {
         assertEquals(2, (long) udf.eval("s.taobao.com", ".", 1L));
 
         assertEquals(0, (long) udf.eval("s.taobao.com", "abc"));
+
+        assertEquals(4L, (long) udf.eval("abcabc", "a", -1L, 1L));
+
+        assertEquals(1L, (long) udf.eval("abcabc", "a", -1L, 2L));
+
+        assertEquals(3L, (long) udf.eval("abcabc", "c", -3L, 1L));
+
+        assertEquals(2L, (long) udf.eval("aaaa", "aa", -1L, 2L));
+
+        assertEquals(0L, (long) udf.eval("abcabc", "a", -7L, 1L));
     }
 
     @Test
@@ -75,5 +85,23 @@ public void testBinaryString() {
         assertEquals(2, (long) udf.eval(BinaryString.fromString("s.taobao.com"), BinaryString.fromString("."), 1L));
 
         assertEquals(0, (long) udf.eval(BinaryString.fromString("s.taobao.com"), BinaryString.fromString("abc")));
+
+        assertEquals(4L, (long) udf.eval(BinaryString.fromString("abcabc"),
+            BinaryString.fromString("a"), -1L, 1L));
+
+        assertEquals(1L, (long) udf.eval(BinaryString.fromString("abcabc"),
+            BinaryString.fromString("a"), -1L, 2L));
+
+        assertEquals(3L, (long) udf.eval(BinaryString.fromString("abcabc"),
+            BinaryString.fromString("c"), -3L, 1L));
+
+        assertEquals(2L, (long) udf.eval(BinaryString.fromString("aaaa"),
+            BinaryString.fromString("aa"), -1L, 2L));
+
+        assertEquals(1L, (long) udf.eval(BinaryString.fromString("甲乙甲乙"),
+            BinaryString.fromString("甲"), -1L, 2L));
+
+        assertEquals(0L, (long) udf.eval(BinaryString.fromString("abcabc"),
+            BinaryString.fromString("a"), -7L, 1L));
     }
 }

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/InstrRuntimeTest.java

@@ -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.
+ */
+
+package org.apache.geaflow.dsl.runtime.query;
+
+import org.testng.annotations.Test;
+
+public class InstrRuntimeTest {
+
+    @Test
+    public void testNegativeFromIndex() throws Exception {
+        QueryTester
+            .build()
+            .withQueryPath("/query/instr_negative_001.sql")
+            .execute()
+            .checkSinkResult();
+    }
+}

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

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

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

@@ -0,0 +1,37 @@
+/*
+ * 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 output (
+    last_occurrence bigint,
+    second_occurrence bigint,
+    bounded_search bigint,
+    overlapping_occurrence bigint,
+    out_of_range bigint
+) WITH (
+    type='file',
+    geaflow.dsl.file.path='${target}'
+);
+
+INSERT INTO output
+SELECT
+    INSTR('abcabc', 'a', -1, 1),
+    INSTR('abcabc', 'a', -1, 2),
+    INSTR('abcabc', 'c', -3, 1),
+    INSTR('aaaa', 'aa', -1, 2),
+    INSTR('abcabc', 'a', -7, 1);