Serialize dynamic filters on execution plan nodes (HashJoin, Aggregate, Sort)

[jayshrivastava]

Which issue does this PR close?

Informs: datafusion-contrib/datafusion-distributed#180
Follow up for: apache#20416

Rationale for this change

I'm interested in serializing a physical plan (post-physical optimizer) and executing it on a remote node. To do so, I need dynamic filters and references/pointers to dynamic filters to be preserved in the plan. Currently, nodes which produce filters such as HashJoinExec, AggregateExec, and SortExec, do not serialize their dynamic filters.

This change intends to update the above nodes to serialize dynamic filters and adds tests for the scenario above.

What changes are included in this PR?

Proto schema (datafusion.proto)

Added PhysicalExprNode dynamic_filter field to:

• HashJoinExecNode (tag 11)
• AggregateExecNode (tag 13)
• SortExecNode (tag 5)

Plan node public API

Added with_dynamic_filter() and dynamic_filter() to HashJoinExec, AggregateExec, SortExec.

with_dynamic_filter() always

• validates that the filter is valid for the plan node's schema
• resets any internal state related to the dynamic filter

Serde

Using the new plan node public APIs above

• Each node's try_from_* serialization function now reads dynamic_filter()
  and serializes it via the proto converter
• Each node's try_into_* deserialization function deserializes the field,
  downcasts to DynamicFilterPhysicalExpr, and sets it on the node

Are these changes tested?

1. Added tests which create this plan and perform round-trip serialization on it (1 test each for HashJoinExec, AggregateExec, and SortExec).

    HashJoinExec ─── dynamic filter
         │               │  
         ▼               │
    FilterExec           | (optimizer pushes down this filter
         │               |
         ▼               ▼ 
    DataSourceExec  ─ dynamic filter 

2. Added tests for with_dynamic_filter() and dynamic_filter() on dynamic_filter() to HashJoinExec, AggregateExec, SortExec.

[jayshrivastava]

Note for reviewers: I'm unsure if I should be using apply_expressions() or expressions() (see apache#20337) instead of with_dynamic_filter() and dynamic_filter()

[LiaCastaneda]

I think it would be cleaner to use apply_expressions (apache#20337), mainly because it's more generic and you can do basically anything with PhysicalExprs inside a plan, including detecting dynamic filters, and you wouldn't need to know beforehand which nodes are producers and consumers -- any custom logic can be done separately in the proto crate. It would also reduce overhead to people who wants to add a new ExecutionPlan that holds a DynamicFilterPhysicalExpr, they'd have to remember to also add the manual dynamic_filter() call here. Implementation for apply_expressions is part of ExecutionPlan and will not be optional, so users will not forget they have to do it in every node.

[LiaCastaneda]

I see we do something similar for every producer/consumer, a more generic way to modify the expressions would probably implementing map_expressions like suggested here in ExecutionPlan to make it more generic?