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

[codecov-commenter]

Codecov Report

❌ Patch coverage is 63.41128% with 266 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.58%. Comparing base (124291e) to head (0a88ca9).
⚠️ Report is 6 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.80%	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.61%   81.58%   -0.03%     
==========================================
  Files        1123     1123              
  Lines      409392   410099     +707     
  Branches   409392   410099     +707     
==========================================
+ Hits       334134   334590     +456     
- Misses      55637    55877     +240     
- Partials    19621    19632      +11     

☔ 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.

[barbarj]

My intent here was to keep the List/Hash choice across the network, though I can see a case for bumping the Hash to "bloom" when serializing to keep the wire size down. But I'd expect we'd need another intra-DataFusion reason to add the bloom filter code besides this.

Related: #16435

[stuhood]

Related: bloom filters coming to spillable hash joins: #24768

[barbarj]

@adriangb or @zhuqi-lucas Can you take a look at this? Do you have any thoughts about this approach vs waiting for bloom filters to ship with #24768 ?

[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 v55.0.0 (current)
       Built [  33.844s] (current)
     Parsing datafusion-physical-plan v55.0.0 (current)
      Parsed [   0.128s] (current)
    Building datafusion-physical-plan v55.0.0 (baseline)
       Built [  32.084s] (baseline)
     Parsing datafusion-physical-plan v55.0.0 (baseline)
      Parsed [   0.126s] (baseline)
    Checking datafusion-physical-plan v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.834s] 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 [  68.501s] datafusion-physical-plan
    Building datafusion-proto v55.0.0 (current)
       Built [  45.294s] (current)
     Parsing datafusion-proto v55.0.0 (current)
      Parsed [   0.015s] (current)
    Building datafusion-proto v55.0.0 (baseline)
       Built [  45.545s] (baseline)
     Parsing datafusion-proto v55.0.0 (baseline)
      Parsed [   0.017s] (baseline)
    Checking datafusion-proto v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   0.119s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  92.526s] datafusion-proto
    Building datafusion-proto-models v55.0.0 (current)
       Built [  21.226s] (current)
     Parsing datafusion-proto-models v55.0.0 (current)
      Parsed [   0.121s] (current)
    Building datafusion-proto-models v55.0.0 (baseline)
       Built [  20.906s] (baseline)
     Parsing datafusion-proto-models v55.0.0 (baseline)
      Parsed [   0.112s] (baseline)
    Checking datafusion-proto-models v55.0.0 -> v55.0.0 (no change; assume patch)
     Checked [   1.935s] 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:1634
  variant ExprType:HashTableLookupExpr in /home/runner/work/datafusion/datafusion/datafusion/proto-models/src/generated/prost.rs:1634

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

[adriangb]

I'm very much 👎🏻 on serializing the in memory map across the wire.

I'm already considering a proposal to disable them by default because the current implementations are slower than no filter if the joint is not selective. Adding more additional work that happens just because the filter exists will only make this worse.

[barbarj]

I'm very much 👎🏻 on serializing the in memory map across the wire.

I'm already considering a proposal to disable them by default because the current implementations are slower than no filter if the joint is not selective. Adding more additional work that happens just because the filter exists will only make this worse.

Noted! That's instructive. If implemented, that proposal would disable the use of HashTableLookupExpr? Is that right?

[adriangb]

No, it would just turn it off by default.