worker: apply merged dynamic filters during execution by jayshrivastava · Pull Request #639 · datafusion-contrib/datafusion-distributed

jayshrivastava · 2026-08-13T13:29:59Z

Stack

This stack of PRs implements distributed dynamic filtering #528

Details

Apply dynamic filter updates on consumer tasks locally to the plan nodes that need them.

This PR contains 2 commits

feat: apply merged dynamic filter updates on worker consumers

Decode and merge dynamic filter updates on workers
update() consumers to use those dynamic filters
A lot of tests.

fix: send producer schema with dynamic filters

When decoding a physical expression, we have to decode it against a schema. Previously, we were decoding producer dynamic filters against the consumer schema, which was causing errors. To fix this, we now store the producer schema in the registry and send it along with dynamic filter updates.

Closes #532
Closes #665

gabotechs · 2026-09-22T14:37:11Z

benchmarks run tpch/sf100 --base main

gabot-0 · 2026-09-22T14:37:39Z

Requested by this comment.

Benchmark results

Compared: Main 0f6fe65e5a5f → PR head 060f044e61e2 · View exact source diff

=== Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=1234.0, new=1234.0, diff=no change (sum of per-query averages)
TOTAL: prev=53919 ms, new=54561 ms, diff=1.01 slower ✖

Show full query output

      q1: prev=1838 ms, new=1759 ms, diff=1.04 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q2: prev=1730 ms, new=1727 ms, diff=1.00 faster ✔, tasks: prev=92.0, new=92.0, diff=no change
      q3: prev=2521 ms, new=2434 ms, diff=1.04 faster ✔, tasks: prev=52.0, new=52.0, diff=no change
      q4: prev=1197 ms, new=1332 ms, diff=1.11 slower ✖, tasks: prev=48.0, new=48.0, diff=no change
      q5: prev=2788 ms, new=2851 ms, diff=1.02 slower ✖, tasks: prev=79.0, new=79.0, diff=no change
      q6: prev=1050 ms, new=1012 ms, diff=1.04 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
      q7: prev=3317 ms, new=3189 ms, diff=1.04 faster ✔, tasks: prev=79.0, new=79.0, diff=no change
      q8: prev=3809 ms, new=4046 ms, diff=1.06 slower ✖, tasks: prev=94.0, new=94.0, diff=no change
      q9: prev=4861 ms, new=4901 ms, diff=1.01 slower ✖, tasks: prev=100.0, new=100.0, diff=no change
     q10: prev=4621 ms, new=4820 ms, diff=1.04 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
     q11: prev=1017 ms, new=1048 ms, diff=1.03 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
     q12: prev=1493 ms, new=1630 ms, diff=1.09 slower ✖, tasks: prev=48.0, new=48.0, diff=no change
     q13: prev=1255 ms, new=1158 ms, diff=1.08 faster ✔, tasks: prev=40.0, new=40.0, diff=no change
     q14: prev=1476 ms, new=1687 ms, diff=1.14 slower ✖, tasks: prev=26.0, new=26.0, diff=no change
     q15: prev=3128 ms, new=2864 ms, diff=1.09 faster ✔, tasks: prev=50.0, new=50.0, diff=no change
     q16: prev= 930 ms, new= 907 ms, diff=1.03 faster ✔, tasks: prev=51.0, new=51.0, diff=no change
     q17: prev=3261 ms, new=3554 ms, diff=1.09 slower ✖, tasks: prev=38.0, new=38.0, diff=no change
     q18: prev=3713 ms, new=3774 ms, diff=1.02 slower ✖, tasks: prev=64.0, new=64.0, diff=no change
     q19: prev=1465 ms, new=1567 ms, diff=1.07 slower ✖, tasks: prev=26.0, new=26.0, diff=no change
     q20: prev=2419 ms, new=2249 ms, diff=1.08 faster ✔, tasks: prev=63.0, new=63.0, diff=no change
     q21: prev=5250 ms, new=5249 ms, diff=1.00 faster ✔, tasks: prev=86.0, new=86.0, diff=no change
     q22: prev= 780 ms, new= 803 ms, diff=1.03 slower ✖, tasks: prev=32.0, new=32.0, diff=no change

Verification and run details

Job 176 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-remote-worker --bin worker target from that checkout.

