refactor: port InListExpr to use try_to_proto/try_from_proto hooks (#22503)

## Which issue does this PR close?

- Closes #22425 

## Rationale for this change

This PR migrates InListExpr to use the new try_to_proto and
try_from_proto hooks, following the modular serialization architecture
established in #21835 . This moves serialization logic into the
expression itself, improving encapsulation and decentralizing the
datafusion-proto logic as part of the broader effort in #22418

## What changes are included in this PR?

 - Implemented `PhysicalExpr::try_to_proto for InListExpr`.
 - Implemented `InListExpr::try_from_proto` inherent method.
- Wired hooks in `from_proto.rs` and removed the central downcast arm in
to_proto.rs.
- Added isolated unit tests in `in_list.rs` using mock drivers to verify
roundtrips and error handling.


## Are these changes tested?

Yes, these changes are covered by both new and existing tests:

- New Unit Tests: Added mod `proto_tests` to in_list.rs using mock
drivers to verify `try_to_proto` and `try_from_proto` in isolation.
These cover successful roundtrips, incorrect node types, and missing
required fields.
- Existing Integration Tests: Verified that the existing InList
roundtrip tests in datafusion-proto continue to pass after removing the
central downcast logic.
- Linting: Verified that the changes pass cargo `clippy --all-targets
--all-features` with zero warnings.

## Are there any user-facing changes?

No.

Author: kkrainov

datafusion/physical-expr/src/expressions/in_list.rs

@@ -246,6 +246,42 @@ impl InListExpr {
 
         Ok(Self::new(expr, list, negated, static_filter))
     }
+
+    #[cfg(feature = "proto")]
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_proto_models::protobuf;
+
+        let node = match &node.expr_type {
+            Some(protobuf::physical_expr_node::ExprType::InList(n)) => n,
+            _ => {
+                return datafusion_common::internal_err!(
+                    "PhysicalExprNode is not an InList"
+                );
+            }
+        };
+
+        let expr = ctx.decode(node.expr.as_deref().ok_or_else(|| {
+            datafusion_common::DataFusionError::Internal(
+                "InList is missing required field 'expr'".to_string(),
+            )
+        })?)?;
+
+        let list = node
+            .list
+            .iter()
+            .map(|e| ctx.decode(e))
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(Arc::new(InListExpr::try_new(
+            expr,
+            list,
+            node.negated,
+            ctx.schema(),
+        )?))
+    }
 }
 impl std::fmt::Display for InListExpr {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
@@ -442,6 +478,29 @@ impl PhysicalExpr for InListExpr {
         }
         write!(f, ")")
     }
+
+    #[cfg(feature = "proto")]
+    fn try_to_proto(
+        &self,
+        ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+    ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+        use datafusion_proto_models::protobuf;
+
+        Ok(Some(protobuf::PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(protobuf::physical_expr_node::ExprType::InList(Box::new(
+                protobuf::PhysicalInListNode {
+                    expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
+                    list: self
+                        .list
+                        .iter()
+                        .map(|e| ctx.encode_child(e))
+                        .collect::<Result<Vec<_>>>()?,
+                    negated: self.negated,
+                },
+            ))),
+        }))
+    }
 }
 
 impl PartialEq for InListExpr {
@@ -3821,3 +3880,163 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(all(test, feature = "proto"))]
+mod proto_tests {
+    use super::*;
+    use crate::expressions::{Column, col, lit};
+    use crate::proto_test_util::{
+        StubDecoder, StubEncoder, UnreachableDecoder, column_node,
+    };
+    use arrow::datatypes::Field;
+    use datafusion_common::DataFusionError;
+    use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
+    use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
+    use datafusion_proto_models::protobuf::{
+        PhysicalExprNode, PhysicalInListNode, physical_expr_node,
+    };
+
+    /// Build an `InListExpr` proto node with the given children.
+    fn in_list_node(
+        expr: Option<Box<PhysicalExprNode>>,
+        list: Vec<PhysicalExprNode>,
+        negated: bool,
+    ) -> PhysicalExprNode {
+        PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(physical_expr_node::ExprType::InList(Box::new(
+                PhysicalInListNode {
+                    expr,
+                    list,
+                    negated,
+                },
+            ))),
+        }
+    }
+
+    /// An `InListExpr` over a column with one literal value.
+    fn in_list_fixture() -> InListExpr {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]);
+        InListExpr::try_new(col("a", &schema).unwrap(), vec![lit(1)], false, &schema)
+            .unwrap()
+    }
+
+    #[test]
+    fn try_to_proto_encodes_in_list() {
+        let in_list = in_list_fixture();
+        let encoder = StubEncoder::ok();
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+
+        let node = in_list
+            .try_to_proto(&ctx)
+            .unwrap()
+            .expect("InListExpr should encode to Some(node)");
+
+        // Built-in exprs never set expr_id; only dynamic filters do.
+        assert!(node.expr_id.is_none());
+        let in_list_node = match node.expr_type {
+            Some(physical_expr_node::ExprType::InList(boxed)) => *boxed,
+            other => panic!("expected an InList node, got {other:?}"),
+        };
+        assert!(!in_list_node.negated);
+        assert!(in_list_node.expr.is_some());
+        assert_eq!(in_list_node.list.len(), 1);
+    }
+
+    #[test]
+    fn try_to_proto_propagates_expr_encode_error() {
+        let in_list = in_list_fixture();
+        let encoder = StubEncoder::failing_on(1);
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+        let err = in_list.try_to_proto(&ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
+    }
+
+    #[test]
+    fn try_to_proto_propagates_list_encode_error() {
+        let in_list = in_list_fixture();
+        // Call 1 is for `expr`, Call 2 is for the first element of `list`
+        let encoder = StubEncoder::failing_on(2);
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+        let err = in_list.try_to_proto(&ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
+    }
+
+    #[test]
+    fn try_from_proto_decodes_in_list() {
+        let node = in_list_node(
+            Some(Box::new(column_node("a"))),
+            vec![column_node("b")],
+            true,
+        );
+        let schema = Schema::new(vec![Field::new("decoded", DataType::Int32, true)]);
+        let decoder = StubDecoder::ok();
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+        let decoded = InListExpr::try_from_proto(&node, &ctx).unwrap();
+        let in_list = decoded
+            .downcast_ref::<InListExpr>()
+            .expect("decoded expr should be an InListExpr");
+
+        assert!(in_list.negated());
+        assert!(in_list.expr().downcast_ref::<Column>().is_some());
+        assert_eq!(in_list.list().len(), 1);
+    }
+
+    #[test]
+    fn try_from_proto_rejects_non_in_list_node() {
+        let node = column_node("a");
+        let schema = Schema::empty();
+        let decoder = UnreachableDecoder;
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+        let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not an InList")
+        ));
+    }
+
+    #[test]
+    fn try_from_proto_rejects_missing_expr() {
+        let node = in_list_node(None, vec![column_node("b")], false);
+        let schema = Schema::empty();
+        let decoder = UnreachableDecoder;
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+        let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("InList is missing required field 'expr'")
+        ));
+    }
+
+    #[test]
+    fn try_from_proto_propagates_expr_decode_error() {
+        let node = in_list_node(
+            Some(Box::new(column_node("a"))),
+            vec![column_node("b")],
+            false,
+        );
+        let schema = Schema::empty();
+        let decoder = StubDecoder::failing_on(1);
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
+    }
+
+    #[test]
+    fn try_from_proto_propagates_list_decode_error() {
+        let node = in_list_node(
+            Some(Box::new(column_node("a"))),
+            vec![column_node("b")],
+            false,
+        );
+        let schema = Schema::empty();
+        // Call 1 is `expr`, Call 2 is the first element of `list`
+        let decoder = StubDecoder::failing_on(2);
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = InListExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
+    }
+}

