fsst like to respect sql escape codes in the pattern (#8107)

## Summary

fsst like dfa implementation to respect escape codes like `%` and `_` in
the pattern

---------

Signed-off-by: Onur Satici <onur@spiraldb.com>

Author: onursatici

encodings/fsst/src/compute/like.rs

@@ -287,6 +287,95 @@ mod tests {
         let result = <FSST as LikeKernel>::like(fsst_v, &pattern, opts, &mut ctx)?;
         assert!(result.is_none(), "ilike should fall back");
 
+        // Suffix patterns are still unsupported, even when the suffix is an escaped literal.
+        let pattern = ConstantArray::new(r"%\%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(result.is_none(), "escaped suffix pattern should fall back");
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_like_kernel_handles_escaped_prefix_and_contains() -> VortexResult<()> {
+        let fsst = make_fsst(
+            &[
+                Some("%front"),
+                Some("_front"),
+                Some("\\front"),
+                Some("middle%value"),
+                Some("middle_value"),
+                Some("middle\\value"),
+                Some("front"),
+            ],
+            Nullability::NonNullable,
+        );
+        let fsst_v = fsst.as_view();
+        let mut ctx = SESSION.create_execution_ctx();
+
+        let pattern = ConstantArray::new(r"\%%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(result.is_some(), "escaped percent prefix should use FSST");
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([true, false, false, false, false, false, false])
+        );
+
+        let pattern = ConstantArray::new(r"\_%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(
+            result.is_some(),
+            "escaped underscore prefix should use FSST"
+        );
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([false, true, false, false, false, false, false])
+        );
+
+        let pattern = ConstantArray::new(r"\\%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(result.is_some(), "escaped backslash prefix should use FSST");
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([false, false, true, false, false, false, false])
+        );
+
+        let pattern = ConstantArray::new(r"%\%%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(result.is_some(), "escaped percent contains should use FSST");
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([true, false, false, true, false, false, false])
+        );
+
+        let pattern = ConstantArray::new(r"%\_%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(
+            result.is_some(),
+            "escaped underscore contains should use FSST"
+        );
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([false, true, false, false, true, false, false])
+        );
+
+        let pattern = ConstantArray::new(r"%\\%", fsst.len()).into_array();
+        let result =
+            <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
+        assert!(
+            result.is_some(),
+            "escaped backslash contains should use FSST"
+        );
+        assert_arrays_eq!(
+            result.unwrap(),
+            BoolArray::from_iter([false, false, true, false, false, true, false])
+        );
+
         Ok(())
     }
 

encodings/fsst/src/dfa/mod.rs

@@ -126,6 +126,8 @@ mod prefix;
 #[cfg(test)]
 mod tests;
 
+use std::borrow::Cow;
+
 use flat_contains::FlatContainsDfa;
 use fsst::ESCAPE_CODE;
 use fsst::Symbol;
@@ -170,18 +172,28 @@ impl FsstMatcher {
         };
 
         let inner = match like_kind {
-            LikeKind::Prefix(b"") | LikeKind::Contains(b"") => MatcherInner::MatchAll,
+            LikeKind::Prefix(pattern) | LikeKind::Contains(pattern) if pattern.is_empty() => {
+                MatcherInner::MatchAll
+            }
             LikeKind::Prefix(prefix) => {
                 if prefix.len() > FlatPrefixDfa::MAX_PREFIX_LEN {
                     return Ok(None);
                 }
-                MatcherInner::Prefix(FlatPrefixDfa::new(symbols, symbol_lengths, prefix)?)
+                MatcherInner::Prefix(FlatPrefixDfa::new(
+                    symbols,
+                    symbol_lengths,
+                    prefix.as_ref(),
+                )?)
             }
             LikeKind::Contains(needle) => {
                 if needle.len() > FlatContainsDfa::MAX_NEEDLE_LEN {
                     return Ok(None);
                 }
-                MatcherInner::Contains(FlatContainsDfa::new(symbols, symbol_lengths, needle)?)
+                MatcherInner::Contains(FlatContainsDfa::new(
+                    symbols,
+                    symbol_lengths,
+                    needle.as_ref(),
+                )?)
             }
         };
 
@@ -201,34 +213,66 @@ impl FsstMatcher {
 /// The subset of LIKE patterns we can handle without decompression.
 enum LikeKind<'a> {
     /// `prefix%`
-    Prefix(&'a [u8]),
+    Prefix(Cow<'a, [u8]>),
     /// `%needle%`
-    Contains(&'a [u8]),
+    Contains(Cow<'a, [u8]>),
 }
 
 impl<'a> LikeKind<'a> {
     fn parse(pattern: &'a [u8]) -> Option<Self> {
-        // The fast-path matchers below do not understand SQL LIKE escape sequences (e.g. `\%`
-        // matching a literal `%`). If the pattern contains a backslash we fall back to the
-        // general implementation, which correctly interprets escapes.
-        if pattern.contains(&b'\\') {
+        Self::parse_prefix(pattern).or_else(|| Self::parse_contains(pattern))
+    }
+
+    fn parse_prefix(pattern: &'a [u8]) -> Option<Self> {
+        Self::parse_literal_until_final_percent(pattern, 0).map(LikeKind::Prefix)
+    }
+
+    fn parse_contains(pattern: &'a [u8]) -> Option<Self> {
+        if !pattern.starts_with(b"%") {
             return None;
         }
 
-        // `prefix%` (including just `%` where prefix is empty)
-        if let Some(prefix) = pattern.strip_suffix(b"%")
-            && !prefix.contains(&b'%')
-            && !prefix.contains(&b'_')
-        {
-            return Some(LikeKind::Prefix(prefix));
-        }
+        Self::parse_literal_until_final_percent(pattern, 1).map(LikeKind::Contains)
+    }
 
-        // `%needle%`
-        let inner = pattern.strip_prefix(b"%")?.strip_suffix(b"%")?;
-        if !inner.contains(&b'%') && !inner.contains(&b'_') {
-            return Some(LikeKind::Contains(inner));
+    /// Parse `pattern[literal_start..]` as a literal terminated by a single
+    /// trailing `%`. Returns `None` if `_` or a non-final `%` is encountered.
+    ///
+    /// `literal` stays `None` until an escape forces us to materialize bytes;
+    /// from then on we push into the owned `Vec`. Otherwise we return a
+    /// borrowed slice straight from `pattern`.
+    fn parse_literal_until_final_percent(
+        pattern: &'a [u8],
+        literal_start: usize,
+    ) -> Option<Cow<'a, [u8]>> {
+        let mut literal: Option<Vec<u8>> = None;
+        let mut idx = literal_start;
+        while idx < pattern.len() {
+            match pattern[idx] {
+                b'\\' => {
+                    // Trailing `\` is treated as a literal backslash.
+                    let escaped = pattern.get(idx + 1).copied().unwrap_or(b'\\');
+                    literal
+                        .get_or_insert_with(|| pattern[literal_start..idx].to_vec())
+                        .push(escaped);
+                    idx = (idx + 2).min(pattern.len());
+                }
+                b'%' if idx + 1 == pattern.len() => {
+                    return Some(match literal {
+                        Some(buf) => Cow::Owned(buf),
+                        None => Cow::Borrowed(&pattern[literal_start..idx]),
+                    });
+                }
+                b'%' | b'_' => return None,
+                byte => {
+                    // No-op on the borrowed path; only push once we've started copying.
+                    if let Some(literal) = &mut literal {
+                        literal.push(byte);
+                    }
+                    idx += 1;
+                }
+            }
         }
-
         None
     }
 }

encodings/fsst/src/dfa/tests.rs

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
+use std::borrow::Cow;
 use std::sync::LazyLock;
 
 use fsst::ESCAPE_CODE;
@@ -50,30 +51,63 @@ fn escaped(bytes: &[u8]) -> Vec<u8> {
     codes
 }
 
+fn assert_borrowed_prefix(pattern: &[u8], expected: &[u8]) {
+    let Some(LikeKind::Prefix(actual)) = LikeKind::parse(pattern) else {
+        panic!("expected borrowed prefix pattern");
+    };
+    assert!(matches!(actual, Cow::Borrowed(_)));
+    assert_eq!(actual.as_ref(), expected);
+}
+
+fn assert_owned_prefix(pattern: &[u8], expected: &[u8]) {
+    let Some(LikeKind::Prefix(actual)) = LikeKind::parse(pattern) else {
+        panic!("expected owned prefix pattern");
+    };
+    assert!(matches!(actual, Cow::Owned(_)));
+    assert_eq!(actual.as_ref(), expected);
+}
+
+fn assert_borrowed_contains(pattern: &[u8], expected: &[u8]) {
+    let Some(LikeKind::Contains(actual)) = LikeKind::parse(pattern) else {
+        panic!("expected borrowed contains pattern");
+    };
+    assert!(matches!(actual, Cow::Borrowed(_)));
+    assert_eq!(actual.as_ref(), expected);
+}
+
+fn assert_owned_contains(pattern: &[u8], expected: &[u8]) {
+    let Some(LikeKind::Contains(actual)) = LikeKind::parse(pattern) else {
+        panic!("expected owned contains pattern");
+    };
+    assert!(matches!(actual, Cow::Owned(_)));
+    assert_eq!(actual.as_ref(), expected);
+}
+
 #[test]
-fn test_like_kind_parse() {
-    assert!(matches!(
-        LikeKind::parse(b"http%"),
-        Some(LikeKind::Prefix(b"http"))
-    ));
-    assert!(matches!(
-        LikeKind::parse(b"%needle%"),
-        Some(LikeKind::Contains(b"needle"))
-    ));
-    assert!(matches!(LikeKind::parse(b"%"), Some(LikeKind::Prefix(b""))));
-    // Suffix and underscore patterns are not supported.
+fn test_like_kind_parse_plain_patterns() {
+    assert_borrowed_prefix(b"http%", b"http");
+    assert_borrowed_contains(b"%needle%", b"needle");
+    assert_borrowed_prefix(b"%", b"");
+}
+
+#[test]
+fn test_like_kind_parse_escaped_patterns() {
+    assert_owned_prefix(br"\%%", b"%");
+    assert_owned_prefix(br"\_%", b"_");
+    assert_owned_prefix(br"\\%", b"\\");
+    assert_owned_prefix(br"has\%middle%", b"has%middle");
+    assert_owned_contains(br"%\%%", b"%");
+    assert_owned_contains(br"%\_%", b"_");
+    assert_owned_contains(br"%\\%", b"\\");
+    assert_owned_contains(br"%has\%middle%", b"has%middle");
+}
+
+#[test]
+fn test_like_kind_parse_unsupported_patterns() {
     assert!(LikeKind::parse(b"%suffix").is_none());
     assert!(LikeKind::parse(b"a_c").is_none());
-
-    // Patterns containing the SQL LIKE escape character must not be parsed by the fast path,
-    // because that path treats `%` and `_` literally and would misinterpret escapes. For
-    // example, `%\%` (the pattern produced by Spark's `endsWith("%")`) means "ends with `%`",
-    // not "contains `\`". The fast path should bail so the general implementation handles it.
     assert!(LikeKind::parse(br"%\%").is_none());
-    assert!(LikeKind::parse(br"\%%").is_none());
-    assert!(LikeKind::parse(br"%\_%").is_none());
-    assert!(LikeKind::parse(br"\_%").is_none());
-    assert!(LikeKind::parse(br"%\\%").is_none());
+    assert!(LikeKind::parse(br"foo\%bar").is_none());
 }
 
 /// No symbols — all bytes escaped. Simplest case to see the two tables.