chore: Fix Scala code warnings

.github/workflows/pr_build_linux.yml

@@ -166,11 +166,47 @@
         run: |
           ./dev/ci/check-working-tree-clean.sh
 
+  # Compiles main and test sources with -Pstrict-warnings, which promotes scalac warnings to
+  # errors, so a change that reintroduces a cleared warning (an Int widened into a Long metric,
+  # a discarded builder result) fails here rather than accumulating. Spark 3.5 only: the profile
+  # passes on the Scala 2.12 profiles, and the Scala 2.13 remainder is tracked in
+  # https://github.com/apache/datafusion-comet/issues/5893.
+  strict-scala-warnings:
+    needs: lint
+    name: Strict Scala warnings (Spark 3.5, JDK 17)
+    runs-on: ubuntu-24.04
+    container:
+      image: amd64/rust
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Setup Rust & Java toolchain
+        uses: ./.github/actions/setup-builder
+        with:
+          rust-version: ${{ env.RUST_VERSION }}
+          jdk-version: 17
+
+      - name: Cache Maven dependencies
+        uses: actions/cache@v6
+        with:
+          path: |
+            ~/.m2/repository
+            /root/.m2/repository
+          key: ${{ runner.os }}-java-maven-${{ hashFiles('**/pom.xml') }}-strict-warnings
+          restore-keys: |
+            ${{ runner.os }}-java-maven-
+
+      - name: Bootstrap Maven
+        uses: ./.github/actions/maven-bootstrap
+
+      - name: Compile with strict warnings
+        run: ./mvnw -B test-compile -Pspark-3.5 -Pstrict-warnings -DskipTests
+
   # Compile-only verification for Spark 4.1. Tests are intentionally skipped: the spark-4.1
   # profile is currently a build target only, and several runtime/test failures are tracked
   # in follow-up PRs. Excluded from lint-java because semanticdb-scalac_2.13.17 is not yet
   # published and the lint job activates -Psemanticdb.
   build-spark-4-1:
     needs: lint
     name: Build Spark 4.1, JDK 17
     runs-on: ubuntu-24.04

pom.xml

@@ -859,29 +859,84 @@ under the License.
         </pluginManagement>
       </build>
     </profile>
+    <!--
+      Compile Scala with warnings promoted to errors. Not active by default; run it
+      explicitly, e.g. `./mvnw test-compile -Pspark-3.5 -Pstrict-warnings`.
+
+      This passes on the Scala 2.12 profiles. The 2.13 profiles (spark-4.0 and later)
+      still report warnings that 2.12 does not raise at all, dominated by
+      `-Xlint:nonlocal-return` (a `return` inside a closure, which the compiler
+      implements by throwing) and non-exhaustive matches. Clearing those means
+      restructuring control flow rather than annotating it, so they are tracked in
+      https://github.com/apache/datafusion-comet/issues/5893 rather than silenced here.
+
+      `args` is configured per execution rather than on the plugin, because main and
+      test sources warrant different flags (see `-Ywarn-value-discard` below). An
+      execution's `args` replaces the plugin-level list instead of appending to it, so
+      each list below is self-contained.
+
+      Two lints are deliberately absent from both lists:
+
+      `-Ywarn-unused:params` reports parameters that are unused because a signature
+      requires them rather than because they are dead: the `@native` declarations in
+      `Native.scala`, which have no body to use them in; cross-version shims under
+      `src/main/spark-*` that match the signature of the Spark version they shim; and
+      overridable defaults and serde helpers whose parameter is part of their public
+      shape. Suppressing these one by one with `@nowarn` does not work across profiles:
+      some shared sources warn under Scala 2.12 but not 2.13 (`CometScanContrib.scala`,
+      for example), so an annotation that silences one profile is unused on the other,
+      which `-Xlint:_` reports via `-Xlint:nowarn` and `-Xfatal-warnings` turns into a
+      build failure.
+
+      `-Ywarn-value-discard` stays on for main sources, where a discarded result is
+      usually a dropped builder or a swallowed return, but is off for test sources,
+      where what it reports is the testing idiom itself: an `assert(...)` in trailing
+      position discards an `org.scalatest.Assertion`, and helpers such as
+      `checkSparkAnswerAndOperator` return plans that most callers ignore.
+    -->
     <profile>
