fix(optimizer): bound planning-time regex compilation

datafusion/core/tests/optimizer/mod.rs

@@ -27,7 +27,9 @@ use arrow::datatypes::{
 };
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::TransformedResult;
-use datafusion_common::{DFSchema, Result, ScalarValue, TableReference, plan_err};
+use datafusion_common::{
+    DFSchema, Result, ScalarValue, TableReference, assert_contains, plan_err,
+};
 use datafusion_expr::interval_arithmetic::{Interval, NullableInterval};
 use datafusion_expr::{
     AggregateUDF, BinaryExpr, Expr, ExprSchemable, HigherOrderUDF, LogicalPlan, Operator,
@@ -136,6 +138,23 @@ fn concat_ws_literals() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn expensive_regexp_like_is_deferred_to_execution() -> Result<()> {
+    let pattern = "a{5}{5}{5}{5}{5}{5}{5}{5}";
+    let plan = test_sql(&format!("SELECT regexp_like('aaaaa', '{pattern}')"))?;
+    let plan = plan.display_indent().to_string();
+
+    assert_contains!(&plan, &format!("Utf8(\"{pattern}\")"));
+    assert_contains!(&plan, " ~ ");
+
+    let null_plan = test_sql(&format!("SELECT regexp_like(NULL, '{pattern}')"))?;
+    assert_contains!(
+        &null_plan.display_indent().to_string(),
+        "Projection: Boolean(NULL)"
+    );
+    Ok(())
+}
+
 fn test_sql(sql: &str) -> Result<LogicalPlan> {
     // parse the SQL
     let dialect = GenericDialect {}; // or AnsiDialect, or your own dialect ...
@@ -147,6 +166,7 @@ fn test_sql(sql: &str) -> Result<LogicalPlan> {
     let context_provider = MyContextProvider::default()
         .with_udf(datetime::now(&config))
         .with_udf(datafusion_functions::core::arrow_cast())
+        .with_udf(datafusion_functions::regex::regexp_like())
         .with_udf(datafusion_functions::string::concat())
         .with_udf(datafusion_functions::string::concat_ws());
     let sql_to_rel = SqlToRel::new(&context_provider);

datafusion/expr/src/simplify.rs

@@ -189,6 +189,11 @@ pub enum ExprSimplifyResult {
     Original(Vec<Expr>),
 }
 
+/// Maximum compiled regex size allowed during constant evaluation in the
+/// planner. Larger regexes are deferred to execution so that planning cannot
+/// spend the runtime regex engine's substantially larger compilation budget.
+pub const REGEX_PLANNING_SIZE_LIMIT_BYTES: usize = 256 * 1024;
+
 #[cfg(test)]
 mod tests {
     use super::*;

datafusion/expr/src/udf.rs

@@ -232,6 +232,12 @@ impl ScalarUDF {
         self.inner.return_field_from_args(args)
     }
 
+    /// Returns whether this scalar function should be evaluated during
+    /// constant folding for the specified literal arguments.
+    pub fn should_evaluate_const(&self, args: &[&ScalarValue]) -> bool {
+        self.inner.should_evaluate_const(args)
+    }
+
     /// Returns this scalar function's simplification result.
     ///
     /// See [`ScalarUDFImpl::simplify`] for more details.
@@ -690,6 +696,17 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync + Any {
     /// to arrays, which will likely be simpler code, but be slower.
     fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue>;
 
+    /// Returns whether this function should be evaluated during constant
+    /// folding for the specified literal arguments.
+    ///
+    /// Implementations can return `false` when evaluation is valid but may be
+    /// too expensive for planning. The expression is then preserved for
+    /// runtime evaluation. The default preserves the existing behavior of
+    /// evaluating immutable functions with literal arguments.
+    fn should_evaluate_const(&self, _args: &[&ScalarValue]) -> bool {
+        true
+    }
+
     /// Optionally apply per-UDF simplification / rewrite rules.
     ///
     /// This can be used to apply function specific simplification rules during
@@ -1094,6 +1111,10 @@ impl ScalarUDFImpl for AliasedScalarUDFImpl {
         self.inner.invoke_with_args(args)
     }
 
+    fn should_evaluate_const(&self, args: &[&ScalarValue]) -> bool {
+        self.inner.should_evaluate_const(args)
+    }
+
     fn with_updated_config(&self, _config: &ConfigOptions) -> Option<ScalarUDF> {
         None
     }

datafusion/functions/src/regex/regexplike.rs

@@ -31,10 +31,13 @@ use datafusion_expr::{
 };
 use datafusion_macros::user_doc;
 
-use datafusion_expr::simplify::{ExprSimplifyResult, SimplifyContext};
+use datafusion_expr::simplify::{
+    ExprSimplifyResult, REGEX_PLANNING_SIZE_LIMIT_BYTES, SimplifyContext,
+};
 use datafusion_expr_common::operator::Operator;
 use datafusion_expr_common::type_coercion::binary::BinaryTypeCoercer;
-use regex::Regex;
+use regex::{Error as RegexError, Regex, RegexBuilder};
+use std::borrow::Cow;
 use std::sync::Arc;
 
 #[user_doc(
@@ -160,6 +163,35 @@ impl ScalarUDFImpl for RegexpLikeFunc {
         }
     }
 
+    fn should_evaluate_const(&self, args: &[&ScalarValue]) -> bool {
+        let Some(pattern) = args.get(1).and_then(|arg| arg.try_as_str()).flatten() else {
+            return true;
+        };
+        let flags = args.get(2).and_then(|arg| arg.try_as_str()).flatten();
+
+        // Preserve normal error handling for unsupported and malformed flags.
+        if flags.is_some_and(|flags| flags.contains('g')) {
+            return true;
+        }
+
+        // regexp_like returns NULL without compiling the pattern when the
+        // value is NULL. Preserve that fast path during constant folding.
+        if args.first().is_some_and(|arg| arg.is_null()) {
+            return true;
+        }
+
+        let pattern = match flags {
+            Some(flags) => Cow::Owned(format!("(?{flags}){pattern}")),
+            None => Cow::Borrowed(pattern),
+        };
+        !matches!(
+            RegexBuilder::new(pattern.as_ref())
+                .size_limit(REGEX_PLANNING_SIZE_LIMIT_BYTES)
+                .build(),
+            Err(RegexError::CompiledTooBig(_))
+        )
+    }
+
     fn simplify(
         &self,
         mut args: Vec<Expr>,
@@ -627,6 +659,39 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_const_evaluation_budget() {
+        let function = RegexpLikeFunc::new();
+        let args = |pattern: &str, flags: Option<&str>| {
+            let mut args = vec![
+                ScalarValue::Utf8(Some("aaaaa".to_string())),
+                ScalarValue::Utf8(Some(pattern.to_string())),
+            ];
+            if let Some(flags) = flags {
+                args.push(ScalarValue::Utf8(Some(flags.to_string())));
+            }
+            args
+        };
+        let should_evaluate = |args: Vec<ScalarValue>| {
+            let args = args.iter().collect::<Vec<_>>();
+            function.should_evaluate_const(&args)
+        };
+
+        assert!(should_evaluate(args("^a+$", None)));
+        assert!(!should_evaluate(args("a{5}{5}{5}{5}{5}{5}", None)));
+        assert!(!should_evaluate(args("a{5}{5}{5}{5}{5}{5}", Some("m"))));
+
+        let null_value = vec![
+            ScalarValue::Utf8(None),
+            ScalarValue::Utf8(Some("a{5}{5}{5}{5}{5}{5}{5}{5}".to_string())),
+        ];
+        assert!(should_evaluate(null_value));
+
+        // Unsupported flags and syntax errors retain their existing error paths.
+        assert!(should_evaluate(args("^a+$", Some("g"))));
+        assert!(should_evaluate(args("[", None)));
+    }
+
     #[test]
     fn test_regexp_like_array_scalar_invoke() {
         let values = Arc::new(StringArray::from(vec!["abc", "xyz"]));

datafusion/optimizer/src/simplify_expressions/expr_simplifier.rs

@@ -45,7 +45,10 @@ use datafusion_expr::{
     BinaryExpr, Case, ColumnarValue, Expr, ExprSchemable, Like, Operator, Volatility,
     and, binary::BinaryTypeCoercer, lit, or, preimage::PreimageResult,
 };
-use datafusion_expr::{Cast, TryCast, simplify::ExprSimplifyResult};
+use datafusion_expr::{
+    Cast, TryCast,
+    simplify::{ExprSimplifyResult, REGEX_PLANNING_SIZE_LIMIT_BYTES},
+};
 use datafusion_expr::{expr::ScalarFunction, interval_arithmetic::NullableInterval};
 use datafusion_expr::{
     expr::{InList, InSubquery},
@@ -69,7 +72,7 @@ use crate::{
 use datafusion_expr::expr_rewriter::rewrite_with_guarantees_map;
 use datafusion_expr_common::casts::try_cast_literal_to_type;
 use indexmap::IndexSet;
-use regex::Regex;
+use regex::{Error as RegexError, Regex, RegexBuilder};
 
 /// This structure handles API for expression simplification
 ///
@@ -494,6 +497,8 @@ enum ConstSimplifyResult {
     NotSimplified(ScalarValue, Option<FieldMetadata>),
     // Evaluation encountered an error, contains the original expression
     SimplifyRuntimeError(DataFusionError, Expr),
+    // Evaluation was deliberately deferred to runtime
+    Deferred(Expr),
 }
 
 impl TreeNodeRewriter for ConstEvaluator {
@@ -552,6 +557,14 @@ impl TreeNodeRewriter for ConstEvaluator {
                     // to allow short-circuit evaluation at execution time
                     Ok(Transformed::yes(expr))
                 }
+                ConstSimplifyResult::Deferred(expr) => {
+                    // Prevent an evaluatable parent (for example, an Alias)
+                    // from evaluating this deferred subtree indirectly.
+                    self.can_evaluate.iter_mut().for_each(|can_evaluate| {
+                        *can_evaluate = false;
+                    });
+                    Ok(Transformed::no(expr))
+                }
             },
             Some(false) => Ok(Transformed::no(expr)),
             _ => internal_err!("Failed to pop can_evaluate"),
@@ -691,6 +704,10 @@ impl ConstEvaluator {
             return ConstSimplifyResult::NotSimplified(s, m);
         }
 
+        if !should_evaluate_const_expr(&expr) {
+            return ConstSimplifyResult::Deferred(expr);
+        }
+
         let phys_expr = match create_physical_expr(
             &expr,
             &DUMMY_DF_SCHEMA,
@@ -747,6 +764,51 @@ impl ConstEvaluator {
     }
 }
 
+fn should_evaluate_const_expr(expr: &Expr) -> bool {
+    match expr {
+        Expr::ScalarFunction(ScalarFunction { func, args }) => {
+            let Some(args) = args
+                .iter()
+                .map(|arg| match arg {
+                    Expr::Literal(value, _) => Some(value),
+                    _ => None,
+                })
+                .collect::<Option<Vec<_>>>()
+            else {
+                return true;
+            };
+            func.should_evaluate_const(&args)
+        }
+        Expr::BinaryExpr(BinaryExpr { op, right, .. })
+            if matches!(
+                op,
+                Operator::RegexMatch
+                    | Operator::RegexNotMatch
+                    | Operator::RegexIMatch
+                    | Operator::RegexNotIMatch
+            ) =>
+        {
+            regex_is_within_planning_budget(op, right)
+        }
+        _ => true,
+    }
+}
+
+fn regex_is_within_planning_budget(op: &Operator, pattern: &Expr) -> bool {
+    let Expr::Literal(pattern, _) = pattern else {
+        return true;
+    };
+    let Some(pattern) = pattern.try_as_str().flatten() else {
+        return true;
+    };
+    let mut builder = RegexBuilder::new(pattern);
+    builder.size_limit(REGEX_PLANNING_SIZE_LIMIT_BYTES);
+    if matches!(op, Operator::RegexIMatch | Operator::RegexNotIMatch) {
+        builder.case_insensitive(true);
+    }
+    !matches!(builder.build(), Err(RegexError::CompiledTooBig(_)))
+}
+
 /// Simplifies [`Expr`]s by applying algebraic transformation rules
 ///
 /// Example transformations that are applied:
@@ -1649,7 +1711,17 @@ impl TreeNodeRewriter for Simplifier<'_> {
                 left,
                 op: op @ (RegexMatch | RegexNotMatch | RegexIMatch | RegexNotIMatch),
                 right,
-            }) => simplify_regex_expr(left, op, right)?,
+            }) => {
+                // Non-constant regexes are not evaluated during planning, so
+                // avoid adding a compilation preflight to their normal path.
+                if !matches!(left.as_ref(), Expr::Literal(_, _))
+                    || regex_is_within_planning_budget(&op, &right)
+                {
+                    simplify_regex_expr(left, op, right)?
+                } else {
+                    Transformed::no(Expr::BinaryExpr(BinaryExpr { left, op, right }))
+                }
+            }
 
             // Rules for Like
             Expr::Like(like) => {
@@ -3458,6 +3530,60 @@ mod tests {
         assert_eq!(simplify(expr_eq), lit(true));
     }
 
+    #[test]
+    fn test_constant_regex_respects_planning_budget() {
+        // Small regexes retain the common-case constant folding behavior.
+        assert_eq!(simplify(regex_match(lit("aaaaa"), lit("^a+$"))), lit(true));
+
+        // Six nested repetitions fit under the runtime regex limit, but exceed
+        // the smaller planning budget and therefore remain for execution.
+        let pattern = "a{5}{5}{5}{5}{5}{5}";
+        let binary_expr = regex_match(lit("aaaaa"), lit(pattern));
+        assert_eq!(simplify(binary_expr.clone()), binary_expr);
+
+        // Deferring a subtree must also prevent an evaluatable parent from
+        // invoking it indirectly during constant folding.
+        let aliased = regex_match(lit("aaaaa"), lit(pattern)).alias("matched");
+        assert_eq!(simplify(aliased.clone()), aliased);
+
+        #[derive(Debug, PartialEq, Eq, Hash)]
+        struct DeferredConstUdf {
+            signature: Signature,
+        }
+
+        impl ScalarUDFImpl for DeferredConstUdf {
+            fn name(&self) -> &str {
+                "deferred_const"
+            }
+
+            fn signature(&self) -> &Signature {
+                &self.signature
+            }
+
+            fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+                Ok(DataType::Boolean)
+            }
+
+            fn invoke_with_args(
+                &self,
+                _args: ScalarFunctionArgs,
+            ) -> Result<ColumnarValue> {
+                panic!("deferred UDF must not be evaluated during planning")
+            }
+
+            fn should_evaluate_const(&self, _args: &[&ScalarValue]) -> bool {
+                false
+            }
+        }
+
+        // UDFs can use the same hook when their literal inputs are expensive.
+        let udf = ScalarUDF::new_from_impl(DeferredConstUdf {
+            signature: Signature::exact(vec![DataType::Utf8], Volatility::Immutable),
+        });
+        let udf_expr = udf.call(vec![lit("value")]);
+        assert_eq!(simplify(udf_expr.clone()), udf_expr);
+    }
+
     #[test]
     fn test_simplify_regex() {
         // malformed regex

datafusion/physical-expr/Cargo.toml

@@ -64,6 +64,7 @@ itertools = { workspace = true, features = ["use_std"] }
 parking_lot = { workspace = true }
 petgraph = "0.8.3"
 recursive = { workspace = true, optional = true }
+regex = { workspace = true }
 tokio = { workspace = true }
 half = { workspace = true }