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

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

[jayshrivastava]

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=80526 ms, new=81697 ms, diff=1.01 slower ✖

Show full query output

      q1: prev=2826 ms, new=2797 ms, diff=1.01 faster ✔, tasks: prev=18.0, new=18.0, diff=no change
      q2: prev=3287 ms, new=3455 ms, diff=1.05 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
      q3: prev=3139 ms, new=3044 ms, diff=1.03 faster ✔, tasks: prev=48.0, new=48.0, diff=no change
      q4: prev=1512 ms, new=1420 ms, diff=1.06 faster ✔, tasks: prev=38.0, new=38.0, diff=no change
      q5: prev=4618 ms, new=4973 ms, diff=1.08 slower ✖, tasks: prev=52.0, new=52.0, diff=no change
      q6: prev=1419 ms, new=1564 ms, diff=1.10 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
      q7: prev=9901 ms, new=9935 ms, diff=1.00 slower ✖, tasks: prev=53.0, new=53.0, diff=no change
      q8: prev=5211 ms, new=5675 ms, diff=1.09 slower ✖, tasks: prev=73.0, new=73.0, diff=no change
      q9: prev=9713 ms, new=10151 ms, diff=1.05 slower ✖, tasks: prev=77.0, new=77.0, diff=no change
     q10: prev=7519 ms, new=8038 ms, diff=1.07 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
     q11: prev=3489 ms, new=3642 ms, diff=1.04 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q12: prev=2223 ms, new=2076 ms, diff=1.07 faster ✔, tasks: prev=44.0, new=44.0, diff=no change
     q13: prev=1917 ms, new=1946 ms, diff=1.02 slower ✖, tasks: prev=32.0, new=32.0, diff=no change
     q14: prev=2150 ms, new=1783 ms, diff=1.21 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q15: prev=4503 ms, new=4086 ms, diff=1.10 faster ✔, tasks: prev=30.0, new=30.0, diff=no change
     q16: prev= 837 ms, new= 863 ms, diff=1.03 slower ✖, tasks: prev=47.0, new=47.0, diff=no change
     q17: prev=5190 ms, new=5221 ms, diff=1.01 slower ✖, tasks: prev=40.0, new=40.0, diff=no change
     q18: prev=5891 ms, new=5917 ms, diff=1.00 slower ✖, tasks: prev=68.0, new=68.0, diff=no change
     q19: prev=2130 ms, new=2126 ms, diff=1.00 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q20: prev=3051 ms, new=2985 ms, diff=1.02 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 47 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	2m 28s	1m 24s
All benchmarks	8m 28s	9m 2s
Benchmark tpch/sf100	8m 28s	9m 2s

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 21m 31s

[jayshrivastava]

benchmarks run tpch/sf10

[gabot-0]

Requested by this comment.

Benchmark results

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

=== Comparing tpch/sf10 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=562.0, new=560.0, diff=2.0 fewer (0.4%) (sum of per-query averages)
TOTAL: prev=18866 ms, new=20422 ms, diff=1.08 slower ✖

