Address review feedback: rename ExtractV2Relation to ExtractV2ScanInfo, collapse single-line match

Author: yyanyy

sql/catalyst/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Relation.scala

@@ -290,7 +290,7 @@ object ExtractV2Scan {
     Some(scanRelation.scan)
 }
 
-object ExtractV2Relation {
+object ExtractV2ScanInfo {
   def unapply(scanRelation: DataSourceV2ScanRelation)
       : Option[(DataSourceV2Relation, Scan, Seq[AttributeReference])] =
     Some((scanRelation.relation, scanRelation.scan, scanRelation.output))

sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala

@@ -60,7 +60,7 @@ import org.apache.spark.sql.execution.aggregate.TypedAggregateExpression
 import org.apache.spark.sql.execution.arrow.{ArrowBatchStreamWriter, ArrowConverters}
 import org.apache.spark.sql.execution.command._
 import org.apache.spark.sql.execution.datasources.LogicalRelationWithTable
-import org.apache.spark.sql.execution.datasources.v2.{ExtractV2Relation, ExtractV2Table, FileTable}
+import org.apache.spark.sql.execution.datasources.v2.{ExtractV2ScanInfo, ExtractV2Table, FileTable}
 import org.apache.spark.sql.execution.python.EvaluatePython
 import org.apache.spark.sql.execution.stat.StatFunctions
 import org.apache.spark.sql.internal.SQLConf
@@ -1732,7 +1732,7 @@ class Dataset[T] private[sql](
         fr.inputFiles
       case r: HiveTableRelation =>
         r.tableMeta.storage.locationUri.map(_.toString).toArray
-      case ExtractV2Relation(ExtractV2Table(table: FileTable), _, _) =>
+      case ExtractV2ScanInfo(ExtractV2Table(table: FileTable), _, _) =>
         table.fileIndex.inputFiles
     }.flatten
     files.toSet.toArray

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

@@ -119,7 +119,7 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
   }
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
-    case PhysicalOperation(project, filters, ExtractV2Relation(
+    case PhysicalOperation(project, filters, ExtractV2ScanInfo(
       v2Relation, V1ScanWrapper(scan, pushed, pushedDownOperators), output)) =>
       val v1Relation = scan.toV1TableScan[BaseRelation with TableScan](session.sqlContext)
       if (v1Relation.schema != scan.readSchema()) {
@@ -146,7 +146,7 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
         project, filters, dsScan, needsUnsafeConversion = false) :: Nil
 
     case PhysicalOperation(project, filters,
-        ExtractV2Relation(_, scan: LocalScan, output)) =>
+        ExtractV2ScanInfo(_, scan: LocalScan, output)) =>
       val localScanExec = LocalTableScanExec(output, scan.rows().toImmutableArraySeq, None)
       DataSourceV2Strategy.withProjectAndFilter(
         project, filters, localScanExec, needsUnsafeConversion = false) :: Nil
@@ -346,7 +346,7 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
 
     case DeleteFromTable(relation, condition) =>
       relation match {
-        case ExtractV2Relation(r, _, output) =>
+        case ExtractV2ScanInfo(r, _, output) =>
           val table = r.table
           if (SubqueryExpression.hasSubquery(condition)) {
             throw QueryCompilationErrors.unsupportedDeleteByConditionWithSubqueryError(condition)

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

@@ -41,7 +41,7 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
   }
 
   private def partitioning(plan: LogicalPlan) = plan.transformDown {
-    case d @ ExtractV2Relation(relation, scan: SupportsReportPartitioning, _)
+    case d @ ExtractV2ScanInfo(relation, scan: SupportsReportPartitioning, _)
         if d.keyGroupedPartitioning.isEmpty =>
       val catalystPartitioning = scan.outputPartitioning() match {
         case kgp: KeyGroupedPartitioning =>
@@ -69,7 +69,7 @@ object V2ScanPartitioningAndOrdering extends Rule[LogicalPlan] with Logging {
   }
 
   private def ordering(plan: LogicalPlan) = plan.transformDown {
-    case d @ ExtractV2Relation(relation, scan: SupportsReportOrdering, _) =>
+    case d @ ExtractV2ScanInfo(relation, scan: SupportsReportOrdering, _) =>
       val ordering =
         V2ExpressionUtils.toCatalystOrdering(scan.outputOrdering(), relation, relation.funCatalog)
       d.copy(ordering = Some(ordering))

sql/core/src/test/scala/org/apache/spark/sql/collation/CollatedFilterPushDownToParquetSuite.scala

@@ -257,8 +257,7 @@ class CollatedFilterPushDownToParquetV2Suite extends CollatedFilterPushDownToPar
 
   override def getPushedDownFilters(query: DataFrame): Seq[Filter] = {
     query.queryExecution.optimizedPlan.collectFirst {
-      case PhysicalOperation(_, _,
-          ExtractV2Scan(scan: ParquetScan)) =>
+      case PhysicalOperation(_, _, ExtractV2Scan(scan: ParquetScan)) =>
         scan.pushedFilters.toSeq
     }.getOrElse(Seq.empty)
   }