Merge branch 'apache:master' into windowFuncFix

Author: xin-aurora

.github/workflows/build_and_test.yml

@@ -1240,7 +1240,7 @@ jobs:
           sudo apt update
           sudo apt-get install r-base
       - name: Start Minikube
-        uses: medyagh/setup-minikube@v0.0.18
+        uses: medyagh/setup-minikube@v0.0.19
         with:
           kubernetes-version: "1.33.0"
           # Github Action limit cpu:2, memory: 6947MB, limit to 2U6G for better resource statistic

.github/workflows/build_infra_images_cache.yml

@@ -36,6 +36,7 @@ on:
     - 'dev/spark-test-image/python-309/Dockerfile'
     - 'dev/spark-test-image/python-310/Dockerfile'
     - 'dev/spark-test-image/python-311/Dockerfile'
+    - 'dev/spark-test-image/python-311-classic-only/Dockerfile'
     - 'dev/spark-test-image/python-312/Dockerfile'
     - 'dev/spark-test-image/python-313/Dockerfile'
     - 'dev/spark-test-image/python-313-nogil/Dockerfile'
@@ -191,6 +192,19 @@ jobs:
       - name: Image digest (PySpark with Python 3.11)
         if: hashFiles('dev/spark-test-image/python-311/Dockerfile') != ''
         run: echo ${{ steps.docker_build_pyspark_python_311.outputs.digest }}
+      - name: Build and push (PySpark Classic Only with Python 3.11)
+        if: hashFiles('dev/spark-test-image/python-311-classic-only/Dockerfile') != ''
+        id: docker_build_pyspark_python_311_classic_only
+        uses: docker/build-push-action@v6
+        with:
+          context: ./dev/spark-test-image/python-311-classic-only/
+          push: true
+          tags: ghcr.io/apache/spark/apache-spark-github-action-image-pyspark-python-311-classic-only-cache:${{ github.ref_name }}-static
+          cache-from: type=registry,ref=ghcr.io/apache/spark/apache-spark-github-action-image-pyspark-python-311-classic-only-cache:${{ github.ref_name }}
+          cache-to: type=registry,ref=ghcr.io/apache/spark/apache-spark-github-action-image-pyspark-python-311-classic-only-cache:${{ github.ref_name }},mode=max
+      - name: Image digest (PySpark Classic Only with Python 3.11)
+        if: hashFiles('dev/spark-test-image/python-311-classic-only/Dockerfile') != ''
+        run: echo ${{ steps.docker_build_pyspark_python_311_classic_only.outputs.digest }}
       - name: Build and push (PySpark with Python 3.12)
         if: hashFiles('dev/spark-test-image/python-312/Dockerfile') != ''
         id: docker_build_pyspark_python_312

.github/workflows/build_python_3.11_classic_only.yml

@@ -0,0 +1,47 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+name: "Build / Python-only Classic-only (master, Python 3.11)"
+
+on:
+  schedule:
+    - cron: '0 0 */3 * *'
+  workflow_dispatch:
+
+jobs:
+  run-build:
+    permissions:
+      packages: write
+    name: Run
+    uses: ./.github/workflows/build_and_test.yml
+    if: github.repository == 'apache/spark'
+    with:
+      java: 17
+      branch: master
+      hadoop: hadoop3
+      envs: >-
+        {
+          "PYSPARK_IMAGE_TO_TEST": "python-311-classic-only",
+          "PYTHON_TO_TEST": "python3.11"
+        }
+      jobs: >-
+        {
+          "pyspark": "true",
+          "pyspark-pandas": "true"
+        }

common/utils/src/main/resources/error/error-conditions.json

@@ -8641,11 +8641,6 @@
       "duration() called on unfinished task"
     ]
   },
