apache / apache/datafusion-comet
Use Spark ResourceProfiles to give native and JVM stages independent memory configs
- Dominant language
- Scala
- Stars
- 1.3k
- Forks
- 373
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 198
Description
### What is the problem the feature request solves?
Provisioning executor memory for Comet is one of the longest-standing operational pain points.
Native execution wants large **off-heap** memory; JVM (fallback) execution wants large **on-heap** memory. Today both are set on the same executor for the whole application, so you either
- over-provision both (wasted memory per executor) or
- pick a compromise that starves whichever path a given query actually uses.
Until recently, a single query stage could interleave native (off-heap) and JVM (on-heap) operators, so no single memory configuration could ever be correct for it.
As of #4519 (stage-based fallback), a query stage is **homogeneous** — it either runs fully native or is reverted entirely to JVM row-based execution at the shuffle boundary. Native and JVM operators no longer interleave within a stage. This makes per-stage memory tuning meaningful.
### Describe the potential solution
A Spark stage maps cleanly onto a Spark **ResourceProfile** (profiles attach to RDDs, and a stage is a set of RDDs bounded by shuffles). Now that stages are homogeneous, Comet can attach a ResourceProfile to the RDD it generates per stage, based on the native-vs-JVM decision:
- **Native stages** → profile with large off-heap, minimal heap
- **Reverted JVM stages** → profile with large heap, minimal off-heap
Each stage then gets memory shaped for how it actually executes, instead of one compromise config for the whole application.
Example profiles:
```scala
// Native stages: off-heap heavy
val nativeProfile = new ResourceProfileBuilder()
.require(new ExecutorResourceRequests()
.memory("2g") // small heap
.offHeap("14g") // large off-heap for native execution
.cores(4))
.build()
// JVM stages: on-heap heavy
val jvmProfile = new ResourceProfileBuilder()
.require(new ExecutorResourceRequests()
.memory("14g") // large heap for JVM row execution
.offHeap("2g") // minimal off-heap
.cores(4))
.build()
```
### Additional context
_No response_
Contributor guide
Research direction
Start by reading the stage-based fallback change in #4519 and the Spark ResourceProfileBuilder, ExecutorResourceRequests, and RDD stage APIs referenced in the issue. Determine how Comet identifies native versus reverted JVM stages and where it generates the per-stage RDD; done means each stage receives the appropriate memory profile without breaking existing execution.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100