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

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

Summary

Closes #530

This PR updates the upstream datafusion SHA to the HEAD of https://github.com/apache/datafusion/commits/branch-55/ (edit: this branch is continuously being updated. I will make sure this PR is at the head before merging)
Rust upgade to 1.94

Changes

In src/protobuf/distributed_codec.rs we now use the proto_converter argument during serde

We still don't use the DeduplicatingProtoConverter, so dynamic filters don't necessarily work. I think this is outside the scope of this PR will be addressed in coordinator: display consumer dynamic filters after execution #623, which will be rebased after the upgrade.

ExecutionPlan::apply_expressions is added for every custom ExecutionPlan in this repo

Wrapper types (MetricsWrapperExec, WorkUnitFileScanConfig, DistributedLeafExec) delegate to the inner type
Other plans takeTreeNodeRecursion::Continue because they have no expressions (ex. SamplerExec)
Note that apply_expressions does not need to yield sort or partitioning expressions in the plan properties

We migrate from partition_statistics to statistics_from_inputs for every ExecutionPlan.

src/distributed_planner/statistics/plan_statistics.rs can just use statistics_from_inputs directly instead of doing the StatisticsWrapper workaround.

Range partitioning is now supported.

CPU costing now includes range-key comparison cost and has a new unit test. See src/distributed_planner/
statistics/complexity_cpu.rs:238.
I think there's open questions about range partitioning. I've opened an issue here to make sure it behaves as expected after the upgrade: tests on range partitioned data #628 (comment)

Peak-memory metrics use the existing gauge wire representation.

DataFusion added MetricValue::PeakMemoryUsage. It is serialized as the existing named-gauge protobuf variant to avoid a wire-format change. See src/protocol/grpc/
metrics_proto.rs:124.

The value and name survive, and aggregation is still additive, but decoding produces a generic Gauge, not PeakMemoryUsage. The practical difference is mainly display formatting: it
may render as a count rather than human-readable bytes. This is the clearest remaining compromise/risk in the upgrade.
File-scan rebalancing changed its discriminator.

DataFusion removed partitioned_by_file_group; output_partitioning.is_some() is now the source of truth. See src/events/defaults/file_scan_config.rs:43. This decides whether files are
round-robin rebalanced or split through FileGroupPartitioner, so it is behavior-sensitive even though it is a one-line migration.
Two previously ignored correctness tests were enabled.

See tests/multi_task_collect_join_repros.rs
These were upstream DataFusion correctness fixes, not fixes made locally in this upgrade.

drop(reporter) was made explicit on the sampler’s empty-input path.

The reporter sends its result on Drop; explicitly dropping it both satisfies the new compiler/lint behavior and guarantees the zero-row EOS report is sent before returning. See src/
execution_plans/sampler.rs:259.
Plan changes

dynamic_rg_pruning=eligible is now displayed on eligible scans: 1,354 occurrences in TPC-DS, 188 in TPC-H, and 12 in ClickBench
DataSourceExec now displays its output partitioning. See tests/join.rs (eventually, someone should delete this test tests on range partitioned data #628)
Project after sort. This looks like some upstream optimizer rule change ex. tests/distributed_unions.rs and tests/distributed_aggregation.rs.

-          │   SortExec: expr=[MinTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true]
-          │     ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday]
+          │   ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday]
+          │     SortExec: expr=[MaxTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true]

LocalLimitExec became more common: TPC-DS went from 0 to 20 occurrences and ClickBench from 1 to 21, reflecting additional local limit pushdown.
Subquery/semi-join plans became more distributed:
- TPC-DS CollectLeft hash joins: 615 → 610
- TPC-DS partitioned hash joins: 98 → 103
- TPC-DS left-semi occurrences: 11 → 25
- TPC-DS network shuffles: 368 → 378
- TPC-H - just a few
These are meaningful topology changes: some subqueries now use partitioned left-semi joins and therefore introduce hash shuffles instead of collecting/broadcasting one side.
- Scalar rendering improved, especially decimal literals: internal forms such as Some(0),7,2 now display as CAST(0.00 AS Decimal128(7, 2)).
Minor changes (Ex. tpcds 21)
- __common_expr_4 became __common_expr_3; that is only an internal alias renumbering.
- The projection that renamed d_date to __common_expr_2 disappeared.
- d_date is retained directly in the join output and referenced directly by partial/final aggregates.
- Column positions changed
File-group allocation changed substantially
- Some explicit RoundRobinBatch repartitions disappeared and scans gained different numbers of file groups
- 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 ]

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.