-  "_LEGACY_ERROR_TEMP_3027" : {
-    "message" : [
-      "Unrecognized <schedulerModeProperty>: <schedulingModeConf>"
-    ]
-  },
   "_LEGACY_ERROR_TEMP_3028" : {
     "message" : [
       "<errorMsg>"

common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala

@@ -57,6 +57,15 @@ private object ConfigHelpers {
     }
   }
 
+  def toEnum[E <: Enum[E]](s: String, enumClass: Class[E], key: String): E = {
+    enumClass.getEnumConstants.find(_.name().equalsIgnoreCase(s.trim)) match {
+      case Some(enum) => enum
+      case None =>
+        throw new IllegalArgumentException(
+          s"$key should be one of ${enumClass.getEnumConstants.mkString(", ")}, but was $s")
+    }
+  }
+
   def stringToSeq[T](str: String, converter: String => T): Seq[T] = {
     SparkStringUtils.stringToSeq(str).map(converter)
   }
@@ -287,6 +296,11 @@ private[spark] case class ConfigBuilder(key: String) {
     new TypedConfigBuilder(this, toEnum(_, e, key))
   }
 
+  def enumConf[E <: Enum[E]](e: Class[E]): TypedConfigBuilder[E] = {
+    checkPrependConfig
+    new TypedConfigBuilder(this, toEnum(_, e, key))
+  }
+
   def timeConf(unit: TimeUnit): TypedConfigBuilder[Long] = {
     checkPrependConfig
     new TypedConfigBuilder(this, timeFromString(_, unit), timeToString(_, unit))

core/src/main/scala/org/apache/spark/deploy/ExternalShuffleService.scala

@@ -31,7 +31,6 @@ import org.apache.spark.network.crypto.AuthServerBootstrap
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.server.{TransportServer, TransportServerBootstrap}
 import org.apache.spark.network.shuffle.ExternalBlockHandler
-import org.apache.spark.network.shuffledb.DBBackend
 import org.apache.spark.network.util.TransportConf
 import org.apache.spark.util.{ShutdownHookManager, Utils}
 
@@ -86,11 +85,11 @@ class ExternalShuffleService(sparkConf: SparkConf, securityManager: SecurityMana
   protected def newShuffleBlockHandler(conf: TransportConf): ExternalBlockHandler = {
     if (sparkConf.get(config.SHUFFLE_SERVICE_DB_ENABLED) && enabled) {
       val shuffleDBName = sparkConf.get(config.SHUFFLE_SERVICE_DB_BACKEND)
-      val dbBackend = DBBackend.byName(shuffleDBName)
-      logInfo(log"Use ${MDC(SHUFFLE_DB_BACKEND_NAME, dbBackend.name())} as the implementation of " +
+      logInfo(
+        log"Use ${MDC(SHUFFLE_DB_BACKEND_NAME, shuffleDBName.name())} as the implementation of " +
         log"${MDC(SHUFFLE_DB_BACKEND_KEY, config.SHUFFLE_SERVICE_DB_BACKEND.key)}")
       new ExternalBlockHandler(conf,
-        findRegisteredExecutorsDBFile(dbBackend.fileName(registeredExecutorsDB)))
+        findRegisteredExecutorsDBFile(shuffleDBName.fileName(registeredExecutorsDB)))
     } else {
       new ExternalBlockHandler(conf, null)
     }

core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala

@@ -267,19 +267,6 @@ private[spark] object SparkCoreErrors {
     new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3026")
   }
 
-  def unrecognizedSchedulerModePropertyError(
-      schedulerModeProperty: String,
-      schedulingModeConf: String): Throwable = {
-    new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3027",
-      messageParameters = Map(
-        "schedulerModeProperty" -> schedulerModeProperty,
-        "schedulingModeConf" -> schedulingModeConf
-      ),
-      cause = null
-    )
-  }
-
   def sparkError(errorMsg: String): Throwable = {
     new SparkException(
       errorClass = "_LEGACY_ERROR_TEMP_3028",

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -807,10 +807,8 @@ package object config {
       .doc("Specifies a disk-based store used in shuffle service local db. " +
         "ROCKSDB or LEVELDB (deprecated).")
       .version("3.4.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .checkValues(DBBackend.values.map(_.toString).toSet)
-      .createWithDefault(DBBackend.ROCKSDB.name)
+      .enumConf(classOf[DBBackend])
+      .createWithDefault(DBBackend.ROCKSDB)
 
   private[spark] val SHUFFLE_SERVICE_PORT =
     ConfigBuilder("spark.shuffle.service.port").version("1.2.0").intConf.createWithDefault(7337)
@@ -2295,9 +2293,8 @@ package object config {
   private[spark] val SCHEDULER_MODE =
     ConfigBuilder("spark.scheduler.mode")
       .version("0.8.0")
-      .stringConf
-      .transform(_.toUpperCase(Locale.ROOT))
-      .createWithDefault(SchedulingMode.FIFO.toString)
+      .enumConf(SchedulingMode)
+      .createWithDefault(SchedulingMode.FIFO)
 
   private[spark] val SCHEDULER_REVIVE_INTERVAL =
     ConfigBuilder("spark.scheduler.revive.interval")

core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala

@@ -89,7 +89,7 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext
             log"${MDC(LogKeys.FILE_NAME, DEFAULT_SCHEDULER_FILE)}")
           Some((is, DEFAULT_SCHEDULER_FILE))
         } else {
-          val schedulingMode = SchedulingMode.withName(sc.conf.get(SCHEDULER_MODE))
+          val schedulingMode = sc.conf.get(SCHEDULER_MODE)
           rootPool.addSchedulable(new Pool(
             DEFAULT_POOL_NAME, schedulingMode, DEFAULT_MINIMUM_SHARE, DEFAULT_WEIGHT))
           logInfo(log"Fair scheduler configuration not found, created default pool: " +

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -181,15 +181,7 @@ private[spark] class TaskSchedulerImpl(
 
   private var schedulableBuilder: SchedulableBuilder = null
   // default scheduler is FIFO
-  private val schedulingModeConf = conf.get(SCHEDULER_MODE)
-  val schedulingMode: SchedulingMode =
-    try {
-      SchedulingMode.withName(schedulingModeConf)
-    } catch {
-      case e: java.util.NoSuchElementException =>
-        throw SparkCoreErrors.unrecognizedSchedulerModePropertyError(SCHEDULER_MODE_PROPERTY,
-          schedulingModeConf)
-    }
+  val schedulingMode: SchedulingMode = conf.get(SCHEDULER_MODE)
 
   val rootPool: Pool = new Pool("", schedulingMode, 0, 0)
 

core/src/main/scala/org/apache/spark/ui/UIWorkloadGenerator.scala

@@ -50,9 +50,7 @@ private[spark] object UIWorkloadGenerator {
     val conf = new SparkConf().setMaster(args(0)).setAppName("Spark UI tester")
 
     val schedulingMode = SchedulingMode.withName(args(1))
-    if (schedulingMode == SchedulingMode.FAIR) {
-      conf.set(SCHEDULER_MODE, "FAIR")
-    }
+    conf.set(SCHEDULER_MODE, schedulingMode)
     val nJobSet = args(2).toInt
     val sc = new SparkContext(conf)
 

core/src/test/scala/org/apache/spark/JobCancellationSuite.scala

@@ -59,7 +59,7 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
   }
 
   test("local mode, FIFO scheduler") {
-    val conf = new SparkConf().set(SCHEDULER_MODE, "FIFO")
+    val conf = new SparkConf().set(SCHEDULER_MODE.key, "FIFO")
     sc = new SparkContext("local[2]", "test", conf)
     testCount()
     testTake()
@@ -68,7 +68,7 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
   }
 
   test("local mode, fair scheduler") {
-    val conf = new SparkConf().set(SCHEDULER_MODE, "FAIR")
+    val conf = new SparkConf().set(SCHEDULER_MODE.key, "FAIR")
     val xmlPath = getClass.getClassLoader.getResource("fairscheduler.xml").getFile()
     conf.set(SCHEDULER_ALLOCATION_FILE, xmlPath)
     sc = new SparkContext("local[2]", "test", conf)
@@ -79,7 +79,7 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
   }
 
   test("cluster mode, FIFO scheduler") {
-    val conf = new SparkConf().set(SCHEDULER_MODE, "FIFO")
+    val conf = new SparkConf().set(SCHEDULER_MODE.key, "FIFO")
     sc = new SparkContext("local-cluster[2,1,1024]", "test", conf)
     testCount()
     testTake()
@@ -88,7 +88,7 @@ class JobCancellationSuite extends SparkFunSuite with Matchers with BeforeAndAft
   }
 
   test("cluster mode, fair scheduler") {
-    val conf = new SparkConf().set(SCHEDULER_MODE, "FAIR")
+    val conf = new SparkConf().set(SCHEDULER_MODE.key, "FAIR")
     val xmlPath = getClass.getClassLoader.getResource("fairscheduler.xml").getFile()
     conf.set(SCHEDULER_ALLOCATION_FILE, xmlPath)
     sc = new SparkContext("local-cluster[2,1,1024]", "test", conf)

core/src/test/scala/org/apache/spark/internal/config/ConfigEntrySuite.scala

@@ -21,6 +21,7 @@ import java.util.Locale
 import java.util.concurrent.TimeUnit
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.network.shuffledb.DBBackend
 import org.apache.spark.network.util.ByteUnit
 import org.apache.spark.util.SparkConfWithEnv
 
@@ -389,7 +390,7 @@ class ConfigEntrySuite extends SparkFunSuite {
   }
 
 
-  test("SPARK-51874: Add Enum support to ConfigBuilder") {
+  test("SPARK-51874: Add Scala Enumeration support to ConfigBuilder") {
     object MyTestEnum extends Enumeration {
       val X, Y, Z = Value
     }
@@ -408,4 +409,20 @@ class ConfigEntrySuite extends SparkFunSuite {
     }
     assert(e.getMessage === s"${enumConf.key} should be one of X, Y, Z, but was A")
   }
+
+  test("SPARK-51896: Add Java enum support to ConfigBuilder") {
+    val conf = new SparkConf()
+    val enumConf = ConfigBuilder("spark.test.java.enum.key")
+      .enumConf(classOf[DBBackend])
+      .createWithDefault(DBBackend.LEVELDB)
+    assert(conf.get(enumConf) === DBBackend.LEVELDB)
+    conf.set(enumConf, DBBackend.ROCKSDB)
+    assert(conf.get(enumConf) === DBBackend.ROCKSDB)
+    val e = intercept[IllegalArgumentException] {
+      conf.set(enumConf.key, "ANYDB")
+      conf.get(enumConf)
+    }
+    assert(e.getMessage ===
+      s"${enumConf.key} should be one of ${DBBackend.values.mkString(", ")}, but was ANYDB")
+  }
 }

dev/spark-test-image/python-311-classic-only/Dockerfile

@@ -0,0 +1,79 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Image for building and testing Spark branches. Based on Ubuntu 22.04.
+# See also in https://hub.docker.com/_/ubuntu
+FROM ubuntu:jammy-20240911.1
+LABEL org.opencontainers.image.authors="Apache Spark project <dev@spark.apache.org>"
+LABEL org.opencontainers.image.licenses="Apache-2.0"
+LABEL org.opencontainers.image.ref.name="Apache Spark Infra Image For PySpark Classic with Python 3.11"
+# Overwrite this label to avoid exposing the underlying Ubuntu OS version label
+LABEL org.opencontainers.image.version=""
+
+ENV FULL_REFRESH_DATE=20250424
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV DEBCONF_NONINTERACTIVE_SEEN=true
+
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    ca-certificates \
+    curl \
+    gfortran \
+    git \
+    gnupg \
+    libcurl4-openssl-dev \
+    libfontconfig1-dev \
+    libfreetype6-dev \
+    libfribidi-dev \
+    libgit2-dev \
+    libharfbuzz-dev \
+    libjpeg-dev \
+    liblapack-dev \
+    libopenblas-dev \
+    libpng-dev \
+    libpython3-dev \
+    libssl-dev \
+    libtiff5-dev \
+    libxml2-dev \
+    openjdk-17-jdk-headless \
+    pkg-config \
+    qpdf \
+    tzdata \
+    software-properties-common \
+    wget \
+    zlib1g-dev
+
+# Install Python 3.11
+RUN add-apt-repository ppa:deadsnakes/ppa
+RUN apt-get update && apt-get install -y \
+    python3.11 \
+    && apt-get autoremove --purge -y \
+    && apt-get clean \
+    && rm -rf /var/lib/apt/lists/*
+
+
+ARG BASIC_PIP_PKGS="numpy pyarrow>=19.0.0 pandas==2.2.3 plotly<6.0.0 matplotlib openpyxl memory-profiler>=0.61.0 mlflow>=2.8.1 scipy scikit-learn>=1.3.2"
+ARG TEST_PIP_PKGS="coverage unittest-xml-reporting"
+
+# Install Python 3.11 packages
+RUN curl -sS https://bootstrap.pypa.io/get-pip.py | python3.11
+RUN python3.11 -m pip install --ignore-installed blinker>=1.6.2 # mlflow needs this
+RUN python3.11 -m pip install $BASIC_PIP_PKGS $TEST_PIP_PKGS && \
+    python3.11 -m pip install 'torch<2.6.0' torchvision --index-url https://download.pytorch.org/whl/cpu && \
+    python3.11 -m pip install deepspeed torcheval && \
+    python3.11 -m pip cache purge