chore: deprecate HashExpr; mirror lit(true) proto serde for MultiMapLookupExpr

HashExpr was only used by the CASE-routing dynamic filter that 82f5ee5
removed. Nothing in the workspace constructs one anymore. Mark it
`#[deprecated(since = "54.0.0", ...)]` per the API health guidelines,
pointing users at HashTableLookupExpr / MultiMapLookupExpr. Keep the
struct, proto wire format, re-exports, and unit tests intact so existing
serialized plans and downstream consumers still work during the
deprecation window; silence the resulting warnings with
`#[expect(deprecated)]` at the impl blocks, re-exports, proto
serializer/deserializer, and roundtrip test.

MultiMapLookupExpr holds Arc<Map>s in exactly the same way
HashTableLookupExpr does — those build-side hash tables can't cross the
wire. Extend the existing lit(true) short-circuit in
`serialize_physical_expr_with_converter` to cover MultiMapLookupExpr,
update the explanatory comment, re-export it from joins/{mod,
hash_join::mod}, and add a `roundtrip_multi_map_lookup_expr_to_lit`
test mirroring the existing HashTableLookupExpr one.

Refs: apache#21931 (comment)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

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

@@ -18,7 +18,11 @@
 //! [`HashJoinExec`] Partitioned Hash Join Operator
 
 pub use exec::{HashJoinExec, HashJoinExecBuilder};
-pub use partitioned_hash_eval::{HashExpr, HashTableLookupExpr, SeededRandomState};
+#[expect(deprecated)]
+pub use partitioned_hash_eval::HashExpr;
+pub use partitioned_hash_eval::{
+    HashTableLookupExpr, MultiMapLookupExpr, SeededRandomState,
+};
 
 mod exec;
 mod inlist_builder;

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

@@ -72,6 +72,10 @@ impl SeededRandomState {
 /// This is used for:
 /// - Computing routing hashes (with RepartitionExec's 0,0,0,0 seeds)
 /// - Computing lookup hashes (with HashJoin's 'J','O','I','N' seeds)
+#[deprecated(
+    since = "54.0.0",
+    note = "Hash-join dynamic filters no longer route rows via a per-row hash + CASE. Use HashTableLookupExpr for single-map membership probes or MultiMapLookupExpr to OR across several maps in one hashing pass."
+)]
 pub struct HashExpr {
     /// Columns to hash
     on_columns: Vec<PhysicalExprRef>,
@@ -81,6 +85,7 @@ pub struct HashExpr {
     description: String,
 }
 
+#[expect(deprecated)]
 impl HashExpr {
     /// Create a new HashExpr
     ///
@@ -116,6 +121,7 @@ impl HashExpr {
     }
 }
 
+#[expect(deprecated)]
 impl std::fmt::Debug for HashExpr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let cols = self
@@ -129,6 +135,7 @@ impl std::fmt::Debug for HashExpr {
     }
 }
 
+#[expect(deprecated)]
 impl Hash for HashExpr {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         self.on_columns.dyn_hash(state);
@@ -137,6 +144,7 @@ impl Hash for HashExpr {
     }
 }
 
+#[expect(deprecated)]
 impl PartialEq for HashExpr {
     fn eq(&self, other: &Self) -> bool {
         self.on_columns == other.on_columns
@@ -145,14 +153,17 @@ impl PartialEq for HashExpr {
     }
 }
 
+#[expect(deprecated)]
 impl Eq for HashExpr {}
 
+#[expect(deprecated)]
 impl Display for HashExpr {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.description)
     }
 }
 