stuhood · 2026-08-06T06:22:16Z

We are also maintaining a version of this patch near the bottom of our patch stack: https://github.com/paradedb/datafusion-distributed/commits/main/ ... if you'd like to pick it, feel free!

jayshrivastava

@asolimando There's some stats changes you may be interested in. Ex. plan_statistics.rs, prepare_dynamic_plan.rs, and stage.rs.

jayshrivastava · 2026-08-12T18:51:19Z

        Partitioning::RoundRobinBatch(p) => Partitioning::RoundRobinBatch(f(*p)),
        Partitioning::Hash(hash, p) => Partitioning::Hash(hash.clone(), f(*p)),
        Partitioning::UnknownPartitioning(p) => Partitioning::UnknownPartitioning(f(*p)),
+        Partitioning::Range(range) => Partitioning::Range(range.clone()),


cc @gene-bordegaray maybe you can advise here

We can't scale up range without changing the split points so I left this as is. There's no test which really exercise range partitioned data today. RepartitionExec doesn't range partition, so I don't think this will ever be called to scale up a `RepartitionExec.

I filed #628 with some open questions we should probably answer. I can look into that issue after the upgrade.

This is what we ended up doing temporarily: paradedb#67 ... there was one shape in our plans which did result in Partitioning::Range at a network boundary. @mdashti might know more.

ya I would back what stu and the parade guys did here, most likely safer to not hide unimplemented capabilities 👍

@jayshrivastava this might need addressing. If we silently don't scale up output partitioning, an upstream NetworkShuffleExec might assume output partitions where actually scaled up, and it will try to reach for non-existing partitions, making queries fail.

It might be safer to fail for now, and address this separately.

Done. I change it to be an execution error and referenced #628

jayshrivastava · 2026-08-12T19:59:13Z

benchmarks run tpch/sf100

jayshrivastava · 2026-08-12T19:59:23Z

benchmarks run tpds/sf10

gabotechs · 2026-08-13T07:01:13Z

benchmarks run tpch/sf100

gabot-0 · 2026-08-13T07:01:30Z

Requested by this comment.

Benchmark job 27 failed for tpch/sf100 while comparing base 3b8a26635352 with head 55b8559e74e7. Full details are available in the controller journal.

gabotechs

👍 All looks good here, it seems like this is going to be a straightforward one

asolimando · 2026-08-13T07:50:25Z

@asolimando There's some stats changes you may be interested in. Ex. plan_statistics.rs, prepare_dynamic_plan.rs, and stage.rs.

Thanks Jay for the ping and for taking care of the upgrade.

I have checked the changes related to the changes in DF I authored and all LGTM, especially happy we could remove StatisticsWrapper!

I can take a closer look early next week when I am back in case the PR will still be open.

gabotechs · 2026-08-13T09:26:04Z

benchmarks run tpch/sf100

gabot-0 · 2026-08-13T09:26:26Z

Requested by this comment.

Benchmark job 33 failed for tpch/sf100. Full details are available in the controller journal.

gabotechs · 2026-08-13T09:52:37Z

The runner does not seem to work well with git dependencies... 😢

gabotechs · 2026-08-13T11:13:40Z

benchmarks run tpch/sf100

gabot-0 · 2026-08-13T11:13:59Z

Requested by this comment.

Run metadata

Phase	Base	PR head
Build and deployment	1m 42s	4m 55s
All benchmarks	7m 3s	7m 31s
Benchmark `tpch/sf100`	7m 3s	7m 31s

Queue: 1s · Dataset validation: 0s · Total: 21m 19s

Capacity: 12 c5n.2xlarge nodes

=== Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] ===
TOTAL: prev=67106 ms, new=68270 ms, diff=1.02 slower ✖

Show full query output

      q1: prev=2506 ms, new=2473 ms, diff=1.01 faster ✔
      q2: prev=1600 ms, new=1632 ms, diff=1.02 slower ✖
      q3: prev=2648 ms, new=2650 ms, diff=1.00 slower ✖
      q4: prev=1182 ms, new=1103 ms, diff=1.07 faster ✔
      q5: prev=3831 ms, new=4312 ms, diff=1.13 slower ✖
      q6: prev=1210 ms, new=1158 ms, diff=1.04 faster ✔
      q7: prev=4452 ms, new=4649 ms, diff=1.04 slower ✖
      q8: prev=4573 ms, new=4729 ms, diff=1.03 slower ✖
      q9: prev=6251 ms, new=6216 ms, diff=1.01 faster ✔
     q10: prev=5814 ms, new=5977 ms, diff=1.03 slower ✖
     q11: prev=1174 ms, new=1145 ms, diff=1.03 faster ✔
     q12: prev=1736 ms, new=1807 ms, diff=1.04 slower ✖
     q13: prev=1802 ms, new=1806 ms, diff=1.00 slower ✖
     q14: prev=1547 ms, new=1684 ms, diff=1.09 slower ✖
     q15: prev=3170 ms, new=3168 ms, diff=1.00 faster ✔
     q16: prev= 820 ms, new= 786 ms, diff=1.04 faster ✔
     q17: prev=4696 ms, new=4776 ms, diff=1.02 slower ✖
     q18: prev=5632 ms, new=5664 ms, diff=1.01 slower ✖
     q19: prev=1987 ms, new=1708 ms, diff=1.16 faster ✔
     q20: prev=2600 ms, new=2487 ms, diff=1.05 faster ✔
     q21: prev=6922 ms, new=7439 ms, diff=1.07 slower ✖
     q22: prev= 953 ms, new= 901 ms, diff=1.06 faster ✔

gabotechs · 2026-08-13T11:35:05Z

The current CI failure (https://github.com/datafusion-contrib/datafusion-distributed/actions/runs/31694133600/job/94427876747?pr=540) is a flaky test that should be fixed by #624

stuhood

Thanks!

FWIW: @barbarj ran our benchmarks with a variant of this upgrade, and they are ~unchanged.

jayshrivastava · 2026-08-13T14:38:54Z

+plans and expressions. See
+[Distribute a custom execution plan](../source/user-guide/04-distribute-custom-plan.md)
+for the complete signatures.
+


We need to change this to 4.0 once 3.0 releases.

Just did a soft launch https://github.com/datafusion-contrib/datafusion-distributed/releases/tag/v3.0.0 (not properly announced yet)

barbarj

Looks fine to me 👍 Thanks!

gabotechs · 2026-08-19T07:35:30Z

        Partitioning::RoundRobinBatch(p) => Partitioning::RoundRobinBatch(f(*p)),
        Partitioning::Hash(hash, p) => Partitioning::Hash(hash.clone(), f(*p)),
        Partitioning::UnknownPartitioning(p) => Partitioning::UnknownPartitioning(f(*p)),
+        Partitioning::Range(range) => Partitioning::Range(range.clone()),


@jayshrivastava this might need addressing. If we silently don't scale up output partitioning, an upstream NetworkShuffleExec might assume output partitions where actually scaled up, and it will try to reach for non-existing partitions, making queries fail.

It might be safer to fail for now, and address this separately.

Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com>

gabotechs · 2026-08-20T08:11:23Z

+            })),
+            partition,
+            labels,
+        }),


This is the reason why CI is failing here: https://github.com/datafusion-contrib/datafusion-distributed/actions/runs/32305173621/job/96236269450?pr=540.

while roundtripping a MetricValue::PeakMemoryUsage through protobuf, we recover the metric back as a MetricValue::Gauge, not a MetricValue::PeakMemoryUsage. We'll need to add native proto serialization support for PeakMemoryUsage.

I've addressed this myself in 9b52a63, as it's pretty contained

gabotechs

The only thing that needs addressing is the metrics proto roundtrip, but besides that, this is pretty much ready to go 🚀

gabotechs · 2026-08-20T08:38:44Z

Thanks @jayshrivastava, @barbarj and @stuhood!

## Summary Closes datafusion-contrib#530 - This PR updates the upstream datafusion SHA to the HEAD of https://github.com/apache/datafusion/commits/branch-55/ (edit: this branch is continuously being updated. I will make sure this PR is at the head before merging) - Rust upgade to 1.94 ## Changes 1. In `src/protobuf/distributed_codec.rs` we now use the `proto_converter` argument during serde - We still don't use the `DeduplicatingProtoConverter`, so dynamic filters don't necessarily work. I think this is outside the scope of this PR will be addressed in datafusion-contrib#623, which will be rebased after the upgrade. 3. `ExecutionPlan::apply_expressions` is added for every custom `ExecutionPlan` in this repo - Wrapper types (`MetricsWrapperExec`, `WorkUnitFileScanConfig`, `DistributedLeafExec`) delegate to the inner type - Other plans take`TreeNodeRecursion::Continue` because they have no expressions (ex. `SamplerExec`) - Note that `apply_expressions` does not need to yield sort or partitioning expressions in the plan properties 3. We migrate from `partition_statistics` to `statistics_from_inputs` for every `ExecutionPlan`. - `src/distributed_planner/statistics/plan_statistics.rs` can just use `statistics_from_inputs` directly instead of doing the `StatisticsWrapper` workaround. 5. Range partitioning is now supported. - CPU costing now includes range-key comparison cost and has a new unit test. See src/distributed_planner/ statistics/complexity_cpu.rs:238. - I think there's open questions about range partitioning. I've opened an issue here to make sure it behaves as expected after the upgrade: datafusion-contrib#628 (comment) 6. Peak-memory metrics use the existing gauge wire representation. DataFusion added MetricValue::PeakMemoryUsage. It is serialized as the existing named-gauge protobuf variant to avoid a wire-format change. See src/protocol/grpc/ metrics_proto.rs:124. The value and name survive, and aggregation is still additive, but decoding produces a generic Gauge, not PeakMemoryUsage. The practical difference is mainly display formatting: it may render as a count rather than human-readable bytes. This is the clearest remaining compromise/risk in the upgrade. 7. File-scan rebalancing changed its discriminator. DataFusion removed partitioned_by_file_group; output_partitioning.is_some() is now the source of truth. See src/events/defaults/file_scan_config.rs:43. This decides whether files are round-robin rebalanced or split through FileGroupPartitioner, so it is behavior-sensitive even though it is a one-line migration. 8. Two previously ignored correctness tests were enabled. - See `tests/multi_task_collect_join_repros.rs` - These were upstream DataFusion correctness fixes, not fixes made locally in this upgrade. 9. drop(reporter) was made explicit on the sampler’s empty-input path. The reporter sends its result on Drop; explicitly dropping it both satisfies the new compiler/lint behavior and guarantees the zero-row EOS report is sent before returning. See src/ execution_plans/sampler.rs:259. 10. Plan changes - `dynamic_rg_pruning=eligible` is now displayed on eligible scans: 1,354 occurrences in TPC-DS, 188 in TPC-H, and 12 in ClickBench - `DataSourceExec` now displays its output partitioning. See `tests/join.rs` (eventually, someone should delete this test datafusion-contrib#628) - Project after sort. This looks like some upstream optimizer rule change ex. `tests/distributed_unions.rs` and `tests/distributed_aggregation.rs`. ``` - │ SortExec: expr=[MinTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true] - │ ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday] + │ ProjectionExec: expr=[MaxTemp@0 as MinTemp, RainToday@1 as RainToday] + │ SortExec: expr=[MaxTemp@0 ASC NULLS LAST, RainToday@1 ASC NULLS LAST], preserve_partitioning=[true] ``` - LocalLimitExec became more common: TPC-DS went from 0 to 20 occurrences and ClickBench from 1 to 21, reflecting additional local limit pushdown. - Subquery/semi-join plans became more distributed: - TPC-DS CollectLeft hash joins: 615 → 610 - TPC-DS partitioned hash joins: 98 → 103 - TPC-DS left-semi occurrences: 11 → 25 - TPC-DS network shuffles: 368 → 378 - TPC-H - just a few - These are meaningful topology changes: some subqueries now use partitioned left-semi joins and therefore introduce hash shuffles instead of collecting/broadcasting one side. - Scalar rendering improved, especially decimal literals: internal forms such as Some(0),7,2 now display as CAST(0.00 AS Decimal128(7, 2)). - Minor changes (Ex. tpcds 21) - `__common_expr_4` became `__common_expr_3`; that is only an internal alias renumbering. - The projection that renamed `d_date` to `__common_expr_2` disappeared. - `d_date` is retained directly in the join output and referenced directly by partial/final aggregates. - Column positions changed - File-group allocation changed substantially - Some explicit RoundRobinBatch repartitions disappeared and scans gained different numbers of file groups - Distribute byte ranges across partitions: 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 <int>....<int>, 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 ] ``` --------- Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com> Co-authored-by: Gabriel <gabriel.musatmestre@datadoghq.com>

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

