feat(physical-plan): Make HashTableLookupExpr serializable

[barbarj]

Which issue does this PR close?

• Closes make HashTableLookupExpr serializable #24277

Rationale for this change

In order for DataFusion Distributed (and presumably other distributed DF projects) to make use of HashTableLookupExpr as a dynamic filter across network boundaries, it needs to be serializable. (for instance, see: datafusion-contrib/datafusion-distributed#623)

What changes are included in this PR?

The only sticky part of serializing HashTableLookupExpr is its Map. The inner members of the two variants of Map, HashMap and ArrayMap both support much more functionality than is needed to evaluate this as a dynamic expression. So, in order to simply serialization and minimize the on-wire size, we serialize only the aspects required for expression evaluation (i.e. membership checks)

We replace HashTableLookupExpr's Map with a local version that includes the membership-only variants. These are implemented such that they are only constructible via deserialization.

Are these changes tested?

The roundtrip_hash_table_lookup_expr_to_lit test is replaced with two regular roundtrip tests, one each for the HashMap and ArrayMap versions of HashTableLookupExpr.

This PR also adds a bunch of tests testing the post-deserialization behavior of the new membership-only variants.

Are there any user-facing changes?

• JoinHashMapType has a new required method hashes

[github-actions]

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-physical-plan v54.1.0 (current)
       Built [  46.834s] (current)
     Parsing datafusion-physical-plan v54.1.0 (current)
      Parsed [   0.152s] (current)
    Building datafusion-physical-plan v54.1.0 (baseline)
       Built [  38.888s] (baseline)
     Parsing datafusion-physical-plan v54.1.0 (baseline)
      Parsed [   0.154s] (baseline)
    Checking datafusion-physical-plan v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   1.005s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure trait_method_added: pub trait method added ---

Description:
A non-sealed public trait added a new method without a default implementation, which breaks downstream implementations of the trait
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#trait-new-item-no-default
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/trait_method_added.ron

Failed in:
  trait method datafusion_physical_plan::joins::utils::JoinHashMapType::hashes in file /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/joins/join_hash_map.rs:151
  trait method datafusion_physical_plan::joins::join_hash_map::JoinHashMapType::hashes in file /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/joins/join_hash_map.rs:151

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  88.622s] datafusion-physical-plan
    Building datafusion-proto v54.1.0 (current)
       Built [  55.586s] (current)
     Parsing datafusion-proto v54.1.0 (current)
      Parsed [   0.022s] (current)
    Building datafusion-proto v54.1.0 (baseline)
       Built [  56.214s] (baseline)
     Parsing datafusion-proto v54.1.0 (baseline)
      Parsed [   0.021s] (baseline)
    Checking datafusion-proto v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   0.173s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 113.267s] datafusion-proto
    Building datafusion-proto-models v54.1.0 (current)
       Built [  25.523s] (current)
     Parsing datafusion-proto-models v54.1.0 (current)
      Parsed [   0.138s] (current)
    Building datafusion-proto-models v54.1.0 (baseline)
       Built [  26.265s] (baseline)
     Parsing datafusion-proto-models v54.1.0 (baseline)
      Parsed [   0.138s] (baseline)
    Checking datafusion-proto-models v54.1.0 -> v54.1.0 (no change; assume patch)
     Checked [   2.888s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure enum_variant_added: enum variant added on exhaustive enum ---

Description:
A publicly-visible enum without #[non_exhaustive] has a new variant.
        ref: https://doc.rust-lang.org/cargo/reference/semver.html#enum-variant-new
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/enum_variant_added.ron

Failed in:
  variant ExprType:HashTableLookupExpr in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:1626
  variant ExprType:HashTableLookupExpr in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:1626

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  56.478s] datafusion-proto-models

[codecov-commenter]

Codecov Report

❌ Patch coverage is 63.46154% with 266 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.16%. Comparing base (30ca39b) to head (f9d7d4e).
⚠️ Report is 47 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/proto-models/src/generated/pbjson.rs	0.00%	234 Missing ⚠️
...-plan/src/joins/hash_join/partitioned_hash_eval.rs	93.81%	22 Missing and 6 partials ⚠️
...usion/physical-plan/src/joins/stream_join_utils.rs	0.00%	3 Missing ⚠️
datafusion/proto/src/physical_plan/from_proto.rs	0.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24382      +/-   ##
==========================================
- Coverage   81.29%   81.16%   -0.14%     
==========================================
  Files        1110     1110              
  Lines      385168   389445    +4277     
  Branches   385168   389445    +4277     
==========================================
+ Hits       313111   316074    +2963     
- Misses      53583    54774    +1191     
- Partials    18474    18597     +123     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[barbarj]

If this doesn't make it into 55.0, this will need to be moved to 55.1

[stuhood]

@jayshrivastava, @barbarj, @adriangb: The deciding line between List and Hash is already configurable: relatively small sets get List, larger sets get Hash.

Given that, should the presence of a Hash like this already suggest that systems which care about serializing a Hash should probably be using a bloom filter instead? Or is the thinking that there are multiple size thresholds, and Hash needs to be serializable for the middle one?:

• small: List
• medium: Hash
• large: "bloom"

[adriangb]

I think the main thought behind the original design and split was that locally at least there's no point in building a Bloom filter if you already have a hash table in memory. It makes sense to me that if you want to serialize across the wire the tradeoff is very different, a bloom filter would be better. I'm not sure if that means you would want to build a Bloom filter upfront or build it when you serialize.