-          <id>strict-warnings</id>
-          <build>
-              <plugins>
-                  <plugin>
-                      <groupId>net.alchim31.maven</groupId>
-                      <artifactId>scala-maven-plugin</artifactId>
-                      <configuration>
-                          <args>
-                              <arg>-deprecation</arg>
-                              <arg>-unchecked</arg>
-                              <arg>-feature</arg>
-                              <arg>-Xlint:_</arg>
-                              <arg>-Ywarn-dead-code</arg>
-                              <arg>-Ywarn-numeric-widen</arg>
-                              <arg>-Ywarn-value-discard</arg>
-                              <arg>-Ywarn-unused:imports,patvars,privates,locals,params,-implicits</arg>
-                              <arg>-Xfatal-warnings</arg>
-                          </args>
-                      </configuration>
-                  </plugin>
-              </plugins>
-          </build>
+      <id>strict-warnings</id>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>net.alchim31.maven</groupId>
+            <artifactId>scala-maven-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>scala-compile-first</id>
+                <configuration>
+                  <args>
+                    <arg>-deprecation</arg>
+                    <arg>-unchecked</arg>
+                    <arg>-feature</arg>
+                    <arg>-Xlint:_</arg>
+                    <arg>-Ywarn-dead-code</arg>
+                    <arg>-Ywarn-numeric-widen</arg>
+                    <arg>-Ywarn-value-discard</arg>
+                    <arg>-Ywarn-unused:imports,patvars,privates,locals,-implicits</arg>
+                    <arg>-Xfatal-warnings</arg>
+                  </args>
+                </configuration>
+              </execution>
+              <execution>
+                <id>scala-test-compile-first</id>
+                <configuration>
+                  <args>
+                    <arg>-deprecation</arg>
+                    <arg>-unchecked</arg>
+                    <arg>-feature</arg>
+                    <arg>-Xlint:_</arg>
+                    <arg>-Ywarn-dead-code</arg>
+                    <arg>-Ywarn-numeric-widen</arg>
+                    <arg>-Ywarn-unused:imports,patvars,privates,locals,-implicits</arg>
+                    <arg>-Xfatal-warnings</arg>
+                  </args>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
     </profile>
   </profiles>
 

spark/src/main/scala/org/apache/comet/CometExecIterator.scala

@@ -86,7 +86,7 @@ class CometExecIterator(
   private val nativeLib = new Native()
   private val nativeUtil = new NativeUtil()
   private val taskAttemptId = TaskContext.get().taskAttemptId()
-  private val taskCPUs = TaskContext.get().cpus()
+  private val taskCPUs = TaskContext.get().cpus().toLong
   private val cometTaskMemoryManager = new CometTaskMemoryManager(id, taskAttemptId)
 
   private val plan = {

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -436,21 +436,21 @@ object IcebergReflection extends Logging {
    * `taskGroups()`, so for staged scans we flatten the groups instead. Both methods are protected
    * and require reflection.
    */
-  def getTasks(scan: Any): Option[java.util.List[_]] =
+  def getTasks(scan: Any): Option[java.util.List[AnyRef]] =
     if (isStagedScan(scan)) tasksFromTaskGroups(scan) else tasksFromTasksAccessor(scan)
 
-  private def tasksFromTasksAccessor(scan: Any): Option[java.util.List[_]] =
+  private def tasksFromTasksAccessor(scan: Any): Option[java.util.List[AnyRef]] =
     findMethodInHierarchy(scan.getClass, "tasks") match {
       case Some(method) =>
-        Some(method.invoke(scan).asInstanceOf[java.util.List[_]])
+        Some(method.invoke(scan).asInstanceOf[java.util.List[AnyRef]])
       case None =>
         logError(
           "Iceberg reflection failure: Failed to get tasks from SparkScan: " +
             s"tasks() not found on ${scan.getClass.getName}")
         None
     }
 
-  private def tasksFromTaskGroups(scan: Any): Option[java.util.List[_]] =
+  private def tasksFromTaskGroups(scan: Any): Option[java.util.List[AnyRef]] =
     findMethodInHierarchy(scan.getClass, "taskGroups") match {
       case Some(method) =>
         try {
@@ -465,7 +465,7 @@ object IcebergReflection extends Logging {
             groups.forEach { group =>
               val groupTasks =
                 groupTasksMethod.invoke(group).asInstanceOf[java.util.Collection[_ <: AnyRef]]
-              flat.addAll(groupTasks)
+              val _ = flat.addAll(groupTasks)
             }
             Some(flat)
           }
@@ -1688,13 +1688,19 @@ object IcebergReflection extends Logging {
   private def newDataManifestFile(inputFile: AnyRef, specId: Int): AnyRef = {
     val inputFileClass = loadClass(ClassNames.INPUT_FILE)
     val cls = loadClass(ClassNames.GENERIC_MANIFEST_FILE)
-    val (ctor, args): (java.lang.reflect.Constructor[_], Array[Object]) =
+    // `Constructor[AnyRef]` rather than `Constructor[_]`: the two `try`/`catch` branches
+    // would otherwise infer a top-level existential, which `-Xlint:existential` rejects.
+    val (ctor, args): (java.lang.reflect.Constructor[AnyRef], Array[Object]) =
       try {
-        val c = cls.getDeclaredConstructor(inputFileClass, classOf[Int], classOf[Long])
+        val c = cls
+          .getDeclaredConstructor(inputFileClass, classOf[Int], classOf[Long])
+          .asInstanceOf[java.lang.reflect.Constructor[AnyRef]]
         (c, Array[Object](inputFile, Integer.valueOf(specId), java.lang.Long.valueOf(0L)))
       } catch {
         case _: NoSuchMethodException =>
-          val c = cls.getDeclaredConstructor(inputFileClass, classOf[Int])
+          val c = cls
+            .getDeclaredConstructor(inputFileClass, classOf[Int])
+            .asInstanceOf[java.lang.reflect.Constructor[AnyRef]]
           (c, Array[Object](inputFile, Integer.valueOf(specId)))
       }
     ctor.setAccessible(true)
@@ -1730,10 +1736,11 @@ object IcebergReflection extends Logging {
       }
       result
     } finally {
-      try reader.getClass.getMethod("close").invoke(reader)
-      catch {
-        case e: Exception => logWarning(s"Failed to close ManifestReader: ${e.getMessage}")
-      }
+      val _ =
+        try reader.getClass.getMethod("close").invoke(reader)
+        catch {
+          case e: Exception => logWarning(s"Failed to close ManifestReader: ${e.getMessage}")
+        }
     }
   }
 

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -911,7 +911,7 @@ case class CometExecRule(session: SparkSession)
       .flatten
       .toSet
     if (reasons.nonEmpty) {
-      withFallbackReasons(op, reasons)
+      val _ = withFallbackReasons(op, reasons)
     }
   }
 
@@ -943,7 +943,7 @@ case class CometExecRule(session: SparkSession)
             "operator or any of its expressions. Add a withFallbackReason call stating why " +
             s"conversion failed. Operator:\n$op")
       }
-      withFallbackReason(op, s"${op.nodeName} is not supported")
+      val _ = withFallbackReason(op, s"${op.nodeName} is not supported")
     }
   }
 