Identity	Main	PR head
Source commit	`0f6fe65e5a5fd305e63dfc535b47a643b7cda49e`	`060f044e61e2a3cc838256cceda8b59892b3391f`

Phase	Base	PR head
Build and deployment	1m 54s	1m 36s
All benchmarks	5m 52s	5m 49s
Benchmark `tpch/sf100`	5m 52s	5m 49s

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

Capacity: 12 c5n.4xlarge nodes for both revisions

Other timings: Queue 1s · Dataset validation 0s · Total 15m 22s

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.

jayshrivastava · 2026-09-22T15:48:40Z

benchmarks run tpch/sf100 --base main --config datafusion.execution.parquet.pushdown_filters=true --config datafusion.execution.parquet.reorder_filters=true

gabot-0 · 2026-09-22T15:49:12Z

Requested by this comment.

Benchmark results

Compared: Main 0f6fe65e5a5f → PR head 08f286fa2c14 · View exact source diff

=== Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=1234.0, new=1234.0, diff=no change (sum of per-query averages)
TOTAL: prev=60457 ms, new=93389 ms, diff=1.54 slower ❌

Show full query output

      q1: prev=1980 ms, new=1924 ms, diff=1.03 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q2: prev=1745 ms, new=2410 ms, diff=1.38 slower ✖, tasks: prev=92.0, new=92.0, diff=no change
      q3: prev=2623 ms, new=3550 ms, diff=1.35 slower ✖, tasks: prev=52.0, new=52.0, diff=no change
      q4: prev=1240 ms, new=2350 ms, diff=1.90 slower ❌, tasks: prev=48.0, new=48.0, diff=no change
      q5: prev=3263 ms, new=4116 ms, diff=1.26 slower ✖, tasks: prev=79.0, new=79.0, diff=no change
      q6: prev=1102 ms, new=1769 ms, diff=1.61 slower ❌, tasks: prev=12.0, new=12.0, diff=no change
      q7: prev=3257 ms, new=4552 ms, diff=1.40 slower ✖, tasks: prev=79.0, new=79.0, diff=no change
      q8: prev=3805 ms, new=7851 ms, diff=2.06 slower ❌, tasks: prev=94.0, new=94.0, diff=no change
      q9: prev=4620 ms, new=9213 ms, diff=1.99 slower ❌, tasks: prev=100.0, new=100.0, diff=no change
     q10: prev=4830 ms, new=6173 ms, diff=1.28 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
     q11: prev=1126 ms, new=1641 ms, diff=1.46 slower ✖, tasks: prev=65.0, new=65.0, diff=no change
     q12: prev=1719 ms, new=3132 ms, diff=1.82 slower ❌, tasks: prev=48.0, new=48.0, diff=no change
     q13: prev=1212 ms, new=1819 ms, diff=1.50 slower ❌, tasks: prev=40.0, new=40.0, diff=no change
     q14: prev=1497 ms, new=1900 ms, diff=1.27 slower ✖, tasks: prev=26.0, new=26.0, diff=no change
     q15: prev=7430 ms, new=3779 ms, diff=1.97 faster ✅, tasks: prev=50.0, new=50.0, diff=no change
     q16: prev=1333 ms, new=1900 ms, diff=1.43 slower ✖, tasks: prev=51.0, new=51.0, diff=no change
     q17: prev=3675 ms, new=13227 ms, diff=3.60 slower ❌, tasks: prev=38.0, new=38.0, diff=no change
     q18: prev=3845 ms, new=4842 ms, diff=1.26 slower ✖, tasks: prev=64.0, new=64.0, diff=no change
     q19: prev=1473 ms, new=3388 ms, diff=2.30 slower ❌, tasks: prev=26.0, new=26.0, diff=no change
     q20: prev=2294 ms, new=4504 ms, diff=1.96 slower ❌, tasks: prev=63.0, new=63.0, diff=no change
     q21: prev=5257 ms, new=8174 ms, diff=1.55 slower ❌, tasks: prev=86.0, new=86.0, diff=no change
     q22: prev=1131 ms, new=1175 ms, diff=1.04 slower ✖, tasks: prev=32.0, new=32.0, diff=no change

Verification and run details

Job 178 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-remote-worker --bin worker target from that checkout.

Identity	Main	PR head
Source commit	`0f6fe65e5a5fd305e63dfc535b47a643b7cda49e`	`08f286fa2c143aa05b7fa2688556f977783bf126`