+#[expect(deprecated)]
 impl PhysicalExpr for HashExpr {
     fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
         self.on_columns.iter().collect()
@@ -546,6 +557,7 @@ fn evaluate_columns(
 }
 
 #[cfg(test)]
+#[expect(deprecated)]
 mod tests {
     use super::*;
     use crate::joins::join_hash_map::JoinHashMapU32;

datafusion/physical-plan/src/joins/mod.rs

@@ -20,8 +20,11 @@
 use arrow::array::BooleanBufferBuilder;
 pub use cross_join::CrossJoinExec;
 use datafusion_physical_expr::PhysicalExprRef;
+#[expect(deprecated)]
+pub use hash_join::HashExpr;
 pub use hash_join::{
-    HashExpr, HashJoinExec, HashJoinExecBuilder, HashTableLookupExpr, SeededRandomState,
+    HashJoinExec, HashJoinExecBuilder, HashTableLookupExpr, MultiMapLookupExpr,
+    SeededRandomState,
 };
 pub use nested_loop_join::{NestedLoopJoinExec, NestedLoopJoinExecBuilder};
 use parking_lot::Mutex;

datafusion/proto/src/physical_plan/from_proto.rs

@@ -46,7 +46,9 @@ use datafusion_physical_plan::expressions::{
     BinaryExpr, CaseExpr, CastExpr, Column, IsNotNullExpr, IsNullExpr, LikeExpr, Literal,
     NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn, in_list,
 };
-use datafusion_physical_plan::joins::{HashExpr, SeededRandomState};
+#[expect(deprecated)]
+use datafusion_physical_plan::joins::HashExpr;
+use datafusion_physical_plan::joins::SeededRandomState;
 use datafusion_physical_plan::windows::{create_window_expr, schema_add_window_field};
 use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
 use datafusion_proto_common::common::proto_error;
@@ -255,6 +257,7 @@ pub fn parse_physical_expr(
 ///   any scoped state needed during recursive deserialization.
 /// * `proto_converter` - Converter hooks used for recursive physical plan and
 ///   expression deserialization.
+#[expect(deprecated)] // HashExpr branch — kept for proto wire compatibility.
 pub fn parse_physical_expr_with_converter(
     proto: &protobuf::PhysicalExprNode,
     input_schema: &Schema,

datafusion/proto/src/physical_plan/to_proto.rs

@@ -40,7 +40,9 @@ use datafusion_physical_plan::expressions::{
     IsNotNullExpr, IsNullExpr, LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr,
     UnKnownColumn,
 };
-use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
+#[expect(deprecated)]
+use datafusion_physical_plan::joins::HashExpr;
+use datafusion_physical_plan::joins::{HashTableLookupExpr, MultiMapLookupExpr};
 use datafusion_physical_plan::udaf::AggregateFunctionExpr;
 use datafusion_physical_plan::windows::{PlainAggregateWindowExpr, WindowUDFExpr};
 use datafusion_physical_plan::{Partitioning, PhysicalExpr, WindowExpr};
@@ -253,27 +255,30 @@ pub fn serialize_physical_expr(
 /// serialization of udfs requiring specialized serialization (see [`PhysicalExtensionCodec::try_encode_udf`]).
 /// A [`PhysicalProtoConverterExtension`] can be provided to handle the
 /// conversion process (see [`PhysicalProtoConverterExtension::physical_expr_to_proto`]).
+#[expect(deprecated)] // HashExpr branch — kept for proto wire compatibility.
 pub fn serialize_physical_expr_with_converter(
     value: &Arc<dyn PhysicalExpr>,
     codec: &dyn PhysicalExtensionCodec,
     proto_converter: &dyn PhysicalProtoConverterExtension,
 ) -> Result<protobuf::PhysicalExprNode> {
     let expr = value.as_ref();
     let expr_id = value.expression_id();
-    // 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.
+    // HashTableLookupExpr and MultiMapLookupExpr are used for dynamic filter
+    // pushdown in hash joins. They contain Arc<Map>s (the build-side hash tables)
+    // which cannot be serialized - the hash tables are runtime structures built
+    // during execution on the build side.
     //
-    // We replace it with lit(true) which is safe because:
+    // We replace them 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() {
+    if expr.downcast_ref::<HashTableLookupExpr>().is_some()
+        || expr.downcast_ref::<MultiMapLookupExpr>().is_some()
+    {
         let value = datafusion_proto_common::ScalarValue {
             value: Some(datafusion_proto_common::scalar_value::Value::BoolValue(
                 true,

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -2466,6 +2466,53 @@ async fn roundtrip_async_func_exec() -> Result<()> {
     Ok(())
 }
 
+/// Test that MultiMapLookupExpr serializes to lit(true)
+///
+/// MultiMapLookupExpr holds runtime build-side hash tables — same reasoning as
+/// [`roundtrip_hash_table_lookup_expr_to_lit`]: the maps cannot cross the wire,
+/// so the optimization is dropped to `lit(true)` and the join still produces
+/// correct results without probe-side prefiltering.
+#[test]
+fn roundtrip_multi_map_lookup_expr_to_lit() -> Result<()> {
+    use datafusion::physical_plan::joins::join_hash_map::JoinHashMapU32;
+    use datafusion::physical_plan::joins::{Map, MultiMapLookupExpr};
+
+    let schema = Arc::new(Schema::new(vec![Field::new("col", DataType::Int64, false)]));
+    let input = Arc::new(EmptyExec::new(schema.clone()));
+
+    let maps = vec![
+        Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))),
+        Arc::new(Map::HashMap(Box::new(JoinHashMapU32::with_capacity(0)))),
+    ];
+    let on_columns = vec![datafusion::physical_plan::expressions::col("col", &schema)?];
+    let lookup_expr: Arc<dyn PhysicalExpr> = Arc::new(MultiMapLookupExpr::new(
+        on_columns,
+        datafusion::physical_plan::joins::SeededRandomState::with_seed(0),
+        maps,
+        "test_multi_lookup".to_string(),
+    ));
+
+    let filter = Arc::new(FilterExec::try_new(lookup_expr, input)?);
+
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+
+    let proto: PhysicalPlanNode =
+        PhysicalPlanNode::try_from_physical_plan(filter.clone(), &codec)
+            .expect("serialization should succeed");
+
+    let result: Arc<dyn ExecutionPlan> = proto
+        .try_into_physical_plan(&ctx.task_ctx(), &codec)
+        .expect("deserialization should succeed");
+
+    let result_filter = result.downcast_ref::<FilterExec>().unwrap();
+    let predicate = result_filter.predicate();
+    let literal = predicate.downcast_ref::<Literal>().unwrap();
+    assert_eq!(*literal.value(), ScalarValue::Boolean(Some(true)));
+
+    Ok(())
+}
+
 /// Test that HashTableLookupExpr serializes to lit(true)
 ///
 /// HashTableLookupExpr contains a runtime hash table that cannot be serialized.
@@ -2517,6 +2564,7 @@ fn roundtrip_hash_table_lookup_expr_to_lit() -> Result<()> {
 }
 
 #[test]
+#[expect(deprecated)] // HashExpr is deprecated but still proto-serialized for wire compatibility.
 fn roundtrip_hash_expr() -> Result<()> {
     use datafusion::physical_plan::joins::{HashExpr, SeededRandomState};