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]

@gabotechs This is ready for another round. I've left some of the conversations open.

I've refactored things quite a bit, mainly moving source code to src/dynamic_filtering and moving tests to tests/dynamic_filtering. I've also added more tests for different topologies.

The main thing that's still up for discussion is dynamic filter serialization. We have to do this to basically deep copy them to break in-memory relations. I've left a lot of comments in /src/dynamic_filtering/display.rs to hopefuly make it more clear.

[gabotechs]

Still an incomplete review, but flushing today's comments, I'll resume review tomorrow. I'm mainly trying to invest some time in coming up with suggestions for improving the more sketchy parts, but for most of the things I'm not finding better alternatives to what you already have here.

Given the cards we are dealt with, this is looking like pretty solid work!

[gabotechs]

It feels a bit strange that we need to do this special case here. I'll keep reading.

[gabotechs]

😂 I really like this long ugly name.

[gabotechs]

Even if this is allowed, it feels like the grpc layer is access an internal detail of the Worker by accessing task_data_entries.

If we were to move the grpc module to a specific crate, this will start giving problems. Just something to keep in mind, I don't think we need to solve this now, what you have here is good enough.

[jayshrivastava]

Sure, let's clean this up whenever it becomes a problem.

[gabotechs]

You did a pretty good job design these functions so that they are isolated from the main code paths. I still see all these as "hacks" we need to do for workarounding the fact that shared stated is being baked into the nodes and expressions.

I think this is a broader topic worth rising upstream (if not already), and all the workarounds we need to do in this project are a strong signal that shared state, in general, should be scoped to a TaskContext, not to individual ExecutionPlans or PhysicalExprs.

Ballista and other distribution layers are going to have the same issues, so I don't think this is a problem only this project will suffer.

[jayshrivastava]

👍🏽 Will look out for ways to improve this

[gabotechs]

This is looking very good! I think this is my last round of comments before approval.

[gabotechs]

Nit: For consistency, it's probably worth it to make the signature and behavior of this function match the one above (wait_for_metrics). Unless you think there's a reason to diverge.

I think we need to clean up all this API for collecting metrics, dyn filters, etc... so this might go away at that point anyways.

[jayshrivastava]

#623 (comment)

[gabotechs]

Very well handled!

[gabotechs]

🤔 this is a behavior change that can blow up on other people's faces...

It will not blow up at Datadog, because this fits how we use DistributedExec, but I wonder if other people are actually relying on retrieving the original children of the DistributedExec after execution.

If you see opportunities for avoiding this change, I'd try to do it.

[jayshrivastava]

Reverted. I think that's fair.

There's a weird lifecycle that the metrics rewrite does that I'm just going to preserve for now.

children() is always the base plan, but rewrite_distributed_plan_with_dynamic_filters and rewrite_distributed_plan_with_metrics replaces both the children and plan_for_viz.

• It replaces the children() so users can iterate over the DistributedExec / Arc<dyn ExecutionPlan and extract metrics
• plan_for_viz is used for display_plan_ascii in this repo

I opened for 2 separate APIs

• with_new_children replaces the children
• with_plan_for_viz replaces plan_for_viz

[gabotechs]

All the conditional logic about different behaviors for when there's a prepared_execution and when there's not is getting a bit messy...

I think there are some opportunities for improving this, but probably worth exploring in another PR.

[jayshrivastava]

Removed the logic. I opted to reset the prepared_plan though. I felt that it doesn't make sense to change the children and leave the old prepared_plan / plan_for_viz.

[gabotechs]

This test is not really asserting that dynamic filters are not getting collected no?

[jayshrivastava]

It happens in execute(). We assert that the plan isn't rewritten because there's no dynamic filters to collect. I'll add a comment.

[gabotechs]

Here's one idea that is comes to mind:

What if we add yet another preparatory step for the plan in distributed_planner/, something like insert_broadcast or normalize_collect_joins, that severs all dynamic filter connections for good?

This would imply that dynamic filters will never be able to work through normal upstream mechanisms, and they should always be updated passing through the coordinator, even in the local case, but I do imagine this can simplify the overall approach, specially for future PRs.

[jayshrivastava]

I would like to leave this for now.

The tricky part is that we want to sever some relationships but not all of them. For example, if there's local dynamic filters, a producer should be able to atomically update them in memory.

In future PRs, I detect local vs remote dynamic filters during static and dynamic planning, so we can think about severing some relationships then.

[jayshrivastava]

Now that TaskCtx is gone, we can now use PreparedPlan directly

[jayshrivastava]

