Merge branch 'master' into limit_pushdown

Author: zml1206

.github/PULL_REQUEST_TEMPLATE

@@ -9,7 +9,7 @@ Thanks for sending a pull request!  Here are some tips for you:
   7. If you want to add a new configuration, please read the guideline first for naming configurations in
      'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
   8. If you want to add or modify an error type or message, please read the guideline first in
-     'core/src/main/resources/error/README.md'.
+     'common/utils/src/main/resources/error/README.md'.
 -->
 
 ### What changes were proposed in this pull request?

.github/workflows/benchmark.yml

@@ -70,7 +70,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Cache Scala, SBT and Maven
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         with:
           path: |
             build/apache-maven-*
@@ -81,15 +81,15 @@ jobs:
           restore-keys: |
             build-
       - name: Cache Coursier local repository
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         with:
           path: ~/.cache/coursier
           key: benchmark-coursier-${{ github.event.inputs.jdk }}-${{ hashFiles('**/pom.xml', '**/plugins.sbt') }}
           restore-keys: |
             benchmark-coursier-${{ github.event.inputs.jdk }}
       - name: Cache TPC-DS generated data
         id: cache-tpcds-sf-1
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         with:
           path: ./tpcds-sf-1
           key: tpcds-${{ hashFiles('.github/workflows/benchmark.yml', 'sql/core/src/test/scala/org/apache/spark/sql/TPCDSSchema.scala') }}
@@ -139,7 +139,7 @@ jobs:
       with:
         fetch-depth: 0
     - name: Cache Scala, SBT and Maven
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: |
           build/apache-maven-*
@@ -150,7 +150,7 @@ jobs:
         restore-keys: |
           build-
     - name: Cache Coursier local repository
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: ~/.cache/coursier
         key: benchmark-coursier-${{ github.event.inputs.jdk }}-${{ hashFiles('**/pom.xml', '**/plugins.sbt') }}
@@ -164,7 +164,7 @@ jobs:
     - name: Cache TPC-DS generated data
       if: contains(github.event.inputs.class, 'TPCDSQueryBenchmark') || contains(github.event.inputs.class, '*')
       id: cache-tpcds-sf-1
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: ./tpcds-sf-1
         key: tpcds-${{ hashFiles('.github/workflows/benchmark.yml', 'sql/core/src/test/scala/org/apache/spark/sql/TPCDSSchema.scala') }}

.github/workflows/build_and_test.yml

@@ -26,8 +26,9 @@ on:
 jobs:
   run-build:
     strategy:
+      fail-fast: false
       matrix:
-        pyversion: ["pypy3,python3.10", "python3.11,python3.12"]
+        pyversion: ["pypy3", "python3.10", "python3.11", "python3.12"]
     permissions:
       packages: write
     name: Run

.github/workflows/build_python.yml

@@ -132,7 +132,7 @@ jobs:
           git -c user.name='Apache Spark Test Account' -c user.email='sparktestacc@gmail.com' commit -m "Merged commit" --allow-empty
       # Cache local repositories. Note that GitHub Actions cache has a 2G limit.
       - name: Cache Scala, SBT and Maven
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         with:
           path: |
             build/apache-maven-*
@@ -143,7 +143,7 @@ jobs:
           restore-keys: |
             build-
       - name: Cache Maven local repository
-        uses: actions/cache@v3
+        uses: actions/cache@v4
         with:
           path: ~/.m2/repository
           key: java${{ matrix.java }}-maven-${{ hashFiles('**/pom.xml') }}

.github/workflows/maven_test.yml

@@ -41,7 +41,7 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           script: |
             const endpoint = 'GET /repos/:owner/:repo/actions/workflows/:id/runs?&branch=:branch'
-            const check_run_endpoint = 'GET /repos/:owner/:repo/commits/:ref/check-runs'
+            const check_run_endpoint = 'GET /repos/:owner/:repo/commits/:ref/check-runs?per_page=100'
 
             // TODO: Should use pull_request.user and pull_request.user.repos_url?
             // If a different person creates a commit to another forked repo,
@@ -115,8 +115,22 @@ jobs:
               }
 
               // Here we get check run ID to provide Check run view instead of Actions view, see also SPARK-37879.