Phase	Base	PR head
Build and deployment	1m 51s	1m 35s
All benchmarks	6m 29s	9m 47s
Benchmark `tpch/sf100`	6m 29s	9m 47s

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

Capacity: 12 c5n.4xlarge nodes for both revisions

PR-head configs: datafusion.execution.parquet.pushdown_filters=true, datafusion.execution.parquet.reorder_filters=true

Other timings: Queue 1s · Dataset validation 0s · Total 19m 51s

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.

jayshrivastava · 2026-09-22T16:27:50Z

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

gabot-0 · 2026-09-22T16:28:15Z

Requested by this comment.

Benchmark results

Compared: Main 0f6fe65e5a5f → PR head 08f286fa2c14 · View exact source diff

=== Comparing clickbench/0-100-date32 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=912.0, new=909.6, diff=2.4 fewer (0.3%) (sum of per-query averages)
TOTAL: prev=92765 ms, new=82931 ms, diff=1.12 faster ✅

Show full query output

      q0: prev=   1 ms, new=   1 ms, diff=1.00 slower ✖, tasks: prev=0.0, new=0.0, diff=no change
      q1: prev= 481 ms, new= 356 ms, diff=1.35 faster ✔, tasks: prev=12.0, new=9.6, diff=2.4 fewer (20.0%)
      q2: prev= 393 ms, new= 446 ms, diff=1.13 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
      q3: prev= 566 ms, new= 354 ms, diff=1.60 faster ✅, tasks: prev=12.0, new=12.0, diff=no change
      q4: prev= 647 ms, new= 473 ms, diff=1.37 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q5: prev= 905 ms, new= 658 ms, diff=1.38 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q6: prev=   2 ms, new=   2 ms, diff=1.00 slower ✖, tasks: prev=0.0, new=0.0, diff=no change
      q7: prev= 544 ms, new= 446 ms, diff=1.22 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q8: prev= 690 ms, new= 517 ms, diff=1.33 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
      q9: prev= 525 ms, new= 479 ms, diff=1.10 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q10: prev= 869 ms, new= 549 ms, diff=1.58 faster ✅, tasks: prev=36.0, new=36.0, diff=no change
     q11: prev= 642 ms, new= 555 ms, diff=1.16 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
     q12: prev=1263 ms, new= 721 ms, diff=1.75 faster ✅, tasks: prev=24.0, new=24.0, diff=no change
     q13: prev= 905 ms, new= 861 ms, diff=1.05 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
     q14: prev= 859 ms, new= 824 ms, diff=1.04 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q15: prev= 545 ms, new= 535 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q16: prev= 802 ms, new= 798 ms, diff=1.01 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q17: prev= 829 ms, new= 715 ms, diff=1.16 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q18: prev= 902 ms, new= 779 ms, diff=1.16 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q19: prev= 466 ms, new= 469 ms, diff=1.01 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q20: prev=8579 ms, new=8572 ms, diff=1.00 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q21: prev=7873 ms, new=6448 ms, diff=1.22 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q22: prev=7228 ms, new=7354 ms, diff=1.02 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q23: prev=21475 ms, new=18992 ms, diff=1.13 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q24: prev= 630 ms, new= 646 ms, diff=1.03 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q25: prev= 670 ms, new= 784 ms, diff=1.17 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q26: prev= 566 ms, new= 595 ms, diff=1.05 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q27: prev=7751 ms, new=6611 ms, diff=1.17 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q28: prev=2789 ms, new=2357 ms, diff=1.18 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q29: prev= 299 ms, new= 232 ms, diff=1.29 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q30: prev= 840 ms, new= 684 ms, diff=1.23 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q31: prev=1261 ms, new= 679 ms, diff=1.86 faster ✅, tasks: prev=24.0, new=24.0, diff=no change
     q32: prev= 717 ms, new= 605 ms, diff=1.19 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q33: prev=8133 ms, new=6787 ms, diff=1.20 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q34: prev=6804 ms, new=6642 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q35: prev= 500 ms, new= 468 ms, diff=1.07 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q36: prev= 850 ms, new= 745 ms, diff=1.14 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q37: prev= 524 ms, new= 605 ms, diff=1.15 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q38: prev= 684 ms, new= 694 ms, diff=1.01 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q39: prev= 875 ms, new= 930 ms, diff=1.06 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q40: prev= 321 ms, new= 305 ms, diff=1.05 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q41: prev= 305 ms, new= 343 ms, diff=1.12 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q42: prev= 255 ms, new= 315 ms, diff=1.24 slower ✖, tasks: prev=24.0, new=24.0, diff=no change

