lazyvalinputRDD: RDD[InternalRow] = {
valreadFile: (PartitionedFile) =>Iterator[InternalRow] =relation.fileFormat.buildReaderWithPartitionValues(
sparkSession=relation.sparkSession,
dataSchema=relation.dataSchema,
partitionSchema=relation.partitionSchema,
requiredSchema=requiredSchema,
filters=pushedDownFilters,
options=relation.options,
hadoopConf=relation.sparkSession.sessionState.newHadoopConfWithOptions(relation.options))
valreadRDD=if (bucketedScan) {
createBucketedReadRDD(relation.bucketSpec.get, readFile, dynamicallySelectedPartitions,
relation)
    } else {
createReadRDD(readFile, dynamicallySelectedPartitions, relation)
    }
sendDriverMetrics()
readRDD  }

/*** Create an RDD for non-bucketed reads.* The bucketed variant of this function is [[createBucketedReadRDD]].** @param readFile a function to read each (part of a) file.* @param selectedPartitions Hive-style partition that are part of the read.* @param fsRelation [[HadoopFsRelation]] associated with the read.*/privatedefcreateReadRDD(
readFile: (PartitionedFile) =>Iterator[InternalRow],
selectedPartitions: Array[PartitionDirectory],
fsRelation: HadoopFsRelation): RDD[InternalRow] = {
// 文件打开开销，每次打开文件最少需要读取的字节    valopenCostInBytes=fsRelation.sparkSession.sessionState.conf.filesOpenCostInBytes// 最大切分分片大小valmaxSplitBytes=FilePartition.maxSplitBytes(fsRelation.sparkSession, selectedPartitions)
logInfo(s"Planning scan with bin packing, max size: $maxSplitBytes bytes, "+s"open cost is considered as scanning $openCostInBytes bytes.")
// Filter files with bucket pruning if possiblevalbucketingEnabled=fsRelation.sparkSession.sessionState.conf.bucketingEnabledvalshouldProcess: Path=>Boolean=optionalBucketSetmatch {
caseSome(bucketSet) ifbucketingEnabled=>// Do not prune the file if bucket file name is invalidfilePath=>BucketingUtils.getBucketId(filePath.getName).forall(bucketSet.get)
case_=>_=>true    }
// 对分区下文件进行切分并按照从大到小进行排序valsplitFiles=selectedPartitions.flatMap { partition=>partition.files.flatMap { file=>// getPath() is very expensive so we only want to call it once in this block:valfilePath=file.getPathif (shouldProcess(filePath)) {
// 文件是否可split，parquet/orc/avro均可被splitvalisSplitable=relation.fileFormat.isSplitable(
relation.sparkSession, relation.options, filePath)
// 切分文件PartitionedFileUtil.splitFiles(
sparkSession=relation.sparkSession,
file=file,
filePath=filePath,
isSplitable=isSplitable,
maxSplitBytes=maxSplitBytes,
partitionValues=partition.values          )
        } else {
Seq.empty        }
      }
    }.sortBy(_.length)(implicitly[Ordering[Long]].reverse)
valpartitions=FilePartition.getFilePartitions(relation.sparkSession, splitFiles, maxSplitBytes)
newFileScanRDD(fsRelation.sparkSession, readFile, partitions)
  }

defmaxSplitBytes(
sparkSession: SparkSession,
selectedPartitions: Seq[PartitionDirectory]): Long= {
// 读取文件时打包成最大的partition大小，默认为128MB，对应一个block大小valdefaultMaxSplitBytes=sparkSession.sessionState.conf.filesMaxPartitionBytes// 打开每个文件的开销，默认为4MBvalopenCostInBytes=sparkSession.sessionState.conf.filesOpenCostInBytes// 建议的（不保证）最小分割文件分区数，默认未设置，从leafNodeDefaultParallelism获取// 代码逻辑调用链 SparkSession#leafNodeDefaultParallelism -> SparkContext#defaultParallelism// -> TaskSchedulerImpl#defaultParallelism -> CoarseGrainedSchedulerBackend#defaultParallelism// -> 总共多少核max(executor core总和, 2)，最少为2valminPartitionNum=sparkSession.sessionState.conf.filesMinPartitionNum      .getOrElse(sparkSession.leafNodeDefaultParallelism)
// 总共读取的大小valtotalBytes=selectedPartitions.flatMap(_.files.map(_.getLen+openCostInBytes)).sum// 单core读取的大小valbytesPerCore=totalBytes/minPartitionNum// 计算大小，不会超过设置的128MBMath.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
  }

defsplitFiles(
sparkSession: SparkSession,
file: FileStatus,
filePath: Path,
isSplitable: Boolean,
maxSplitBytes: Long,
partitionValues: InternalRow): Seq[PartitionedFile] = {
if (isSplitable) {
// 切分为多个分片      (0Luntilfile.getLenbymaxSplitBytes).map { offset=>valremaining=file.getLen-offsetvalsize=if (remaining>maxSplitBytes) maxSplitByteselseremainingvalhosts=getBlockHosts(getBlockLocations(file), offset, size)
PartitionedFile(partitionValues, filePath.toUri.toString, offset, size, hosts)
      }
    } else {
Seq(getPartitionedFile(file, filePath, partitionValues))
    }
  }

defgetFilePartitions(
sparkSession: SparkSession,
partitionedFiles: Seq[PartitionedFile],
maxSplitBytes: Long): Seq[FilePartition] = {
valpartitions=newArrayBuffer[FilePartition]
valcurrentFiles=newArrayBuffer[PartitionedFile]
varcurrentSize=0L/** Close the current partition and move to the next. */defclosePartition(): Unit= {
if (currentFiles.nonEmpty) {
// Copy to a new Array.// 重新生成一个新的PartitionFilevalnewPartition=FilePartition(partitions.size, currentFiles.toArray)
partitions+=newPartition      }
currentFiles.clear()
currentSize=0    }
// 打开文件开销，默认为4MBvalopenCostInBytes=sparkSession.sessionState.conf.filesOpenCostInBytes// Assign files to partitions using "Next Fit Decreasing"partitionedFiles.foreach { file=>if (currentSize+file.length>maxSplitBytes) {
// 如果累加的文件大小大于的最大切分大小，则关闭该分区，表示完成一个Task读取的数据切分closePartition()
      }
// Add the given file to the current partition.currentSize+=file.length+openCostInBytescurrentFiles+=file    }
// 最后关闭一次分区，文件可能较小closePartition()
partitions.toSeq  }