feat(like): support LIKE over Binary arrays with byte semantics

vortex-array/src/scalar_fn/fns/like/mod.rs

@@ -123,12 +123,19 @@ impl ScalarFnVTable for Like {
         Display::fmt(expr.display_child(1), f)
     }
 
-    fn return_dtype(&self, _options: &Self::Options, arg_dtypes: &[DType]) -> VortexResult<DType> {
+    fn return_dtype(&self, options: &Self::Options, arg_dtypes: &[DType]) -> VortexResult<DType> {
         let input = &arg_dtypes[0];
         let pattern = &arg_dtypes[1];
 
-        if !input.is_utf8() {
-            vortex_bail!("LIKE expression requires UTF8 input dtype, got {}", input);
+        // Utf8 input is always supported. Binary input is supported for case-sensitive LIKE, where
+        // matching uses SQL byte semantics; ILIKE over Binary is rejected (case folding is only
+        // well-defined over text).
+        let input_ok = input.is_utf8() || (input.is_binary() && !options.case_insensitive);
+        if !input_ok {
+            vortex_bail!(
+                "LIKE expression requires UTF8 input dtype (or Binary for case-sensitive LIKE), got {}",
+                input
+            );
         }
         if !pattern.is_utf8() {
             vortex_bail!(
@@ -191,6 +198,8 @@ pub(crate) fn execute_like(
         array.encoding_id()
     );
     let len = array.len();
+    // Binary haystacks match with byte semantics; Utf8 with Unicode codepoint semantics.
+    let byte_mode = array.dtype().is_binary();
     let nullability =
         Nullability::from(array.dtype().is_nullable() || pattern.dtype().is_nullable());
 
@@ -216,6 +225,7 @@ pub(crate) fn execute_like(
             pattern_str.as_str(),
             options.case_insensitive,
             ascii_haystack,
+            byte_mode,
         )?;
         let bits = eval_pattern(&haystack, &compiled, options.negated);
         let validity = values.validity()?.union_nullability(nullability);
@@ -244,6 +254,7 @@ pub(crate) fn execute_like(
                     pattern_str,
                     options.case_insensitive,
                     ascii_haystack && pattern_str.is_ascii(),
+                    byte_mode,
                 )?;
                 &cached.insert((pattern_bytes, compiled)).1
             }
@@ -591,6 +602,72 @@ mod tests {
         assert_arrays_eq!(result, BoolArray::from_iter([true, true, false]), &mut ctx);
     }
 
+    #[test]
+    fn test_like_binary_byte_semantics() {
+        let mut ctx = array_session().create_execution_ctx();
+        // `Ж` (U+0416) encodes as bytes D0 96 (2 bytes); 0xFF is an invalid-UTF-8 byte.
+        let rows: Vec<&[u8]> = vec![
+            b"aXb",                                  // a + 1 byte  + b
+            &[0x61, 0xD0, 0x96, 0x62],               // a + Ж (2 bytes) + b
+            &[0x61, 0xD0, 0x96, 0xD0, 0x96, 0x62],   // a + ЖЖ (4 bytes) + b
+            b"ab",                                   // a + 0 bytes + b
+            &[0x61, 0xFF, 0x62],                     // a + invalid-UTF-8 byte + b
+        ];
+        let array = VarBinViewArray::from_iter_bin(rows).into_array();
+        assert!(array.dtype().is_binary());
+
+        // Over Binary, `_` matches exactly one BYTE (over Utf8 it matches one codepoint, see
+        // test_like_unicode): the 2-byte `Ж` is not matched by a single `_`.
+        let result = run_like(
+            array.clone(),
+            ConstantArray::new("a_b", 5).into_array(),
+            LikeOptions::default(),
+        );
+        assert_arrays_eq!(
+            result,
+            BoolArray::from_iter([true, false, false, false, true]),
+            &mut ctx
+        );
+
+        // Two underscores match the two bytes of `Ж`.
+        let result = run_like(
+            array.clone(),
+            ConstantArray::new("a__b", 5).into_array(),
+            LikeOptions::default(),
+        );
+        assert_arrays_eq!(
+            result,
+            BoolArray::from_iter([false, true, false, false, false]),
+            &mut ctx
+        );
+
+        // `%` spans any bytes, including invalid UTF-8.
+        let result = run_like(
+            array,
+            ConstantArray::new("a%b", 5).into_array(),
+            LikeOptions::default(),
+        );
+        assert_arrays_eq!(
+            result,
+            BoolArray::from_iter([true, true, true, true, true]),
+            &mut ctx
+        );
+    }
+
+    #[test]
+    fn test_ilike_over_binary_is_rejected() {
+        let mut ctx = array_session().create_execution_ctx();
+        let array = VarBinViewArray::from_iter_bin([b"abc".as_slice()]).into_array();
+        // Case folding is only well-defined over text, so ILIKE over Binary must be rejected.
+        let built = Like.try_new_array(
+            1,
+            LikeOptions { negated: false, case_insensitive: true },
+            [array, ConstantArray::new("a%c", 1).into_array()],
+        );
+        let rejected = built.map_or(true, |e| e.execute::<BoolArray>(&mut ctx).is_err());
+        assert!(rejected, "ILIKE over Binary must be rejected");
+    }
+
     #[test]
     fn test_nlike() {
         let mut ctx = array_session().create_execution_ctx();

vortex-array/src/scalar_fn/fns/like/pattern.rs

@@ -54,6 +54,7 @@ impl LikePattern {
         pattern: &str,
         case_insensitive: bool,
         ascii_haystack: bool,
+        byte_mode: bool,
     ) -> VortexResult<Self> {
         if case_insensitive {
             if ascii_haystack && pattern.is_ascii() {
@@ -69,7 +70,7 @@ impl LikePattern {
                     return Ok(Self::IEndsWithAscii(pattern.as_bytes()[1..].to_vec()));
                 }
             }
-            return Ok(Self::Regex(regex_like(pattern, true)?));
+            return Ok(Self::Regex(regex_like(pattern, true, byte_mode)?));
         }
 
         Ok(if !contains_like_pattern(pattern) {
@@ -87,7 +88,7 @@ impl LikePattern {
                 pattern.len() - 2,
             )
         } else {
-            Self::Regex(regex_like(pattern, false)?)
+            Self::Regex(regex_like(pattern, false, byte_mode)?)
         })
     }
 
@@ -135,9 +136,11 @@ fn contains_like_pattern(pattern: &str) -> bool {
 /// 3. regex meta characters are escaped so they match literally;
 /// 4. `\x` matches `x` literally (a trailing `\` matches a literal backslash).
 ///
-/// The regex runs over the haystack bytes directly (`regex::bytes`) in Unicode mode, which
-/// matches identically to a `&str` regex on valid UTF-8 input.
-fn regex_like(pattern: &str, case_insensitive: bool) -> VortexResult<Regex> {
+/// The regex runs over the haystack bytes directly (`regex::bytes`). For Utf8 haystacks it uses
+/// Unicode mode, matching identically to a `&str` regex on valid UTF-8 input. For Binary haystacks
+/// (`byte_mode`) Unicode is disabled, so `.`/`.*` operate on single bytes and match arbitrary byte
+/// sequences (including invalid UTF-8), giving SQL byte semantics for `_`/`%`.
+fn regex_like(pattern: &str, case_insensitive: bool, byte_mode: bool) -> VortexResult<Regex> {
     let mut result = String::with_capacity(pattern.len() * 2);
     let mut chars_iter = pattern.chars().peekable();
     match chars_iter.peek() {
@@ -188,6 +191,9 @@ fn regex_like(pattern: &str, case_insensitive: bool) -> VortexResult<Regex> {
     RegexBuilder::new(&result)
         .case_insensitive(case_insensitive)
         .dot_matches_new_line(true)
+        // Binary haystacks use SQL byte semantics: `.`/`_` match one byte and `.*`/`%` span
+        // arbitrary bytes (including invalid UTF-8). Utf8 keeps Unicode codepoint semantics.
+        .unicode(!byte_mode)
         .build()
         .map_err(|e| vortex_err!("Unable to build regex from LIKE pattern: {e}"))
 }
@@ -197,7 +203,7 @@ mod tests {
     use super::*;
 
     fn like(pattern: &str) -> LikePattern {
-        LikePattern::compile(pattern, false, false).unwrap()
+        LikePattern::compile(pattern, false, false, false).unwrap()
     }
 
     #[test]
@@ -229,7 +235,7 @@ mod tests {
             (r"\\", r"^\\$"),
         ];
         for (pattern, expected) in cases {
-            assert_eq!(regex_like(pattern, false).unwrap().to_string(), expected);
+            assert_eq!(regex_like(pattern, false, false).unwrap().to_string(), expected);
         }
     }
 
@@ -257,7 +263,7 @@ mod tests {
 
     #[test]
     fn matches_case_insensitive() {
-        let ilike = |pattern: &str| LikePattern::compile(pattern, true, false).unwrap();
+        let ilike = |pattern: &str| LikePattern::compile(pattern, true, false, false).unwrap();
         assert!(ilike("hello%").matches(b"HELLO WORLD"));
         assert!(ilike("%WORLD").matches(b"hello world"));
         // Full case folding: the ASCII pattern `k` matches U+212A KELVIN SIGN.
@@ -267,7 +273,7 @@ mod tests {
         assert!(ilike("\u{03c3}").matches("\u{03c2}".as_bytes()));
 
         // The ASCII fast paths agree with the regex on ASCII haystacks.
-        let ascii = |pattern: &str| LikePattern::compile(pattern, true, true).unwrap();
+        let ascii = |pattern: &str| LikePattern::compile(pattern, true, true, false).unwrap();
         assert!(matches!(ascii("abc"), LikePattern::IEqAscii(_)));
         assert!(matches!(ascii("abc%"), LikePattern::IStartsWithAscii(_)));
         assert!(matches!(ascii("%abc"), LikePattern::IEndsWithAscii(_)));