Fix compiler warning

Author: szehon-ho

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleFunction.java

@@ -37,5 +37,6 @@ public interface ReducibleFunction<T, A> extends ScalarFunction<T> {
      * @param otherArgument argument for other function instance
      * @return a reduction function if it is reducible, none if not
      */
-    Option<Reducer<A>> reducer(ReducibleFunction<?, ?> other, Option<?> thisArgument, Option<?> otherArgument);
+    Option<Reducer<A>> reducer(ReducibleFunction<?, ?> other, Option<?> thisArgument,
+                               Option<?> otherArgument);
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala

@@ -69,8 +69,7 @@ case class TransformExpression(
       true
     } else {
       (function, other.function) match {
-        case (f: ReducibleFunction[Any, Any] @unchecked,
-          o: ReducibleFunction[Any, Any] @unchecked) =>
+        case (f: ReducibleFunction[_, _], o: ReducibleFunction[_, _]) =>
           val reducer = f.reducer(o, numBucketsOpt, other.numBucketsOpt)
           val otherReducer = o.reducer(f, other.numBucketsOpt, numBucketsOpt)
           reducer.isDefined || otherReducer.isDefined

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala

@@ -651,7 +651,7 @@ trait ShuffleSpec {
    * Returning none also indicates that none of the partition expressions can be reduced on the
    * corresponding expression on the other shuffle spec.
    */
-  def reducers(spec: ShuffleSpec): Option[Seq[Option[Reducer[Any]]]] = None
+  def reducers(spec: ShuffleSpec): Option[Seq[Option[Reducer[_]]]] = None
 }
 
 case object SinglePartitionShuffleSpec extends ShuffleSpec {
@@ -854,17 +854,16 @@ case class KeyGroupedShuffleSpec(
     KeyGroupedPartitioning(clustering, partitioning.numPartitions, partitioning.partitionValues)
   }
 
-  override def reducers(other: ShuffleSpec): Option[Seq[Option[Reducer[Any]]]] = {
+  override def reducers(other: ShuffleSpec): Option[Seq[Option[Reducer[_]]]] = {
     other match {
       case otherSpec: KeyGroupedShuffleSpec =>
         val results = partitioning.expressions.zip(otherSpec.partitioning.expressions).map {
           case (e1: TransformExpression, e2: TransformExpression)
-            if e1.function.isInstanceOf[ReducibleFunction[Any, Any]@unchecked]
-              && e2.function.isInstanceOf[ReducibleFunction[Any, Any]@unchecked] =>
-            e1.function.asInstanceOf[ReducibleFunction[Any, Any]].reducer(
-              e2.function.asInstanceOf[ReducibleFunction[Any, Any]],
-              e1.numBucketsOpt.map(a => a.asInstanceOf[Any]),
-              e2.numBucketsOpt.map(a => a.asInstanceOf[Any]))
+            if e1.function.isInstanceOf[ReducibleFunction[_, _]]
+              && e2.function.isInstanceOf[ReducibleFunction[_, _]] =>
+            e1.function.asInstanceOf[ReducibleFunction[_, _]].reducer(
+              e2.function.asInstanceOf[ReducibleFunction[_, _]],
+              e1.numBucketsOpt, e2.numBucketsOpt)
           case (_, _) => None
         }
 
@@ -892,11 +891,11 @@ case class KeyGroupedShuffleSpec(
 object KeyGroupedShuffleSpec {
   def reducePartitionValue(row: InternalRow,
                            expressions: Seq[Expression],
-                           reducers: Seq[Option[Reducer[Any]]]):
+                           reducers: Seq[Option[Reducer[_]]]):
     InternalRowComparableWrapper = {
     val partitionVals = row.toSeq(expressions.map(_.dataType))
     val reducedRow = partitionVals.zip(reducers).map{
-      case (v, Some(reducer)) => reducer.reduce(v)
+      case (v, Some(reducer: Reducer[Any])) => reducer.reduce(v)
       case (v, _) => v
     }.toArray
     InternalRowComparableWrapper(new GenericInternalRow(reducedRow), expressions)

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

@@ -272,7 +272,7 @@ case class StoragePartitionJoinParams(
     keyGroupedPartitioning: Option[Seq[Expression]] = None,
     joinKeyPositions: Option[Seq[Int]] = None,
     commonPartitionValues: Option[Seq[(InternalRow, Int)]] = None,
-    reducers: Option[Seq[Option[Reducer[Any]]]] = None,
+    reducers: Option[Seq[Option[Reducer[_]]]] = None,
     applyPartialClustering: Boolean = false,
     replicatePartitions: Boolean = false) {
   override def equals(other: Any): Boolean = other match {

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

@@ -550,7 +550,7 @@ case class EnsureRequirements(
       plan: SparkPlan,
       values: Seq[(InternalRow, Int)],
       joinKeyPositions: Option[Seq[Int]],
-      reducers: Option[Seq[Option[Reducer[Any]]]],
+      reducers: Option[Seq[Option[Reducer[_]]]],
       applyPartialClustering: Boolean,
       replicatePartitions: Boolean): SparkPlan = plan match {
     case scan: BatchScanExec =>
@@ -570,7 +570,7 @@ case class EnsureRequirements(
 
   private def reduceCommonPartValues(commonPartValues: Seq[(InternalRow, Int)],
                                      expressions: Seq[Expression],
-                                     reducers: Option[Seq[Option[Reducer[Any]]]]) = {
+                                     reducers: Option[Seq[Option[Reducer[_]]]]) = {
     reducers match {
       case Some(reducers) => commonPartValues.groupBy { case (row, _) =>
         KeyGroupedShuffleSpec.reducePartitionValue(row, expressions, reducers)