chore: Fix Scala code warnings

.github/workflows/pr_build_linux.yml

@@ -240,6 +240,54 @@ jobs:
             /root/.m2/repository
           key: ${{ runner.os }}-java-maven-${{ hashFiles('**/pom.xml') }}-lint
 
+  # 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)
+    if: ${{ !inputs.cache-refresh-only && inputs.profiles != 'nightly' }}
+    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: Restore Maven dependencies
+        id: maven-cache
+        uses: actions/cache/restore@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: Compile with strict warnings
+        run: ./mvnw -B test-compile -Pspark-3.5 -Pstrict-warnings -DskipTests
+
+      # Saved only on main: an entry written from a pull request or a
+      # `gh-readonly-queue/*` branch cannot be restored by any later run, and
+      # it evicts main's from the shared budget. See "Large caches are written
+      # on main only" in README.md.
+      - name: Save Maven dependencies
+        if: ${{ github.ref == 'refs/heads/main' && steps.maven-cache.outputs.cache-hit != 'true' }}
+        uses: actions/cache/save@v6
+        with:
+          path: |
+            ~/.m2/repository
+            /root/.m2/repository
+          key: ${{ runner.os }}-java-maven-${{ hashFiles('**/pom.xml') }}-strict-warnings
+
   # 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

docs/source/contributor-guide/ci.md

@@ -410,7 +410,7 @@ actionlint --shellcheck=off
 ```
 
 A new test suite has to be registered in the workflow files by hand; see
-[Register New Test Suites in CI](development.md#5-register-new-test-suites-in-ci). A new job in
+[Register New Test Suites in CI](development.md#6-register-new-test-suites-in-ci). A new job in
 `ci.yml` needs an entry in both `FILTERS` and `POLICY` in `dev/ci/compute-changes.py`, a case in
 `dev/ci/check-ci-config.py`, and a line in `required_checks.needs`. The check will tell you which
 of those is missing.

docs/source/contributor-guide/development.md

@@ -613,7 +613,17 @@ cargo clippy --color=never --all-targets --workspace -- -D warnings
 
 Make sure to resolve any Clippy warnings before submitting your pull request, as the CI/CD pipeline will fail if warnings are present.
 
-### 4. Run Tests
+### 4. Compile With Strict Scala Warnings (Recommended)
+
+The `Strict Scala warnings` job runs on every pull request and in the merge queue. It compiles the main and test sources with scalac warnings promoted to errors, so anything it reports fails the build — an `Int` widened into a `Long` metric, or a discarded builder result, for example. Reproduce it locally with:
+
+```sh
+./mvnw test-compile -Pspark-3.5 -Pstrict-warnings -DskipTests
+```
+
+Use the Spark 3.5 profile: it is the one the job runs, and the default build profile will not reproduce it. The default is Spark 4.1 on Scala 2.13, where `-Pstrict-warnings` still fails on warnings unrelated to your change (tracked in [#5893](https://github.com/apache/datafusion-comet/issues/5893)), and where the compiler reports a different set — an adapted argument list, for instance, is flagged under Scala 2.12 but not under 2.13.
+
+### 5. Run Tests
 
 Run the relevant tests for your changes:
 
@@ -628,7 +638,7 @@ make test-rust
 make test-jvm
 ```
 
-### 5. Register New Test Suites in CI
+### 6. Register New Test Suites in CI
 
 Comet's CI does not automatically discover test suites. Instead, test suites are explicitly listed
 in the GitHub Actions workflow files so they can be grouped by category and run as separate parallel

pom.xml

@@ -856,29 +856,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 through `-Ywarn-unused: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

@@ -96,7 +96,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 = {
@@ -411,7 +411,7 @@ object CometExecIterator extends Logging {
       if (intervalMs > 0) {
         val nativeLib = new Native()
         val limitBytes = nativeMemoryLimit(conf)
-        Executors
+        val _ = Executors
           .newSingleThreadScheduledExecutor(new ThreadFactory {
             override def newThread(runnable: Runnable): Thread = {
               val thread = new Thread(runnable, "comet-memory-usage-log")

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

@@ -444,21 +444,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 {
@@ -473,7 +473,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)
           }
@@ -1705,13 +1705,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)
@@ -1747,10 +1753,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

@@ -992,7 +992,7 @@ case class CometExecRule(session: SparkSession)
       .flatten
       .toSet
     if (reasons.nonEmpty) {
-      withFallbackReasons(op, reasons)
+      val _ = withFallbackReasons(op, reasons)
     }
   }
 
@@ -1024,7 +1024,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")
     }
   }
 
@@ -1053,7 +1053,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
@@ -1001,9 +1001,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

@@ -979,7 +979,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

@@ -357,7 +357,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 =>
         }
@@ -519,7 +519,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}")
@@ -579,7 +579,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