-              const check_runs = await github.request(check_run_endpoint, check_run_params)
-              const check_run_head = check_runs.data.check_runs.filter(r => r.name === "Run / Check changes")[0]
+              let retryCount = 0;
+              let check_run_head;
+              while (retryCount < 3) {
+                const check_runs = await github.request(check_run_endpoint, check_run_params);
+                check_run_head = check_runs.data.check_runs.find(r => r.name === "Run / Check changes");
+                if (check_run_head) {
+                  break;
+                }
+                retryCount++;
+                if (retryCount < 3) {
+                  await new Promise(resolve => setTimeout(resolve, 3000));
+                }
+              }
+              if (!check_run_head) {
+                throw new Error('Failed to retrieve check_run_head after 3 attempts');
+              }
 
               if (check_run_head.head_sha != context.payload.pull_request.head.sha) {
                 throw new Error('There was a new unsynced commit pushed. Please retrigger the workflow.');
@@ -126,11 +140,13 @@ jobs:
                 + context.payload.pull_request.head.repo.full_name
                 + '/runs/'
                 + check_run_head.id
+              console.log('Check run URL: ' + check_run_url)
 
               const actions_url = 'https://github.com/'
                 + context.payload.pull_request.head.repo.full_name
                 + '/actions/runs/'
                 + run_id
+              console.log('Actions URL: ' + actions_url)
 
               github.rest.checks.create({
                 owner: context.repo.owner,

.github/workflows/notify_test_workflow.yml

@@ -45,7 +45,7 @@ jobs:
       with:
         ref: ${{ matrix.branch }}
     - name: Cache Maven local repository
-      uses: actions/cache@v3
+      uses: actions/cache@v4
       with:
         path: ~/.m2/repository
         key: snapshot-maven-${{ hashFiles('**/pom.xml') }}

.github/workflows/publish_snapshot.yml

@@ -1091,6 +1091,34 @@ setMethod("dayofyear",
             column(jc)
           })
 
+#' @details
+#' \code{monthname}: Extracts the three-letter abbreviated month name from a
+#' given date/timestamp/string.
+#'
+#' @rdname column_datetime_functions
+#' @aliases monthname monthname,Column-method
+#' @note monthname since 4.0.0
+setMethod("monthname",
+          signature(x = "Column"),
+          function(x) {
+            jc <- callJStatic("org.apache.spark.sql.functions", "monthname", x@jc)
+            column(jc)
+          })
+
+#' @details
+#' \code{dayname}: Extracts the three-letter abbreviated day name from a
+#' given date/timestamp/string.
+#'
+#' @rdname column_datetime_functions
+#' @aliases dayname dayname,Column-method
+#' @note dayname since 4.0.0
+setMethod("dayname",
+          signature(x = "Column"),
+          function(x) {
+            jc <- callJStatic("org.apache.spark.sql.functions", "dayname", x@jc)
+            column(jc)
+          })
+
 #' @details
 #' \code{decode}: Computes the first argument into a string from a binary using the provided
 #' character set.

R/pkg/R/functions.R

@@ -1020,6 +1020,14 @@ setGeneric("dayofweek", function(x) { standardGeneric("dayofweek") })
 #' @name NULL
 setGeneric("dayofyear", function(x) { standardGeneric("dayofyear") })
 
+#' @rdname column_datetime_functions
+#' @name NULL
+setGeneric("monthname", function(x) { standardGeneric("monthname") })
+
+#' @rdname column_datetime_functions
+#' @name NULL
+setGeneric("dayname", function(x) { standardGeneric("dayname") })
+
 #' @rdname column_string_functions
 #' @name NULL
 setGeneric("decode", function(x, charset) { standardGeneric("decode") })

R/pkg/R/generics.R

@@ -1414,7 +1414,7 @@ test_that("test HiveContext", {
 
     # Invalid mode
     expect_error(saveAsTable(df, "parquetest", "parquet", mode = "abc", path = parquetDataPath),
-                 "illegal argument - Unknown save mode: abc")
+                 "Error in mode : analysis error - \\[INVALID_SAVE_MODE\\].*")
     unsetHiveContext()
   }
 })
