Add procedure declaration checker since it's fully deprecated in Scal…

…a 2.13 and will cause compilation to fail

Signed-off-by: Navin Kumar <[email protected]>

pom.xml

@@ -559,7 +559,7 @@
         <profile>
             <id>scala-2.12</id>
             <properties>
-                <scala.version>2.12.15</scala.version>
+                <scala.version>2.13.8</scala.version>
             </properties>
             <build>
                 <pluginManagement>
@@ -572,7 +572,7 @@
             <id>scala-2.13</id>
             <properties>
                 <scala.version>2.13.8</scala.version>
-                <scala.binary.version>2.13</scala.binary.version>
+                <scala.binary.version>2.12</scala.binary.version>
             </properties>
             <dependencyManagement>
                 <dependencies>

scalastyle-config.xml

@@ -129,6 +129,10 @@ You can also disable only one rule, by specifying its rule id, as specified in:
         <customMessage>procedure syntax is deprecated for constructors in Scala 2.13: add `=`, as in method definition</customMessage>
     </check>
 
+    <check level="error" class="org.scalastyle.scalariform.ProcedureDeclarationChecker" enabled="true">
+        <customMessage>procedure syntax is deprecated in Scala 2.13: add return type `: Unit` and `=`</customMessage>
+    </check>
+
     <check level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
         <parameters>
             <parameter name="regex">ArrayBuilder.make\[(.+)\]\(\)</parameter>

sql-plugin/src/main/scala/com/nvidia/spark/rapids/Arm.scala

@@ -15,7 +15,7 @@
  */
 package com.nvidia.spark.rapids
 
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable.{ArrayBuffer, ListBuffer}
 import scala.util.control.ControlThrowable
 
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
@@ -68,6 +68,15 @@ object Arm {
     }
   }
 
+  /** Executes the provided code block and then closes the list buffer of resources */
+  def withResource[T <: AutoCloseable, V](r: ListBuffer[T])(block: ListBuffer[T] => V): V = {
+    try {
+      block(r)
+    } finally {
+      r.safeClose()
+    }
+  }
+
   /** Executes the provided code block and then closes the value if it is AutoCloseable */
   def withResourceIfAllowed[T, V](r: T)(block: T => V): V = {
     try {

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuShuffledHashJoinExec.scala

@@ -357,7 +357,7 @@ object GpuShuffledHashJoinExec extends Logging {
         // The size still overflows after filtering or sub-partitioning is enabled for test.
         logDebug("Return multiple batches as the build side data for the following " +
           "sub-partitioning join in null-filtering mode.")
-        val safeIter = GpuSubPartitionHashJoin.safeIteratorFromSeq(spillBuf).map { sp =>
+        val safeIter = GpuSubPartitionHashJoin.safeIteratorFromSeq(spillBuf.toSeq).map { sp =>
           withResource(sp)(_.getColumnarBatch())
         } ++ filteredIter
         Right(new CollectTimeIterator("hash join build", safeIter, buildTime))
@@ -369,7 +369,7 @@ object GpuShuffledHashJoinExec extends Logging {
           spillBuf.append(
             SpillableColumnarBatch(filteredIter.next(), SpillPriorities.ACTIVE_BATCHING_PRIORITY))
         }
-        val spill = GpuSubPartitionHashJoin.concatSpillBatchesAndClose(spillBuf)
+        val spill = GpuSubPartitionHashJoin.concatSpillBatchesAndClose(spillBuf.toSeq)
         // There is a prior empty check so this `spill` can not be a None.
         assert(spill.isDefined, "The build data iterator should not be empty.")
         withResource(spill) { _ =>

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuSortExec.scala

@@ -23,7 +23,7 @@ import scala.collection.mutable.ArrayBuffer
 import ai.rapids.cudf.{ColumnVector, ContiguousTable, NvtxColor, NvtxRange, Table}
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
 import com.nvidia.spark.rapids.GpuMetric._
-import com.nvidia.spark.rapids.RapidsPluginImplicits.AutoCloseableProducingSeq
+import com.nvidia.spark.rapids.RapidsPluginImplicits.{ArrayBufferAsStack, AutoCloseableProducingSeq}
 import com.nvidia.spark.rapids.RmmRapidsRetryIterator.{splitSpillableInHalfByRows, withRetry, withRetryNoSplit}
 import com.nvidia.spark.rapids.ScalableTaskCompletion.onTaskCompletion
 import com.nvidia.spark.rapids.shims.ShimUnaryExecNode
@@ -377,7 +377,7 @@ case class GpuOutOfCoreSortIterator(
         }
       }
     }
-    (sortedCb, pendingObs)
+    (sortedCb, pendingObs.toSeq)
   }
 
   /** Save the splitting result returned from `splitAfterSort` into the cache */

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuTextBasedPartitionReader.scala

@@ -361,7 +361,7 @@ abstract class GpuTextBasedPartitionReader[BUFF <: LineBufferer, FACT <: LineBuf
               }
               columns += castColumn
             }
-            new Table(columns: _*)
+            new Table(columns.toSeq: _*)
           }
         }
 