@@ -972,7 +972,8 @@ case class CometExecRule(session: SparkSession)
       CometExplainInfo.collectExprTagValues(allExprs, CometExplainInfo.CODEGEN_DISPATCH_EXPRS)
     appendTagValues(exec, CometExplainInfo.CODEGEN_DISPATCH_EXPRS, routedNames)
     if (routedNames.nonEmpty && CometConf.COMET_EXPLAIN_CODEGEN_ENABLED.get()) {
-      withInfo(exec, s"JVM codegen dispatcher: ${routedNames.toSeq.sorted.mkString(", ")}")
+      val _ =
+        withInfo(exec, s"JVM codegen dispatcher: ${routedNames.toSeq.sorted.mkString(", ")}")
     }
   }
 

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -31,7 +31,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.internal.Logging
 import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName, PlanExpression}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, DynamicPruningExpression, Expression, GenericInternalRow, InputFileBlockLength, InputFileBlockStart, InputFileName}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.util.{sideBySide, ArrayBasedMapData, GenericArrayData, MetadataColumnHelper}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getExistenceDefaultValues
@@ -1043,9 +1043,6 @@ case class CometScanRule(session: SparkSession)
     }
   }
 
-  private def isDynamicPruningFilter(e: Expression): Boolean =
-    e.exists(_.isInstanceOf[PlanExpression[_]])
-
   /**
    * Detects AQE DPP (SubqueryAdaptiveBroadcastExec), as opposed to non-AQE DPP.
    *

spark/src/main/scala/org/apache/comet/serde/CometScalaUDF.scala

@@ -181,7 +181,7 @@ object CometScalaUDF extends CometExpressionSerde[ScalaUDF] {
     target.foreach {
       case _: AttributeReference | _: Literal =>
       case node if !(node eq target) =>
-        withCodegenDispatchExpr(expr, CometExplainInfo.exprDisplayName(node))
+        val _ = withCodegenDispatchExpr(expr, CometExplainInfo.exprDisplayName(node))
       case _ =>
     }
     Some(

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -911,7 +911,7 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       e.getTagValue(CometExplainInfo.FALLBACK_REASONS).foreach(reasons ++= _)
     }
     if (reasons.nonEmpty) {
-      withFallbackReasons(to, reasons.toSet)
+      val _ = withFallbackReasons(to, reasons.toSet)
     }
   }
 

spark/src/main/scala/org/apache/comet/serde/literals.scala

@@ -86,8 +86,8 @@ object CometLiteral extends CometExpressionSerde[Literal] with CometTypeShim wit
       exprBuilder.setIsNull(false)
       dataType match {
         case _: BooleanType => exprBuilder.setBoolVal(value.asInstanceOf[Boolean])
-        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte])
-        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short])
+        case _: ByteType => exprBuilder.setByteVal(value.asInstanceOf[Byte].toInt)
+        case _: ShortType => exprBuilder.setShortVal(value.asInstanceOf[Short].toInt)
         case _: IntegerType | _: DateType => exprBuilder.setIntVal(value.asInstanceOf[Int])
         case _: LongType | _: TimestampType | _: TimestampNTZType | _: DayTimeIntervalType =>
           exprBuilder.setLongVal(value.asInstanceOf[Long])

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala

@@ -353,7 +353,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
           val equalityIds = equalityIdsMethod
             .invoke(deleteFile)
             .asInstanceOf[java.util.List[Integer]]
-          equalityIds.forEach(id => deleteBuilder.addEqualityIds(id))
+          deleteBuilder.addAllEqualityIds(equalityIds)
         } catch {
           case _: Exception =>
         }
@@ -515,7 +515,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                 commonBuilder.addPartitionTypePool(partitionTypeJson)
                 idx
               })
-            taskBuilder.setPartitionSpecIdx(specIdx)
+            val _ = taskBuilder.setPartitionSpecIdx(specIdx)
           } catch {
             case e: Exception =>
               logWarning(s"Failed to serialize partition spec to JSON: ${e.getMessage}")
@@ -575,7 +575,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
               commonBuilder.addPartitionDataPool(partitionDataProto)
               idx
             })
-          taskBuilder.setPartitionDataIdx(partitionDataIdx)
+          val _ = taskBuilder.setPartitionDataIdx(partitionDataIdx)
         } else {
           // Defensive: ContentScanTask.partition() returns an empty struct (never null) for
           // unpartitioned tables in practice. If it is ever null we cannot compute values, so

spark/src/main/scala/org/apache/comet/shuffle/CelebornShufflePusherFactory.scala

@@ -246,7 +246,7 @@ object CelebornShufflePusherFactory {
       client: AnyRef,
       taskContext: TaskContext,
       handle: ShuffleHandle): Unit = {
-    sparkUtilsClass
+    val _ = sparkUtilsClass
       .getMethod(
         "addFailureListenerIfBarrierTask",
         shuffleClientClass,
@@ -336,7 +336,7 @@ object CelebornShufflePusherFactory {
   /** Remove task-independent state for one unregistered Celeborn generation. */
   def cleanupShuffle(client: AnyRef, celebornShuffleId: Int): Unit = {
     try {
-      client.getClass
+      val _ = client.getClass
         .getMethod("cleanupShuffle", java.lang.Integer.TYPE)
         .invoke(client, Int.box(celebornShuffleId))
     } catch {
@@ -354,7 +354,8 @@ object CelebornShufflePusherFactory {
     try {
       val shuffleClientClass = ClassLoaders.loadClass(CELEBORN_SHUFFLE_CLIENT)
       try {
-        shuffleClientClass.getMethod("removeInstance", shuffleClientClass).invoke(null, client)
+        val _ =
+          shuffleClientClass.getMethod("removeInstance", shuffleClientClass).invoke(null, client)
       } catch {
         case _: NoSuchMethodException =>
         // Celeborn 0.6 owns one shared application client and cannot remove a single instance.

spark/src/main/scala/org/apache/comet/udf/codegen/CometScalaUDFCodegen.scala

@@ -296,7 +296,7 @@ object CometScalaUDFCodegen {
   private[codegen] def recordCompiledSignature(
       specs: IndexedSeq[ArrowColumnSpec],
       outputType: DataType): Unit = {
-    compiledSignatures.add((specs.map(_.vectorClass), outputType))
+    val _ = compiledSignatures.add((specs.map(_.vectorClass), outputType))
   }
 
   /**

spark/src/main/scala/org/apache/spark/CometSource.scala

@@ -54,9 +54,9 @@ object CometSource extends Source {
     })
 
   def recordStats(stats: CometCoverageStats): Unit = {
-    NATIVE_OPERATORS.inc(stats.cometOperators)
-    SPARK_OPERATORS.inc(stats.sparkOperators)
-    TRANSITIONS.inc(stats.transitions)
+    NATIVE_OPERATORS.inc(stats.cometOperators.toLong)
+    SPARK_OPERATORS.inc(stats.sparkOperators.toLong)
+    TRANSITIONS.inc(stats.transitions.toLong)
     QUERIES_PLANNED.inc()
   }
 }