jayshrivastava force-pushed the branch-55 branch 2 times, most recently from 78b65f1 to b8eddac Compare August 12, 2026 17:08

jayshrivastava commented Aug 12, 2026

View reviewed changes

gene-bordegaray reviewed Aug 12, 2026

View reviewed changes

Comment thread src/distributed_planner/statistics/complexity_cpu.rs Outdated

gabotechs reviewed Aug 13, 2026

View reviewed changes

Comment thread src/distributed_planner/statistics/plan_statistics.rs

Comment thread src/distributed_planner/distributed_query_planner.rs

Comment thread src/execution_plans/distributed_leaf.rs

stuhood mentioned this pull request Aug 13, 2026

Partitioning M1: Benchmark and Choose Partition Targets paradedb/paradedb#5734

Closed

6 tasks

gabotechs force-pushed the branch-55 branch from 55b8559 to f5aedab Compare August 13, 2026 11:07

stuhood approved these changes Aug 13, 2026

View reviewed changes

jayshrivastava commented Aug 13, 2026

View reviewed changes

gabotechs mentioned this pull request Aug 13, 2026

Release DataFusion 55.0.0 (Jul / Aug 2026) apache/datafusion#22393

Closed

50 tasks

barbarj approved these changes Aug 14, 2026

View reviewed changes