datafusion/proto/src/physical_plan/from_proto.rs

@@ -43,8 +43,8 @@ use datafusion_physical_expr::projection::{ProjectionExpr, ProjectionExprs};
 use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
 use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr, ScalarFunctionExpr};
 use datafusion_physical_plan::expressions::{
-    BinaryExpr, CaseExpr, CastExpr, Column, IsNotNullExpr, IsNullExpr, LikeExpr, Literal,
-    NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn, in_list,
+    BinaryExpr, CaseExpr, CastExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr,
+    LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
 };
 use datafusion_physical_plan::joins::{HashExpr, SeededRandomState};
 use datafusion_physical_plan::windows::{create_window_expr, schema_add_window_field};
@@ -328,18 +328,7 @@ pub fn parse_physical_expr_with_converter(
                 proto_converter,
             )?))
         }
-        ExprType::InList(e) => in_list(
-            parse_required_physical_expr(
-                e.expr.as_deref(),
-                ctx,
-                "expr",
-                input_schema,
-                proto_converter,
-            )?,
-            parse_physical_exprs(&e.list, ctx, input_schema, proto_converter)?,
-            &e.negated,
-            input_schema,
-        )?,
+        ExprType::InList(_) => InListExpr::try_from_proto(proto, &decode_ctx)?,
         ExprType::Case(e) => Arc::new(CaseExpr::try_new(
             e.expr
                 .as_ref()

datafusion/proto/src/physical_plan/to_proto.rs

@@ -36,8 +36,8 @@ use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
 use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindowExpr};
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use datafusion_physical_plan::expressions::{
-    CaseExpr, CastExpr, DynamicFilterPhysicalExpr, InListExpr, IsNotNullExpr, IsNullExpr,
-    Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
+    CaseExpr, CastExpr, DynamicFilterPhysicalExpr, IsNotNullExpr, IsNullExpr, Literal,
+    NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
 };
 use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
 use datafusion_physical_plan::udaf::AggregateFunctionExpr;
@@ -412,19 +412,6 @@ pub fn serialize_physical_expr_with_converter(
                 }),
             )),
         })
-    } else if let Some(expr) = expr.downcast_ref::<InListExpr>() {
-        Ok(protobuf::PhysicalExprNode {
-            expr_id,
-            expr_type: Some(protobuf::physical_expr_node::ExprType::InList(Box::new(
-                protobuf::PhysicalInListNode {
-                    expr: Some(Box::new(
-                        proto_converter.physical_expr_to_proto(expr.expr(), codec)?,
-                    )),
-                    list: serialize_physical_exprs(expr.list(), codec, proto_converter)?,
-                    negated: expr.negated(),
-                },
-            ))),
-        })
     } else if let Some(expr) = expr.downcast_ref::<NegativeExpr>() {
         Ok(protobuf::PhysicalExprNode {
             expr_id,