Verification and run details

Job 179 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-remote-worker --bin worker target from that checkout.

Identity	Main	PR head
Source commit	`0f6fe65e5a5fd305e63dfc535b47a643b7cda49e`	`08f286fa2c143aa05b7fa2688556f977783bf126`

Phase	Base	PR head
Build and deployment	1m 37s	1m 36s
All benchmarks	10m 33s	9m 2s
Benchmark `clickbench/0-100-date32`	10m 33s	9m 2s

Workload: clickbench/0-100-date32 · all queries · 1 warmup + 5 measured iterations per query

Capacity: 12 c5n.4xlarge nodes for both revisions

Other timings: Queue 1s · Dataset validation 0s · Total 22m 56s

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.

jayshrivastava · 2026-09-22T18:03:36Z

benchmarks run clickbench/0-100-date32 --iterations 20

gabot-0 · 2026-09-22T18:03:50Z

Requested by this comment.

Benchmark results

Compared: PR base 8e449ba3a75c → PR head 08f286fa2c14 · View exact source diff

=== Comparing clickbench/0-100-date32 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TASKS: prev=912.0, new=910.8, diff=1.2 fewer (0.1%) (sum of per-query averages)
TOTAL: prev=86729 ms, new=68257 ms, diff=1.27 faster ✅

Show full query output

      q0: prev=   1 ms, new=   1 ms, diff=1.00 slower ✖, tasks: prev=0.0, new=0.0, diff=no change
      q1: prev= 441 ms, new= 321 ms, diff=1.37 faster ✔, tasks: prev=12.0, new=10.8, diff=1.2 fewer (10.0%)
      q2: prev= 323 ms, new= 275 ms, diff=1.17 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
      q3: prev= 397 ms, new= 286 ms, diff=1.39 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
      q4: prev= 481 ms, new= 372 ms, diff=1.29 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q5: prev= 787 ms, new= 673 ms, diff=1.17 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q6: prev=   2 ms, new=   2 ms, diff=1.00 slower ✖, tasks: prev=0.0, new=0.0, diff=no change
      q7: prev= 524 ms, new= 360 ms, diff=1.46 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
      q8: prev= 707 ms, new= 565 ms, diff=1.25 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
      q9: prev= 598 ms, new= 589 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q10: prev= 680 ms, new= 573 ms, diff=1.19 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
     q11: prev= 633 ms, new= 607 ms, diff=1.04 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
     q12: prev= 871 ms, new= 739 ms, diff=1.18 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q13: prev= 919 ms, new= 851 ms, diff=1.08 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
     q14: prev= 779 ms, new= 761 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q15: prev= 494 ms, new= 431 ms, diff=1.15 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q16: prev= 780 ms, new= 749 ms, diff=1.04 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q17: prev= 744 ms, new= 726 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q18: prev= 874 ms, new= 783 ms, diff=1.12 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q19: prev= 431 ms, new= 568 ms, diff=1.32 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q20: prev=7640 ms, new=6527 ms, diff=1.17 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q21: prev=7233 ms, new=4960 ms, diff=1.46 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q22: prev=8105 ms, new=4797 ms, diff=1.69 faster ✅, tasks: prev=24.0, new=24.0, diff=no change
     q23: prev=20184 ms, new=15523 ms, diff=1.30 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q24: prev= 685 ms, new= 548 ms, diff=1.25 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q25: prev= 611 ms, new= 650 ms, diff=1.06 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
     q26: prev= 587 ms, new= 546 ms, diff=1.08 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q27: prev=6082 ms, new=4927 ms, diff=1.23 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q28: prev=2606 ms, new=2165 ms, diff=1.20 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q29: prev= 333 ms, new= 237 ms, diff=1.41 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
     q30: prev= 787 ms, new= 722 ms, diff=1.09 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q31: prev= 730 ms, new= 693 ms, diff=1.05 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q32: prev= 685 ms, new= 684 ms, diff=1.00 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q33: prev=7454 ms, new=5314 ms, diff=1.40 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q34: prev=7098 ms, new=5462 ms, diff=1.30 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q35: prev= 445 ms, new= 399 ms, diff=1.12 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q36: prev= 784 ms, new= 759 ms, diff=1.03 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q37: prev= 537 ms, new= 550 ms, diff=1.02 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q38: prev= 734 ms, new= 695 ms, diff=1.06 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q39: prev= 943 ms, new= 881 ms, diff=1.07 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q40: prev= 335 ms, new= 311 ms, diff=1.08 faster ✔, tasks: prev=24.0, new=24.0, diff=no change
     q41: prev= 334 ms, new= 352 ms, diff=1.05 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
     q42: prev= 331 ms, new= 323 ms, diff=1.02 faster ✔, tasks: prev=24.0, new=24.0, diff=no change

