make HashTableLookupExpr serializable

[jayshrivastava]

Is your feature request related to a problem or challenge?

See

datafusion/datafusion/proto/tests/cases/roundtrip_physical_plan.rs

Lines 3963 to 3969 in 5d6146c

	/// Test that HashTableLookupExpr serializes to lit(true)
	///
	/// HashTableLookupExpr contains a runtime hash table that cannot be serialized.
	/// The serialization code replaces it with lit(true) which is safe because
	/// it's a performance optimization filter, not a correctness requirement.
	#[test]
	fn roundtrip_hash_table_lookup_expr_to_lit() -> Result<()> {

If you take an ExecutionPlan and serialize it, all occurrences of HashTableLookupExpr are replaced with lit(true). This makes it difficult to send dynamic filters across the network.

Describe the solution you'd like

Ideally, there's some way to serialize and deserialize these expressions, potentially by serializing the seed

  message HashTableLookupExpr {
    repeated PhysicalExprNode on_columns = 1;
    uint64 seed = 2;
    repeated fixed64 build_hashes = 3;
  }

Describe alternatives you've considered

No response

Additional context

No response

[barbarj]

take

[barbarj]

I'll implement this soon, but at high-level, the design I'm thinking is:

Serialization

For evaluation purposes, we only need the hashes in the HashMap case. The protobuf then looks like this, with the contents of ArrayMapNode to be determined:

message PhysicalHashTableLookupExprNode {
  repeated PhysicalExprNode on_columns = 1;
  uint64 seed = 2;
  string description = 3;
  oneof map {
    HashSetMapNode hash_map = 4;
    ArrayMapNode array_map = 5;
  }
}

message HashSetMapNode {
  repeated fixed64 build_hashes = 1;
}

We add a hashes access method:

pub trait JoinHashMapType {
    pub fn hashes(&self) -> Vec<u64>;
}

Deserialization

For the ArrayMapNode case, we just deserialize to the ArrayMap variant of HashTableLookupExpr. For the HashSetMapNode case, we add a new impl of JoinHashMapType with contain_hashes/len/is_empty implemented. It will return internal errors from the build-side mutation and get_matched_indices methods. This will allow it be evaluated after deserialization but not updated. (Will document that this should be constructed during deserialization only and is an immutable snapshot of the expression).