upgrade to datafusion 55 by jayshrivastava · Pull Request #540 · datafusion-contrib/datafusion-distributed

jayshrivastava · 2026-07-07T16:02:21Z

Goal

Closes #530

I would like to create a development branch to work on dynamic filtering. Datafusion 54, which is currently used in main does not have this change: apache/datafusion#21055.

This PR aims to create a development branch using apache/datafusion@01bf68c, which is the latest commit on apache/datafusion:main as of writing.

Once Datafusion 55 releases, we update this PR to the correct SHA and merge it in to datafusion-distributed:main. Hopefully the changes will be minimal because we already have all datafusion 55 changes up to apache/datafusion@01bf68c.

Summary of Significant Changes

In src/protobuf/distributed_codec.rs we now use the proto_converter argument during serde to ensure dynamic filters retain referential integrity. This automatically makes dynamic filters work.
Distribute byte ranges across partitions: feat: lower repartition_file_min_size default from 10 MiB to 1 MiB apache/datafusion#22439

Lowers repartition_file_min_size from 10 MiB to 1 MiB. The PR explicitly calls out TPC-DS SF1 dimension tables. Files may be duplicated across multiple partitions where but each partition reads a different byte range (this is hidden by ...., but we know from the correctness tests that nothing broke). A lot of tpcds queries now split across target_partitions instead of staying under-partitioned. In the tpcds plan tests, we use target_partitions=3.

Example:

-                │     t0: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
-                │     t1: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
-                │     t2: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
-                │     t3: DataSourceExec: file_groups={2 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
+                │     t0: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
+                │     t1: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
+                │     t2: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]
+                │     t3: DataSourceExec: file_groups={3 groups: [[/testdata/tpcds/plans_sf1_partitions4/date_dim/part-0.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-1.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>, /testdata/tpcds/plans_sf1_partitions4/date_dim/part-2.parquet:<int>..<int>], [/testdata/tpcds/plans_sf1_partitions4/date_dim/part-3.parquet:<int>..<int>]]}, projection=[d_date_sk, d_week_seq, d_day_name], file_type=parquet, predicate=DynamicFilter [ empty ]

dynamic_rg_pruning=eligible display changes feat(parquet): intra-file early stopping via statistics + dynamic filters apache/datafusion#22450

Adds dynamic_rg_pruning=eligible to leaf nodes. Lots of snapshots change because of this.

convert inner joins to semi joins when equivalent perf: Convert inner joins to semi joins when equivalent apache/datafusion#22652.

Left semi joins cannot be broadcast. insert_broadcast_execs correctly rejects these. However, in _inject_network_boundaries, we would assume that every CollectLeft HashJoin can be broadcast. So, I added some code (look for collect_left_hash_join_requires_single_task) to check for a BroadcastExec before committing to doing a broadcast join. Without this change, correctness tests fail. I honestly think we previously did not have good coverage for LeftSemi, which is why this bug never appeared before.
Why can't LeftSemi be broadcast? A left semi join means "emit a row on the left side of the join if it matches any row on the right". You cannot broadcast the left side of a LeftSemi join because you duplicate the left side across tasks, which may result in duplicate rows.

Example: See TPCDS query 1 (note the new LeftSemi join)

introduces the proto converter to the PhysicalExtensionCodec trait which helps dedupe dynamic filters

jayshrivastava · 2026-07-07T19:52:05Z

branch-55 vs main using --iterations 5

TPCH