@@ -2062,6 +2062,8 @@ test_that("date functions on a DataFrame", {
   expect_equal(collect(select(df, weekofyear(df$b)))[, 1], c(50, 50, 51))
   expect_equal(collect(select(df, year(df$b)))[, 1], c(2012, 2013, 2014))
   expect_equal(collect(select(df, month(df$b)))[, 1], c(12, 12, 12))
+  expect_equal(collect(select(df, monthname(df$b)))[, 1], c("Dec", "Dec", "Dec"))
+  expect_equal(collect(select(df, dayname(df$b)))[, 1], c("Thu", "Sat", "Mon"))
   expect_equal(collect(select(df, last_day(df$b)))[, 1],
                c(as.Date("2012-12-31"), as.Date("2013-12-31"), as.Date("2014-12-31")))
   expect_equal(collect(select(df, next_day(df$b, "MONDAY")))[, 1],

R/pkg/tests/fulltests/test_sparkSQL.R

@@ -149,6 +149,12 @@
           <groupId>org.apache.spark</groupId>
           <artifactId>spark-connect_${scala.binary.version}</artifactId>
           <version>${project.version}</version>
+          <exclusions>
+            <exclusion>
+              <groupId>org.apache.spark</groupId>
+              <artifactId>spark-connect-common_${scala.binary.version}</artifactId>
+            </exclusion>
+          </exclusions>
         </dependency>
         <dependency>
           <groupId>org.apache.spark</groupId>
@@ -204,6 +210,16 @@
         </dependency>
       </dependencies>
     </profile>
+    <profile>
+      <id>jvm-profiler</id>
+      <dependencies>
+        <dependency>
+          <groupId>org.apache.spark</groupId>
+          <artifactId>spark-profiler_${scala.binary.version}</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
     <profile>
       <id>bigtop-dist</id>
       <!-- This profile uses the assembly plugin to create a special "dist" package for BigTop

assembly/pom.xml

@@ -48,6 +48,33 @@ export PYSPARK_PYTHON
 export PYSPARK_DRIVER_PYTHON
 export PYSPARK_DRIVER_PYTHON_OPTS
 
+# Attempt to find JAVA_HOME.
+# If JAVA_HOME not set, install JDK 17 and set JAVA_HOME using a temp dir, and adding the
+# temp dir to the PYTHONPATH.
+if [ -n "${JAVA_HOME}" ]; then
+  RUNNER="${JAVA_HOME}/bin/java"
+else
+  if [ "$(command -v java)" ]; then
+    RUNNER="java"
+  else
+    echo -n "JAVA_HOME is not set. Would you like to install JDK 17 and set JAVA_HOME? (Y/N) " >&2
+
+    read -r input
+
+    if [[ "${input,,}" == "y" ]]; then
+        TEMP_DIR=$(mktemp -d)
+        $PYSPARK_DRIVER_PYTHON -m pip install --target="$TEMP_DIR" install-jdk
+        export JAVA_HOME=$(PYTHONPATH="$TEMP_DIR" $PYSPARK_DRIVER_PYTHON -c 'import jdk; print(jdk.install("17"))')
+        RUNNER="${JAVA_HOME}/bin/java"
+        echo "JDK was installed to the path \"$JAVA_HOME\""
+        echo "You can avoid needing to re-install JDK by setting your JAVA_HOME environment variable to \"$JAVA_HOME\""
+    else
+        echo "JDK installation skipped. You can manually install JDK (17 or later) and set JAVA_HOME in your environment."
+        exit 1
+    fi
+  fi
+fi
+
 # Add the PySpark classes to the Python path:
 export PYTHONPATH="${SPARK_HOME}/python/:$PYTHONPATH"
 export PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.9.7-src.zip:$PYTHONPATH"

bin/pyspark

@@ -54,4 +54,9 @@ public int hashCode() {
         result = 31 * result + minor;
         return result;
     }
+
+    @Override
+    public String toString() {
+      return "StoreVersion[" + major + "." + minor + ']';
+    }
 }

common/network-common/src/main/java/org/apache/spark/network/shuffledb/StoreVersion.java

@@ -61,7 +61,6 @@ public final class ReloadingX509TrustManager
   protected volatile int needsReloadCheckCounts;
   private final AtomicReference<X509TrustManager> trustManagerRef;
 
-  private volatile boolean running;
   private Thread reloader;
 
   /**
@@ -98,15 +97,13 @@ public ReloadingX509TrustManager(
   public void init() {
     reloader = new Thread(this, "Truststore reloader thread");
     reloader.setDaemon(true);
-    running = true;
     reloader.start();
   }
 
   /**
    * Stops the reloader thread.
    */
   public void destroy() throws InterruptedException {
-    running = false;
     reloader.interrupt();
     reloader.join();
   }
@@ -200,11 +197,12 @@ X509TrustManager loadTrustManager()
 
   @Override
   public void run() {
+    boolean running = true;
     while (running) {
       try {
         Thread.sleep(reloadInterval);
       } catch (InterruptedException e) {
-        //NOP
+        running = false;
       }
       try {
         if (running && needsReload()) {

common/network-common/src/main/java/org/apache/spark/network/ssl/ReloadingX509TrustManager.java

@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.network.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.SystemUtils;
+import org.apache.spark.network.shuffledb.DBBackend;
+import org.apache.spark.network.shuffledb.StoreVersion;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.jupiter.api.Assumptions.assumeFalse;
+
+public class DBProviderSuite {
+
+  @Test
+  public void testRockDBCheckVersionFailed() throws IOException {
+    testCheckVersionFailed(DBBackend.ROCKSDB, "rocksdb");
+  }
+
+  @Test
+  public void testLevelDBCheckVersionFailed() throws IOException {
+    assumeFalse(SystemUtils.IS_OS_MAC_OSX && SystemUtils.OS_ARCH.equals("aarch64"));
+    testCheckVersionFailed(DBBackend.LEVELDB, "leveldb");
+  }
+
+  private void testCheckVersionFailed(DBBackend dbBackend, String namePrefix) throws IOException {
+    String root = System.getProperty("java.io.tmpdir");
+    File dbFile = JavaUtils.createDirectory(root, namePrefix);
+    try {
+      StoreVersion v1 = new StoreVersion(1, 0);
+      ObjectMapper mapper = new ObjectMapper();
+      DBProvider.initDB(dbBackend, dbFile, v1, mapper).close();
+      StoreVersion v2 = new StoreVersion(2, 0);
+      IOException ioe = Assertions.assertThrows(IOException.class, () ->
+        DBProvider.initDB(dbBackend, dbFile, v2, mapper));
+      Assertions.assertTrue(
+        ioe.getMessage().contains("incompatible with current version StoreVersion[2.0]"));
+    } finally {
+      JavaUtils.deleteRecursively(dbFile);
+    }
+  }
+}