Fix: Infer placeholder type from subquery (#22436)

## Which issue does this PR close?

Closes #15979.

## Rationale for this change

`$1 IN (SELECT ...)` left the placeholder untyped because
`infer_placeholder_types` had no arm for `InSubquery` and made
`get_parameter_types()` return `None` for these placeholders.

## What changes are included in this PR?

Adds the `InSubquery` arm to `infer_placeholder_types`, reading the type
from the subquery's projected column. Covers both `IN` and `NOT IN`.

## How are these changes tested?

Unit tests for `IN` and `NOT IN` placeholder inference, plus end-to-end
sqllogictests with `PREPARE`/`EXECUTE`.

## Are there any user-facing changes?

No.

Author: HairstonE

datafusion/expr/src/expr.rs

@@ -2186,6 +2186,32 @@ impl Expr {
                         rewrite_placeholder(item, expr.as_ref(), schema)?;
                     }
                 }
+                Expr::InSubquery(InSubquery {
+                    expr,
+                    subquery,
+                    negated: _,
+                }) => {
+                    let subquery_schema = subquery.subquery.schema();
+                      match &subquery_schema.fields()[..] {
+                          [subquery_field] => {
+                              let column = Expr::Column(Column::new_unqualified(
+                                  subquery_field.name().clone(),
+                              ));
+                              rewrite_placeholder(
+                                  expr.as_mut(),
+                                  &column,
+                                  subquery_schema,
+                              )?;
+                          }
+                          _ => {
+                              return plan_err!(
+                                  "InSubquery should only return one column, but found {}: {}",
+                                  subquery_schema.fields().len(),
+                                  subquery_schema.field_names().join(", ")
+                              );
+                          }
+                      }
+                }
                 Expr::Like(Like { expr, pattern, .. })
                 | Expr::SimilarTo(Like { expr, pattern, .. }) => {
                     rewrite_placeholder(pattern.as_mut(), expr.as_ref(), schema)?;
@@ -3817,6 +3843,108 @@ mod test {
         }
     }
 
+    #[test]
+    fn infer_placeholder_in_subquery() {
+        // WHERE $1 IN (SELECT a FROM t)
+        let subquery_field = Field::new("a", DataType::Int32, false);
+        let subquery_schema = Arc::new(
+            DFSchema::from_unqualified_fields(
+                vec![subquery_field].into(),
+                Default::default(),
+            )
+            .unwrap(),
+        );
+        let subquery = Subquery {
+            subquery: Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
+                produce_one_row: false,
+                schema: subquery_schema,
+            })),
+            outer_ref_columns: vec![],
+            spans: Spans::new(),
+        };
+
+        let in_subquery = Expr::InSubquery(InSubquery {
+            expr: Box::new(Expr::Placeholder(Placeholder {
+                id: "$1".to_string(),
+                field: None,
+            })),
+            subquery,
+            negated: false,
+        });
+
+        let outer_schema = DFSchema::empty();
+        let (inferred_expr, contains_placeholder) =
+            in_subquery.infer_placeholder_types(&outer_schema).unwrap();
+
+        assert!(contains_placeholder);
+
+        match inferred_expr {
+            Expr::InSubquery(in_subquery) => match *in_subquery.expr {
+                Expr::Placeholder(placeholder) => {
+                    let inferred = placeholder.field.expect("placeholder field");
+                    assert_eq!(inferred.data_type(), &DataType::Int32);
+                    assert!(inferred.is_nullable());
+                }
+                _ => panic!("Expected Placeholder expression in InSubquery"),
+            },
+            _ => panic!("Expected InSubquery expression"),
+        }
+    }
+
+    #[test]
+    fn infer_placeholder_not_in_subquery() {
+        // WHERE $1 NOT IN (SELECT a FROM t)
+        let subquery_field = Field::new("a", DataType::Int32, false);
+        let subquery_schema = Arc::new(
+            DFSchema::from_unqualified_fields(
+                vec![subquery_field].into(),
+                Default::default(),
+            )
+            .unwrap(),
+        );
+        let subquery = Subquery {
+            subquery: Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
+                produce_one_row: false,
+                schema: subquery_schema,
+            })),
+            outer_ref_columns: vec![],
+            spans: Spans::new(),
+        };
+
+        let not_in_subquery = Expr::InSubquery(InSubquery {
+            expr: Box::new(Expr::Placeholder(Placeholder {
+                id: "$1".to_string(),
+                field: None,
+            })),
+            subquery,
+            negated: true,
+        });
+
+        let outer_schema = DFSchema::empty();
+        let (inferred_expr, contains_placeholder) = not_in_subquery
+            .infer_placeholder_types(&outer_schema)
+            .unwrap();
+
+        assert!(contains_placeholder);
+
+        match inferred_expr {
+            Expr::InSubquery(in_subquery) => {
+                assert!(in_subquery.negated, "negated flag must be preserved");
+                match *in_subquery.expr {
+                    Expr::Placeholder(placeholder) => {
+                        let inferred = placeholder.field.expect("placeholder field");
+                        assert_eq!(inferred.data_type(), &DataType::Int32);
+                        assert!(inferred.is_nullable());
+                    }
+                    _ => {
+                        panic!("Expected Placeholder expression in InSubquery")
+                    }
+                }
+            }
+            _ => panic!("Expected InSubquery expression"),
+        }
+    }
+
     #[test]
     fn infer_placeholder_like_and_similar_to() {
         // name LIKE $1

datafusion/sqllogictest/test_files/prepare.slt

@@ -107,6 +107,38 @@ EXECUTE my_plan('j%');
 statement ok
 DEALLOCATE my_plan
 
+# Allow prepare $1 IN (subquery)
+statement ok
+PREPARE my_plan AS SELECT id FROM person WHERE $1 IN (SELECT age FROM person);
+
+query I rowsort
+EXECUTE my_plan(20);
+----
+1
+
+query I rowsort
+EXECUTE my_plan(99);
+----
+
+statement ok
+DEALLOCATE my_plan
+
+# Allow prepare $1 NOT IN (subquery)
+statement ok
+PREPARE my_plan AS SELECT id FROM person WHERE $1 NOT IN (SELECT age FROM person);
+
+query I rowsort
+EXECUTE my_plan(99);
+----
+1
+
+query I rowsort
+EXECUTE my_plan(20);
+----
+
+statement ok
+DEALLOCATE my_plan
+
 # Check for missing parameters
 statement ok
 PREPARE my_plan AS SELECT * FROM person WHERE id < $1;