[SPARK-58513][CORE] Init UnifiedMemeoryManager should check driver config in driver , check executor config in executor

[AngersZhuuuu]

What changes were proposed in this pull request?

This patch updates UnifiedMemoryManager to validate memory according to the current process role.

• SparkEnv passes the current process role to UnifiedMemoryManager.
• Driver validates Driver JVM memory only.
• Executor validates spark.executor.memory only.

Why are the changes needed?

When spark.executor.memory is set to 500m, the Executor may fail with the following error:

Exception in thread "main" java.lang.IllegalArgumentException: System memory 466092032 must be at least 471859200. Please increase heap size using the --driver-memory option or spark.driver.memory in Spark configuration.

The error is misleading because UnifiedMemoryManager is initialized in the Executor, but the shared validation logic checks the runtime system memory as Driver memory and reports Driver-specific configuration guidance.

The Executor should validate its own executor memory configuration instead of applying the Driver memory check.

Does this PR introduce any user-facing change?

Yes.

Previously, an Executor could report a Driver memory error when its runtime system memory was below the Driver threshold.

After this change:

• Driver-side memory failures continue to report Driver memory configuration guidance.
• Executor-side memory failures validate and report Executor memory configuration.
• An Executor with sufficient spark.executor.memory will not fail because of the Driver heap validation.

How was this patch tested?

Added UT

Was this patch authored or co-authored using generative AI tooling?

Yea

[uros-b]

Thank you @AngersZhuuuu!

[AngersZhuuuu]

This fix have a problem that may cause gluten failed to init, since we change teh method's parameter. So there have two way to fix this:

1. Add a compatible method and set its default value to true.

def apply(conf: SparkConf, numCores: Int): UnifiedMemoryManager = {
    apply(conf, numCores, isDriver = true)
  }

  def apply(
      conf: SparkConf,
      numCores: Int,
      isDriver: Boolean = true): UnifiedMemoryManager = {
      isDriver: Boolean): UnifiedMemoryManager = {
    val maxMemory = getMaxMemory(conf, isDriver)
    new UnifiedMemoryManager(
      conf,
      maxHeapMemory = maxMemory,
      onHeapStorageRegionSize =
        (maxMemory * conf.get(config.MEMORY_STORAGE_FRACTION)).toLong,
      numCores = numCores)
  }

2. add

// Compatible with legacy calls, such as Gluten
def apply(conf: SparkConf, numCores: Int): UnifiedMemoryManager = {
  applyInternal(conf, numCores, None)
}


// Spark internally explicitly passes in roles
private def applyInternal(
    conf: SparkConf,
    numCores: Int,
    isDriver: Option[Boolean]): UnifiedMemoryManager = {
  val maxMemory = getMaxMemory(conf, isDriver)
  ...
}

isDriver match {
  case Some(true) =>
    // Only check Driver memory

  case Some(false) =>
    // Only check Executor memory

  case None =>
    // both check：Driver & Executor 
}

Which one do you think is better? @uros-b