=== Comparing tpch_sf10 results from engine 'datafusion-distributed-main' [prev] with 'datafusion-distributed-branch-55' [new] ===
      q1: prev= 519 ms, new= 512 ms, diff=1.01 faster ✔
      q2: prev=1943 ms, new=1432 ms, diff=1.36 faster ✅
      q3: prev=1312 ms, new=1129 ms, diff=1.16 faster ✔
      q4: prev= 371 ms, new= 522 ms, diff=1.41 slower ❌
      q5: prev=1155 ms, new=1225 ms, diff=1.06 slower ✖
      q6: prev= 345 ms, new= 402 ms, diff=1.17 slower ✖
      q7: prev=1428 ms, new=1371 ms, diff=1.04 faster ✔
      q8: prev=1687 ms, new=1608 ms, diff=1.05 faster ✔
      q9: prev=1768 ms, new=1624 ms, diff=1.09 faster ✔
     q10: prev=1457 ms, new=1279 ms, diff=1.14 faster ✔
     q11: prev= 643 ms, new= 624 ms, diff=1.03 faster ✔
     q12: prev= 564 ms, new= 614 ms, diff=1.09 slower ✖
     q13: prev= 728 ms, new= 796 ms, diff=1.09 slower ✖
     q14: prev= 627 ms, new= 730 ms, diff=1.16 slower ✖
     q15: prev= 863 ms, new= 914 ms, diff=1.06 slower ✖
     q16: prev= 671 ms, new= 502 ms, diff=1.34 faster ✅
     q17: prev= 914 ms, new=1039 ms, diff=1.14 slower ✖
     q18: prev=1258 ms, new=1337 ms, diff=1.06 slower ✖
     q19: prev= 500 ms, new= 686 ms, diff=1.37 slower ❌
     q20: prev= 908 ms, new=1087 ms, diff=1.20 slower ✖
     q21: prev=1243 ms, new=1309 ms, diff=1.05 slower ✖
     q22: prev= 371 ms, new= 361 ms, diff=1.03 faster ✔
   TOTAL: prev=21275 ms, new=21103 ms, diff=1.01 faster ✔

