Upgrade DataFusion to v54.0.0

We can't actually merge an upgrade yet because the version has not been published in crates.io yet. We can start testing in the meantime and then merge the upgrade once the new version is in crates.io.

DF issue: apache/datafusion#21080

Testing new version in #465

There are some issues we are investigating now

@gabotechs

After bisecting, we found two breaking changes not compatible with some assumptions we make in DDF:

Dynamic work scheduling in FileStream apache/datafusion#21351 -- breaks some tests in tpch and tpcds. It also breaks custom_extension_codec with correctness issues :

│ +---------+-----------+
│ | MinTemp | RainToday |
│ +---------+-----------+
│  | 20.9    | No        |
│+| 20.9    | No        |
│+| 20.9    | No        |
+---------+-----------+

We see 3 results when there should be 1, this only happens in distributed execution. Talking with @gabotechs the suspicion is that this breaks PartitionIsolatorExec assumption that each ExecutionPlan partition only reads its corresponding files. With #21351 the parquet files are pulled from a shared queue all partitions share. In distributed mode, because each task gets its own DataSourceExec instance, it can read a file that was already read by another DataSourceExec, resulting in duplicated results.

perf: Implement physical execution of uncorrelated scalar subqueries apache/datafusion#21240: This PR introduces a node to execute subqueries, ScalarSubqueryExec which structure holds a shared state:

pub struct ScalarSubqueryExec {
    ...
    results: ScalarSubqueryResults,

}

pub struct ScalarSubqueryResults {
    slots: Arc<Vec<Mutex<Option<ScalarValue>>>>,
}

where the a given subquery has an index to fill it's corresponding slot at runtime:

pub struct ScalarSubqueryExpr {
   ...
    results: ScalarSubqueryResults,
}

If ScalarSubqueryExec and the node holding the ScalarSubqueryExpr are in different stages, the query fails during deserialization with the following error Internal("ScalarSubqueryExpr can only be deserialized as part of a surrounding ScalarSubqueryExec") . Even if we edited the decoder to create a fresh ScalarSubqueryResults on the worker, the result wouldn't be correct I think, the worker's slots are a different Arc from the coordinator, so the writer would never fill them. This looks like a similar problem to Dynamic Filtering -- although DynamicFilterPhysicalExpr does not fail on serialization and we will never see correctness issues since it's an optimization.

Opened apache/datafusion#22530 for point 2.

We also found the issue described in the following revert PR: DataDog/datafusion#135 -- not really sure what's going on, I will open a follow-up issue soon

Introduce DistributedLeafExec (#467)

Opened an issue to solve permanently point 2 #471

We also found the issue described in the following revert PR: DataDog/datafusion#135 -- not really sure what's going on, I will open a follow-up issue soon

I managed to get this fixed I think, it's not really an upstream bug, it's in this codebase, will post a PR soon

Fixed!

Fix early worker partition drop in RemoteWorkerConnection demux #473

Gate new ScalarSubqueryExec node behind session property (apache#22530)

Upgrade branch is green now #465 🎉 . Once DataFusion releases 54, I think we are done here.

I'm seeing some deadlocks in the CI...

I verified that main does not deadlock, so it must be something related to the upgrade...

Root-caused the deadlocks (task_estimator_test::custom_routing_join* and friends): it's DataFusion #21068 — "Skip probe-side consumption when hash join build side is empty", new in branch-54.

When a partitioned HashJoinExec partition has an empty build side it now short-circuits to Completed and drops its probe-side input without ever polling it. When that probe is a NetworkShuffleExec, the remote request is issued lazily on first poll — so dropping it before the first poll means the request is never sent, the upstream worker never execute()s that partition of its shared RepartitionExec, that output channel is never drained, the distributor latches its global gate (closes once all output channels are non-empty), the pump blocks, and every sibling consumer of that shared RepartitionExec deadlocks.

This is distinct from the demux fix in #473: that keeps an already-running demux alive when a receiver drops; here the demux for the abandoned partition never starts.

This should solve it: #482 (comment)

I don't love that approach though, it feels a bit like a workaround, but I don't have a better idea

Upgrade DF to v54 (#465)

Navigation Menu

Repository navigation

Upgrade DataFusion to v54.0.0 #460

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Upgrade DataFusion to v54.0.0 #460

Description

Activity

LiaCastaneda commented on May 25, 2026

LiaCastaneda commented on May 25, 2026

LiaCastaneda commented on May 26, 2026

LiaCastaneda commented on May 27, 2026

LiaCastaneda commented on May 28, 2026

gabotechs commented on May 28, 2026

gabotechs commented on May 28, 2026

gabotechs commented on Jun 2, 2026

gabotechs commented on Jun 3, 2026

gabotechs commented on Jun 3, 2026

gabotechs commented on Jun 3, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions