[SPARK-41097][CORE][SQL][SS][PROTOBUF] Remove redundant collection conversion base on Scala 2.13 code

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala

@@ -243,7 +243,7 @@ private[kafka010] class KafkaOffsetReaderAdmin(
       }
 
       tp -> offset
-    }.toMap
+    }
   }
 
   private def fetchSpecificOffsets0(

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderConsumer.scala

@@ -291,7 +291,7 @@ private[kafka010] class KafkaOffsetReaderConsumer(
       }
 
       tp -> offset
-    }.toMap
+    }
   }
 
   private def fetchSpecificOffsets0(

connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala

@@ -99,7 +99,7 @@ object SchemaConverters {
 
         Option(
           fd.getMessageType.getFields.asScala
-            .flatMap(structFieldFor(_, newRecordNames.toSet))
+            .flatMap(structFieldFor(_, newRecordNames))
             .toSeq)
           .filter(_.nonEmpty)
           .map(StructType.apply)

core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

@@ -568,7 +568,7 @@ private[spark] class ExecutorAllocationManager(
       // We don't want to change our target number of executors, because we already did that
       // when the task backlog decreased.
       if (decommissionEnabled) {
-        val executorIdsWithoutHostLoss = executorIdsToBeRemoved.toSeq.map(
+        val executorIdsWithoutHostLoss = executorIdsToBeRemoved.map(
           id => (id, ExecutorDecommissionInfo("spark scale down"))).toArray
         client.decommissionExecutors(
           executorIdsWithoutHostLoss,

core/src/main/scala/org/apache/spark/status/api/v1/api.scala

@@ -424,7 +424,7 @@ class ExecutorPeakMetricsDistributions private[spark](
   /** Returns the distributions for the specified metric. */
   def getMetricDistribution(metricName: String): IndexedSeq[Double] = {
     val sorted = executorMetrics.map(_.getMetricValue(metricName)).sorted
-    indices.map(i => sorted(i.toInt).toDouble).toIndexedSeq
+    indices.map(i => sorted(i.toInt).toDouble)
   }
 }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/types/StructType.scala

@@ -555,7 +555,7 @@ object StructType extends AbstractDataType {
 
   def apply(fields: java.util.List[StructField]): StructType = {
     import scala.collection.JavaConverters._
-    StructType(fields.asScala.toSeq)
+    StructType(fields.asScala.toArray)
   }
 
   private[sql] def fromAttributes(attributes: Seq[Attribute]): StructType =

sql/core/src/main/scala/org/apache/spark/sql/UDFRegistration.scala

@@ -57,7 +57,7 @@ class UDFRegistration private[sql] (functionRegistry: FunctionRegistry) extends
       s"""
         | Registering new PythonUDF:
         | name: $name
-        | command: ${udf.func.command.toSeq}
+        | command: ${udf.func.command}
         | envVars: ${udf.func.envVars}
         | pythonIncludes: ${udf.func.pythonIncludes}
         | pythonExec: ${udf.func.pythonExec}

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -545,7 +545,7 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     } else {
       CatalogTableType.MANAGED
     }
-    val (partitionColumns, maybeBucketSpec) = partitioning.toSeq.convertTransforms
+    val (partitionColumns, maybeBucketSpec) = partitioning.convertTransforms
 
     CatalogTable(
       identifier = table,

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala

@@ -305,9 +305,9 @@ object ShufflePartitionsUtil extends Logging {
           val dataSize = spec.startReducerIndex.until(spec.endReducerIndex)
             .map(mapStats.bytesByPartitionId).sum
           spec.copy(dataSize = Some(dataSize))
-        }.toSeq
-      case None => partitionSpecs.map(_.copy(dataSize = Some(0))).toSeq
-    }.toSeq
+        }
+      case None => partitionSpecs.map(_.copy(dataSize = Some(0)))
+    }
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -939,7 +939,7 @@ case class ShowTablePropertiesCommand(
           }
         case None =>
           properties.filterKeys(!_.startsWith(CatalogTable.VIEW_PREFIX))
-            .toSeq.sortBy(_._1).map(p => Row(p._1, p._2)).toSeq
+            .toSeq.sortBy(_._1).map(p => Row(p._1, p._2))
       }
     }
   }

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

@@ -39,7 +39,7 @@ case class ShowFunctionsExec(
     pattern: Option[String]) extends V2CommandExec with LeafExecNode {
 
   private def applyPattern(names: Seq[String]): Seq[String] = {
-    StringUtils.filterPattern(names.toSeq, pattern.getOrElse("*"))
+    StringUtils.filterPattern(names, pattern.getOrElse("*"))
   }
 
   override protected def run(): Seq[InternalRow] = {

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

@@ -47,7 +47,7 @@ case class ShowTablePropertiesExec(
         }
       case None =>
         properties.toSeq.sortBy(_._1).map(kv =>
-          toCatalystRow(kv._1, kv._2)).toSeq
+          toCatalystRow(kv._1, kv._2))
     }
   }
 }

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -776,7 +776,7 @@ private[hive] class HiveClientImpl(
           assert(s.values.forall(_.nonEmpty), s"partition spec '$s' is invalid")
           shim.getPartitionNames(client, table.database, table.identifier.table, s.asJava, -1)
       }
-    hivePartitionNames.sorted.toSeq
+    hivePartitionNames.sorted
   }
 
   override def getPartitionOption(
@@ -812,7 +812,7 @@ private[hive] class HiveClientImpl(
     val parts = shim.getPartitions(client, hiveTable, partSpec.asJava)
       .map(fromHivePartition(_, absoluteUri))
     HiveCatalogMetrics.incrementFetchedPartitions(parts.length)
-    parts.toSeq
+    parts
   }
 
   override def getPartitionsByFilter(