shinzoxD mentioned this pull request Aug 15, 2026

fix: preserve range partitions when scaling file scans #648

Closed

jayshrivastava mentioned this pull request Aug 17, 2026

tests on range partitioned data #628

Open

gabotechs reviewed Aug 19, 2026

View reviewed changes

jayshrivastava mentioned this pull request Aug 19, 2026

implement replace_children for all ExecutinoPlan implementations #657

Open

jayshrivastava requested a review from gabotechs August 19, 2026 19:36

shinzoxD mentioned this pull request Aug 19, 2026

fix: implement replace_children on distributed plan nodes #658

Closed

stuhood mentioned this pull request Aug 19, 2026

chore: Update to datafusion@7e015b74af0928187a77cdf9df23b587e349ebf9. paradedb/datafusion-distributed#77

Closed

jayshrivastava and others added 13 commits August 19, 2026 21:33

feat: update DataFusion to 46bd88a

39b7468

feat: update DataFusion to d555234

d51622c

test: update dynamic pruning snapshots

19cb79b

fix: preserve expression identity in encoded plans

9e937ae

refactor: centralize physical plan proto conversion

6b0829f

refactor: clarify key complexity helper name

5784448

fix: use replace_children for DataFusion 55 compatibility

0492df5

build: use the DataFusion 55 crates.io release

