Current TaskEstimator API challenges

[gabotechs]

The TaskEstimator API was introduced in

• Rework distributed planning logic #259

For allowing a better task assignation control than just static config numbers. The approach followed by this API is the following:

• Nodes can decide how many tasks are appropriate for them
• The amount of tasks appropriate for each individual node is calculated from a bottom to top fashion, taking into account how many tasks each node decided
• Upon reaching a network boundary, all the nodes in each stage reach a consensus on whats the appropriate final task count

The current API has a big limitation:

The current TaskEstimator API lets nodes decide how many tasks are appropriate for them taking into account how much data they are going to return, but it does not take into account the amount of compute that will happen during the query.

For example:

Plan that currently gets distributed but it shouldnt

                           ┌───────────────────┐                           
                           │  ProjectionExec   │                           
                           └─────────▲─────────┘                           
                                     │                                     
                           ┌─────────┴─────────┐                           
                           │     UnionExec     │                           
                           └─────▲─▲───▲─▲─────┘                           
         ┌──────────────────┬────┴─┘   └─┴─────┬──────────────────┐        
         │                  │                  │                  │        
┌────────────────┐ ┌────────────────┐ ┌────────────────┐ ┌────────────────┐
│   DataSource   │ │   DataSource   │ │   DataSource   │ │   DataSource   │
└────────────────┘ └────────────────┘ └────────────────┘ └────────────────┘

This plan has the capacity of pulling a lot of data up, but computationally speaking is very inexpensive, as it's almost a pure IO query. This plan should not be distributed, it's fine to run this in a single node.

Another example:

Plan that currently does not get distributed but it should

      ┌───────────────────┐      
      │  ProjectionExec   │      
      └─────────▲─────────┘      
                │                
┌───────────────┴───────────────┐
│ ExtremelyExpensiveComputeExec │
└───────────────▲───────────────┘
                │                
   ┌────────────┴───────────┐    
   │    SmallDataSource     │    
   └────────────────────────┘    

The current TaskEstimator API might decide that, as not a huge amount of data is flowing through SmallDataSource, then it does not need to be distributed, but it doesn't account for the fact that ExtremelyExpensiveComputeExec might be doing something crazy computationally speaking that might be nice to distribute.

---

Right now the amount of tasks is determined by looking at the data size, but it should take into account the product of "data size" * "compute", so that pure IO plans do not get unnecessarily distributed, and compute heavy plans are not left undistributed by mistake.

[gabotechs]

Take

[jayshrivastava]

Are there any examples of ExtremelyExpensiveComputeExec? Maybe these?

• Array aggregations
• GPU aggregations

I'm also curious what other databases do to measure how expensive operations are.

It sounds like each ExecutionPlan needs a weight that effectively measures "instructions per row" (including allocs and whatever other sys calls which also take cpu time). Or maybe an enum like
FullyIOBound, SmallCPUBound, MediumCPUBound, HeavyCPUBound.

[jayshrivastava]

Separate but related problem - since you brought up that sending data over then network is expensive. If we have a case where a stage outputs a LOT of rows, we may benefit from pushing down some node which reduces the cardinality of the rows into that stage so it outputs less data.

[gabotechs]

Are there any examples of ExtremelyExpensiveComputeExec? Maybe these?

No idea to be honest. I do imagine that array aggregations are expensive, and DISTINCT ALL aggregations should also be pretty heavy. I wonder how do cost-based optimizer do this, I imagine they do not to associate a "cost" of doing things, maybe looking at Calcite or other systems for inspiration could yield something useful (cc @asolimando)

It sounds like each ExecutionPlan needs a weight that effectively measures "instructions per row" (including allocs and whatever other sys calls which also take cpu time). Or maybe an enum like
FullyIOBound, SmallCPUBound, MediumCPUBound, HeavyCPUBound.

I imagine that every node that doesn't do network requests is fully CPU bound if you observe it in isolation. Learn from cost-based system how do they represent these kind of things might be a good approach.

