proto: serialize and dedupe dynamic filters

[jayshrivastava]

Which issue does this PR close?

Informs: datafusion-contrib/datafusion-distributed#180
Informs: #21207
Closes: #20418

Rationale for this change

Consider you have a plan with a HashJoinExec and DataSourceExec

HashJoinExec(dynamic_filter_1 on a@0)
  (...left side of join)
  ProjectionExec(a := Column("a", source_index))
    DataSourceExec
      ParquetSource(predicate = dynamic_filter_2)

You serialize the plan, deserialize it, and execute it. What should happen is that the dynamic filter should "work", meaning:

1. When you deserialize the plan, both the HashJoinExec and DataSourceExec should have pointers to the same DynamicFilterPhysicalExpr
2. The DynamicFilterPhysicalExpr should be updated during execution by the HashJoinExec and the DataSourceExec should filter out rows

This does not happen today for a few reasons, a couple of which this PR aims to address

1. DynamicFilterPhysicalExpr is not survive round-tripping. The internal exprs get inlined (ex. it may be serialized as Literal) due to the PhysicalExpr::snapshot() API
2. Even if DynamicFilterPhysicalExpr survives round-tripping, the one pushed down to the DataSourceExec often has different children. In this case, you have two DynamicFilterPhysicalExpr which
   do not survive deduping, causing referential integrity to be lost.

This PR aims to fix those problems by:

1. Removing the snapshot() call from the serialization process
2. Adding protos for DynamicFilterPhysicalExpr so it can be serialized and deserialized
3. Adding a new concept, a PhysicalExprId, which has two identifiers,
   a "shallow" identifier to indicate two equal expressions which may
   have different children, and an "exact" identifier to indicate two
   exprs that are exactly the same.
4. Updating the deduping deserializer and protos to now be aware of the
   new "shallow" id, deduping exprs which are the same but have
   different children accordingly.

This change adds tests which roundtrip dynamic filters and assert that
referential integrity is maintained.

Future work:

1. Serialize dynamic filters in HashJoinExec and other ExecutionPlans which produce dynamic filters
2. Add tests which actually execute plans after deserialization and assert that dynamic filtering is functional
3. Add proto converters to the PhysicalExtensionCodec trait so implementors can utilize deduping logic

Are these changes tested?

Yes. This change adds tests which roundtrip dynamic filters and assert that referential integrity is maintained.

Are there any user-facing changes?

• The default codec does not call snapshot() on PhysicalExpr during serialization anymore. This means that DynamicFilterPhysicalExpr are now serialized and deserialized without snapshotting

[NGA-TRAN]

Glad you tracked down the root cause. Great explanation of the two issue.
The changes align well with your description, and the tests look solid. I’ll let the folks who know the Dynamic Filtering details comment further on the specifics.

[jayshrivastava]

Added this type because I thought it looked cleaner / simpler than having getters on the DynamicFilterPhysicalExpr itself.

[jayshrivastava]

Maybe we should just add the getters and setters for the internal fields...

[adriangb]

It seems like we end up exposing all of the internals anyway: if there was a way to return a dyn Serializable it'd be one thing, but the proto machinery still needs to know all of the fields to serialize.

[adriangb]

I also think the name snapshot makes it sounds like this is what would be returned from PhysicalExpr::snapshot, but that is not the case.

[gabotechs]

Could be... you kind of have already this, just to an intermediate structure instead

[gene-bordegaray]

overall looks good, had one idea as to making the api more familiar to those new with it. Let me know what you think

[gabotechs]

Overall, I think we should find alternative ways of doing this that do not imply special-casing the dynamic filter serialization process.

I shared some ideas on how to do this, but they probably need to be fleshed out a bit more.

The things that I think this PR should be achieving is:

• No special casing for dynamic filters in serialization/deserialization code
• No changes to global protobuf messages just for the sake of dynamic filters, just a new normal entry in the enum PhysicalDynamicFilterNode

Things that IMO we should have, probably in a separate PR:

• Stop playing with raw pointer addresses in dynamic filters and assign proper unique identifiers, probably using the uuid crate.