[AURON #2017] [BUILD] Add Spark 4.x support to dev/reformat script.

Signed-off-by: slfan1989 <slfan1989@apache.org>

Author: slfan1989

.github/workflows/style.yml

@@ -46,5 +46,16 @@ jobs:
           java-version: 8
           cache: 'maven'
           check-latest: false
+      - name: Capture JDK 8 home
+        run: echo "JAVA_HOME_8_X64=$JAVA_HOME" >> "$GITHUB_ENV"
+      - name: Setup JDK 17
+        uses: actions/setup-java@v5
+        with:
+          distribution: 'adopt-hotspot'
+          java-version: 17
+          cache: 'maven'
+          check-latest: false
+      - name: Capture JDK 17 home
+        run: echo "JAVA_HOME_17_X64=$JAVA_HOME" >> "$GITHUB_ENV"
       - run: |
           ./dev/reformat --check

dev/reformat

@@ -32,6 +32,37 @@ done
 
 MODE=pre
 SCALA_PROFILE=scala-2.12
+ORIGINAL_PATH="${PATH}"
+function set_java_home() {
+  local version="$1"
+  local mac_version="$version"
+  local candidate=""
+  if [[ "${version}" == "8" ]]; then
+    mac_version="1.8"
+  fi
+
+  if [[ "$(uname)" == "Darwin" ]] && [[ -x /usr/libexec/java_home ]]; then
+    candidate=$(/usr/libexec/java_home -v "${mac_version}" 2>/dev/null || true)
+  else
+    local java_home_version_x64_var="JAVA_HOME_${version}_X64"
+    local java_home_version_var="JAVA_HOME_${version}"
+
+    if [[ -n "${!java_home_version_x64_var:-}" ]]; then
+      candidate="${!java_home_version_x64_var}"
+    elif [[ -n "${!java_home_version_var:-}" ]]; then
+      candidate="${!java_home_version_var}"
+    fi
+  fi
+
+  if [[ -n "${candidate}" ]]; then
+    export JAVA_HOME="${candidate}"
+    export PATH="${candidate}/bin:${ORIGINAL_PATH}"
+  else
+    echo "JAVA_HOME for JDK ${version} not found."
+    exit 1
+  fi
+}
+
 function run_maven() {
   if [[ "$CHECK" == "true" ]]; then
     "${PROJECT_DIR}"/build/mvn spotless:check compile test-compile scalafix:scalafix -Dscalafix.mode=CHECK -Dscalafix.skipTest=true -DskipBuildNative -P"${MODE}" -P"${SCALA_PROFILE}" "$@"
@@ -52,12 +83,23 @@ fi
 sparkver=spark-3.5
 for celebornver in celeborn-0.5 celeborn-0.6
 do
-  run_maven -P"${sparkver}" -Pceleborn,"${celebornver}" -Puniffle,uniffle-0.10 -Ppaimon,paimon-1.2 -Pflink,flink-1.18 -Piceberg,iceberg-1.9
+  run_maven -P"${sparkver}" -Pceleborn,"${celebornver}" -Puniffle,uniffle-0.10 -Ppaimon,paimon-1.2 -Pflink-1.18 -Piceberg-1.9
 
 done
 
-sparkvers=(spark-3.0 spark-3.1 spark-3.2 spark-3.3 spark-3.4)
+sparkvers=(spark-3.0 spark-3.1 spark-3.2 spark-3.3 spark-3.4 spark-4.0 spark-4.1)
 for sparkver in "${sparkvers[@]}"
 do
+  if [[ $sparkver == spark-4.* ]]; then
+    SCALA_PROFILE=scala-2.13
+  else
+    SCALA_PROFILE=scala-2.12
+  fi
+  if [[ $sparkver == spark-4.* ]]; then
+    # Use JDK 17 for Spark 4.x to match the Spark 4.x requirement
+    set_java_home 17
+  else
+    set_java_home 8
+  fi
   run_maven -P"${sparkver}"
 done

spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronConverters.scala

@@ -424,7 +424,7 @@ object AuronConverters extends Logging {
           assert(
             !exec.requiredSchema.exists(e => existTimestampType(e.dataType)),
             s"Parquet scan with timestamp type is not supported for table: ${tableIdentifier
-              .getOrElse("unknown")}. " +
+                .getOrElse("unknown")}. " +
               "Set spark.auron.enable.scan.parquet.timestamp=true to enable timestamp support " +
               "or remove timestamp columns from the query.")
         }
