chore(query): Implement Oracle-compatible INSTR Function in Databend (#18084)

This PR implements Oracle-compatible `INSTR` string search functionality in Databend, supporting both three-parameter and four-parameter variants to enhance SQL compatibility with Oracle databases.

```sql
INSTR(string, substring [, pos] [, occurrence])
```

Finds the position of a substring within a string starting from a specified position.

**Parameters:**
- `string`: The input string to search within (can be NULL)
- `substring`: The substring to locate (can be NULL)
- `pos`: The starting position for the search (1-based index)
  - Positive number: Search forward from left to right
  - Negative number: Search backward from right to left
  - Zero: Treated as 1 (start from beginning)

**Returns:**
- The 1-based position where the substring is found
- Returns 0 if:
  - Either string or substring is NULL
  - Substring is not found
  - pos is 0 (after being converted to 1) and substring isn't at start

Finds the nth occurrence of a substring within a string starting from a specified position.

**Additional Parameter:**
- `occurrence`: Which occurrence of the substring to find (must be positive integer)
  - 1 = first occurrence (default behavior)
  - 2 = second occurrence, etc.

**Returns:**
- Same return rules as three-parameter version
- For occurrence > 1, continues searching after previous matches
- If occurrence exceeds actual number of matches, returns 0

Author: TCeason

src/query/functions/src/scalars/string.rs

@@ -324,12 +324,105 @@ pub fn register(registry: &mut FunctionRegistry) {
         }
     };
 
+    fn instr(s: &str, sub: &str, position: i64, occurrence: u64) -> u64 {
+        if occurrence == 0 {
+            return 0;
+        }
+
+        let s_len_chars = s.chars().count();
+        let sub_len_chars = sub.chars().count();
+        if sub_len_chars > s_len_chars {
+            return 0;
+        }
+
+        if sub.is_empty() {
+            return if position > 0 {
+                // For forward search, the empty string matches at the position
+                // such as: INSTR (' ABC ', ', 1) - > 1, INSTR (' ABC ', ' ', 4) - > 4
+                // The effective range is from 1 to s_len_chars + 1
+                if (position as usize) <= s_len_chars + 1 {
+                    position as u64
+                } else {
+                    0
+                }
+            } else {
+                // For reverse search, empty strings match positions after the end of the string
+                // such as: INSTR('abc', '', -1) -> 4 (s_len_chars + 1)
+                (s_len_chars + 1) as u64
+            };
+        }
+
+        let mut char_to_byte_map = Vec::with_capacity(s_len_chars + 1);
+        for (byte_idx, _) in s.char_indices() {
+            char_to_byte_map.push(byte_idx);
+        }
+        char_to_byte_map.push(s.len());
+
+        let get_slice_by_char_idx = |start_char_idx: usize, num_chars: usize| -> Option<&str> {
+            if start_char_idx + num_chars > s_len_chars {
+                return None;
+            }
+            let byte_start = char_to_byte_map[start_char_idx];
+            let byte_end = char_to_byte_map[start_char_idx + num_chars];
+            s.get(byte_start..byte_end)
+        };
+
+        let mut found_count = 0;
+
+        if position > 0 {
+            let start_char_idx_0_indexed = (position - 1) as usize;
+            for current_char_idx in
+                start_char_idx_0_indexed..=s_len_chars.saturating_sub(sub_len_chars)
+            {
+                if let Some(slice) = get_slice_by_char_idx(current_char_idx, sub_len_chars) {
+                    if slice == sub {
+                        found_count += 1;
+                        if found_count == occurrence {
+                            return (current_char_idx + 1) as u64;
+                        }
+                    }
+                }
+            }
+        } else {
+            let search_start_char_0_indexed =
+                s_len_chars.saturating_sub(position.unsigned_abs() as usize);
+
+            let max_possible_match_start_idx =
+                search_start_char_0_indexed.min(s_len_chars.saturating_sub(sub_len_chars));
+
+            for current_char_idx in (0..=max_possible_match_start_idx).rev() {
+                if let Some(slice) = get_slice_by_char_idx(current_char_idx, sub_len_chars) {
+                    if slice == sub {
+                        found_count += 1;
+                        if found_count == occurrence {
+                            return (current_char_idx + 1) as u64;
+                        }
+                    }
+                }
+            }
+        }
+
+        0
+    }
+
     registry.register_2_arg::<StringType, StringType, NumberType<u64>, _, _>(
         "instr",
         |_, _, _| FunctionDomain::Full,
         move |s: &str, substr: &str, _| find_at(s, substr, 1),
     );
 
+    registry.register_3_arg::<StringType, StringType, NumberType<i64>, NumberType<u64>, _, _>(
+        "instr",
+        |_, _, _, _| FunctionDomain::Full,
+        move |s: &str, substr: &str, pos, _| instr(s, substr, pos, 1),
+    );
+
+    registry.register_4_arg::<StringType, StringType, NumberType<i64>, NumberType<u64>, NumberType<u64>, _, _>(
+        "instr",
+        |_, _,_, _, _| FunctionDomain::Full,
+        move |s: &str, substr: &str, pos, occurrence, _| instr(s, substr, pos, occurrence),
+    );
+
     registry.register_2_arg::<StringType, StringType, NumberType<u64>, _, _>(
         "position",
         |_, _, _| FunctionDomain::Full,

src/query/functions/tests/it/scalars/testdata/function_list.txt

@@ -2144,6 +2144,10 @@ Functions overloads:
 1 insert(String NULL, Int64 NULL, Int64 NULL, String NULL) :: String NULL
 0 instr(String, String) :: UInt64
 1 instr(String NULL, String NULL) :: UInt64 NULL
+2 instr(String, String, Int64) :: UInt64
+3 instr(String NULL, String NULL, Int64 NULL) :: UInt64 NULL
+4 instr(String, String, Int64, UInt64) :: UInt64
+5 instr(String NULL, String NULL, Int64 NULL, UInt64 NULL) :: UInt64 NULL
 0 is_array(Variant) :: Boolean
 1 is_array(Variant NULL) :: Boolean NULL
 0 is_binary(Variant) :: Boolean

tests/sqllogictests/suites/query/functions/02_0025_function_strings_locate.test

@@ -67,3 +67,59 @@ query I
 SELECT INSTR('', '')
 ----
 1
+
+query T
+SELECT INSTR('你你好你', '你', 1);
+----
+1
+
+query T
+SELECT INSTR('你你好你', '你', 1, 1);
+----
+1
+
+query T
+SELECT INSTR('你你好你', '你', 1, 2);
+----
+2
+
+query T
+SELECT INSTR('你你好你', '你', -1, 1);
+----
+4
+
+query T
+SELECT INSTR('你你好你', '你', -1, 2);
+----
+2
+
+query T
+SELECT INSTR('CORPORATE FLOOR','OR', 3, 2);
+----
+14
+
+query T
+SELECT INSTR('CORPORATE FLOOR','OR', -3, 2)
+----
+2
+
+statement ok
+create or replace table t(c1 string);
+
+statement ok
+insert into t values('CORPORATE FLOOR'),('你你好你');
+
+query T
+select instr(c1,'OR', -3, 2) from t;
+----
+2
+0
+
+query T
+select instr(c1,'你', -1) from t;
+----
+0
+4
+
+statement ok
+drop table if exists t;