3df9d77

move upgrade guide changes from 3.0.0 to 4.0.0

a6d4a0a

return error in producer_head for range partitioning

dbe9445

Explicitly report stats in sampler.rs when reporter is dropped

c757bed

Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com>

restore example

5760182

rebase and fix proto import

e114458

jayshrivastava force-pushed the branch-55 branch from 300b5af to e114458 Compare August 19, 2026 21:41

stuhood mentioned this pull request Aug 20, 2026

chore: Update datafusion to get support for dynamic filters for range-partitioned data paradedb/paradedb#5943

Closed

gabotechs mentioned this pull request Aug 20, 2026

Avoid redundant LocalLimitExec when fetch is separated by a ProjectionExec #659

Closed

gabotechs reviewed Aug 20, 2026

View reviewed changes

gabotechs approved these changes Aug 20, 2026

View reviewed changes

Add PeakMemoryUsage to proto

9b52a63

gabotechs merged commit 743bd0a into main Aug 20, 2026
31 checks passed

gabotechs deleted the branch-55 branch August 20, 2026 08:38

gabotechs mentioned this pull request Aug 20, 2026

Upgrade Rust toolchain #593

Closed

shinzoxD mentioned this pull request Aug 24, 2026

fix: implement replace_children on distributed plan nodes #668

Closed

Uh oh!

Conversation

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

Uh oh!

Summary

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!

stuhood commented Aug 6, 2026

Uh oh!

jayshrivastava left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jayshrivastava commented Aug 12, 2026

Uh oh!

jayshrivastava commented Aug 12, 2026

Uh oh!

Uh oh!

gabotechs commented Aug 13, 2026

Uh oh!

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

Uh oh!

Uh oh!

gabotechs left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

asolimando commented Aug 13, 2026

Uh oh!

gabotechs commented Aug 13, 2026

Uh oh!

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

Uh oh!

Uh oh!

gabotechs commented Aug 13, 2026

Uh oh!

gabotechs commented Aug 13, 2026

Uh oh!

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

Uh oh!

Uh oh!

gabotechs commented Aug 13, 2026

Uh oh!

stuhood left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

barbarj left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jayshrivastava commented Jul 7, 2026 •

edited

Loading

jayshrivastava commented Jul 7, 2026 •

edited

Loading

gabot-0 commented Aug 13, 2026 •

edited

Loading

gabot-0 commented Aug 13, 2026 •

edited

Loading

gabot-0 commented Aug 13, 2026 •

edited

Loading