Add cost model by gabotechs · Pull Request #486 · datafusion-contrib/datafusion-distributed

gabotechs · 2026-06-08T10:50:53Z

This is one PR from the following stack of PRs:

Mostly rescued from:

Cost based planning #311

This PR is pretty big, but the whole purpose is to expose a very simple function:

pub(crate) fn calculate_cost(plan: &Arc<dyn ExecutionPlan>, cfg: &DistributedConfig) -> Result<Cost>;

While keeping everything else private to the rest of the codebase.

This PR introduces a cost model used in Adaptive task count assignation, that attributes a compute cost to a plan based on:

The Statistics available by normal DataFusion mechanisms (ExecutionPlan::partition_statistics())
The compute time complexity of the different operators in a plan

The whole meat of this PR lives in the src/distributed_planner/statistics/compute_per_node.rs file: it's in charge of attributing a time complexity to the different ExecutionPlan implementations using big O notation. For example:

        let plan = plan_costs(r#"SELECT * FROM weather WHERE "MinTemp" > 5"#, 1).await;
        assert_snapshot!(plan, @r"
        O((out_Cols+Col0)) | FilterExec: MinTemp@0 > 5
         O(out_Cols) | DataSourceExec: file_groups={...}
        ");

In this example:

DataSourceExec needs to allocate all the RecordBatches for all the output columns, as it's the data source, so it's time complexity is O(out_Cols), as its compute complexity linearly scales with the amount of columns it needs to allocate.
FilterExec needs to:
1. Perform a linear predicate evaluation over the MinTemp column (O(Col0))
2. Allocate full new RecordBatches for all output columns based on rows filtered by the evaulated predicate (O(out_Cols))

The tests in complexity_cpu.rs and complexity_memory.rs contains several examples of this for the different operators, which can go from pretty simple (FilterExec, ProjectionExec), to very complex (HashJoinExec).

Once the time complexity in big O notation is attributed to a node, doing a simple multiplication math of the columns appearing in the notation by the estimated byte_size of those columns (given by Statistics), gives the compute cost measured in bytes.

This is one PR from the following stack of PRs: - #477 - #463 <- you are here - #464 - #478 - #479 - #486 - #432 This PR introduces a NetworkBoundaryBuilder argument to the network boundary injection logic, allowing more flexible and configurable strategies for determining which exchanges require network communication. This enables better optimization of data movement across distributed tasks.

This is one PR from the following stack of PRs: - #477 - #463 - #464 <- you are here - #478 - #479 - #486 - #432 This PR introduces a MaxGauge metric to provide better tracking of peak values in distributed metrics collection. This enables more accurate monitoring of resource utilization and helps identify bottlenecks in the execution pipeline.

This is one PR from the following stack of PRs: - #477 - #463 - #464 - #478 <- you are here - #479 - #486 - #432 --- Introduces the `ProducerHead` type: ```rust pub enum ProducerHead { /// No specific head node is necessary. None, /// The head node should be a [BroadcastExec]. BroadcastExec { output_partitions: usize }, /// The head node should be a [RepartitionExec]. RepartitionExec { partitioning: Partitioning }, } ``` Which is passed over the network while remotely executing tasks in order to set the appropriate node at the head of a stage. Today, this is a noop because the right head node in stages is ensured statically at planning time, but in follow up PRs, network boundaries can get swamped and reorganized arbitrarily. One example that happens in AQE: 1. A JOIN is planned as a CollectLeft ```js HashJoinExec: mode=CollectLeft CoalescePartitionsExec: [Stage 1] => NetworkBroadcastExec BroadcastExec DistributedLeafExec: unknown size DistributedLeafExec: unknown size ``` 2. While collecting runtime statistics, it happens that `Stage 1` is huge, and during AQE the JOINs are swapped ```js HashJoinExec: mode=CollectLeft DistributedLeafExec: small size CoalescePartitionsExec: [Stage 1] => NetworkBroadcastExec BroadcastExec DistributedLeafExec: big size ``` 3. The `Stage 1` is now on the probe side, so it needs to be rewritten to a `NetworkShuffleExec`, otherwise duplicate data will be returned: ```js HashJoinExec: mode=CollectLeft DistributedLeafExec: small size CoalescePartitionsExec: [Stage 1] => NetworkShuffleExec RepartitionExec // <- dynamically swapped at runtime based on the passed `ProducerHead` DistributedLeafExec: big size ``` Passing a `ProducerHead` at execution time unlocks two things: 1. dynamically set the fanout width accounting for a dynamically scaled upper stage 2. dynamically set the correct operator `BroadcastExec` or `RepartitionExec` based on the network boundary above, which might have changed because of AQE

This is one PR from the following stack of PRs: - #477 - #463 - #464 - #478 - #479 <- you are here - #486 - #432 --- This is a pure refactor PR with a couple of implementation detail changes: # Reducing the bloat of complex functions in already complicated parts of the codebase. For this, two key structs are introduced: - `QueryCoordinator`: scoped to a whole distributed query, it handles references to pieces of data global to a query's lifetime, like the `TaskContext`, the metrics, the `JoinSet` used for spawning tasks, etc... it's in charge also of building `StageCoordinator` instances, which are scoped per-stage instead of per-query. - `StageCoordinator`: this is the old `CoordinatorToWorkerTaskSpawner`, but with some more methods that allow reusability and some better naming. It handles all the comms between workers and coordinator needed for driving a stage forward. This allows reducing the bloat in `prepare_static_plan` and the future `prepare_dynamic_plan` functions. # Ensuring a coordinator->worker channel is held active for as long as the `DistributedExec` node is executing the query on the coordinator. For the static planner, this is a noop, as the previous model worked fine before, but this will become important in the future for the dynamic planner. In the dynamic planner, the plan can be set for some stages, but they might never reach execution, so instead of coupling the task entry cache invalidation to the task execution finish, it's coupled instead of the coordinator->channel lifetime. This has one collateral effect: `WorkUnit` feeds can no longer rely on the global `coordinator->worker` EOS signal for ensuring that no further `WorkUnit` feed is going to be sent by the coordinator, so they need an explicit EOS message that signals that no further `WorkUnit`s will be received, even though the `coordinator->worker` channel will still be alive for a while. # Add a `plan_for_viz` field in `DistributedExec`. This is a new slot in `DistributedExec` that holds a reference of the plan that is supposed to be rewritten with metrics for visualization purposes. Again, for the static planner this is a noop, because the plan meant for visualization is equal to the plan that arrived as child to `DistributedExec` on the first place. However, during dynamic planning, the plan that arrives to `DistributedExec` is not going to be the same as the final one after execution, so we need a slot for storing that final plan.

asolimando

I did a first pass on the PR, it's surely heading in the right direction but there are a couple of choices around how to deal with missing values and how to represent cost that will probably benefit from more discussion.

asolimando

Thanks for this new iteration, Gabriel, the split between the three complexity_* files reinforces the need to keep the three dimensions separate, not only to write a better cost model, but also to tame complexity for maintainers.

I think the only must-have before merge is switching the remaining +/* on estimates to saturating_add/saturating_mul.

The broadcast fan-out and printing per-node costs I'd put as (very) nice to have. Both are additive and can land in a follow-up without touching the design, so no pressure to fit them here.

Everything else (the NDV ratio, the stale comments, naming) is minor.

Overall it's in great shape, thanks for the quick turnaround on the vector split.

asolimando

LGTM, one last point to address in #486 (comment) but I think the direction is clear now and all major comments have been addressed!

gabotechs · 2026-06-30T10:48:36Z

Thanks @asolimando for the thorough review! this looks much better with your suggestions

This is one PR from the following stack of PRs: - #477 - #463 - #464 - #478 - #479 - #486 - #432 <- you are here --- This PR implements support for dynamic task count assignation, allowing the distributed planner to determine the optimal number of tasks for each stage based on runtime characteristics and available resources. This enables adaptive query execution where task counts are optimized for the specific execution context rather than being determined statically. There are two fundamental pieces delivered in this PR that make adaptive task count assignation possible: 1. [SamplerExec](https://github.com/datafusion-contrib/datafusion-distributed/pull/432/changes#diff-17bd8d290d703ba259143955ade0265fd22463f45cfe752c6071ab136331fe7f) 2. [prepare_dynamic_plan](https://github.com/datafusion-contrib/datafusion-distributed/pull/432/changes#diff-01ffef8aa9ef0e8104dfe1c133320f1c1edd157fd968798b9206a3e04dd95228) ## SamplerExec This is a new `ExecutionPlan` implementation that peeks a few initial record batches before execution, gathers statistics over them, and then reports a `LoadInfo` message to the `prepare_dynamic_plan` running on the coordinator ```rust ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌─────────────────────┐ │ │ │ │ ┌───────────────────▶│ DistributedExec │◀──────────────────────┐ │ │ │ │ │ │ │ └─────────────────────┘ │ ┌──┴─────┐ │ ┌─────────────────────┐ │ ┌─────┴──┐ │LoadInfo│ │ │ │LoadInfo│ └──┬─────┘ │ │ ... │ │ └─────┬──┘ │ │ │ │ │ │ └─────────────────────┘ │ │ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ │ │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐│ │ ┌─────────────────────┐ ┌─────────────────────┐ │ │ │ │ ProducerHead │ │ │ │ ProducerHead │ ││ │ │ (RepartitionExec or │ │ (RepartitionExec or │ │ │ │ │ BroadcastExec) │ │ │ │ BroadcastExec) │ ││ │ └─────────────────────┘ └─────────────────────┘ │ │ │ ┌─────────────────────┐ │ │ ┌─────────────────────┐ ││ │ │ │ │ │ │ └─┼───│ SamplerExec │ │ │ │ SamplerExec │───┼┘ │ │ │ │ │ └─────────────────────┘ │ │ └─────────────────────┘ │ ┌─────────────────────┐ ┌─────────────────────┐ │ │ │ │ │ │ │ │ │ ... │ │ ... │ │ │ │ │ │ │ │ │ └─────────────────────┘ └─────────────────────┘ │ ┌─────────────────────┐ │ │ ┌─────────────────────┐ │ │ │ │ │ │ │ DataSourceExec │ │ │ │ DataSourceExec │ │ │ │ │ │ │ └─────────────────────┘ │ │ └─────────────────────┘ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ``` Each `SamplerExec` node contains one `PartitionSampler` per partition, and each partition sampling is treated independently: ```rust ┌────────────────────────────────────────────────────────────────────────────┐ │ SamplerExec │ │ ┌────────────────┐┌────────────────┐ ┌────────────────┐ │ │ │ ││ │ │ │ │ │ │PartitionSampler││PartitionSampler│ ... │PartitionSampler│ │ │ │ ││ │ │ │ │ │ └────────────────┘└────────────────┘ └────────────────┘ │ └────────────────────────────────────────────────────────────────────────────┘ ``` The way a `PartitionSampler` works is that, before actual execution, it will peek a representative number of `RecordBatch`es and gather stats over them (per column byte count, per column NDV, per column null count, total row count, etc...). For that the following things happen: 1. Pulls all the `RecordBatch`es that are available synchronously without `async` gaps. Some operators like `Aggregate(Partial)` retain big internal representations in memory that can yield more than 1 `RecordBatch` synchronously, because they were already computed and stored in memory. ```rust ┌───────────────────────────────────────────────────────────────┐ │ PartitionSampler │ │ │ │┌───────────┬───────────┬───────────┐ │ ││ │ │ │ │ ││RecordBatch│RecordBatch│RecordBatch│ │ ││ │ │ │ │ │└───────────┴───────────┴───────────┘ │ └───────────────────────────────────────────────────────────────┘ ``` 2. Once an `async` gap is reached, it pulls just 1 more record batch, and it measures the time it takes in this `async` gap: ```rust ┌───────────────────────────────────────────────────────────────┐ │ PartitionSampler │ │ │ │┌───────────┬───────────┬───────────┐ ┌───────────┐│ ││ │ │ │ │ ││ ││RecordBatch│RecordBatch│RecordBatch│ async wait │RecordBatch││ ││ │ │ │ │ ││ │└───────────┴───────────┴───────────┘ └───────────┘│ └───────────────────────────────────────────────────────────────┘ ``` 3. With this information, two sources of useful information are present: - Bytes ready to be yielded immediately - The velocity at with new `RecordBatch`es are arriving ```rust ┌───────────────────────────────────────────────────────────────┐ │ PartitionSampler │ │ │ │┌───────────┬───────────┬───────────┐ ┌───────────┐│ ││ │ │ │ │ ││ ││RecordBatch│RecordBatch│RecordBatch│ async wait │RecordBatch││ ││ │ │ │ │ ││ │└─────┬─────┴─────┬─────┴───┬─────┬─┘ └────▲────┬─┘│ └──────┼───────────┼─────────┼─────┼────────────────────┼────┼──┘ │ │ │ ─────────────────────│ │ │ │ │ │ │ │ │ RecordBatch velocity │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ┌─────────────────────────┘ ▼ ▼ ▼ ▼ Batches to be yielded immediately ``` This velocity information is what really matters towards deciding how many task count should be yielded. - Scanning a lot of data is not problematic if you can do it in a streaming fashion and the DataSources are bottleneck on IO, in this situations, 1 node might be enough. - If DataSources yield data very quickly, or if the compute cost over that data is very expensive, even if it's not a lot of data, it might still bottleneck the query en CPU power, so we'll benefit from more CPUs (more machines). # prepare_dynamic_plan This is analogous to the existing `prepare_static_plan`, but with a runtime based dynamic task count assignation. In order to understand how it works, let's use the classical aggregation example: ```rust ┌───────────────────┐ │CoalescePartitions │ └───────────────────┘ ┌───────────────────┐ │ ProjectionExec │ └───────────────────┘ ┌───────────────────┐ │ Aggregate(final) │ └───────────────────┘ ┌───────────────────┐ │ RepartitionExec │ └───────────────────┘ ┌───────────────────┐ │Aggregate(partial) │ └───────────────────┘ ┌───────────────────┐ │ FilterExec │ └───────────────────┘ ┌───────────────────┐ │ DataSourceExec │ └───────────────────┘ ``` The plan walks from bottom to top, until we delimit a stage, same as before: ```rust Stage 1 ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌───────────────────┐ │ │ RepartitionExec │ │ └───────────────────┘ │ ┌───────────────────┐ │ │Aggregate(partial) │ │ └───────────────────┘ │ ┌───────────────────┐ │ │ FilterExec │ │ └───────────────────┘ │ ┌───────────────────┐ │ │ DataSourceExec │ │ └───────────────────┘ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ``` We'll use runtime information of "Stage 1" in order to determine the task count for the future "Stage 2". For that, the plan is modified and a `SamplerExec` is inserted just below the producer head (`RepartitionExec`): ```rust Stage 1 ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┐ ┌───────────────────┐ │ │ RepartitionExec │ │ └───────────────────┘ │ ┌───────────────────┐ │ │ SamplerExec │◀──────────── │ └───────────────────┘ │ ┌───────────────────┐ │ │Aggregate(partial) │ │ └───────────────────┘ │ ┌───────────────────┐ │ │ FilterExec │ │ └───────────────────┘ │ ┌───────────────────┐ │ │ DataSourceExec │ │ └───────────────────┘ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ ``` Let's say that the `TaskEstimator` for the leaf node `DataSourceExec` decided a `Desired(3)` task estimation. This subplan is then immediately sent to three workers, and upon setting the plan on the workers, all the `SamplerExec`s are kicked off and they start sampling even before any call to `.execute()` in any node: ```rust ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─Stage 1 ┌ ─ ─ ─ ─ ─ ─ ─Worker 1 ┌ ─ ─ ─ ─ ─ ─ ─Worker 2 ┌ ─ ─ ─ ─ ─ ─ ─Worker 3 │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││ RepartitionExec ││ ││ RepartitionExec ││ ││ RepartitionExec ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │ SamplerExec │ │ SamplerExec │ │ SamplerExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ │ │ │ │ │ Eager kick off │ │ Eager kick off │ │ Eager kick off │ │ ▼ ▼ ▼ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │Aggregate(partial) │ │Aggregate(partial) │ │Aggregate(partial) │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││ FilterExec ││ ││ FilterExec ││ ││ FilterExec ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │ DataSourceExec │ │ DataSourceExec │ │ DataSourceExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ``` All the `SamplerExec`s start sampling, and they start reporting `LoadInfo` messages to the `prepare_dynamic_plan` function, which builds the next `NetworkShuffleExec` boundary with some pre-defined `Statistics`, the ones that where collected at runtime by the `SamplerExec`s below: ```rust ┌───────────────────┐ * │ ProjectionExec │ * └───────────────────┘ * ┌───────────────────┐ * │ Aggregate(final) │ * └───────────────────┘ * Still not planned ┌───────────────────┐ * │NetworkShuffleExec │ * │ (+ runtime stats) │ * └───────▲──▲──▲─────┘ * │ │ │ ┌──────────────────────────────────────┘ │ └────────┐ │ ┌────────────┘ │ │ │ │ │ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─Stage 1 │ ┌ ─ ─ ─ ─ ─ ─ ─Worker 1 │┌ ─ ─ ─ ─ ─ ─ ─Worker 2 │┌ ─ ─ ─ ─ ─ ─ ─Worker 3 │ │ ┌───────────────────┐ │ ┌───────────────────┐ │ ┌───────────────────┐ │ │ ││ RepartitionExec ││ │││ RepartitionExec ││ │││ RepartitionExec ││ │ │ └───────────────────┘ │ └───────────────────┘ │ └───────────────────┘ │ │ │┌───────────────────┐│ ││┌───────────────────┐│ ││┌───────────────────┐│ └─┼───┤ SamplerExec │ └─┤ SamplerExec │ └─┤ SamplerExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ │ │ │ │ │ Eager kick off │ │ Eager kick off │ │ Eager kick off │ │ ▼ ▼ ▼ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │Aggregate(partial) │ │Aggregate(partial) │ │Aggregate(partial) │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││ FilterExec ││ ││ FilterExec ││ ││ FilterExec ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │ DataSourceExec │ │ DataSourceExec │ │ DataSourceExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ``` While delimiting the next stage, the one that contains the `ProjectionExec` and the `Aggregate(final)`, the task count for that stage is decided purely over its compute cost, based on stats. At this point, the stats we have in that slice of the plan are very accurate, as we manage to gather them and condense them in `NetworkShuffleExec`, which at that point acts as a leaf node (as it's stage was already set to `Stage::Remote` because it already was sent and started sampling). Based on the compute cost inferred from the runtime stats, we suddenly realize that there's very very little data flowing through the `SamplerExec`, and as a consequence, the compute cost of "Stage 2" is estimated to be super low, so we collapse early to a single node: ```rust ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─Stage 2 ┌ ─ ─ ─ ─ ─ ─ ─Worker 1 │ ┌───────────────────┐ │ ││ ProjectionExec ││ │ └───────────────────┘ │ │┌───────────────────┐│ │ │ Aggregate(final) │ Very little data flowing, │ │└───────────────────┘│ 1 task is enough │ ┌───────────────────┐ │ ││NetworkShuffleExec ││ │ │ (+ runtime stats) │ │ │└───────────────────┘│ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─Stage 1 ┌ ─ ─ ─ ─ ─ ─ ─Worker 1 ┌ ─ ─ ─ ─ ─ ─ ─Worker 2 ┌ ─ ─ ─ ─ ─ ─ ─Worker 3 │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││ RepartitionExec ││ ││ RepartitionExec ││ ││ RepartitionExec ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │ SamplerExec │ │ SamplerExec │ │ SamplerExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││Aggregate(partial) ││ ││Aggregate(partial) ││ ││Aggregate(partial) ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ │┌───────────────────┐│ │┌───────────────────┐│ │┌───────────────────┐│ │ │ FilterExec │ │ FilterExec │ │ FilterExec │ │ │└───────────────────┘│ │└───────────────────┘│ │└───────────────────┘│ │ ┌───────────────────┐ ┌───────────────────┐ ┌───────────────────┐ │ ││ DataSourceExec ││ ││ DataSourceExec ││ ││ DataSourceExec ││ │ └───────────────────┘ └───────────────────┘ └───────────────────┘ │ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘ ``` # Benchmarks Extracted comparing runs of this same branch with `--dynamic false` VS `--dynamic true`: <details><summary>TPCH SF1: Tasks 279 -> 318, prev=14631 ms, new=6727 ms, diff=2.17 faster ✅</summary> ``` q1: prev= 319 ms, new= 307 ms, diff=1.04 faster ✔ q2: prev=1054 ms, new= 393 ms, diff=2.68 faster ✅ q3: prev= 989 ms, new= 504 ms, diff=1.96 faster ✅ q4: prev= 436 ms, new= 449 ms, diff=1.03 slower ✖ q5: prev=1019 ms, new= 314 ms, diff=3.25 faster ✅ q6: prev= 255 ms, new= 206 ms, diff=1.24 faster ✅ q7: prev= 934 ms, new= 310 ms, diff=3.01 faster ✅ q8: prev=1307 ms, new= 467 ms, diff=2.80 faster ✅ q9: prev=1208 ms, new= 331 ms, diff=3.65 faster ✅ q10: prev= 585 ms, new= 381 ms, diff=1.54 faster ✅ q11: prev= 380 ms, new= 183 ms, diff=2.08 faster ✅ q12: prev= 328 ms, new= 170 ms, diff=1.93 faster ✅ q13: prev= 477 ms, new= 176 ms, diff=2.71 faster ✅ q14: prev= 644 ms, new= 222 ms, diff=2.90 faster ✅ q15: prev= 946 ms, new= 175 ms, diff=5.41 faster ✅ q16: prev= 268 ms, new= 297 ms, diff=1.11 slower ✖ q17: prev= 828 ms, new= 281 ms, diff=2.95 faster ✅ q18: prev= 838 ms, new= 261 ms, diff=3.21 faster ✅ q19: prev= 424 ms, new= 313 ms, diff=1.35 faster ✅ q20: prev= 374 ms, new= 200 ms, diff=1.87 faster ✅ q21: prev= 815 ms, new= 608 ms, diff=1.34 faster ✅ q22: prev= 203 ms, new= 179 ms, diff=1.13 faster ✔ TOTAL: prev=14631 ms, new=6727 ms, diff=2.17 faster ✅ ``` </details> <details><summary>TPCH SF10: Tasks 363 -> 734, prev=35577 ms, new=13014 ms, diff=2.73 faster ✅</summary> ``` q1: prev=1498 ms, new= 604 ms, diff=2.48 faster ✅ q2: prev=2258 ms, new= 529 ms, diff=4.27 faster ✅ q3: prev=2005 ms, new= 567 ms, diff=3.54 faster ✅ q4: prev= 841 ms, new= 468 ms, diff=1.80 faster ✅ q5: prev=2051 ms, new= 701 ms, diff=2.93 faster ✅ q6: prev= 679 ms, new= 575 ms, diff=1.18 faster ✔ q7: prev=1810 ms, new= 723 ms, diff=2.50 faster ✅ q8: prev=3028 ms, new= 769 ms, diff=3.94 faster ✅ q9: prev=3011 ms, new= 885 ms, diff=3.40 faster ✅ q10: prev=1699 ms, new= 764 ms, diff=2.22 faster ✅ q11: prev=1142 ms, new= 304 ms, diff=3.76 faster ✅ q12: prev= 913 ms, new= 427 ms, diff=2.14 faster ✅ q13: prev= 789 ms, new= 925 ms, diff=1.17 slower ✖ q14: prev=1261 ms, new= 489 ms, diff=2.58 faster ✅ q15: prev=2132 ms, new= 541 ms, diff=3.94 faster ✅ q16: prev= 610 ms, new= 242 ms, diff=2.52 faster ✅ q17: prev=1771 ms, new= 589 ms, diff=3.01 faster ✅ q18: prev=2162 ms, new= 795 ms, diff=2.72 faster ✅ q19: prev=1157 ms, new= 480 ms, diff=2.41 faster ✅ q20: prev=1350 ms, new= 669 ms, diff=2.02 faster ✅ q21: prev=2830 ms, new= 747 ms, diff=3.79 faster ✅ q22: prev= 580 ms, new= 221 ms, diff=2.62 faster ✅ TOTAL: prev=35577 ms, new=13014 ms, diff=2.73 faster ✅ ``` </details> <details><summary>TPCH SF100: Tasks 387 -> 1175, prev=307187 ms, new=62943 ms, diff=4.88 faster ✅</summary> ``` q1: prev=16724 ms, new=3086 ms, diff=5.42 faster ✅ q2: prev=3282 ms, new=1038 ms, diff=3.16 faster ✅ q3: prev=14599 ms, new=2683 ms, diff=5.44 faster ✅ q4: prev=9351 ms, new=1378 ms, diff=6.79 faster ✅ q5: prev=16400 ms, new=3628 ms, diff=4.52 faster ✅ q6: prev=9168 ms, new=1552 ms, diff=5.91 faster ✅ q7: prev=17370 ms, new=4049 ms, diff=4.29 faster ✅ q8: prev=22325 ms, new=4329 ms, diff=5.16 faster ✅ q9: prev=24036 ms, new=5788 ms, diff=4.15 faster ✅ q10: prev=14957 ms, new=5010 ms, diff=2.99 faster ✅ q11: prev=2642 ms, new= 852 ms, diff=3.10 faster ✅ q12: prev=10561 ms, new=1726 ms, diff=6.12 faster ✅ q13: prev=4839 ms, new=1503 ms, diff=3.22 faster ✅ q14: prev=9423 ms, new=1625 ms, diff=5.80 faster ✅ q15: prev=26224 ms, new=3120 ms, diff=8.41 faster ✅ q16: prev=1426 ms, new= 566 ms, diff=2.52 faster ✅ q17: prev=21121 ms, new=4623 ms, diff=4.57 faster ✅ q18: prev=25218 ms, new=5256 ms, diff=4.80 faster ✅ q19: prev=11387 ms, new=1751 ms, diff=6.50 faster ✅ q20: prev=13517 ms, new=2141 ms, diff=6.31 faster ✅ q21: prev=29777 ms, new=6310 ms, diff=4.72 faster ✅ q22: prev=2840 ms, new= 929 ms, diff=3.06 faster ✅ TOTAL: prev=307187 ms, new=62943 ms, diff=4.88 faster ✅ ``` </details> <details><summary>TPCDS SF1: Tasks 3079->2770, prev=62815 ms, new=44737 ms, diff=1.40 faster ✅</summary> ``` q1: prev=1039 ms, new= 531 ms, diff=1.96 faster ✅ q2: prev= 677 ms, new= 396 ms, diff=1.71 faster ✅ q3: prev= 618 ms, new= 288 ms, diff=2.15 faster ✅ q4: prev=3420 ms, new=2582 ms, diff=1.32 faster ✅ q5: prev= 767 ms, new= 554 ms, diff=1.38 faster ✅ q6: prev=1002 ms, new= 967 ms, diff=1.04 faster ✔ q7: prev= 614 ms, new= 421 ms, diff=1.46 faster ✅ q8: prev= 484 ms, new= 422 ms, diff=1.15 faster ✔ q9: prev= 267 ms, new= 341 ms, diff=1.28 slower ❌ q10: prev= 648 ms, new= 826 ms, diff=1.27 slower ❌ q11: prev=2281 ms, new=1663 ms, diff=1.37 faster ✅ q12: prev= 287 ms, new= 229 ms, diff=1.25 faster ✅ q13: prev= 962 ms, new= 720 ms, diff=1.34 faster ✅ q14: prev=1095 ms, new= 737 ms, diff=1.49 faster ✅ q15: prev= 314 ms, new= 212 ms, diff=1.48 faster ✅ q16: prev= 528 ms, new= 597 ms, diff=1.13 slower ✖ q17: prev= 768 ms, new= 257 ms, diff=2.99 faster ✅ q18: prev= 780 ms, new= 339 ms, diff=2.30 faster ✅ q19: prev= 280 ms, new= 269 ms, diff=1.04 faster ✔ q20: prev= 180 ms, new= 169 ms, diff=1.07 faster ✔ q21: prev= 708 ms, new= 382 ms, diff=1.85 faster ✅ q22: prev= 490 ms, new= 435 ms, diff=1.13 faster ✔ q23: prev=1149 ms, new= 567 ms, diff=2.03 faster ✅ q24: prev=1148 ms, new= 586 ms, diff=1.96 faster ✅ q25: prev= 558 ms, new= 275 ms, diff=2.03 faster ✅ q26: prev= 346 ms, new= 265 ms, diff=1.31 faster ✅ q27: prev= 480 ms, new= 328 ms, diff=1.46 faster ✅ q28: prev= 329 ms, new= 221 ms, diff=1.49 faster ✅ q29: prev= 663 ms, new= 251 ms, diff=2.64 faster ✅ q31: prev=1268 ms, new= 218 ms, diff=5.82 faster ✅ q32: prev= 426 ms, new= 190 ms, diff=2.24 faster ✅ q33: prev= 429 ms, new= 321 ms, diff=1.34 faster ✅ q34: prev= 424 ms, new= 373 ms, diff=1.14 faster ✔ q35: prev= 655 ms, new= 546 ms, diff=1.20 faster ✔ q36: prev= 388 ms, new= 290 ms, diff=1.34 faster ✅ q37: prev= 389 ms, new= 384 ms, diff=1.01 faster ✔ q38: prev= 568 ms, new= 252 ms, diff=2.25 faster ✅ q39: prev= 547 ms, new= 479 ms, diff=1.14 faster ✔ q40: prev= 612 ms, new= 417 ms, diff=1.47 faster ✅ q41: prev= 128 ms, new= 143 ms, diff=1.12 slower ✖ q42: prev= 331 ms, new= 139 ms, diff=2.38 faster ✅ q43: prev= 158 ms, new= 204 ms, diff=1.29 slower ❌ q44: prev= 514 ms, new= 353 ms, diff=1.46 faster ✅ q45: prev= 398 ms, new= 237 ms, diff=1.68 faster ✅ q46: prev= 641 ms, new= 495 ms, diff=1.29 faster ✅ q47: prev= 964 ms, new= 488 ms, diff=1.98 faster ✅ q48: prev= 514 ms, new= 418 ms, diff=1.23 faster ✅ q49: prev= 550 ms, new= 314 ms, diff=1.75 faster ✅ q50: prev= 549 ms, new= 445 ms, diff=1.23 faster ✅ q51: prev= 425 ms, new= 226 ms, diff=1.88 faster ✅ q52: prev= 252 ms, new= 117 ms, diff=2.15 faster ✅ q53: prev= 202 ms, new= 223 ms, diff=1.10 slower ✖ q54: prev= 679 ms, new= 383 ms, diff=1.77 faster ✅ q55: prev= 191 ms, new= 155 ms, diff=1.23 faster ✅ q56: prev= 448 ms, new= 332 ms, diff=1.35 faster ✅ q57: prev= 673 ms, new= 288 ms, diff=2.34 faster ✅ q58: prev= 644 ms, new= 289 ms, diff=2.23 faster ✅ q59: prev= 447 ms, new= 294 ms, diff=1.52 faster ✅ q60: prev= 301 ms, new= 291 ms, diff=1.03 faster ✔ q61: prev=1017 ms, new=1095 ms, diff=1.08 slower ✖ q62: prev= 776 ms, new= 761 ms, diff=1.02 faster ✔ q63: prev= 243 ms, new= 223 ms, diff=1.09 faster ✔ q64: prev=1907 ms, new=1370 ms, diff=1.39 faster ✅ q65: prev= 432 ms, new= 259 ms, diff=1.67 faster ✅ q66: prev= 839 ms, new= 781 ms, diff=1.07 faster ✔ q67: prev= 486 ms, new= 389 ms, diff=1.25 faster ✅ q68: prev= 432 ms, new= 373 ms, diff=1.16 faster ✔ q69: prev= 596 ms, new= 618 ms, diff=1.04 slower ✖ q70: prev= 440 ms, new= 541 ms, diff=1.23 slower ❌ q71: prev= 406 ms, new= 298 ms, diff=1.36 faster ✅ q72: prev=5276 ms, new=3156 ms, diff=1.67 faster ✅ q73: prev= 259 ms, new= 294 ms, diff=1.14 slower ✖ q74: prev= 733 ms, new= 656 ms, diff=1.12 faster ✔ q75: prev=1004 ms, new= 542 ms, diff=1.85 faster ✅ q76: prev= 362 ms, new= 232 ms, diff=1.56 faster ✅ q77: prev= 549 ms, new= 305 ms, diff=1.80 faster ✅ q78: prev= 954 ms, new= 299 ms, diff=3.19 faster ✅ q79: prev= 278 ms, new= 266 ms, diff=1.05 faster ✔ q80: prev= 689 ms, new= 398 ms, diff=1.73 faster ✅ q81: prev= 441 ms, new= 344 ms, diff=1.28 faster ✅ q82: prev= 315 ms, new= 310 ms, diff=1.02 faster ✔ q83: prev= 551 ms, new= 299 ms, diff=1.84 faster ✅ q84: prev= 321 ms, new= 359 ms, diff=1.12 slower ✖ q85: prev= 594 ms, new= 545 ms, diff=1.09 faster ✔ q86: prev= 168 ms, new= 186 ms, diff=1.11 slower ✖ q87: prev= 394 ms, new= 283 ms, diff=1.39 faster ✅ q88: prev= 479 ms, new= 484 ms, diff=1.01 slower ✖ q89: prev= 262 ms, new= 207 ms, diff=1.27 faster ✅ q90: prev= 352 ms, new= 274 ms, diff=1.28 faster ✅ q91: prev= 541 ms, new= 404 ms, diff=1.34 faster ✅ q92: prev= 326 ms, new= 207 ms, diff=1.57 faster ✅ q93: prev= 287 ms, new= 226 ms, diff=1.27 faster ✅ q94: prev= 401 ms, new= 451 ms, diff=1.12 slower ✖ q95: prev= 444 ms, new= 354 ms, diff=1.25 faster ✅ q96: prev= 193 ms, new= 235 ms, diff=1.22 slower ❌ q97: prev= 268 ms, new= 174 ms, diff=1.54 faster ✅ q98: prev= 165 ms, new= 176 ms, diff=1.07 slower ✖ q99: prev=1038 ms, new=1229 ms, diff=1.18 slower ✖ TOTAL: prev=62815 ms, new=44737 ms, diff=1.40 faster ✅ ``` </details> <details><summary>ClickBench 0-100: Tasks 912->609, prev=35116 ms, new=34185 ms, diff=1.03 faster ✔</summary> ``` q0: prev= 2 ms, new= 2 ms, diff=1.00 slower ✖ q1: prev= 506 ms, new= 396 ms, diff=1.28 faster ✅ q2: prev= 477 ms, new= 337 ms, diff=1.42 faster ✅ q3: prev= 323 ms, new= 452 ms, diff=1.40 slower ❌ q4: prev= 405 ms, new= 439 ms, diff=1.08 slower ✖ q5: prev= 585 ms, new= 555 ms, diff=1.05 faster ✔ q6: prev= 2 ms, new= 2 ms, diff=1.00 slower ✖ q7: prev= 253 ms, new= 363 ms, diff=1.43 slower ❌ q8: prev= 492 ms, new= 477 ms, diff=1.03 faster ✔ q9: prev= 620 ms, new= 635 ms, diff=1.02 slower ✖ q10: prev= 478 ms, new= 461 ms, diff=1.04 faster ✔ q11: prev= 449 ms, new= 392 ms, diff=1.15 faster ✔ q12: prev= 596 ms, new= 550 ms, diff=1.08 faster ✔ q13: prev= 771 ms, new= 792 ms, diff=1.03 slower ✖ q14: prev= 536 ms, new= 549 ms, diff=1.02 slower ✖ q15: prev= 477 ms, new= 414 ms, diff=1.15 faster ✔ q16: prev= 775 ms, new= 753 ms, diff=1.03 faster ✔ q17: prev= 749 ms, new= 726 ms, diff=1.03 faster ✔ q18: prev=1114 ms, new=1038 ms, diff=1.07 faster ✔ q19: prev= 380 ms, new= 391 ms, diff=1.03 slower ✖ q20: prev=1886 ms, new=1786 ms, diff=1.06 faster ✔ q21: prev=1807 ms, new=1775 ms, diff=1.02 faster ✔ q22: prev=2144 ms, new=2053 ms, diff=1.04 faster ✔ q23: prev=5267 ms, new=5354 ms, diff=1.02 slower ✖ q24: prev= 461 ms, new= 554 ms, diff=1.20 slower ❌ q25: prev= 409 ms, new= 482 ms, diff=1.18 slower ✖ q26: prev= 582 ms, new= 463 ms, diff=1.26 faster ✅ q27: prev=1958 ms, new=1955 ms, diff=1.00 faster ✔ q28: prev=2504 ms, new=2367 ms, diff=1.06 faster ✔ q29: prev= 297 ms, new= 261 ms, diff=1.14 faster ✔ q30: prev= 614 ms, new= 589 ms, diff=1.04 faster ✔ q31: prev= 714 ms, new= 656 ms, diff=1.09 faster ✔ q32: prev= 946 ms, new= 891 ms, diff=1.06 faster ✔ q33: prev=2557 ms, new=2397 ms, diff=1.07 faster ✔ q34: prev=2521 ms, new=2440 ms, diff=1.03 faster ✔ q35: prev= 459 ms, new= 438 ms, diff=1.05 faster ✔ TOTAL: prev=35116 ms, new=34185 ms, diff=1.03 faster ✔ ``` </details>

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from bda1c2d to 2879131 Compare June 8, 2026 12:19

gabotechs force-pushed the gabrielmusat/cost-calculation branch from bda802f to 4c66ecb Compare June 8, 2026 12:19

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from 2879131 to 4e96136 Compare June 8, 2026 14:57

gabotechs force-pushed the gabrielmusat/cost-calculation branch from 4c66ecb to 26eeb03 Compare June 8, 2026 15:11

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch 2 times, most recently from 5ad9e3f to 4cb23c0 Compare June 9, 2026 06:42

gabotechs force-pushed the gabrielmusat/cost-calculation branch 2 times, most recently from 3db9e03 to 20ed9aa Compare June 9, 2026 09:28

gabotechs marked this pull request as ready for review June 10, 2026 06:45

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch 2 times, most recently from 4c906e5 to 6b7d1a4 Compare June 11, 2026 08:00

gabotechs force-pushed the gabrielmusat/cost-calculation branch 2 times, most recently from b176c44 to 8422609 Compare June 11, 2026 08:27

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from 6b7d1a4 to fc261f1 Compare June 12, 2026 08:12

gabotechs force-pushed the gabrielmusat/cost-calculation branch 2 times, most recently from 156eaf7 to dad19f2 Compare June 12, 2026 09:36

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from fc261f1 to a4ceac5 Compare June 12, 2026 11:03

gabotechs force-pushed the gabrielmusat/cost-calculation branch from dad19f2 to b8e0d65 Compare June 12, 2026 11:03

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from a4ceac5 to 1c864d5 Compare June 14, 2026 17:20

gabotechs force-pushed the gabrielmusat/cost-calculation branch from b8e0d65 to 5575ad2 Compare June 14, 2026 17:20

gabotechs force-pushed the gabrielmusat/task-spawner-refactor-and-cache-invalidation branch from 1c864d5 to 3477c45 Compare June 17, 2026 06:25

Base automatically changed from gabrielmusat/task-spawner-refactor-and-cache-invalidation to main June 17, 2026 07:08

gabotechs force-pushed the gabrielmusat/cost-calculation branch 3 times, most recently from fd4e808 to 740e86e Compare June 21, 2026 19:05

asolimando reviewed Jun 24, 2026

View reviewed changes

gabotechs requested a review from asolimando June 25, 2026 12:45

asolimando mentioned this pull request Jun 26, 2026

Add AQE to DataFusion apache/datafusion#23194

Open

asolimando reviewed Jun 29, 2026

View reviewed changes

gabotechs requested a review from asolimando June 29, 2026 10:57

gabotechs added 9 commits June 29, 2026 18:12

Add NetworkBoundaryBuilder argument to inject_network_boundaries.rs

becc5be

Add a cost model for the physical layer

77b00f6

Use saturating_add and saturating_mul

5f586ce

Remove default workarounds

1257d58

Factor out complexity

8934aa2

Introduce memory and network complexity

9517fbd

Remove unused file

f8fb0f3

Bring back defensive defaults

74c74f2

Address comments

a06c956

gabotechs force-pushed the gabrielmusat/cost-calculation branch from beb1f72 to a06c956 Compare June 29, 2026 16:13

asolimando approved these changes Jun 30, 2026

View reviewed changes

Use Precision for Cost

b764c99

gabotechs merged commit 462739d into main Jun 30, 2026
17 checks passed

gabotechs deleted the gabrielmusat/cost-calculation branch June 30, 2026 10:48

Navigation Menu

Repository navigation

Add cost model - #486

Add cost model#486
gabotechs merged 10 commits into
mainfrom
gabrielmusat/cost-calculation

gabotechs commented Jun 8, 2026 •

edited

Loading

Uh oh!

asolimando left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

asolimando left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

asolimando left a comment

Uh oh!

gabotechs commented Jun 30, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

gabotechs commented Jun 8, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

asolimando left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

asolimando left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

asolimando left a comment

Choose a reason for hiding this comment

Uh oh!

gabotechs commented Jun 30, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

gabotechs commented Jun 8, 2026 •

edited

Loading