refactor: add try_to_proto to HashTableLookupExpr (#22451)

## Which issue does this PR close?
Part of #22435

## What changes are included?
Adds `try_to_proto` to `HashTableLookupExpr` so it participates in the
expression-local serialization pattern introduced in #21929.

`HashTableLookupExpr` holds a runtime `Arc<Map>` that cannot be
serialized, so `try_to_proto` replaces it with `lit(true)`. This is
safe because the filter is a performance optimisation only — `lit(true)`
passes all rows and the join produces correct results either way.

The centralized arm in `to_proto.rs` remains as a fallback for now.
Cleanup can follow in a separate PR once this lands.

## Are these changes tested?
Yes — covered by the existing `roundtrip_hash_table_lookup_expr_to_lit`
test in `datafusion/proto/tests/cases/roundtrip_physical_plan.rs`.

## Are there any user-facing changes?
No.

---------

Co-authored-by: Anurag Tryambak Raut <anuragtryambakraut@Anurags-MacBook-Air.local>

Author: AnuragRaut08

Cargo.lock

@@ -42,6 +42,11 @@ force_hash_collisions = []
 test_utils = ["arrow/test_utils"]
 tokio_coop = []
 tokio_coop_fallback = []
+proto = [
+    "dep:datafusion-proto-models",
+    "dep:datafusion-proto-common",
+    "datafusion-physical-expr-common/proto",
+]
 
 [lib]
 name = "datafusion_physical_plan"
@@ -65,6 +70,8 @@ datafusion-functions-aggregate-common = { workspace = true }
 datafusion-functions-window-common = { workspace = true }
 datafusion-physical-expr = { workspace = true, default-features = true }
 datafusion-physical-expr-common = { workspace = true }
+datafusion-proto-common = { workspace = true, optional = true }
+datafusion-proto-models = { workspace = true, optional = true }
 futures = { workspace = true }
 half = { workspace = true }
 hashbrown = { workspace = true }

datafusion/physical-plan/Cargo.toml

@@ -215,7 +215,6 @@ pub struct HashTableLookupExpr {
     /// Description for display
     description: String,
 }
-
 impl HashTableLookupExpr {
     /// Create a new HashTableLookupExpr
     ///
@@ -241,7 +240,6 @@ impl HashTableLookupExpr {
         }
     }
 }
-
 impl std::fmt::Debug for HashTableLookupExpr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let cols = self
@@ -337,7 +335,38 @@ impl PhysicalExpr for HashTableLookupExpr {
             }
         }
     }
-
+    #[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;
+        use datafusion_proto_models::protobuf::physical_expr_node::ExprType;
+
+        // HashTableLookupExpr holds a runtime Arc<Map> (the build-side hash
+        // table) that cannot be serialized, so it is replaced with lit(true).
+        //
+        // Dynamic filtering is a performance optimisation only — replacing the
+        // lookup with lit(true) preserves correctness by allowing all rows
+        // through.
+        //
+        // If a plan is serialized before execution, HashTableLookupExpr is not
+        // yet present in the dynamic filter expression.
+        //
+        // If a plan is serialized after execution, any runtime-created
+        // HashTableLookupExpr is replaced during serialization. Re-executing
+        // the plan requires reset_state(), after which HashJoinExec rebuilds
+        // fresh dynamic filters at runtime.
+        let value = datafusion_proto_common::ScalarValue {
+            value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
+                true,
+            )),
+        };
+        Ok(Some(protobuf::PhysicalExprNode {
+            expr_id: None,
+            expr_type: Some(ExprType::Literal(value)),
+        }))
+    }
     fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.description)
     }

datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs

@@ -65,7 +65,7 @@ datafusion-expr = { workspace = true }
 datafusion-functions-table = { workspace = true }
 datafusion-physical-expr = { workspace = true, features = ["proto"] }
 datafusion-physical-expr-common = { workspace = true, features = ["proto"] }
-datafusion-physical-plan = { workspace = true }
+datafusion-physical-plan = { workspace = true, features = ["proto"] }
 datafusion-proto-common = { workspace = true }
 datafusion-proto-models = { workspace = true }
 object_store = { workspace = true }

datafusion/proto/Cargo.toml

@@ -39,7 +39,7 @@ use datafusion_physical_plan::expressions::{
     CaseExpr, CastExpr, DynamicFilterPhysicalExpr, IsNotNullExpr, IsNullExpr, Literal,
     NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
 };
-use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
+use datafusion_physical_plan::joins::HashExpr;
 use datafusion_physical_plan::udaf::AggregateFunctionExpr;
 use datafusion_physical_plan::windows::{PlainAggregateWindowExpr, WindowUDFExpr};
 use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
@@ -302,31 +302,6 @@ pub fn serialize_physical_expr_with_converter(
         return Ok(node);
     }
 
-    // HashTableLookupExpr is used for dynamic filter pushdown in hash joins.
-    // It contains an Arc<dyn JoinHashMapType> (the build-side hash table) which
-    // cannot be serialized - the hash table is a runtime structure built during
-    // execution on the build side.
-    //
-    // We replace it with lit(true) which is safe because:
-    // 1. The filter is a performance optimization, not a correctness requirement
-    // 2. lit(true) passes all rows, so no valid rows are incorrectly filtered out
-    // 3. The join itself will still produce correct results, just without the
-    //    benefit of early filtering on the probe side
-    //
-    // In distributed execution, the remote worker won't have access to the hash
-    // table anyway, so the best we can do is skip this optimization.
-    if expr.downcast_ref::<HashTableLookupExpr>().is_some() {
-        let value = datafusion_proto_common::ScalarValue {
-            value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
-                true,
-            )),
-        };
-        return Ok(protobuf::PhysicalExprNode {
-            expr_id,
-            expr_type: Some(protobuf::physical_expr_node::ExprType::Literal(value)),
-        });
-    }
-
     if let Some(expr) = expr.downcast_ref::<UnKnownColumn>() {
         Ok(protobuf::PhysicalExprNode {
             expr_id,