risingwave_expr/scalar/
like.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::borrow::Cow;
16
17use const_currying::const_currying;
18use itertools::Itertools;
19use risingwave_common::util::recursive::{self, Recurse};
20use risingwave_expr::function;
21
22use crate::{ExprError, Result};
23
24/// Converts a SQL `LIKE` pattern to the internal backslash escape convention.
25///
26/// This mirrors PostgreSQL's `do_like_escape`: `ESCAPE ''` disables escaping and
27/// doubles backslashes, `ESCAPE '\'` leaves the pattern unchanged, and any other
28/// escape character is rewritten to the internal `\` escape marker.
29fn normalize_pattern(p: &str, escape: Option<u8>) -> Cow<'_, [u8]> {
30    if escape == Some(b'\\') {
31        return Cow::Borrowed(p.as_bytes());
32    }
33
34    let mut normalized = Vec::with_capacity(p.len());
35    let mut after_escape = false;
36
37    for c in p.bytes() {
38        if Some(c) == escape && !after_escape {
39            normalized.push(b'\\');
40            after_escape = true;
41        } else if c == b'\\' {
42            normalized.push(b'\\');
43            if !after_escape {
44                normalized.push(b'\\');
45            }
46            after_escape = false;
47        } else {
48            normalized.push(c);
49            after_escape = false;
50        }
51    }
52
53    Cow::Owned(normalized)
54}
55
56fn like_impl_inner<const CASE_INSENSITIVE: bool>(
57    s: &str,
58    p: &str,
59    escape: Option<u8>,
60) -> Result<bool> {
61    let pattern = normalize_pattern(p, escape);
62    Ok(match_text::<CASE_INSENSITIVE>(s.as_bytes(), pattern.as_ref())?.into_bool())
63}
64
65#[const_currying]
66fn like_impl<const CASE_INSENSITIVE: bool>(
67    s: &str,
68    p: &str,
69    #[maybe_const(consts = [b'\\'])] escape: u8,
70) -> Result<bool> {
71    like_impl_inner::<CASE_INSENSITIVE>(s, p, Some(escape))
72}
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub enum MatchResult {
76    True,
77    False,
78    Abort,
79}
80
81// PostgreSQL checks the remaining native stack space with `check_stack_depth`.
82// Use a conservative recursion limit because Rust does not expose an equivalent
83// portable stack-depth check.
84const MAX_RECURSION_DEPTH: usize = 4096;
85
86impl MatchResult {
87    #[inline]
88    fn into_bool(self) -> bool {
89        matches!(self, MatchResult::True)
90    }
91}
92
93#[inline]
94fn normalize<const CASE_INSENSITIVE: bool>(b: u8) -> u8 {
95    if CASE_INSENSITIVE {
96        b.to_ascii_lowercase()
97    } else {
98        b
99    }
100}
101
102#[inline]
103fn byte_eq<const CASE_INSENSITIVE: bool>(a: u8, b: u8) -> bool {
104    normalize::<CASE_INSENSITIVE>(a) == normalize::<CASE_INSENSITIVE>(b)
105}
106
107pub fn match_text<const CASE_INSENSITIVE: bool>(
108    text: &[u8],
109    pattern: &[u8],
110) -> Result<MatchResult, ExprError> {
111    recursive::tracker!().recurse(|tracker| {
112        if tracker.depth() > MAX_RECURSION_DEPTH {
113            return Err(ExprError::InvalidParam {
114                name: "pattern",
115                reason: "stack depth limit exceeded".into(),
116            });
117        }
118
119        match_text_inner::<CASE_INSENSITIVE>(text, pattern)
120    })
121}
122
123fn match_text_inner<const CASE_INSENSITIVE: bool>(
124    mut text: &[u8],
125    mut pattern: &[u8],
126) -> Result<MatchResult, ExprError> {
127    // Fast path for match-everything pattern
128    if pattern == b"%" {
129        return Ok(MatchResult::True);
130    }
131
132    while let (Some((&p, rest_pattern)), Some((&t, rest_text))) =
133        (pattern.split_first(), text.split_first())
134    {
135        match p {
136            b'%' => {
137                pattern = rest_pattern;
138
139                while let Some((&p, rest_pattern)) = pattern.split_first() {
140                    match p {
141                        b'%' => {
142                            // N's % == one's %
143                            pattern = rest_pattern;
144                        }
145                        b'_' => {
146                            let Some((_, rest_text)) = text.split_first() else {
147                                // If not enough text left to match the pattern, ABORT
148                                return Ok(MatchResult::Abort);
149                            };
150
151                            text = rest_text;
152                            pattern = rest_pattern;
153                        }
154                        _ => break, // Reached a non-wildcard pattern char
155                    }
156                }
157
158                // If we're at end of pattern, match: we have a trailing % which
159                // matches any remaining text string.
160                let Some((&first, _)) = pattern.split_first() else {
161                    return Ok(MatchResult::True);
162                };
163
164                // Get the first literal character in the remaining pattern for fast text scanning.
165                let first = if first == b'\\' {
166                    let Some(&escaped) = pattern.get(1) else {
167                        return Err(ExprError::InvalidParam {
168                            name: "pattern",
169                            reason: "LIKE pattern must not end with escape character".into(),
170                        });
171                    };
172                    escaped
173                } else {
174                    first
175                };
176
177                while let Some((&t, rest_text)) = text.split_first() {
178                    if byte_eq::<CASE_INSENSITIVE>(t, first) {
179                        let matched = match_text::<CASE_INSENSITIVE>(text, pattern)?;
180
181                        if matched != MatchResult::False {
182                            return Ok(matched); /* TRUE or ABORT */
183                        }
184                    }
185
186                    text = rest_text;
187                }
188
189                // End of text with no match, so no point in trying later places
190                // to start matching this pattern.
191                return Ok(MatchResult::Abort);
192            }
193            b'_' => {
194                text = rest_text;
195                pattern = rest_pattern;
196            }
197            b'\\' => {
198                let Some((&escaped, rest_pattern)) = rest_pattern.split_first() else {
199                    return Err(ExprError::InvalidParam {
200                        name: "pattern",
201                        reason: "LIKE pattern must not end with escape character".into(),
202                    });
203                };
204                if !byte_eq::<CASE_INSENSITIVE>(t, escaped) {
205                    return Ok(MatchResult::False);
206                }
207                text = rest_text;
208                pattern = rest_pattern;
209            }
210            literal => {
211                if !byte_eq::<CASE_INSENSITIVE>(t, literal) {
212                    // non-wildcard pattern char fails to match text char
213                    return Ok(MatchResult::False);
214                }
215
216                text = rest_text;
217                pattern = rest_pattern;
218            }
219        }
220    }
221
222    if !text.is_empty() {
223        // end of pattern, but not of text
224        return Ok(MatchResult::False);
225    }
226
227    // End of text, but perhaps not of pattern.
228    // Match iff the remaining pattern can match a zero-length string,
229    // ie, it's zero or more %'s.
230    while pattern.first() == Some(&b'%') {
231        pattern = &pattern[1..];
232    }
233
234    // End of text with no match, so no point in trying later places to start
235    // matching this pattern.
236    if pattern.is_empty() {
237        Ok(MatchResult::True)
238    } else {
239        Ok(MatchResult::Abort)
240    }
241}
242
243#[function("like(varchar, varchar) -> boolean")]
244pub fn like_default(s: &str, p: &str) -> Result<bool> {
245    like_impl_escape::<false, b'\\'>(s, p)
246}
247
248#[function("i_like(varchar, varchar) -> boolean")]
249pub fn i_like_default(s: &str, p: &str) -> Result<bool> {
250    like_impl_escape::<true, b'\\'>(s, p)
251}
252
253#[function(
254    "like(varchar, varchar, varchar) -> boolean",
255    prebuild = "EscapeChar::from_str($2)?"
256)]
257fn like(s: &str, p: &str, escape: &EscapeChar) -> Result<bool> {
258    like_impl_inner::<false>(s, p, escape.0)
259}
260
261#[function(
262    "i_like(varchar, varchar, varchar) -> boolean",
263    prebuild = "EscapeChar::from_str($2)?"
264)]
265fn i_like(s: &str, p: &str, escape: &EscapeChar) -> Result<bool> {
266    like_impl_inner::<true>(s, p, escape.0)
267}
268
269// TODO: We should support any UTF-8 character as escape character.
270#[derive(Copy, Clone, Debug)]
271struct EscapeChar(Option<u8>);
272
273impl EscapeChar {
274    fn from_str(escape: &str) -> Result<Self> {
275        if escape.is_empty() {
276            return Ok(Self(None));
277        }
278
279        Itertools::exactly_one(escape.chars())
280            .ok()
281            .and_then(|c| c.as_ascii().map(|c| c.to_u8()))
282            .map(Some)
283            .ok_or_else(|| ExprError::InvalidParam {
284                name: "escape",
285                reason: "only empty or single ascii character is supported now".into(),
286            })
287            .map(Self)
288    }
289}
290
291#[cfg(test)]
292mod tests {
293    use risingwave_expr::scalar::like::EscapeChar;
294
295    use super::{
296        MAX_RECURSION_DEPTH, i_like, i_like_default, like, like_default, normalize_pattern,
297    };
298
299    static CASES: &[(&str, &str, bool, bool)] = &[
300        (r#"ABCDE"#, r#"%abcde%"#, false, false),
301        (r#"Like, expression"#, r#"Like, expression"#, false, true),
302        (r#"Like, expression"#, r#"Like, %"#, false, true),
303        (r#"Like, expression"#, r#"%, expression"#, false, true),
304        (r#"like"#, r#"li%ke"#, false, true),
305        (r#"like"#, r#"l%ik%e"#, false, true),
306        (r#"like"#, r#"%like%"#, false, true),
307        (r#"like"#, r#"l%i%k%e%"#, false, true),
308        (r#"like"#, r#"_%_e"#, false, true),
309        (r#"like"#, r#"l%__"#, false, true),
310        (r#"like"#, r#"_%_%_%_"#, false, true),
311        (r#"abctest"#, r#"__test"#, false, false),
312        (r#"abctest"#, r#"%_test"#, false, true),
313        (r#"aaaaabbb"#, r#"a%a%a%a%a%a%b"#, false, false),
314        (
315            r#"blush thistle blue yellow saddle"#,
316            r#"%yellow%"#,
317            false,
318            true,
319        ),
320        (r#"ABC_123"#, r#"ABC_123"#, false, true),
321        (r#"ABCD123"#, r#"ABC_123"#, false, true),
322        (r#"ABC_123"#, r"ABC\_123", false, true),
323        (r#"ABCD123"#, r"ABC\_123", false, false),
324        (r"ABC\123", r#"ABC_123"#, false, true),
325        (r"ABC\123", r"ABC\\123", false, true),
326        (r"ABC\123", r"ABC\123", false, false),
327        ("apple", r#"App%"#, true, true),
328        ("banana", r#"B%nana"#, true, true),
329        ("apple", r#"B%nana"#, true, false),
330        ("grape", "Gr_P_", true, true),
331    ];
332
333    #[test]
334    fn test_like() {
335        for (target, pattern, case_insensitive, expected) in CASES {
336            let output = if *case_insensitive {
337                i_like_default(target, pattern)
338            } else {
339                like_default(target, pattern)
340            };
341            assert!(output.is_ok());
342            assert_eq!(
343                output.unwrap(),
344                *expected,
345                "target={}, pattern={}, case_insensitive={}",
346                target,
347                pattern,
348                case_insensitive
349            );
350        }
351    }
352
353    fn assert_normalized_pattern(pattern: &str, escape: &str, expected: &str) {
354        let escape = EscapeChar::from_str(escape).unwrap();
355        let normalized = normalize_pattern(pattern, escape.0);
356        assert_eq!(
357            normalized,
358            expected.as_bytes(),
359            "pattern={pattern}, escape={escape:?}"
360        );
361    }
362
363    #[test]
364    fn test_normalize_pattern() {
365        let testcases = [
366            // Default PostgreSQL escape behavior is already the internal representation.
367            (r"ABC\_123", r"\", r"ABC\_123"),
368            (r"ABC\\123", r"\", r"ABC\\123"),
369            // Unused escape characters keep wildcard semantics.
370            ("h%", "#", "h%"),
371            ("ind_o", "$", "ind_o"),
372            // Escaped percent and underscore become internal backslash escapes.
373            ("h#%", "#", r"h\%"),
374            ("h#%%", "#", r"h\%%"),
375            ("i$_d_o", "$", r"i\_d_o"),
376            ("i$_d%o", "$", r"i\_d%o"),
377            // PostgreSQL regression cases where the escape is also a wildcard.
378            ("m%aca", "%", r"m\aca"),
379            ("m%a%%a", "%", r"m\a\%a"),
380            ("b_ear", "_", r"b\ear"),
381            ("b_e__r", "_", r"b\e\_r"),
382            ("__e__r", "_", r"\_e\_r"),
383            ("____r", "_", r"\_\_r"),
384            // ESCAPE '' disables escaping, including backslash escaping.
385            ("a_c", "", "a_c"),
386            (r"a\_c", "", r"a\\_c"),
387        ];
388
389        for (pattern, escape, expected) in testcases {
390            assert_normalized_pattern(pattern, escape, expected);
391        }
392    }
393
394    #[test]
395    fn test_normalize_pattern_ends_with_escape() {
396        let testcases = [
397            (r"abc\", r"\", r"abc\"),
398            ("h#", "#", r"h\"),
399            ("_____", "_", r"\_\_\"),
400            ("m%", "%", r"m\"),
401        ];
402
403        for (pattern, escape, expected) in testcases {
404            assert_normalized_pattern(pattern, escape, expected);
405        }
406
407        assert_normalized_pattern(r"abc\", "", r"abc\\");
408    }
409
410    static ESCAPE_CASES: &[(&str, &str, &str, bool)] = &[
411        (r"bear", r"b_ear", r"_", true),
412        (r"be_r", r"b_e__r", r"_", true),
413        (r"be__r", r"b_e___r", r"_", false),
414        (r"be__r", r"b_e____r", r"_", true),
415        (r"be___r", r"b_e_____r", r"_", false),
416        (r"be___r", r"b_e______r", r"_", true),
417        (r"be_r", r"__e__r", r"_", false),
418        (r"___r", r"____r", r"_", false),
419        (r"__r", r"____r", r"_", true),
420        (r"maca", r"m%aca", r"%", true),
421        (r"ma%a", r"m%a%%a", r"%", true),
422        (r"abc", r"a_c", r"", true),
423        (r"a_c", r"a\_c", r"", false),
424        (r"a\_c", r"a\_c", r"", true),
425        (r"abc", r"abc\", r"\", false),
426        (r"h", r"h#", "#", false),
427        (r"__", r"_____", "_", false),
428        (r"m", r"m%", "%", false),
429    ];
430
431    #[test]
432    fn test_escape_like() {
433        for (target, pattern, escape, expected) in ESCAPE_CASES {
434            let output = like(target, pattern, &EscapeChar::from_str(escape).unwrap());
435            assert!(output.is_ok());
436            assert_eq!(
437                output.unwrap(),
438                *expected,
439                "target={}, pattern={}, escape={}",
440                target,
441                pattern,
442                escape
443            );
444        }
445    }
446
447    #[test]
448    fn test_like_pattern_ends_with_escape() {
449        let escape = EscapeChar::from_str("_").unwrap();
450        assert!(like("____", "_____", &escape).is_err());
451        assert!(i_like("____", "_____", &escape).is_err());
452    }
453
454    #[test]
455    fn test_like_recursion_depth_limit() {
456        let text = "a".repeat(MAX_RECURSION_DEPTH - 1);
457        let pattern = "%a".repeat(MAX_RECURSION_DEPTH - 1);
458        assert!(like_default(&text, &pattern).unwrap());
459
460        let text = "a".repeat(MAX_RECURSION_DEPTH);
461        let pattern = "%a".repeat(MAX_RECURSION_DEPTH);
462
463        assert!(like_default(&text, &pattern).is_err());
464    }
465}
risingwave_expr/scalar/like.rs

risingwave_expr/scalar/
like.rs
