Port LikeExpr to use try_to_proto / try_from_proto (apache#22471)

## Which issue does this PR close?

  - Closes apache#22431.

  ## Rationale for this change

`LikeExpr` still serialized through the central physical proto
conversion chain. This change moves serialization and deserialization
responsibility into the expression
type itself, following the established migration pattern used for
`Column` and `BinaryExpr`.

  ## What changes are included in this PR?

  - Add `LikeExpr::try_to_proto`
  - Add `LikeExpr::try_from_proto`
  - Route physical proto decode through `LikeExpr::try_from_proto`
  - Remove the central `LikeExpr` encode and decode branches
  - Keep behavior and wire shape unchanged
- Rely on existing physical proto roundtrip coverage instead of an extra
expression-local roundtrip test

  ## Are these changes tested?

  Yes.

  Existing physical proto roundtrip coverage already covers `LikeExpr`.

  Verification run:
  - `cargo fmt --all`
  - `cargo test -p datafusion-physical-expr --features proto`
  - `cargo check -p datafusion-proto --all-features`
- `cargo clippy -p datafusion-physical-expr -p datafusion-proto
--all-targets --all-features -- -D warnings`

  ## Are there any user-facing changes?

  No.

---------

Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

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

@@ -145,6 +145,69 @@ impl PhysicalExpr for LikeExpr {
         write!(f, " {} ", self.op_name())?;
         self.pattern.fmt_sql(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::LikeExpr(Box::new(
+                protobuf::PhysicalLikeExprNode {
+                    negated: self.negated,
+                    case_insensitive: self.case_insensitive,
+                    expr: Some(Box::new(ctx.encode_child(&self.expr)?)),
+                    pattern: Some(Box::new(ctx.encode_child(&self.pattern)?)),
+                },
+            ))),
+        }))
+    }
+}
+
+#[cfg(feature = "proto")]
+impl LikeExpr {
+    /// Reconstruct a [`LikeExpr`] from its protobuf representation.
+    ///
+    /// Takes the whole [`PhysicalExprNode`] so the decode signature matches
+    /// other migrated expressions and can inspect outer-node metadata if
+    /// needed in the future.
+    ///
+    /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+    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_common::internal_err;
+        use datafusion_proto_models::protobuf;
+
+        let like_expr = match &node.expr_type {
+            Some(protobuf::physical_expr_node::ExprType::LikeExpr(like_expr)) => {
+                like_expr.as_ref()
+            }
+            _ => return internal_err!("PhysicalExprNode is not a LikeExpr"),
+        };
+
+        let expr = like_expr.expr.as_deref().ok_or_else(|| {
+            datafusion_common::DataFusionError::Internal(
+                "LikeExpr is missing required field 'expr'".to_string(),
+            )
+        })?;
+        let pattern = like_expr.pattern.as_deref().ok_or_else(|| {
+            datafusion_common::DataFusionError::Internal(
+                "LikeExpr is missing required field 'pattern'".to_string(),
+            )
+        })?;
+
+        Ok(Arc::new(LikeExpr::new(
+            like_expr.negated,
+            like_expr.case_insensitive,
+            ctx.decode(expr)?,
+            ctx.decode(pattern)?,
+        )))
+    }
 }
 
 /// used for optimize Dictionary like
@@ -283,3 +346,187 @@ mod test {
         Ok(())
     }
 }
