coordinator: display consumer dynamic filters after execution

[jayshrivastava]

Stack

This stack of PRs implements distributed dynamic filtering #528

1. coordinator: display consumer dynamic filters after execution #623 <- you are here
2. feat: plan distributed dynamic filters #634
3. feat: forward remote dynamic filter updates to coordinator #635
4. coordinator: merge partial dynamic filters  #636
5. coordinator: forward merged dynamic filters to consumers #637
6. [do not review] 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. This will be fixed later.

Displaying Dynamic Filters

Protocol is as similar to the metrics protocol as possible. Due to double wrapping (MetricsWrapperExec wraps DistributedLeafExec, it's tricky to do the dynamic filter rewrite after doing the metrics rewrite. So rewrite_distributed_plan_with_dynamic_filters has to be called first.

let plan = rewrite_distributed_plan_with_dynamic_filters(plan).await?;
let plan = rewrite_distributed_plan_with_metrics(plan, DistributedMetricsFormat::Aggregated).await?;
println!("{}", display_plan_ascii(plan.as_ref(), true));

Testing

• Tests in tests/dynamic_filtering.rs

[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?

[jayshrivastava]

🤔 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?

The suggestion is basically to consider returning a string representation from the worker to the coordinator for displaying rather than a serialized PhysicalExpr. We will need to do some hacking to make that work.

I think we can consider this later. The serialization in this PR is simpler and uses the native PhysicalExpr serialization.

[gabotechs]

It's looking very good! love the new Store and how that generalizes to dyn filtering.

Also, really good choice having this PR shipped first before the other ones.

Left a first round of comments:

[gabotechs]

The worker protocol should not be coupled to protobuf, this should depend on either vanilla datafusion types or just raw Vec<u8>.

[gabotechs]

Same as the other datafusion related dependencies in this file, this should be an MaybeEncoded<Arc<dyn PhysicalExpr>> instead, and not be coupled to protobuf specifically.

Implementations of the WorkerChannel trait might not want to build a protobuf message here at all, or they might want to just pass an in-memory Arc<dyn PhysicalExpr>.

I think this change should be trivial.

[gabotechs]

Aren't we swallowing the error here? if you think that's fine, we can just leave a comment explaining why is it fine.

[gabotechs]

Why do we need this proto roundtrip here?

[gabotechs]

This seems to be wrong. The base plan is supposed to be the base plan that came after physical optimization, not the plan meant for visualization.

[gabotechs]

In the same way that metrics collection is optional, it'd be also consistent to make the completed dynamic filters optional. If there's people with a reason for not collecting metrics, for that same reason they might also not want to collect dynamic filters.

[gabotechs]

Why do we need to rewrite the plan here?

I imagine just mutating the dynamic filters in-place should be enough, we don't need to create a new plan.

[gabotechs]

🤔 I don't think I understand why we need this function. If there's a DistributedLeafExec present in the plan, it means that the leaf variants where already isolated