For queries that got faster, I would expect that they have local dynamic filters, which should work in this branch.
I can't verify if pushdown is happening for sure because we don't display dynamic filters (will fix soon #529), but I at least know where the local dynamic filters should be active.

Also, I could actually try to run individual queries with metrics but I'm not sure if it's worth the time since the overall diff is 1.01x faster.

q2, q3, q10, q16 are all faster, likely due to dynamic filtering. They all have a HashJoinExec local to a DataSourceExec.
q17, q20 are slower, likely because network broadcasts were removed.
q4, q6, and q19 are slower, but they also have a HashJoinExec local to a DataSourceExec 🤔. These queries are pretty short, 400-600ms. Maybe they spend time building / waiting for dynamic filters, but the filters don't prune enough rows for it to be worth it. 1.41x for q4 and 1.37x slower for q19 is quite bad though :/

TPCDS

=== Comparing tpcds_sf1 results from engine 'datafusion-distributed-main' [prev] with 'datafusion-distributed-branch-55' [new] ===
      q1: prev= 334 ms, new= 320 ms, diff=1.04 faster ✔
      q2: prev=1304 ms, new= 963 ms, diff=1.35 faster ✅
      q3: prev= 719 ms, new= 585 ms, diff=1.23 faster ✅
      q4: prev=2930 ms, new=3059 ms, diff=1.04 slower ✖
      q5: prev= 723 ms, new= 611 ms, diff=1.18 faster ✔
      q6: prev= 960 ms, new= 857 ms, diff=1.12 faster ✔
      q7: prev= 584 ms, new= 719 ms, diff=1.23 slower ❌
      q8: prev= 393 ms, new= 342 ms, diff=1.15 faster ✔
      q9: prev= 320 ms, new= 284 ms, diff=1.13 faster ✔
     q10: prev= 710 ms, new= 522 ms, diff=1.36 faster ✅
     q11: prev=2165 ms, new=2018 ms, diff=1.07 faster ✔
     q12: prev= 183 ms, new= 137 ms, diff=1.34 faster ✅
     q13: prev= 752 ms, new= 753 ms, diff=1.00 slower ✖
     q14: prev=1105 ms, new=1236 ms, diff=1.12 slower ✖
     q15: prev= 350 ms, new= 302 ms, diff=1.16 faster ✔
     q16: prev= 332 ms, new= 380 ms, diff=1.14 slower ✖
     q17: prev= 731 ms, new= 505 ms, diff=1.45 faster ✅
     q18: prev= 573 ms, new= 534 ms, diff=1.07 faster ✔
     q19: prev= 281 ms, new= 288 ms, diff=1.02 slower ✖
     q20: prev= 141 ms, new= 162 ms, diff=1.15 slower ✖
     q21: prev= 145 ms, new= 179 ms, diff=1.23 slower ❌
     q22: prev= 459 ms, new= 475 ms, diff=1.03 slower ✖
     q23: prev=1043 ms, new=1013 ms, diff=1.03 faster ✔
     q24: prev=1119 ms, new=1204 ms, diff=1.08 slower ✖
     q25: prev= 539 ms, new= 481 ms, diff=1.12 faster ✔
     q26: prev= 327 ms, new= 258 ms, diff=1.27 faster ✅
     q27: prev= 576 ms, new= 519 ms, diff=1.11 faster ✔
     q28: prev= 187 ms, new= 323 ms, diff=1.73 slower ❌
     q29: prev= 688 ms, new= 528 ms, diff=1.30 faster ✅
     q31: prev=1252 ms, new=1495 ms, diff=1.19 slower ✖
     q32: prev= 242 ms, new= 185 ms, diff=1.31 faster ✅
     q33: prev= 447 ms, new= 352 ms, diff=1.27 faster ✅
     q34: prev= 348 ms, new= 413 ms, diff=1.19 slower ✖
     q35: prev= 767 ms, new= 504 ms, diff=1.52 faster ✅
     q36: prev= 348 ms, new= 501 ms, diff=1.44 slower ❌
     q37: prev= 454 ms, new= 348 ms, diff=1.30 faster ✅
     q38: prev= 599 ms, new= 656 ms, diff=1.10 slower ✖
     q39: prev= 416 ms, new= 330 ms, diff=1.26 faster ✅
     q40: prev= 525 ms, new= 447 ms, diff=1.17 faster ✔
     q41: prev=  99 ms, new= 114 ms, diff=1.15 slower ✖
     q42: prev= 251 ms, new= 290 ms, diff=1.16 slower ✖
     q43: prev= 187 ms, new= 204 ms, diff=1.09 slower ✖
     q44: prev= 276 ms, new= 321 ms, diff=1.16 slower ✖
     q45: prev= 403 ms, new= 387 ms, diff=1.04 faster ✔
     q46: prev= 559 ms, new= 478 ms, diff=1.17 faster ✔
     q47: prev= 709 ms, new= 845 ms, diff=1.19 slower ✖
     q48: prev= 509 ms, new= 539 ms, diff=1.06 slower ✖
     q49: prev= 429 ms, new= 604 ms, diff=1.41 slower ❌
     q50: prev= 586 ms, new= 555 ms, diff=1.06 faster ✔
     q51: prev= 525 ms, new= 687 ms, diff=1.31 slower ❌
     q52: prev= 214 ms, new= 272 ms, diff=1.27 slower ❌
     q53: prev= 364 ms, new= 220 ms, diff=1.65 faster ✅
     q54: prev= 898 ms, new= 664 ms, diff=1.35 faster ✅
     q55: prev= 217 ms, new= 192 ms, diff=1.13 faster ✔
     q56: prev= 341 ms, new= 361 ms, diff=1.06 slower ✖
     q57: prev= 531 ms, new= 563 ms, diff=1.06 slower ✖
     q58: prev= 738 ms, new= 946 ms, diff=1.28 slower ❌
     q59: prev= 399 ms, new= 389 ms, diff=1.03 faster ✔
     q60: prev= 353 ms, new= 342 ms, diff=1.03 faster ✔
     q61: prev=1197 ms, new=1013 ms, diff=1.18 faster ✔
     q62: prev= 438 ms, new= 505 ms, diff=1.15 slower ✖
     q63: prev= 219 ms, new= 275 ms, diff=1.26 slower ❌
     q64: prev=2014 ms, new=3278 ms, diff=1.63 slower ❌
     q65: prev= 498 ms, new= 556 ms, diff=1.12 slower ✖
     q66: prev= 503 ms, new= 461 ms, diff=1.09 faster ✔
     q67: prev= 645 ms, new= 579 ms, diff=1.11 faster ✔
     q68: prev= 507 ms, new= 485 ms, diff=1.05 faster ✔
     q69: prev= 569 ms, new= 620 ms, diff=1.09 slower ✖
     q70: prev= 508 ms, new= 499 ms, diff=1.02 faster ✔
     q71: prev= 471 ms, new= 468 ms, diff=1.01 faster ✔
     q72: prev=11347 ms, new=11450 ms, diff=1.01 slower ✖
     q73: prev= 291 ms, new= 389 ms, diff=1.34 slower ❌
     q74: prev= 901 ms, new= 757 ms, diff=1.19 faster ✔
     q75: prev= 991 ms, new=1048 ms, diff=1.06 slower ✖
     q76: prev= 283 ms, new= 346 ms, diff=1.22 slower ❌
     q77: prev= 516 ms, new= 501 ms, diff=1.03 faster ✔
     q78: prev=1143 ms, new= 938 ms, diff=1.22 faster ✅
     q79: prev= 325 ms, new= 299 ms, diff=1.09 faster ✔
     q80: prev= 707 ms, new= 523 ms, diff=1.35 faster ✅
     q81: prev= 290 ms, new= 334 ms, diff=1.15 slower ✖
     q82: prev= 572 ms, new= 430 ms, diff=1.33 faster ✅
     q83: prev= 320 ms, new= 388 ms, diff=1.21 slower ❌
     q84: prev= 343 ms, new= 354 ms, diff=1.03 slower ✖
     q85: prev= 609 ms, new= 748 ms, diff=1.23 slower ❌
     q86: prev= 144 ms, new= 180 ms, diff=1.25 slower ❌
     q87: prev= 533 ms, new= 685 ms, diff=1.29 slower ❌
     q88: prev= 462 ms, new= 504 ms, diff=1.09 slower ✖
     q89: prev= 331 ms, new= 304 ms, diff=1.09 faster ✔
     q90: prev= 206 ms, new= 218 ms, diff=1.06 slower ✖
     q91: prev= 523 ms, new= 550 ms, diff=1.05 slower ✖
     q92: prev= 317 ms, new= 313 ms, diff=1.01 faster ✔
     q93: prev= 422 ms, new= 470 ms, diff=1.11 slower ✖
     q94: prev= 440 ms, new= 370 ms, diff=1.19 faster ✔
     q95: prev= 679 ms, new= 481 ms, diff=1.41 faster ✅
     q96: prev= 235 ms, new= 310 ms, diff=1.32 slower ❌
     q97: prev= 359 ms, new= 368 ms, diff=1.03 slower ✖
     q98: prev= 194 ms, new= 249 ms, diff=1.28 slower ❌
     q99: prev= 748 ms, new= 751 ms, diff=1.00 slower ✖
   TOTAL: prev=66459 ms, new=66758 ms, diff=1.00 slower ✖

jayshrivastava · 2026-07-08T19:47:17Z

@gabotechs What do you think about the benchmarks above? Unremarkable? I'm not sure if it's worth analyzing why things are slow or fast because on average, the improvement is 1x. Seems like the variance is high though. Some queries are 30-40% faster, some are 30-40% slower. I ran these with --iterations 5.

gabotechs · 2026-07-09T08:24:00Z

I see them within the typical noise. You can try running them with --iterations 10 and see if you reduce the noise.

Typically, the most relevant bits you want to look at for overall performance are these:

   TOTAL: prev=21275 ms, new=21103 ms, diff=1.01 faster ✔

   TOTAL: prev=66459 ms, new=66758 ms, diff=1.00 slower ✖

So with those benchmarks my read is that there's just no impact on performance.

jayshrivastava added 2 commits July 7, 2026 15:32

update datafusion to 01bf68cdc46ffbf4c226e760b0ba8a63fdb89a6b

44d85d8

introduces the proto converter to the PhysicalExtensionCodec trait which helps dedupe dynamic filters

poke

a6c3268

jayshrivastava changed the title ~~Branch 55~~ upgrade to datafusion 55 Jul 7, 2026

jayshrivastava marked this pull request as ready for review July 7, 2026 16:04

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

upgrade to datafusion 55#540

upgrade to datafusion 55#540
jayshrivastava wants to merge 2 commits into
mainfrom
branch-55

jayshrivastava commented Jul 7, 2026 •

edited

Loading

Uh oh!

jayshrivastava commented Jul 7, 2026 •

edited

Loading

Uh oh!

jayshrivastava commented Jul 8, 2026

Uh oh!

gabotechs commented Jul 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

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

Uh oh!

Goal

Summary of Significant Changes

Uh oh!

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

Uh oh!

Uh oh!

jayshrivastava commented Jul 8, 2026

Uh oh!

gabotechs commented Jul 9, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

jayshrivastava commented Jul 7, 2026 •

edited

Loading

jayshrivastava commented Jul 7, 2026 •

edited

Loading