FSST LIKE: ILIKE / ASCII case-insensitive matching

Extend the DFA construction to optionally fold ASCII letter case.
Adds `FsstMatcher::try_new_with(symbols, lengths, pattern,
case_insensitive)`; the `like.rs` kernel now plumbs
`options.case_insensitive` through instead of bailing out.

Mechanism:
- `dfa/mod.rs`: add `ascii_to_lower`, `pattern_eq(a, b, ci)`,
  `pattern_matches_byte(p, b, ci)`, and a `set_advance` helper that,
  when `ci` is true, sets both case variants of an ASCII letter in
  the byte table. `kmp_byte_transitions` and `kmp_failure_table`
  now take `ci`; the fold is at construction time so the hot loop
  stays a single table lookup per byte.
- `dfa/prefix.rs::build_prefix_byte_table`, `dfa/suffix.rs::
  build_suffix_byte_table`, `dfa/multi_contains.rs::
  chained_kmp_byte_transitions`: same pattern.
- Each DFA's `new()` takes `case_insensitive: bool`. Threaded
  through `FsstMatcher::try_new_with` from `LikeKernel::like`.
- Escape-only memmem fast path is gated to wildcard-free,
  case-sensitive needles (the encoded pattern is byte-exact).

Adds 6 ILIKE tests covering prefix, suffix, contains,
multi-contains, ILIKE + `_` wildcard, and ILIKE with FSST
symbol expansions in mixed case. 169 tests pass.

Signed-off-by: Claude <noreply@anthropic.com>

Author: claude

encodings/fsst/src/compute/like.rs

@@ -68,9 +68,8 @@ impl LikeKernel for FSST {
             return Ok(None);
         };
 
