Add recursive protection on planner's create_physical_expr (#19299)

## Which issue does this PR close?

None, as this PR is rather self-describing.

## Rationale for this change

We recently encountered stack overflow issues in the physical planner's
`create_physical_expr` function, when dealing with deeply nested
expression trees.

In the rest of the codebase, it seems the `recursive::recursive`
attribute is used to prevent this.

## What changes are included in this PR?

This PR adds the `recursive::recursive` attribute on
`create_physical_expr` whenever the `recursive_protection` feature is
enabled.

## Are these changes tested?

Yes, but mostly for the sake of manually testing the change.

I don't think it should be merged, as it has to be `#[ignore]`d anyway,
as the `recursive_protection` is not enabled in tests (which means it
would systematically crash).

I've left it in this PR however, to make review / manual testing easier.

## Are there any user-facing changes?

None.

---------

Co-authored-by: Andrew Lamb <[email protected]>

Author: rgehan

Cargo.lock

@@ -87,6 +87,7 @@ recursive_protection = [
     "datafusion-expr/recursive_protection",
     "datafusion-optimizer/recursive_protection",
     "datafusion-physical-optimizer/recursive_protection",
+    "datafusion-physical-expr/recursive_protection",
     "datafusion-sql/recursive_protection",
     "sqlparser/recursive-protection",
 ]

datafusion/core/Cargo.toml

@@ -40,6 +40,9 @@ workspace = true
 [lib]
 name = "datafusion_physical_expr"
 
+[features]
+recursive_protection = ["dep:recursive"]
+
 [dependencies]
 ahash = { workspace = true }
 arrow = { workspace = true }
@@ -54,6 +57,7 @@ itertools = { workspace = true, features = ["use_std"] }
 parking_lot = { workspace = true }
 paste = { workspace = true }
 petgraph = "0.8.3"
+recursive = { workspace = true, optional = true }
 tokio = { workspace = true }
 half = { workspace = true }
 

datafusion/physical-expr/Cargo.toml

@@ -105,6 +105,7 @@ use datafusion_expr::{
 /// * `e` - The logical expression
 /// * `input_dfschema` - The DataFusion schema for the input, used to resolve `Column` references
 ///   to qualified or unqualified fields by name.
+#[cfg_attr(feature = "recursive_protection", recursive::recursive)]
 pub fn create_physical_expr(
     e: &Expr,
     input_dfschema: &DFSchema,
@@ -417,7 +418,7 @@ mod tests {
     use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
     use arrow::datatypes::{DataType, Field};
 
-    use datafusion_expr::{col, lit};
+    use datafusion_expr::{Operator, col, lit};
 
     use super::*;
 
@@ -445,4 +446,34 @@ mod tests {
 
         Ok(())
     }
+
+    /// Test that deeply nested expressions do not cause a stack overflow.
+    ///
+    /// This test only runs when the `recursive_protection` feature is enabled,
+    /// as it would overflow the stack otherwise.
+    #[test]
+    #[cfg_attr(not(feature = "recursive_protection"), ignore)]
+    fn test_deeply_nested_binary_expr() -> Result<()> {
+        // Create a deeply nested binary expression tree: ((((a + a) + a) + a) + ... )
+        // With 1000 levels of nesting, this would overflow the stack without recursion protection.
+        let depth = 1000;
+
+        let mut expr = col("a");
+        for _ in 0..depth {
+            expr = Expr::BinaryExpr(BinaryExpr {
+                left: Box::new(expr),
+                op: Operator::Plus,
+                right: Box::new(col("a")),
+            });
+        }
+
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema)?;
+
+        // This should not stack overflow
+        let _physical_expr =
+            create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
+
+        Ok(())
+    }
 }