coordinator: display dynamic filters after execution

[jayshrivastava]

Stack

This stack of PRs implements distributed dynamic filtering #528

1. coordinator: display dynamic filters after execution #623 <- you are here
2. worker: forward partial dynamic filters to coordinator #634
3. coordinator: plan distributed dynamic filters #635
4. coordinator: merge partial dynamic filters  #636
5. coordinator: forward merged dynamic filters to consumers #637
6. worker: apply merged dynamic filters during execution #639

Closes: #529

Problem

Post df-55 upgrade, dynamic filters should work in the worker-local case. There's no way to observe them working other than looking at metrics.

  ┌───── Stage 2 ── tasks=1
  │ AggregateExec: Final COUNT(*)
  │   [Stage 1] => NetworkCoalesceExec
  └──────────────────────────────────────────────────
    ┌───── Stage 1 ── tasks=2
    │ HashJoinExec: orders.customer_id = selected_customers.customer_id
    │   DistributedLeafExec:
    |     ...
    │   DistributedLeafExec:
    │     t0: DataSourceExec: predicate=DynamicFilter [ empty ]
    │     t1: DataSourceExec: predicate=DynamicFilter [ empty ]
    └────────────────────────────────────────────────

Ideally we want the final filters visible when displaying plans.

Solution

This PR adds a new protocol which is basically identical to the metrics protocol. Even the MetricsStore is now just Store and is generic over TaskMetrics and TaskCompletedDynamicFilters (contains completed dynamic filters for a task).

pub(crate) type MetricsStore = Store<TaskMetrics>;
pub(crate) type CompletedDynamicFilterStore = Store<TaskCompletedDynamicFilters>;

Similar to the metrics protocol, workers now collect completed dynamic filters and send them back to the coordinator.

Coordinator                                               Worker
-----------                                               ------
       Create independent display copies
                    |
                    +-- SetPlan(task 0, filter IDs) -------> Decode plan
                    |                                       |
                    |                                       | execute
                    |                                       |
                    |                                       |
                    |                                       |
                    |                                       |
                    |                                       | task finishes
                    |                                       v
                    |<----- TaskDynamicFilters ----- Serialize completed filters from the consumers
                    |
                    v

Then, at display time, we call apply_reports_to_distributed_leaves which traverses the plan_for_viz and updates the dynamic filters for all the variants:

DistributedLeafExec
  task 0: DynamicFilter [ key@0 >= 1 AND key@0 <= 10 ]
  task 1: DynamicFilter [ empty ]

Notes

Duplicate RPC Messages

We will eventually have more dynamic filter RPCs which manage the worker -> coordinator -> merge -> worker flow mentioned in #553.

In theory, the coordinator will know at merge time what the completed filters are, making the TaskCompletedDynamicFilters and final worker -> coordinator message in this PR irrelevant.

However, I think having these mechanisms be separate is good because a) it helps us validate that the dynamic filter coordinator -> worker flow work using external "oracle", and b) there's no guarantee that the coordinator -> worker propagation happens before the query is done (ex. the DataSourceExec may not block execution waiting for dynamic filters), so it's good to have a separate way to know if the final DataSourceExec applied a filter or not.

AND true and empty filters

DynamicFilter [ sr_returned_date_sk@0 >= 2451545 AND sr_returned_date_sk@0 <= 2451910 AND true ] AND DynamicFilter [ empty ]

In this filter AND true occurs because of apache/datafusion#24277. The first DynamicFilter is active but we lose the HashTableLookupExpr when serializing it to send back to the coordinator.

The 2nd filter is DynamicFilter [ empty ] because this is a dynamic filter produced by a remote producer, which does not get propagated to this node yet.

Testing

[stuhood]

Thanks for working on this: this will be very useful!

One quick thought: the dynamic filter will sometimes be much, much larger than what you would actually want to display in an EXPLAIN plan (a large InList or Hash). That suggests that rather than sending the whole filter, what would actually make sense to send back is some sort of human readable summary of the filter?

Also, I have a draft of a related change on our codebase, and it seemed like the easiest mechanism for transferring this kind of information back is via metrics... but the most natural/obvious thing that seemed to be missing in that case was essentially a "string" metric type (we would use it to display a chosen strategy/enum from a scan). Do you think that that might be worth pursuing upstream?

[jayshrivastava]

One quick thought: the dynamic filter will sometimes be much, much larger than what you would actually want to display in an EXPLAIN plan (a large InList or Hash). That suggests that rather than sending the whole filter, what would actually make sense to send back is some sort of human readable summary of the filter?

Also, I have a draft of a related change on our codebase, and it seemed like the easiest mechanism for transferring this kind of information back is via metrics... but the most natural/obvious thing that seemed to be missing in that case was essentially a "string" metric type (we would use it to display a chosen strategy/enum from a scan). Do you think that that might be worth pursuing upstream?

Serializing them as a string is reasonable. Rather than a metric, I think we can implement a PhysicalExpr which just wraps a string and inject it into the play for display using DynamicFilterPhysicalExpr::update(string_expr). @gabotechs what do you think?

[gabotechs]

🤔 I'm not sure if I'm understanding the suggestion. Updating a filter with DynamicFilterPhysicalExpr::update is not really related to visualization, it's how you actually update the filter no?