apache / apache/datafusion-comet

Use Spark ResourceProfiles to give native and JVM stages independent memory configs

Open
#5,116 1 comment 0 reactions 0 assignees View on GitHub
enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.