If we have a case where a stage outputs a LOT of rows, we may benefit from pushing down some node which reduces the cardinality of the rows into that stage so it outputs less data.

Definitely! the more we push down the better, although I do imagine that vanilla DataFusion is already making the right decisions here, and I think it does a decent job. I get the impression that in this project is fine to assume that vanilla DataFusion is already doing its best in this front, and if it isn't, we should probably go contribute improvements there.

[asolimando]

Thanks for the ping @gabotechs, I've been looking at how different systems approach this.

TLDR: Research and production systems consistently show that cardinality estimation matters far more than detailed cost models.

Research

The main paper supporting this thesis is Leis et al. 2015 (and its follow-up Leis et al. 2025 from the same authors), which found that cardinality estimation has a larger impact on plan quality than cost model sophistication. Even with a perfect cost model, bad cardinality estimates lead to bad plans. The converse isn't true, as good cardinality estimates with a simple cost model still produce good plans. This was proven by using main open-source and commercial databases, it's not a purely theoretical result.

You are rightfully asking how successful distributed databases do this, I took a look at Apache Spark and Trino, but also double-checked how Apache Calcite handles this.

Spark

Spark explicitly states this in their code (CostBasedJoinReorder.scala#L140-L141):

/**
 * For cost evaluation, since physical costs for operators are not available
 * currently, we use cardinalities and sizes to compute costs.
 */

Their Cost class is just Cost(card: BigInt, size: BigInt), cardinality and byte size, there are no per-operator CPU weights.

Trino

Trino has a more elaborate PlanCostEstimate with cpuCost, maxMemory, networkCost (PlanCostEstimate.java#L34-L37), but look at what these actually are (CostCalculatorUsingExchanges.java#L131, #L173-L174):

double cpuCost = stats.getOutputSizeInBytes(symbols);

double cpuCost = sourceStats.getOutputSizeInBytes(node.getSource().getOutputSymbols());
double memoryCost = aggregationStats.getOutputSizeInBytes(node.getOutputSymbols());

• CPU cost = bytes processed
• Memory cost = bytes materialized
• Network cost = bytes transferred

They are all derived from cardinality * row size, not from per-expression complexity analysis.

The CostComparator then does a weighted sum with configurable weights, allowing some level of INFRA-specific tuning, without changes to the cost model itself:

double leftCost = left.getCpuCost() * cpuWeight
        + left.getMaxMemory() * memoryWeight
        + left.getNetworkCost() * networkWeight;

Calcite

Calcite similarly focuses on row counts, the RelOptCost class has rows/cpu/io dimensions, but cpu is typically just rowCount * SOME_FACTOR in most implementations I have seen.

Conclusion

I am not against per-expression cost modeling but research and successful distributed systems confirm that this is not needed, I'd rather focus on improving cardinality first, then use a simple cost-model, and only then explore per-expression modeling only if all this proves insufficient for us.

For network boundary decisions (our datafusion-distributed context here), what matters is if a given operator reduces or increases data volume before the shuffle, which can satisfactorily be answered with cardinality + selectivity.

Another argument is that if your cost-model gets too close to the implementation details, then you become tightly coupled to them, making it harder to evolve the execution layer independently.

Next Steps

For this project, the valuable short-term improvements would be (IMO):

1. Better cardinality estimation: especially for aggregates with GROUP BY. it needs distinct count computation/propagation (#15265) and start using it (#18628)
2. Output byte size estimation : rows * avg_row_width for network cost awareness (we discussed this offline, avg_row_width can be computed based on the row type and avg_len/avg_size for collections or length varying types)
3. Selectivity estimation: for filters, to know which operators reduce data

Since all these would benefit single-node DataFusion too, I think it makes sense to improve them in DataFusion itself. Based on the discussions I've followed over the last few months, I see more and more decision points that could use better statistics and cost-based decisions, so I don't expect any pushback from the community there.