apache / apache/datafusion-comet

Explore cost-based optimizations

Open
#833 0 comments 3 reactions 0 assignees View on GitHub
enhancement
Dominant language
Scala
Stars
1.3k
Forks
373
Avg merge
2d 6h
Merged PRs (30d)
190

Description

### What is the problem the feature request solves?

I recently worked on a few experimental PRs around CBO that I am going to close because they did not provide an immediate benefit, but I wanted to document the work so am using this issue for that.

## Using DataFusion's Physical Optimizer

DataFusion provides a physical optimizer and there may be benefit in the future from applying DataFusion's rules or custom Comet rules. For example, injecting CopyExec into the plan would be ideal for an optimizer rule.

In `jni_api.rs` we would need to add the rules that we want to enable:

```rust
let state = SessionStateBuilder::new()
.with_config(session_config)
.with_runtime_env(Arc::new(runtime))
.with_default_features()
.with_physical_optimizer_rules(vec![Arc::new(TopKAggregation::new())])
.build();
```

Then in `planner.rs` we could add the call to optimize the plan:

```rust
pub fn optimize_plan(
&self,
plan: Arc,
) -> Result, ExecutionError> {
// optimize the physical plan
let datafusion_planner = DefaultPhysicalPlanner::default();
datafusion_planner
.optimize_physical_plan(plan, &self.session_ctx.state(), |_, _| {})
.map_err(|e| e.into())
}
```

Because we receive a plan that is already optimized by Spark, there is no immediate benefit in enabling the current rules from DataFusion.

## Passing statistics down to the native plan

It is possible for use to pass Spark statistics down to the native plan. For example, we can add this `QueryPlanSerde.scala`:

```scala
op match {
case qs: QueryStageExec =>
qs.computeStats() match {
case Some(stats) =>
val statsBuilder = OperatorOuterClass.Statistics.newBuilder()
stats.rowCount.foreach(c => statsBuilder.setRowCount(c.toFloat))
statsBuilder.setSizeInBytes(stats.sizeInBytes.toFloat)
scanBuilder.setStatistics(statsBuilder.build())
case _ =>
}
case _ =>
}
```

It is also possible to get size in bytes from any Hadoop input relations and infer row count based on schema.

There is no value in doing this though until we have optimizer rules that can make use of this data.

## Cost-model to determine when to fall back to Spark

We could implement a cost-model with the relative costs of Comet vs Spark operators and expressions and fall back to Spark in the case where we estimate that Comet would be more expensive.

### Describe the potential solution

_No response_

### Additional context

_No response_

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.