expr, storage: Fix reduce error-propagation, LIKE, and CSV decoder bugs

src/expr/src/scalar.rs

@@ -2475,6 +2475,35 @@ mod tests {
     use super::*;
     use crate::scalar::func::variadic::Coalesce;
 
+    #[mz_ore::test]
+    fn test_reduce_and_or_does_not_propagate_operand_error() {
+        // Regression: AND/OR are non-strict, so a `false`/`true` operand
+        // dominates an erroring operand at runtime (`false AND <error>` is
+        // `false`). `reduce` must therefore not fold `x AND <error>` (or
+        // `x OR <error>`) to the error, which would make a query fail that
+        // otherwise succeeds. Found by the mir_scalar_reduce fuzz target.
+        let bool_typ = ReprScalarType::Bool;
+        let types = vec![bool_typ.clone().nullable(true)];
+        let err = || MirScalarExpr::literal(Err(EvalError::DivisionByZero), bool_typ.clone());
+        let arena = RowArena::new();
+
+        let mut and = MirScalarExpr::call_variadic(And, vec![MirScalarExpr::column(0), err()]);
+        and.reduce(&types);
+        assert!(
+            !and.is_literal_err(),
+            "reduce folded AND with an error operand to an error: {and:?}"
+        );
+        assert_eq!(and.eval(&[Datum::False], &arena), Ok(Datum::False));
+
+        let mut or = MirScalarExpr::call_variadic(Or, vec![MirScalarExpr::column(0), err()]);
+        or.reduce(&types);
+        assert!(
+            !or.is_literal_err(),
+            "reduce folded OR with an error operand to an error: {or:?}"
+        );
+        assert_eq!(or.eval(&[Datum::True], &arena), Ok(Datum::True));
+    }
+
     #[mz_ore::test]
     #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
     fn test_reduce() {

src/expr/src/scalar/like_pattern.rs

@@ -149,7 +149,18 @@ pub fn compile(pattern: &str, case_insensitive: bool) -> Result<Matcher, EvalErr
         return Err(EvalError::LikePatternTooLong);
     }
     let subpatterns = build_subpatterns(pattern)?;
-    let matcher_impl = match case_insensitive || subpatterns.len() > MAX_SUBPATTERNS {
+    // `is_match_subpatterns` resolves each `%` (a `many` subpattern) by searching
+    // for the following suffix and backtracking over every candidate position. A
+    // single `%` is near-linear, but with two or more the backtracking nests and
+    // the cost becomes super-linear in the text length — an adversarial pattern
+    // like `%a%a%a` against a long run of `a`s takes time proportional to
+    // `len(text)^(number of %)`, which can stall a worker for many minutes. The
+    // regex engine matches the same patterns in linear time with no backtracking,
+    // so fall back to it whenever more than one `many` subpattern is present (and
+    // for the existing case-insensitive / too-many-subpatterns reasons).
+    let many_subpatterns = subpatterns.iter().filter(|s| s.many).count();
+    let use_regex = case_insensitive || subpatterns.len() > MAX_SUBPATTERNS || many_subpatterns > 1;
+    let matcher_impl = match use_regex {
         false => MatcherImpl::String(subpatterns),
         true => MatcherImpl::Regex(build_regex(&subpatterns, case_insensitive)?),
     };
@@ -499,4 +510,41 @@ mod test {
             }
         }
     }
+
+    // Patterns with two or more `%` wildcards now compile to the linear regex
+    // matcher instead of the back-tracking string matcher, which was
+    // super-linear: an adversarial pattern like `%a%a%a` against a long run of
+    // `a`s took time proportional to `len(text)^(number of %)`. Verify the
+    // routing change still yields correct match results (the regex matcher and
+    // the string matcher must agree), including the previously-pathological
+    // shape, which now completes instantly regardless of text length.
+    #[mz_ore::test]
+    fn test_many_wildcards_match_correctly() {
+        let long_a = "a".repeat(64);
+        let cases: &[(&str, &str, bool)] = &[
+            ("%a%a%a", "xaxaxa", true),
+            ("%a%a%a", "aaa", true),
+            ("%a%a%a", "aa", false),
+            ("%a%b%c%", "zzabqcz", true),
+            ("%a%b%c%", "cba", false),
+            ("a%b%c", "abc", true),
+            ("a%b%c", "axxbxxc", true),
+            ("a%b%c", "abcd", false),
+            // A single `%` still uses the (near-linear) string matcher.
+            ("%_%_%", "ab", true),
+            ("%%%%", "", true),
+            // The exact pathological shape from the fuzzer; super-linear before
+            // the fix, instant after.
+            ("%a%a%a%a%a", &long_a, true),
+            ("%a%a%a%a%a", "aaaa", false),
+        ];
+        for (pat, text, expected) in cases {
+            let m = compile(pat, false).unwrap();
+            assert_eq!(
+                m.is_match(text),
+                *expected,
+                "pattern {pat:?} against {text:?}"
+            );
+        }
+    }
 }

src/expr/src/scalar/reduce/variadic.rs

@@ -51,9 +51,17 @@ pub(super) fn reduce_call_variadic(
         *e = MirScalarExpr::literal_null(e.typ(column_types).scalar_type);
         return;
     }
-    if let Some(err) = exprs.iter().find_map(|x| x.as_literal_err()) {
-        *e = MirScalarExpr::literal(Err(err.clone()), e.typ(column_types).scalar_type);
-        return;
+    // Only a strict (null-propagating) function propagates an operand's error
+    // unconditionally. A non-strict function such as AND/OR has a dominating
+    // operand (`false`/`true`) that absorbs another operand's error at runtime
+    // — e.g. `false AND <error>` evaluates to `false` — so folding the whole
+    // call to the error here would introduce an error the evaluated expression
+    // never raises. (Coalesce, also non-strict, bailed out above.)
+    if func.propagates_nulls() {
+        if let Some(err) = exprs.iter().find_map(|x| x.as_literal_err()) {
+            *e = MirScalarExpr::literal(Err(err.clone()), e.typ(column_types).scalar_type);
+            return;
+        }
     }
 
     // Per-function dispatch. Arms are mutually exclusive on discriminant; the

src/storage-types/Cargo.toml

@@ -25,6 +25,7 @@ aws-smithy-runtime-api.workspace = true
 aws-types.workspace = true
 bytes.workspace = true
 columnation.workspace = true
+csv-core.workspace = true
 dec.workspace = true
 derivative.workspace = true
 differential-dataflow.workspace = true