I decided this is the cleanest API: "wait until all task datas are present and then return them". It lets us remove the whole get() API on the store and just work with HashMap<...> directly

[gabotechs]

👍 yeap, this does look clean indeed, thanks!

[jayshrivastava]

Basically a copy paste of gather_stage_header_metrics removed below

[jayshrivastava]

@gabotechs Addressed the comments in the last few commits. I left the nontrivial comments open. I can check CI and rebase tomorrow 🫡

[jayshrivastava]

We lose these fields when doing the proto roundtrip before displaying. So, this means we actually lose these fields during execution since we serialize this plan before executing.

In a way, roundtripping the plan before displaying gives us a better picture of what's actually happening during execution.

I've addressed this here: apache/datafusion#24930, so these optimizations will come back in df56.

[gabotechs]

benchmarks run tpch/sf100

[gabot-0]

Requested by this comment.

Benchmark results

Compared: PR base 417f80977d7f → PR head 7f006309f5b8 · View exact source diff

=== Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=849.0, new=849.0, diff=no change (sum of per-query averages)
TOTAL: prev=82049 ms, new=81259 ms, diff=1.01 faster ✔

Show full query output

      q1: prev=3251 ms, new=2890 ms, diff=1.12 faster ✔, tasks: prev=18.0, new=18.0, diff=no change
      q2: prev=3485 ms, new=3668 ms, diff=1.05 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
      q3: prev=3397 ms, new=2968 ms, diff=1.14 faster ✔, tasks: prev=48.0, new=48.0, diff=no change
      q4: prev=1441 ms, new=1437 ms, diff=1.00 faster ✔, tasks: prev=38.0, new=38.0, diff=no change
      q5: prev=4820 ms, new=4810 ms, diff=1.00 faster ✔, tasks: prev=52.0, new=52.0, diff=no change
      q6: prev=1510 ms, new=1758 ms, diff=1.16 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
      q7: prev=10115 ms, new=10316 ms, diff=1.02 slower ✖, tasks: prev=53.0, new=53.0, diff=no change
      q8: prev=5518 ms, new=5374 ms, diff=1.03 faster ✔, tasks: prev=73.0, new=73.0, diff=no change
      q9: prev=9982 ms, new=10005 ms, diff=1.00 slower ✖, tasks: prev=77.0, new=77.0, diff=no change
     q10: prev=7865 ms, new=7664 ms, diff=1.03 faster ✔, tasks: prev=40.0, new=40.0, diff=no change
     q11: prev=3471 ms, new=3581 ms, diff=1.03 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q12: prev=2239 ms, new=2062 ms, diff=1.09 faster ✔, tasks: prev=44.0, new=44.0, diff=no change
     q13: prev=1991 ms, new=1959 ms, diff=1.02 faster ✔, tasks: prev=32.0, new=32.0, diff=no change
     q14: prev=1852 ms, new=1829 ms, diff=1.01 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q15: prev=4271 ms, new=4101 ms, diff=1.04 faster ✔, tasks: prev=30.0, new=30.0, diff=no change
     q16: prev= 826 ms, new= 972 ms, diff=1.18 slower ✖, tasks: prev=47.0, new=47.0, diff=no change
     q17: prev=5002 ms, new=5125 ms, diff=1.02 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
     q18: prev=5771 ms, new=5682 ms, diff=1.02 faster ✔, tasks: prev=68.0, new=68.0, diff=no change
     q19: prev=2179 ms, new=2209 ms, diff=1.01 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q20: prev=3063 ms, new=2849 ms, diff=1.08 faster ✔, tasks: prev=65.0, new=65.0, diff=no change
q21: Previously failed, and now also failed ❌
q22: Previously failed, and now also failed ❌

Verification and run details

Job 46 captured both immutable revisions when the request was queued. The bot fetched and checked out each full commit SHA in detached HEAD, then built and deployed the datafusion-distributed-benchmarks --bin worker target from that checkout.

Identity	PR base	PR head
Source commit	417f80977d7fa5b6c4be0e4d1fa7f30864c5ca94	7f006309f5b8664ca838251a0ac0053559dbeb0c

Phase	Base	PR head
Build and deployment	1m 43s	1m 23s
All benchmarks	8m 39s	8m 28s
Benchmark tpch/sf100	8m 39s	8m 28s

Workload: tpch/sf100 · all queries · 1 warmup + 5 measured iterations per query

Capacity: 12 c5n.2xlarge nodes for both revisions

Other timings: Queue 1s · Dataset validation 0s · Total 20m 21s

[gabotechs]

💯 let's go!