@@ -759,7 +759,7 @@ object GpuTextBasedDateUtils {
       }
       formats += fmt
     }
-    formats
+    formats.toSeq
   }
 
 }

sql-plugin/src/main/scala/com/nvidia/spark/rapids/PlanUtils.scala

@@ -63,7 +63,7 @@ object PlanUtils {
         case other =>
           other.children.flatMap(p => recurse(p, predicate, accum)).headOption
       }
-      accum
+      accum.toSeq
     }
     recurse(exp, predicate, new ListBuffer[Expression]())
   }
@@ -86,7 +86,7 @@ object PlanUtils {
         case qs: ShuffleQueryStageExec => recurse(qs.shuffle, predicate, accum)
         case other => other.children.flatMap(p => recurse(p, predicate, accum)).headOption
       }
-      accum
+      accum.toSeq
     }
     recurse(plan, predicate, new ListBuffer[SparkPlan]())
   }

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RapidsConf.scala

@@ -2185,7 +2185,7 @@ class RapidsConf(conf: Map[String, String]) extends Logging {
   }
 
   lazy val rapidsConfMap: util.Map[String, String] = conf.filterKeys(
-    _.startsWith("spark.rapids.")).asJava
+    _.startsWith("spark.rapids.")).toMap.asJava
 
   lazy val metricsLevel: String = get(METRICS_LEVEL)
 

sql-plugin/src/main/scala/com/nvidia/spark/rapids/shuffle/RapidsShuffleClient.scala

@@ -345,7 +345,7 @@ class RapidsShuffleClient(
     }
 
     if (ptrs.nonEmpty) {
-      transport.queuePending(ptrs)
+      transport.queuePending(ptrs.toSeq)
     }
   }
 

sql-plugin/src/main/scala/com/nvidia/spark/rapids/shuffle/RapidsShuffleServer.scala

@@ -201,7 +201,7 @@ class RapidsShuffleServer(transport: RapidsShuffleTransport,
           }
         }
         if (bssToIssue.nonEmpty) {
-          doHandleTransferRequest(bssToIssue)
+          doHandleTransferRequest(bssToIssue.toSeq)
         }
       }
 
@@ -368,7 +368,7 @@ class RapidsShuffleServer(transport: RapidsShuffleTransport,
 
         // If we are still able to handle at least one `BufferSendState`, add any
         // others that also failed due back to the queue.
-        addToContinueQueue(toTryAgain)
+        addToContinueQueue(toTryAgain.toSeq)
       }
 
       serverStream.sync()

sql-plugin/src/main/scala/com/nvidia/spark/rapids/shuffle/WindowedBlockIterator.scala

@@ -154,7 +154,7 @@ class WindowedBlockIterator[T <: BlockWithSize](blocks: Seq[T], windowSize: Long
     }
     val lastBlock = blockRangesInWindow.last
     BlocksForWindow(lastBlockIndex,
-      blockRangesInWindow,
+      blockRangesInWindow.toSeq,
       !continue || !lastBlock.isComplete())
   }
 

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuFileFormatDataWriter.scala

@@ -562,7 +562,7 @@ class GpuDynamicPartitionDataSingleWriter(
           // clear the caches
           savedStatus.get.tableCaches.clear()
 
-          withRetryNoSplit(toConcat) { spillables: ListBuffer[SpillableColumnarBatch] =>
+          withRetryNoSplit(toConcat) { spillables =>
             withResource(spillables.toSeq.safeMap(_.getColumnarBatch())) { batches =>
               withResource(batches.map(GpuColumnVector.from)) { subTables =>
                 Table.concatenate(subTables: _*)
@@ -998,8 +998,7 @@ class GpuDynamicPartitionDataConcurrentWriter(
     // get concat table or the single table
     val spillableToWrite = if (status.tableCaches.length >= 2) {
       // concat the sub batches to write in once.
-      val concatted = withRetryNoSplit(status.tableCaches) { 
-          spillableSubBatches: ListBuffer[SpillableColumnarBatch] =>
+      val concatted = withRetryNoSplit(status.tableCaches) { spillableSubBatches =>
         withResource(spillableSubBatches.toSeq.safeMap(_.getColumnarBatch())) { subBatches =>
           withResource(subBatches.map(GpuColumnVector.from)) { subTables =>
             Table.concatenate(subTables: _*)

sql-plugin/src/main/spark320/scala/com/nvidia/spark/rapids/shims/Spark320PlusShims.scala

@@ -316,7 +316,7 @@ trait Spark320PlusShims extends SparkShims with RebaseShims with Logging {
         case c: CommandResultExec => recurse(c.commandPhysicalPlan, predicate, accum)
         case other => other.children.flatMap(p => recurse(p, predicate, accum)).headOption
       }
-      accum
+      accum.toSeq
     }
 
     recurse(plan, predicate, new ListBuffer[SparkPlan]())