-        if options.case_insensitive {
-            return Ok(None);
-        }
+        // `case_insensitive` (SQL `ILIKE`) is plumbed through the
+        // matcher; only ASCII letter case is folded.
 
         let pattern_bytes: &[u8] = if let Some(s) = pattern_scalar.as_utf8_opt() {
             let Some(v) = s.value() else {
@@ -101,8 +100,12 @@ impl LikeKernel for FSST {
         let layout_us = phase_us(phase_t);
         phase_t = trace.then(std::time::Instant::now);
 
-        let Some(matcher) =
-            FsstMatcher::try_new(symbols.as_slice(), symbol_lengths.as_slice(), pattern_bytes)?
+        let Some(matcher) = FsstMatcher::try_new_with(
+            symbols.as_slice(),
+            symbol_lengths.as_slice(),
+            pattern_bytes,
+            options.case_insensitive,
+        )?
         else {
             return Ok(None);
         };
@@ -364,21 +367,22 @@ mod tests {
         let fsst = make_fsst(&[Some("abc"), Some("def")], Nullability::NonNullable);
         let mut ctx = SESSION.create_execution_ctx();
 
-        // Underscore wildcard -- not handled.
+        // Anchored exact pattern (no bookend `%`) is still unsupported —
+        // mixed-anchor work is a separate item.
         let pattern = ConstantArray::new("a_c", fsst.len()).into_array();
         let fsst_v = fsst.as_view();
         let result =
             <FSST as LikeKernel>::like(fsst_v, &pattern, LikeOptions::default(), &mut ctx)?;
-        assert!(result.is_none(), "underscore pattern should fall back");
+        assert!(result.is_none(), "anchored exact pattern should fall back");
 
-        // Case-insensitive -- not handled.
+        // ILIKE is now handled by the matcher's case-insensitive path.
         let pattern = ConstantArray::new("abc%", fsst.len()).into_array();
         let opts = LikeOptions {
             negated: false,
             case_insensitive: true,
         };
         let result = <FSST as LikeKernel>::like(fsst_v, &pattern, opts, &mut ctx)?;
-        assert!(result.is_none(), "ilike should fall back");
+        assert!(result.is_some(), "ilike should now be handled");
 
         Ok(())
     }

encodings/fsst/src/dfa/flat_contains.rs

@@ -68,6 +68,7 @@ impl FlatContainsDfa {
         symbols: &[Symbol],
         symbol_lengths: &[u8],
         needle: &[u8],
+        case_insensitive: bool,
     ) -> VortexResult<Self> {
         if needle.len() > Self::MAX_NEEDLE_LEN {
             vortex_bail!(
@@ -82,7 +83,7 @@ impl FlatContainsDfa {
         let n_states = accept_state + 1;
         let sentinel = n_states;
 
-        let byte_table = kmp_byte_transitions(needle);
+        let byte_table = kmp_byte_transitions(needle, case_insensitive);
         let sym_trans =
             build_symbol_transitions(symbols, symbol_lengths, &byte_table, n_states, accept_state);
         let transitions = build_fused_table(&sym_trans, symbols.len(), n_states, |_| sentinel, 0);
@@ -103,7 +104,8 @@ impl FlatContainsDfa {
         // and `scan_to_bitbuf` can prefilter with a single `memmem` over
         // `all_bytes` whose length matches the encoded needle. Disabled
         // for L < 2 (no possible win over the existing scan path).
-        let escape_only_pattern = (needle.len() >= 2
+        let escape_only_pattern = (!case_insensitive
+            && needle.len() >= 2
             && super::needle_is_literal(needle)
             && needle_bytes_absent_from_all_symbols(symbols, symbol_lengths, needle))
         .then(|| build_escape_only_encoded_pattern(needle));

encodings/fsst/src/dfa/folded_contains.rs

@@ -219,6 +219,7 @@ impl FoldedContainsDfa {
         symbols: &[Symbol],
         symbol_lengths: &[u8],
         needle: &[u8],
+        case_insensitive: bool,
     ) -> VortexResult<Self> {
         if needle.len() > Self::MAX_NEEDLE_LEN {
             vortex_bail!(
@@ -235,7 +236,7 @@ impl FoldedContainsDfa {
         // Total states: 2N+1 (normal 0..=N, escape N+1..=2N for base 0..=N-1).
         let n_states_usize = 2 * usize::from(accept_state) + 1;
 
-        let byte_table = kmp_byte_transitions(needle);
+        let byte_table = kmp_byte_transitions(needle, case_insensitive);
         let sym_trans =
             build_symbol_transitions(symbols, symbol_lengths, &byte_table, n_normal, accept_state);
 
@@ -327,13 +328,10 @@ impl FoldedContainsDfa {
             if v.is_empty() { None } else { Some(v) }
         });
 
-        // Escape-only fast path: when no symbol's expansion contains any
-        // needle byte AND the needle has no `_` wildcards, the only
-        // DFA-accepting compressed sequence is the 2L-byte pattern
-        // `[ESCAPE, needle[0], …, ESCAPE, needle[L-1]]`. With wildcards
-        // present the encoded pattern is no longer unique and the
-        // memmem prefilter is disabled.
-        let escape_only_pattern = (needle.len() >= 2
+        // Escape-only fast path: only when the needle is wildcard-free,
+        // case-sensitive, and no symbol contains any literal needle byte.
+        let escape_only_pattern = (!case_insensitive
+            && needle.len() >= 2
             && super::needle_is_literal(needle)
             && needle_bytes_absent_from_all_symbols(symbols, symbol_lengths, needle))
         .then(|| build_escape_only_encoded_pattern(needle));

encodings/fsst/src/dfa/mod.rs

@@ -180,11 +180,24 @@ impl FsstMatcher {
         symbols: &[Symbol],
         symbol_lengths: &[u8],
         pattern: &[u8],
+    ) -> VortexResult<Option<Self>> {
+        Self::try_new_with(symbols, symbol_lengths, pattern, false)
+    }
+
+    /// Variant of [`Self::try_new`] that opts in to ASCII case-insensitive
+    /// matching (SQL `ILIKE`). Letter bytes in the needle then accept
+    /// either case at every position.
+    pub fn try_new_with(
+        symbols: &[Symbol],
+        symbol_lengths: &[u8],
+        pattern: &[u8],
+        case_insensitive: bool,
     ) -> VortexResult<Option<Self>> {
         let Some(like_kind) = LikeKind::parse(pattern) else {
             return Ok(None);
         };
 
+        let ci = case_insensitive;
         let inner = match like_kind {
             LikeKind::Prefix(b"") | LikeKind::Contains(b"") | LikeKind::Suffix(b"") => {
                 MatcherInner::MatchAll
@@ -193,23 +206,29 @@ impl FsstMatcher {
                 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, ci)?)
             }
             LikeKind::Suffix(suffix) => {
                 if suffix.len() > SuffixMatcher::MAX_SUFFIX_LEN {
                     return Ok(None);
                 }
-                MatcherInner::Suffix(SuffixMatcher::new(symbols, symbol_lengths, suffix)?)
+                MatcherInner::Suffix(SuffixMatcher::new(symbols, symbol_lengths, suffix, ci)?)
             }
             LikeKind::Contains(needle) => {
                 if needle.len() <= FoldedContainsDfa::MAX_NEEDLE_LEN {
                     MatcherInner::FoldedContains(FoldedContainsDfa::new(
                         symbols,
                         symbol_lengths,
                         needle,
+                        ci,
                     )?)
                 } else if needle.len() <= FlatContainsDfa::MAX_NEEDLE_LEN {
-                    MatcherInner::Contains(FlatContainsDfa::new(symbols, symbol_lengths, needle)?)
+                    MatcherInner::Contains(FlatContainsDfa::new(
+                        symbols,
+                        symbol_lengths,
+                        needle,
+                        ci,
+                    )?)
                 } else {
                     return Ok(None);
                 }
@@ -223,6 +242,7 @@ impl FsstMatcher {
                     symbols,
                     symbol_lengths,
                     &segments,
+                    ci,
                 )?))
             }
         };
@@ -610,20 +630,57 @@ fn build_fused_table(
 /// expressed.
 pub(super) const WILDCARD: u8 = b'_';
 
+/// ASCII case fold to lowercase. Non-letters pass through.
+#[inline]
+fn ascii_to_lower(b: u8) -> u8 {
+    if b.is_ascii_uppercase() { b + 32 } else { b }
+}
+
 /// Pattern-position byte equality with wildcard semantics. Returns
 /// `true` if `a` or `b` is the [`WILDCARD`] byte, or both bytes are
-/// equal.
+/// equal. When `ci` is true, ASCII letter case is ignored.
 #[inline]
-fn pattern_eq(a: u8, b: u8) -> bool {
-    a == WILDCARD || b == WILDCARD || a == b
+fn pattern_eq(a: u8, b: u8, ci: bool) -> bool {
+    if a == WILDCARD || b == WILDCARD {
+        return true;
+    }
+    if ci {
+        ascii_to_lower(a) == ascii_to_lower(b)
+    } else {
+        a == b
+    }
 }
 
 /// Concrete-input byte match against a needle position. The pattern
 /// position `p` is one of the needle bytes (possibly the wildcard);
-/// the input byte `b` is always concrete (never a wildcard).
+/// the input byte `b` is always concrete (never a wildcard). When `ci`
+/// is true, ASCII letter case is ignored.
+#[inline]
+#[expect(
+    dead_code,
+    reason = "Reserved for the future correct contains-wildcard DFA."
+)]
+fn pattern_matches_byte(p: u8, b: u8, ci: bool) -> bool {
+    if p == WILDCARD {
+        return true;
+    }
+    if ci {
+        ascii_to_lower(p) == ascii_to_lower(b)
+    } else {
+        p == b
+    }
+}
+
+/// For an advancing transition on byte `needle_byte`, set the table
+/// row entry. With `ci` true, also set the entry for the case-flipped
+/// byte so either case of the same ASCII letter advances.
 #[inline]
-fn pattern_matches_byte(p: u8, b: u8) -> bool {
-    p == WILDCARD || p == b
+fn set_advance(table: &mut [u8], row_start: usize, needle_byte: u8, new_state: u8, ci: bool) {
+    table[row_start + usize::from(needle_byte)] = new_state;
+    if ci && needle_byte.is_ascii_alphabetic() {
+        let flipped = needle_byte ^ 0x20;
+        table[row_start + usize::from(flipped)] = new_state;
+    }
 }
 
 /// Build the `(state × byte) → state` KMP transition table.
@@ -641,11 +698,11 @@ fn pattern_matches_byte(p: u8, b: u8) -> bool {
 ///
 /// This is one 256-byte memcpy + a single override per state, instead
 /// of running the KMP fallback loop at every cell.
-fn kmp_byte_transitions(needle: &[u8]) -> Vec<u8> {
+fn kmp_byte_transitions(needle: &[u8], ci: bool) -> Vec<u8> {
     let n_states = u8::try_from(needle.len() + 1)
         .vortex_expect("kmp_byte_transitions: must have needle.len() ≤ 255");
     let accept = n_states - 1;
-    let failure = kmp_failure_table(needle);
+    let failure = kmp_failure_table(needle, ci);
 
     let mut table = vec![0u8; usize::from(n_states) * 256];
 
@@ -654,7 +711,7 @@ fn kmp_byte_transitions(needle: &[u8]) -> Vec<u8> {
         if first == WILDCARD {
             table[0..256].fill(1);
         } else {
-            table[usize::from(first)] = 1;
+            set_advance(&mut table, 0, first, 1, ci);
         }
     }
 
@@ -672,7 +729,7 @@ fn kmp_byte_transitions(needle: &[u8]) -> Vec<u8> {
             // Wildcard at position s: every byte advances.
             table[state_row..state_row + 256].fill(state + 1);
         } else {
-            table[state_row + usize::from(needle[s])] = state + 1;
+            set_advance(&mut table, state_row, needle[s], state + 1, ci);
         }
     }
 
@@ -685,14 +742,14 @@ fn kmp_byte_transitions(needle: &[u8]) -> Vec<u8> {
     table
 }
 
-fn kmp_failure_table(needle: &[u8]) -> Vec<u8> {
+fn kmp_failure_table(needle: &[u8], ci: bool) -> Vec<u8> {
     let mut failure = vec![0u8; needle.len()];
     let mut k = 0u8;
     for i in 1..needle.len() {
-        while k > 0 && !pattern_eq(needle[usize::from(k)], needle[i]) {
+        while k > 0 && !pattern_eq(needle[usize::from(k)], needle[i], ci) {
             k = failure[usize::from(k) - 1];
         }
-        if pattern_eq(needle[usize::from(k)], needle[i]) {
+        if pattern_eq(needle[usize::from(k)], needle[i], ci) {
             k += 1;
         }
         failure[i] = k;

encodings/fsst/src/dfa/multi_contains.rs

@@ -101,6 +101,7 @@ impl MultiContainsDfa {
         symbols: &[Symbol],
         symbol_lengths: &[u8],
         segments: &[&[u8]],
+        case_insensitive: bool,
     ) -> VortexResult<Self> {
         let total_len: usize = segments.iter().map(|s| s.len()).sum();
         if total_len > Self::MAX_TOTAL_LEN {
@@ -116,7 +117,7 @@ impl MultiContainsDfa {
         let n_states = accept_state + 1;
         let sentinel = n_states;
 
-        let byte_table = chained_kmp_byte_transitions(segments, accept_state);
+        let byte_table = chained_kmp_byte_transitions(segments, accept_state, case_insensitive);
         let sym_trans =
             build_symbol_transitions(symbols, symbol_lengths, &byte_table, n_states, accept_state);
         let transitions = build_fused_table(&sym_trans, symbols.len(), n_states, |_| sentinel, 0);
@@ -160,8 +161,13 @@ impl MultiContainsDfa {
         // requires every segment to appear in order, so the longest
         // segment's encoded pattern (the most selective single test) is
         // a sound row-level prefilter — rows without it can't match.
-        let escape_only_anchor_pattern =
-            compute_escape_only_anchor(symbols, symbol_lengths, segments);
+        // Case-insensitive patterns disable this fast path; the encoded
+        // pattern is byte-exact and wouldn't match case-flipped bytes.
+        let escape_only_anchor_pattern = if case_insensitive {
+            None
+        } else {
+            compute_escape_only_anchor(symbols, symbol_lengths, segments)
+        };
 
         Ok(Self {
             transitions,
@@ -351,7 +357,11 @@ impl MultiContainsDfa {
 /// - The final phase's accept is the global accept state (sticky)
 ///
 /// Each phase has its own KMP failure function for intra-segment backtracking.
-fn chained_kmp_byte_transitions(segments: &[&[u8]], accept_state: u8) -> Vec<u8> {
+fn chained_kmp_byte_transitions(
+    segments: &[&[u8]],
+    accept_state: u8,
+    case_insensitive: bool,
+) -> Vec<u8> {
     let n_states = accept_state + 1;
     let mut table = vec![0u8; usize::from(n_states) * 256];
 
@@ -366,14 +376,21 @@ fn chained_kmp_byte_transitions(segments: &[&[u8]], accept_state: u8) -> Vec<u8>
 
     for (k, segment) in segments.iter().enumerate() {
         let base = offsets[k];
-        let failure = kmp_failure_table(segment);
+        let failure = kmp_failure_table(segment, case_insensitive);
 
         for local_s in 0..segment.len() {
             let global_s = base + local_s;
             for byte in 0..256usize {
                 let mut s = local_s;
                 loop {
-                    if byte == usize::from(segment[s]) {
+                    let pattern_byte = segment[s];
+                    let matches = if case_insensitive {
+                        super::ascii_to_lower(pattern_byte)
+                            == super::ascii_to_lower(u8::try_from(byte).expect("0..256"))
+                    } else {
+                        byte == usize::from(pattern_byte)
+                    };
+                    if matches {
                         s += 1;
                         break;
                     }