@@ -435,15 +435,15 @@ object AuronConverters extends Logging {
           assert(
             !exec.requiredSchema.exists(e => existTimestampType(e.dataType)),
             s"ORC scan with timestamp type is not supported for tableIdentifier: ${tableIdentifier
-              .getOrElse("unknown")}. " +
+                .getOrElse("unknown")}. " +
               "Set spark.auron.enable.scan.orc.timestamp=true to enable timestamp support " +
               "or remove timestamp columns from the query.")
         }
         addRenameColumnsExec(Shims.get.createNativeOrcScanExec(exec))
       case p =>
         throw new NotImplementedError(
           s"Cannot convert FileSourceScanExec tableIdentifier: ${tableIdentifier.getOrElse(
-            "unknown")}, class: ${p.getClass.getName}")
+              "unknown")}, class: ${p.getClass.getName}")
     }
   }
 

spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala

@@ -91,7 +91,7 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
         dumpSimpleSparkPlanTreeNode(sparkPlanTransformed)
 
         logInfo(s"Transformed spark plan after preColumnarTransitions:\n${sparkPlanTransformed
-          .treeString(verbose = true, addSuffix = true)}")
+            .treeString(verbose = true, addSuffix = true)}")
 
         // post-transform
         Shims.get.postTransform(sparkPlanTransformed, sparkSession.sparkContext)

spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeHelper.scala

@@ -74,7 +74,7 @@ object NativeHelper extends Logging {
     val heapMemory = Runtime.getRuntime.maxMemory()
     val offheapMemory = totalMemory - heapMemory
     logWarning(s"memory total: ${Utils.bytesToString(totalMemory)}, onheap: ${Utils.bytesToString(
-      heapMemory)}, offheap: ${Utils.bytesToString(offheapMemory)}")
+        heapMemory)}, offheap: ${Utils.bytesToString(offheapMemory)}")
     offheapMemory
   }
 

spark-extension/src/main/scala/org/apache/spark/sql/auron/util/TaskContextHelper.scala

@@ -47,7 +47,7 @@ object TaskContextHelper extends Logging {
     val thread = Thread.currentThread()
     val threadName = if (context != null) {
       s"auron native task ${context.partitionId()}.${context.attemptNumber()} in stage ${context
-        .stageId()}.${context.stageAttemptNumber()} (TID ${context.taskAttemptId()})"
+          .stageId()}.${context.stageAttemptNumber()} (TID ${context.taskAttemptId()})"
     } else {
       "auron native task " + thread.getName
     }

spark-extension/src/main/scala/org/apache/spark/sql/execution/auron/plan/NativeParquetInsertIntoHiveTableBase.scala

@@ -69,9 +69,10 @@ abstract class NativeParquetInsertIntoHiveTableBase(
         .filterKeys(Set("stage_id", "output_rows", "elapsed_compute"))
         .toSeq
         :+ ("io_time", SQLMetrics.createNanoTimingMetric(sparkContext, "Native.io_time"))
-        :+ ("bytes_written",
-        SQLMetrics
-          .createSizeMetric(sparkContext, "Native.bytes_written")): _*)
+        :+ (
+          "bytes_written",
+          SQLMetrics
+            .createSizeMetric(sparkContext, "Native.bytes_written")): _*)
 
   def check(): Unit = {
     val hadoopConf = sparkContext.hadoopConfiguration

thirdparty/auron-uniffle/src/main/scala/org/apache/spark/sql/execution/auron/shuffle/uniffle/AuronUniffleShuffleReader.scala

@@ -166,7 +166,7 @@ class AuronUniffleShuffleReader[K, C](
       }
       if (!emptyPartitionIds.isEmpty) {
         logDebug(s"Found ${emptyPartitionIds
-          .size()} empty shuffle partitions: ${emptyPartitionIds.asScala.mkString(",")}")
+            .size()} empty shuffle partitions: ${emptyPartitionIds.asScala.mkString(",")}")
       }
       iterators = shuffleDataIterList.iterator()
       if (iterators.hasNext) {