fix: handle empty patterns in regexp_instr

datafusion/functions/src/regex/regexpinstr.rs

@@ -302,20 +302,12 @@ where
             continue;
         }
         let regex = regex_array.value(i);
-        if regex.is_empty() {
-            result.append_value(0);
-            continue;
-        }
 
         if values.is_null(i) {
             result.append_null();
             continue;
         }
         let value = values.value(i);
-        if value.is_empty() {
-            result.append_value(0);
-            continue;
-        }
 
         let flags = match flags_array {
             Some(flags) if !flags.is_null(i) => Some(flags.value(i)),
@@ -376,7 +368,7 @@ impl<'a> RegexCache<'a> {
 /// Returns the 1-based character position of the `n`-th match of `pattern` in
 /// `value`, or 0 if there is no such match. The search begins at the 1-based
 /// character position `start`. A positive `subexpr` selects that capture group
-/// of the first match instead of the `n`-th match. `value` is non-empty.
+/// of the first match instead of the `n`-th match.
 fn get_index(
     value: &str,
     pattern: &Regex,
@@ -396,9 +388,16 @@ fn get_index(
         ));
     }
 
-    // Byte offset of the `start`-th character. A `start` past the end of the
-    // string leaves nothing to search, so no match is possible.
-    let Some((byte_start_offset, _)) = value.char_indices().nth((start - 1) as usize)
+    let Ok(start_index) = usize::try_from(start - 1) else {
+        return Ok(0);
+    };
+    // Include the terminal byte boundary so an empty pattern can match after
+    // the last character, including in an empty string.
+    let Some(byte_start_offset) = value
+        .char_indices()
+        .map(|(offset, _)| offset)
+        .chain(std::iter::once(value.len()))
+        .nth(start_index)
     else {
         return Ok(0);
     };
@@ -451,6 +450,14 @@ mod tests {
         test_case_sensitive_regexp_instr_array_nth::<GenericStringArray<i32>>();
         test_case_sensitive_regexp_instr_array_nth::<GenericStringArray<i64>>();
         test_case_sensitive_regexp_instr_array_nth::<StringViewArray>();
+
+        test_case_sensitive_regexp_instr_empty_pattern::<GenericStringArray<i32>>();
+        test_case_sensitive_regexp_instr_empty_pattern::<GenericStringArray<i64>>();
+        test_case_sensitive_regexp_instr_empty_pattern::<StringViewArray>();
+
+        test_case_sensitive_regexp_instr_zero_width_pattern::<GenericStringArray<i32>>();
+        test_case_sensitive_regexp_instr_zero_width_pattern::<GenericStringArray<i64>>();
+        test_case_sensitive_regexp_instr_zero_width_pattern::<StringViewArray>();
     }
 
     fn regexp_instr_with_scalar_values(args: &[ScalarValue]) -> Result<ColumnarValue> {
@@ -479,7 +486,7 @@ mod tests {
     fn test_case_sensitive_regexp_instr_nulls() {
         let v = "";
         let r = "";
-        let expected = 0;
+        let expected = 1;
         let regex_sv = ScalarValue::Utf8(Some(r.to_string()));
         let re = regexp_instr_with_scalar_values(&[v.to_string().into(), regex_sv]);
         // let res_exp = re.unwrap();
@@ -489,6 +496,29 @@ mod tests {
             }
             _ => panic!("Unexpected result"),
         }
+
+        for (value, regex) in [
+            (
+                ScalarValue::Utf8(None),
+                ScalarValue::Utf8(Some(String::new())),
+            ),
+            (
+                ScalarValue::LargeUtf8(None),
+                ScalarValue::LargeUtf8(Some(String::new())),
+            ),
+            (
+                ScalarValue::Utf8View(None),
+                ScalarValue::Utf8View(Some(String::new())),
+            ),
+        ] {
+            let re = regexp_instr_with_scalar_values(&[value, regex]);
+            match re {
+                Ok(ColumnarValue::Scalar(ScalarValue::Int64(v))) => {
+                    assert_eq!(v, None, "regexp_instr NULL scalar test failed");
+                }
+                _ => panic!("Unexpected result"),
+            }
+        }
     }
     fn test_case_sensitive_regexp_instr_scalar() {
         let values = [
@@ -800,4 +830,38 @@ mod tests {
         .unwrap();
         assert_eq!(re.as_ref(), &expected);
     }
+
+    fn test_case_sensitive_regexp_instr_empty_pattern<A>()
+    where
+        A: From<Vec<&'static str>> + Array + 'static,
+    {
+        let values = A::from(vec!["abc", "", "abc", "abc", "😀"]);
+        let regex = A::from(vec!["", "", "", "", ""]);
+        let start = Int64Array::from(vec![1, 1, 4, 5, 1]);
+        let nth = Int64Array::from(vec![1, 1, 1, 1, 2]);
+        let expected = Int64Array::from(vec![1, 1, 4, 0, 2]);
+
+        let re = regexp_instr_func(&[
+            Arc::new(values),
+            Arc::new(regex),
+            Arc::new(start),
+            Arc::new(nth),
+        ])
+        .unwrap();
+        assert_eq!(re.as_ref(), &expected);
+    }
+
+    fn test_case_sensitive_regexp_instr_zero_width_pattern<A>()
+    where
+        A: From<Vec<&'static str>> + Array + 'static,
+    {
+        let values = A::from(vec!["abc"]);
+        let regex = A::from(vec!["x*"]);
+        let start = Int64Array::from(vec![4]);
+        let expected = Int64Array::from(vec![4]);
+
+        let re = regexp_instr_func(&[Arc::new(values), Arc::new(regex), Arc::new(start)])
+            .unwrap();
+        assert_eq!(re.as_ref(), &expected);
+    }
 }

datafusion/sqllogictest/test_files/regexp/regexp_instr.slt

@@ -23,6 +23,18 @@ SELECT regexp_instr('123123123123123', '(12)3');
 ----
 1
 
+query IIIIIII
+SELECT
+  regexp_instr('abc', ''),
+  regexp_instr('', ''),
+  regexp_instr('abc', '', 4),
+  regexp_instr('abc', '', 5),
+  regexp_instr('😀', '', 1, 2),
+  regexp_instr('abc', 'x*', 4),
+  regexp_instr(NULL, '');
+----
+1 1 4 0 2 4 NULL
+
 query I
 SELECT regexp_instr('123123123123', '123', 1);
 ----