Verification and run details

Job 182 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-remote-worker --bin worker target from that checkout.

Identity	PR base	PR head
Source commit	`8e449ba3a75cac1b9b86432dab3e99b5655aba4d`	`08f286fa2c143aa05b7fa2688556f977783bf126`

Phase	Base	PR head
Build and deployment	1m 54s	1m 36s
All benchmarks	33m 45s	25m 41s
Benchmark `clickbench/0-100-date32`	33m 45s	25m 41s

Workload: clickbench/0-100-date32 · all queries · 1 warmup + 20 measured iterations per query for both revisions

Capacity: 12 c5n.4xlarge nodes for both revisions

Other timings: Queue 1s · Dataset validation 0s · Total 1h 3m 4s

How to use the benchmark bot

Post a comment whose first non-empty line is:

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

For example:

benchmarks run tpch/sf10 tpch/sf100 --instance-type m5.2xlarge --nodes 24 --iterations 20 --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.
`--iterations <count>`	`5`	A positive safe integer. Measured iterations per query for both revisions; warmup is excluded.
`--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 is not supported. By default, every query uses 1 warmup and 5 measured iterations for both revisions.

Carry full dynamic filters through MaybeEncoded without exposing protobuf types in the transport API. Encode each merged snapshot once, enqueue updates under the registry lock, and send the latest state to late consumers. Keep delivery fail-open and release senders when the query ends.

jayshrivastava · 2026-09-23T01:40:39Z

+            task_count: stage.tasks,
+        };
+        // Union children can use different task indices from their enclosing stage.
+        stage.plan.apply_with_dt_ctx(d_ctx, |node, d_ctx| {


This is the main change in this function. We previously did not update the children of unions correctly.

Apply successive full dynamic filter snapshots in producer coordinates and mark completion only when reported. Reuse deduplicated consumer discovery, preserve alias remapping and task-local relationships, and verify remote join, TopK, and aggregate filters with SQL integration tests and enabled/disabled result comparisons.

When decoding a physical expression, we have to decode it against a schema. Previously, we were decoding producer dynamic filters against the consumer schema, which was causing errors. Example: A producer may send a filter on `my_col@2` but the consumer data source may have this column at a different index ex. `my_col@8`. To fix this, we now store the producer schema in the registry and send it along with dynamic filter updates.

jayshrivastava · 2026-09-23T21:21:59Z

                r: None,
-                op: "Or".to_owned(),
+                op: if mode == DynamicFilterMergeMode::Incremental {
+                    "And"


For aggregates and sorts, this is tested via the integration test. Also TPCDS has a ton of queries that should exercise this: q1, q7, q8, q15, q26, q27, q37, q43, q50, q58, q69, q74, q82, q83, q84

Replay completed task predicates into isolated per-task display variants. Add SQL regression snapshots for local and cross-stage FilterExec consumers.

jayshrivastava changed the base branch from main to js/5-forward-dynamic-filters August 13, 2026 13:30

jayshrivastava force-pushed the js/6-apply-merged-filters branch from fe0486a to 76f01c5 Compare August 13, 2026 19:27

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 3229903 to ad77338 Compare August 13, 2026 19:27

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 76f01c5 to c82e668 Compare August 17, 2026 18:54

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from ad77338 to 15e6602 Compare August 17, 2026 18:54

jayshrivastava force-pushed the js/6-apply-merged-filters branch from c82e668 to 720a540 Compare August 17, 2026 19:05

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from 56fee53 to 1572836 Compare August 17, 2026 19:38

jayshrivastava force-pushed the js/6-apply-merged-filters branch 2 times, most recently from 30ebeec to 3cfb41a Compare August 18, 2026 18:15

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from 296026d to e04ceac Compare August 18, 2026 18:52

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 3cfb41a to 67b5a4b Compare August 18, 2026 18:52

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from e04ceac to e720301 Compare August 21, 2026 16:28

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 67b5a4b to db3f95a Compare August 21, 2026 16:28

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from e720301 to 7a74a70 Compare August 21, 2026 16:48

jayshrivastava force-pushed the js/6-apply-merged-filters branch from db3f95a to 5378b1e Compare August 21, 2026 16:48

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 7a74a70 to c1b6398 Compare August 21, 2026 20:58

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 5378b1e to 6cf8e33 Compare August 21, 2026 20:58

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from c1b6398 to 6a70c73 Compare August 22, 2026 15:00

jayshrivastava force-pushed the js/6-apply-merged-filters branch 2 times, most recently from 51940d0 to 86dfdfb Compare August 23, 2026 00:21

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from 0a68111 to 42cb2ac Compare August 23, 2026 15:27

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 86dfdfb to 79e4299 Compare August 23, 2026 15:27

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 96fe420 to c884cec Compare September 21, 2026 23:31

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch 2 times, most recently from acdb19a to da1fa7f Compare September 22, 2026 03:55

jayshrivastava force-pushed the js/6-apply-merged-filters branch from c884cec to 060f044 Compare September 22, 2026 04:24

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from da1fa7f to 8e449ba Compare September 22, 2026 14:46

jayshrivastava force-pushed the js/6-apply-merged-filters branch from 060f044 to 08f286f Compare September 22, 2026 14:46

datafusion-contrib deleted a comment from gabot-0 Sep 22, 2026

jayshrivastava force-pushed the js/6-apply-merged-filters branch from add27be to efb5037 Compare September 23, 2026 01:22

jayshrivastava added 2 commits September 23, 2026 01:23

handle errors and query cancellation better

abd172e

jayshrivastava force-pushed the js/5-forward-dynamic-filters branch from 8e449ba to abd172e Compare September 23, 2026 01:23

jayshrivastava force-pushed the js/6-apply-merged-filters branch from efb5037 to c8de0ff Compare September 23, 2026 01:23

jayshrivastava commented Sep 23, 2026

View reviewed changes

jayshrivastava added 2 commits September 23, 2026 21:20

jayshrivastava commented Sep 23, 2026

View reviewed changes

fix: display dynamic filters on intermediate FilterExec nodes

c78f9e1

Replay completed task predicates into isolated per-task display variants. Add SQL regression snapshots for local and cross-stage FilterExec consumers.

barbarj mentioned this pull request Sep 25, 2026

Adopt native dynamic-filter propagation in MPP paradedb/paradedb#5559

Open

3 tasks

Navigation Menu

worker: apply merged dynamic filters during execution - #639

worker: apply merged dynamic filters during execution#639
jayshrivastava wants to merge 5 commits into
js/5-forward-dynamic-filtersfrom
js/6-apply-merged-filters

jayshrivastava commented Aug 13, 2026 •

edited

Loading

Uh oh!

gabotechs commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 •

edited

Loading

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 •

edited

Loading

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 •

edited

Loading

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 •

edited

Loading

Uh oh!

jayshrivastava Sep 23, 2026

Uh oh!

jayshrivastava Sep 23, 2026

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

jayshrivastava commented Aug 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Stack

Details

Uh oh!

gabotechs commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmark results

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmark results

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmark results

Uh oh!

jayshrivastava commented Sep 22, 2026

Uh oh!

gabot-0 commented Sep 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Benchmark results

Uh oh!

jayshrivastava Sep 23, 2026

Choose a reason for hiding this comment

Uh oh!

jayshrivastava Sep 23, 2026

Choose a reason for hiding this comment

Uh oh!

This branch has not been deployed

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

jayshrivastava commented Aug 13, 2026 •

edited

Loading

gabot-0 commented Sep 22, 2026 •

edited

Loading

gabot-0 commented Sep 22, 2026 •

edited

Loading

gabot-0 commented Sep 22, 2026 •

edited

Loading

gabot-0 commented Sep 22, 2026 •

edited

Loading