address comments

Author: gengliangwang

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

@@ -176,16 +176,16 @@ case class ParquetPartitionReaderFactory(
     reader
   }
 
-  private def createRowBaseReader(file: PartitionedFile): ParquetRecordReader[UnsafeRow] = {
-    buildReaderBase(file, createRowBaseReader0).asInstanceOf[ParquetRecordReader[UnsafeRow]]
+  private def createRowBaseReader(file: PartitionedFile): RecordReader[Void, UnsafeRow] = {
+    buildReaderBase(file, createRowBaseReader0)
   }
 
   private def createRowBaseReader0(
       split: ParquetInputSplit,
       partitionValues: InternalRow,
       hadoopAttemptContext: TaskAttemptContextImpl,
       pushed: Option[FilterPredicate],
-      convertTz: Option[TimeZone]): ParquetRecordReader[UnsafeRow] = {
+      convertTz: Option[TimeZone]): RecordReader[Void, UnsafeRow] = {
     logDebug(s"Falling back to parquet-mr")
     val taskContext = Option(TaskContext.get())
     // ParquetRecordReader returns UnsafeRow

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

@@ -63,9 +63,9 @@ case class ParquetScanBuilder(
     this.filters
   }
 
-  // The actual filter push down happens in [[ParquetPartitionReaderFactory]].
+  // Note: for Parquet, the actual filter push down happens in [[ParquetPartitionReaderFactory]].
   // It requires the Parquet physical schema to determine whether a filter is convertible.
-  // So here we simply mark that all the filters are pushed down.
+  // All filters that can be converted to Parquet are pushed down.
   override def pushedFilters(): Array[Filter] = _pushedFilters
 
   override def build(): Scan = {