+
+/// Tests for the `try_to_proto` / `try_from_proto` hooks.
+#[cfg(all(test, feature = "proto"))]
+mod proto_tests {
+    use super::*;
+    use crate::expressions::{Column, col};
+    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, PhysicalLikeExprNode, physical_expr_node,
+    };
+
+    /// Build a `LikeExpr` proto node with the given children.
+    fn like_node(
+        negated: bool,
+        case_insensitive: bool,
+        expr: Option<Box<PhysicalExprNode>>,
+        pattern: Option<Box<PhysicalExprNode>>,
+    ) -> PhysicalExprNode {
+        PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(physical_expr_node::ExprType::LikeExpr(Box::new(
+                PhysicalLikeExprNode {
+                    negated,
+                    case_insensitive,
+                    expr,
+                    pattern,
+                },
+            ))),
+        }
+    }
+
+    /// A `LikeExpr` over two `Utf8` columns with both flags set, so the
+    /// `negated` / `case_insensitive` wiring is actually exercised.
+    fn like_fixture() -> LikeExpr {
+        let schema = Schema::new(vec![
+            Field::new("a", DataType::Utf8, false),
+            Field::new("b", DataType::Utf8, false),
+        ]);
+        LikeExpr::new(
+            true,
+            true,
+            col("a", &schema).unwrap(),
+            col("b", &schema).unwrap(),
+        )
+    }
+
+    #[test]
+    fn try_to_proto_encodes_like_expr() {
+        let like = like_fixture();
+        let encoder = StubEncoder::ok();
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+
+        let node = like
+            .try_to_proto(&ctx)
+            .unwrap()
+            .expect("LikeExpr should encode to Some(node)");
+
+        // Built-in exprs never set expr_id; only dynamic filters do.
+        assert!(node.expr_id.is_none());
+        let like_node = match node.expr_type {
+            Some(physical_expr_node::ExprType::LikeExpr(boxed)) => *boxed,
+            other => panic!("expected a LikeExpr node, got {other:?}"),
+        };
+        assert!(like_node.negated);
+        assert!(like_node.case_insensitive);
+        assert!(like_node.expr.is_some());
+        assert!(like_node.pattern.is_some());
+    }
+
+    #[test]
+    fn try_to_proto_propagates_expr_encode_error() {
+        let like = like_fixture();
+        let encoder = StubEncoder::failing_on(1);
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+        let err = like.try_to_proto(&ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
+    }
+
+    #[test]
+    fn try_to_proto_propagates_pattern_encode_error() {
+        let like = like_fixture();
+        let encoder = StubEncoder::failing_on(2);
+        let ctx = PhysicalExprEncodeCtx::new(&encoder);
+        let err = like.try_to_proto(&ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
+    }
+
+    #[test]
+    fn try_from_proto_decodes_like_expr() {
+        let node = like_node(
+            true,
+            true,
+            Some(Box::new(column_node("a"))),
+            Some(Box::new(column_node("b"))),
+        );
+        let schema = Schema::empty();
+        let decoder = StubDecoder::ok();
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+
+        let decoded = LikeExpr::try_from_proto(&node, &ctx).unwrap();
+        let like = decoded
+            .downcast_ref::<LikeExpr>()
+            .expect("decoded expr should be a LikeExpr");
+        assert!(like.negated());
+        assert!(like.case_insensitive());
+        assert!(like.expr().downcast_ref::<Column>().is_some());
+        assert!(like.pattern().downcast_ref::<Column>().is_some());
+    }
+
+    #[test]
+    fn try_from_proto_rejects_non_like_node() {
+        let node = column_node("a");
+        let schema = Schema::empty();
+        let decoder = UnreachableDecoder;
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("PhysicalExprNode is not a LikeExpr")
+        ));
+    }
+
+    #[test]
+    fn try_from_proto_rejects_missing_expr() {
+        let node = like_node(false, false, None, Some(Box::new(column_node("b"))));
+        let schema = Schema::empty();
+        let decoder = UnreachableDecoder;
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'expr'")
+        ));
+    }
+
+    #[test]
+    fn try_from_proto_rejects_missing_pattern() {
+        let node = like_node(false, false, Some(Box::new(column_node("a"))), None);
+        let schema = Schema::empty();
+        let decoder = UnreachableDecoder;
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(
+            err,
+            DataFusionError::Internal(msg) if msg.contains("LikeExpr is missing required field 'pattern'")
+        ));
+    }
+
+    #[test]
+    fn try_from_proto_propagates_expr_decode_error() {
+        let node = like_node(
+            false,
+            false,
+            Some(Box::new(column_node("a"))),
+            Some(Box::new(column_node("b"))),
+        );
+        let schema = Schema::empty();
+        let decoder = StubDecoder::failing_on(1);
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 1")));
+    }
+
+    #[test]
+    fn try_from_proto_propagates_pattern_decode_error() {
+        let node = like_node(
+            false,
+            false,
+            Some(Box::new(column_node("a"))),
+            Some(Box::new(column_node("b"))),
+        );
+        let schema = Schema::empty();
+        let decoder = StubDecoder::failing_on(2);
+        let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
+        let err = LikeExpr::try_from_proto(&node, &ctx).unwrap_err();
+        assert!(matches!(err, DataFusionError::Internal(msg) if msg.contains("call 2")));
+    }
+}

datafusion/physical-expr/src/lib.rs

@@ -40,6 +40,9 @@ mod partitioning;
 mod physical_expr;
 pub mod planner;
 pub mod projection;
+/// Shared test helpers for the `try_to_proto` / `try_from_proto` unit tests
+#[cfg(all(test, feature = "proto"))]
+pub(crate) mod proto_test_util;
 mod scalar_function;
 pub mod scalar_subquery;
 pub mod simplifier;