Show full query output

      q1: prev= 344 ms, new= 630 ms, diff=1.83 slower ❌, tasks: prev=18.0, new=16.0, diff=2.0 fewer (11.1%)
      q2: prev= 816 ms, new= 887 ms, diff=1.09 slower ✖, tasks: prev=6.0, new=6.0, diff=no change
      q3: prev= 728 ms, new= 970 ms, diff=1.33 slower ✖, tasks: prev=28.0, new=28.0, diff=no change
      q4: prev= 326 ms, new= 586 ms, diff=1.80 slower ❌, tasks: prev=30.0, new=30.0, diff=no change
      q5: prev= 937 ms, new=1022 ms, diff=1.09 slower ✖, tasks: prev=32.0, new=32.0, diff=no change
      q6: prev= 526 ms, new= 341 ms, diff=1.54 faster ✅, tasks: prev=12.0, new=12.0, diff=no change
      q7: prev=1294 ms, new=1346 ms, diff=1.04 slower ✖, tasks: prev=33.0, new=33.0, diff=no change
      q8: prev=1156 ms, new=1352 ms, diff=1.17 slower ✖, tasks: prev=54.0, new=54.0, diff=no change
      q9: prev=1510 ms, new=1595 ms, diff=1.06 slower ✖, tasks: prev=57.0, new=57.0, diff=no change
     q10: prev=1196 ms, new=1239 ms, diff=1.04 slower ✖, tasks: prev=20.0, new=20.0, diff=no change
     q11: prev= 587 ms, new= 560 ms, diff=1.05 faster ✔, tasks: prev=6.0, new=6.0, diff=no change
     q12: prev= 452 ms, new= 681 ms, diff=1.51 slower ❌, tasks: prev=30.0, new=30.0, diff=no change
     q13: prev= 592 ms, new= 627 ms, diff=1.06 slower ✖, tasks: prev=10.0, new=10.0, diff=no change
     q14: prev= 479 ms, new= 398 ms, diff=1.20 faster ✔, tasks: prev=21.0, new=21.0, diff=no change
     q15: prev=1074 ms, new=1035 ms, diff=1.04 faster ✔, tasks: prev=30.0, new=30.0, diff=no change
     q16: prev= 409 ms, new= 487 ms, diff=1.19 slower ✖, tasks: prev=3.0, new=3.0, diff=no change
     q17: prev= 769 ms, new= 743 ms, diff=1.03 faster ✔, tasks: prev=37.0, new=37.0, diff=no change
     q18: prev=1172 ms, new=1184 ms, diff=1.01 slower ✖, tasks: prev=45.0, new=45.0, diff=no change
     q19: prev= 427 ms, new= 569 ms, diff=1.33 slower ✖, tasks: prev=21.0, new=21.0, diff=no change
     q20: prev= 755 ms, new= 810 ms, diff=1.07 slower ✖, tasks: prev=32.0, new=32.0, diff=no change
     q21: prev=3020 ms, new=3054 ms, diff=1.01 slower ✖, tasks: prev=28.0, new=28.0, diff=no change
     q22: prev= 297 ms, new= 306 ms, diff=1.03 slower ✖, tasks: prev=9.0, new=9.0, diff=no change

Verification and run details

Job 48 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 22s	1m 22s
All benchmarks	2m 3s	2m 13s
Benchmark tpch/sf10	2m 3s	2m 13s

Workload: tpch/sf10 · 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 7m 9s

[jayshrivastava]

benchmarks run clickbench/0-100-date32 --base e5a4a77

[gabot-0]

@jayshrivastava Invalid base e5a4a779d191206c1acd5cbfb31f3fed87c57934; only main is supported.

How to use the benchmark bot

Post a comment whose first non-empty line is:

benchmarks run <suite>/<variant>... [--instance-type <type>] [--nodes <count>] [--base main] [--config <key=value>]...

For example:

benchmarks run tpch/sf10 tpch/sf100 --instance-type m5.2xlarge --nodes 24 --base main --config distributed.collect_dynamic_filters=false

Currently available datasets: clickbench/0-100, tpcds/sf1, tpch/sf1, tpch/sf10, tpch/sf100. Request one or more, without duplicates. The bot validates availability before provisioning and runs every query in each dataset.

Option	Default	Supported values and behavior
--instance-type <type>	c5n.4xlarge	One of the supported instance types listed below, subject to availability within the cluster's availability zones and quota.
--nodes <count>	12	An integer from 1 to 60. The deployment uses one benchmark worker per node.
--base main	PR base	Compare against a snapshot of main; no other explicit base is supported.
--config <key=value>	none	Apply a safe DataFusion session setting to the PR head only. Repeat for distinct keys; spaces and shell syntax are not supported.

Supported instances and per-worker limits:

Instance type	EC2 capacity	Worker requests and limits
c5n.2xlarge	8 vCPU, 21 GiB	7 vCPU, 17Gi
c5n.4xlarge	16 vCPU, 42 GiB	15 vCPU, 38Gi
m5.2xlarge	8 vCPU, 32 GiB	7 vCPU, 28Gi
m5.4xlarge	16 vCPU, 64 GiB	15 vCPU, 60Gi
r5.2xlarge	8 vCPU, 64 GiB	7 vCPU, 60Gi
r5.4xlarge	16 vCPU, 128 GiB	15 vCPU, 124Gi

Each node runs one worker. The worker allocation reserves 1 vCPU and 4 GiB for Kubernetes and system processes, then makes the rest of the selected instance available to the benchmark.

Limits: Only authorized users can enqueue jobs. Jobs run serially, the queue holds 20 active jobs, and each requester may have 3. Query selection and iteration overrides are not supported; every query uses 1 warmup and 5 measured iterations for both revisions.