Make MiMa happy with the -Xsource:3 -Xsource-features:v2.13.15,-case-companion-function options

Author: joan38

mllib/src/main/scala/org/apache/spark/mllib/classification/LogisticRegression.scala

@@ -221,7 +221,8 @@ class LogisticRegressionWithSGD private[mllib] (
     .setNumIterations(numIterations)
     .setRegParam(regParam)
     .setMiniBatchFraction(miniBatchFraction)
-  override protected val validators = List(DataValidators.binaryLabelValidator)
+  override protected val validators: List[RDD[LabeledPoint] => Boolean] =
+    List(DataValidators.binaryLabelValidator)
 
   override protected[mllib] def createModel(
     weights: Vector,

pom.xml

@@ -2736,7 +2736,8 @@
               <arg>-explaintypes</arg>
               <arg>-release</arg>
               <arg>17</arg>
-              <arg>-Xsource:3-cross</arg>
+              <arg>-Xsource:3</arg>
+              <arg>-Xsource-features:v2.13.15,-case-companion-function</arg>
               <arg>-Wconf:any:e</arg>
               <arg>-Wconf:cat=deprecation:wv</arg>
               <arg>-Wunused:imports</arg>

project/MimaExcludes.scala

@@ -40,7 +40,7 @@ object MimaExcludes {
   )
 
   // Exclude rules for 4.0.x from 3.5.0
-  lazy val v40excludes = defaultExcludes ++ Seq(
+  lazy val v40excludes = defaultExcludes ++ scala3Excludes ++ Seq(
     // [SPARK-44863][UI] Add a button to download thread dump as a txt in Spark UI
     ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ThreadStackTrace.*"),
     ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.ThreadStackTrace$"),
@@ -238,6 +238,37 @@ object MimaExcludes {
     loggingExcludes("org.apache.spark.sql.streaming.DataStreamReader") ++
     loggingExcludes("org.apache.spark.sql.SparkSession#Builder")
 
+  // Enable -Xsource:3 compiler flag
+  lazy val scala3Excludes = Seq(
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Metric.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Metric.tupled"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Metric.curried"),
+    ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.sql.Metric$"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.apply$default$3"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.tupled"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.curried"),
+    ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.sql.Node$"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.paths.SparkPath.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.paths.SparkPath.copy"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.paths.SparkPath.copy$default$1"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.paths.SparkPath.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.AccumulableInfo.apply$default$7"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.AccumulableInfo.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.AccumulableInfo.tupled"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.AccumulableInfo.curried"),
+    ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.AccumulableInfo$"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.apply$default$7"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.tupled"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo.curried"),
+    ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.ApplicationAttemptInfo$"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationInfo.apply"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationInfo.tupled"),
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ApplicationInfo.curried"),
+    ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.ApplicationInfo$"),
+  )
+
   // Default exclude rules
   lazy val defaultExcludes = Seq(
     // Spark Internals

project/SparkBuild.scala

@@ -229,7 +229,8 @@ object SparkBuild extends PomBuild {
   lazy val compilerWarningSettings: Seq[sbt.Def.Setting[_]] = Seq(
     (Compile / scalacOptions) ++= {
       Seq(
-        "-Xsource:3-cross",
+        "-Xsource:3",
+        "-Xsource-features:v2.13.15,-case-companion-function",
         // replace -Xfatal-warnings with fine-grained configuration, since 2.13.2
         // verbose warning on deprecation, error on all others
         // see `scalac -Wconf:help` for details

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotSuite.scala

@@ -38,14 +38,14 @@ class ExecutorPodsSnapshotSuite extends SparkFunSuite {
   test("States are interpreted correctly from pod metadata.") {
     ExecutorPodsSnapshot.setShouldCheckAllContainers(false)
     val testCases = Seq(
-      testCase(pendingExecutor(0), PodPending),
-      testCase(runningExecutor(1), PodRunning),
-      testCase(succeededExecutor(2), PodSucceeded),
-      testCase(failedExecutorWithoutDeletion(3), PodFailed),
-      testCase(deletedExecutor(4), PodDeleted),
-      testCase(unknownExecutor(5), PodUnknown),
-      testCase(finishedExecutorWithRunningSidecar(6, 0), PodSucceeded),
-      testCase(finishedExecutorWithRunningSidecar(7, 1), PodFailed)
+      testCase(pendingExecutor(0), PodPending(_)),
+      testCase(runningExecutor(1), PodRunning(_)),
+      testCase(succeededExecutor(2), PodSucceeded(_)),
+      testCase(failedExecutorWithoutDeletion(3), PodFailed(_)),
+      testCase(deletedExecutor(4), PodDeleted(_)),
+      testCase(unknownExecutor(5), PodUnknown(_)),
+      testCase(finishedExecutorWithRunningSidecar(6, 0), PodSucceeded(_)),
+      testCase(finishedExecutorWithRunningSidecar(7, 1), PodFailed(_))
     )
     doTest(testCases)
   }
@@ -54,13 +54,13 @@ class ExecutorPodsSnapshotSuite extends SparkFunSuite {
     + " when configured to check all containers.") {
     ExecutorPodsSnapshot.setShouldCheckAllContainers(true)
     val testCases = Seq(
-      testCase(pendingExecutor(0), PodPending),
-      testCase(runningExecutor(1), PodRunning),
-      testCase(runningExecutorWithFailedContainer(2), PodFailed),
-      testCase(succeededExecutor(3), PodSucceeded),
-      testCase(failedExecutorWithoutDeletion(4), PodFailed),
-      testCase(deletedExecutor(5), PodDeleted),
-      testCase(unknownExecutor(6), PodUnknown)
+      testCase(pendingExecutor(0), PodPending(_)),
+      testCase(runningExecutor(1), PodRunning(_)),
+      testCase(runningExecutorWithFailedContainer(2), PodFailed(_)),
+      testCase(succeededExecutor(3), PodSucceeded(_)),
+      testCase(failedExecutorWithoutDeletion(4), PodFailed(_)),
+      testCase(deletedExecutor(5), PodDeleted(_)),
+      testCase(unknownExecutor(6), PodUnknown(_))
     )
     doTest(testCases)
   }

sql/api/src/main/scala/org/apache/spark/sql/MergeIntoWriter.scala

@@ -250,9 +250,7 @@ case class WhenMatched[T] private[sql] (
  * @tparam T
  *   The type of data in the MergeIntoWriter.
  */
-case class WhenNotMatched[T](
-    mergeIntoWriter: MergeIntoWriter[T],
-    condition: Option[Column]) {
+case class WhenNotMatched[T](mergeIntoWriter: MergeIntoWriter[T], condition: Option[Column]) {
 
   /**
    * Specifies an action to insert all non-matched rows into the DataFrame.

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/arrow/ArrowEncoderSuite.scala

@@ -816,7 +816,7 @@ class ArrowEncoderSuite extends ConnectFunSuite with BeforeAndAfterAll {
   test("REPL generated classes") {
     val encoder = ScalaReflection.encoderFor[MyTestClass]
     roundTripAndCheckIdentical(encoder) { () =>
-      Iterator.tabulate(10)(MyTestClass)
+      Iterator.tabulate(10)(MyTestClass(_))
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala

@@ -134,7 +134,7 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
     val numBuckets = bucketSpec.numBuckets
 
     val normalizedFiltersAndExpr = normalizedFilters
-      .reduce(expressions.And)
+      .reduce(expressions.And(_, _))
     val matchedBuckets = getExpressionBuckets(normalizedFiltersAndExpr, bucketColumnName,
       numBuckets)
 
@@ -353,7 +353,8 @@ object FileSourceStrategy extends Strategy with PredicateHelper with Logging {
       }.getOrElse(scan)
 
       // bottom-most filters are put in the left of the list.
-      val finalFilters = afterScanFilters.toSeq.reduceOption(expressions.And).toSeq ++ stayUpFilters
+      val finalFilters =
+        afterScanFilters.toSeq.reduceOption(expressions.And(_, _)).toSeq ++ stayUpFilters
       val withFilter = finalFilters.foldLeft(withMetadataProjections)((plan, cond) => {
         execution.FilterExec(cond, plan)
       })

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala

@@ -184,7 +184,7 @@ abstract class PartitioningAwareFileIndex(
     }
 
     if (partitionPruningPredicates.nonEmpty) {
-      val predicate = partitionPruningPredicates.reduce(expressions.And)
+      val predicate = partitionPruningPredicates.reduce(expressions.And(_, _))
 
       val boundPredicate = Predicate.createInterpreted(predicate.transform {
         case a: AttributeReference =>

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/sources/TextSocketMicroBatchStream.scala

@@ -128,7 +128,7 @@ class TextSocketMicroBatchStream(host: String, port: Int, numPartitions: Int)
       slices(idx % numPartitions).append(r)
     }
 
-    slices.map(TextSocketInputPartition)
+    slices.map(TextSocketInputPartition(_))
   }
 
   override def createReaderFactory(): PartitionReaderFactory =