private def createNonBucketedReadRDD(
    readFile: (PartitionedFile) => Iterator[InternalRow],
    selectedPartitions: Array[PartitionDirectory],
    fsRelation: HadoopFsRelation): RDD[InternalRow] = {
  val openCostInBytes = fsRelation.sparkSession.sessionState.conf.filesOpenCostInBytes
  val maxSplitBytes =
    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.")

  val splitFiles = selectedPartitions.flatMap { partition =>
    partition.files.flatMap { file =>
      // getPath() is very expensive so we only want to call it once in this block:
      val filePath = file.getPath
      val isSplitable = relation.fileFormat.isSplitable(
        relation.sparkSession, relation.options, filePath)
      PartitionedFileUtil.splitFiles(
        sparkSession = relation.sparkSession,
        file = file,
        filePath = filePath,
        isSplitable = isSplitable,
        maxSplitBytes = maxSplitBytes,
        partitionValues = partition.values
      )
    }
  }.sortBy(_.length)(implicitly[Ordering[Long]].reverse)

  val partitions =
    FilePartition.getFilePartitions(relation.sparkSession, splitFiles, maxSplitBytes)

  // 使用分配的分区进而创建FileScanRDD
  new FileScanRDD(fsRelation.sparkSession, readFile, partitions)
}

def maxSplitBytes(
    sparkSession: SparkSession,
    selectedPartitions: Seq[PartitionDirectory]): Long = {
  // spark.sql.files.maxPartitionBytes
  val defaultMaxSplitBytes = sparkSession.sessionState.conf.filesMaxPartitionBytes
  // spark.sql.files.openCostInBytes
  val openCostInBytes = sparkSession.sessionState.conf.filesOpenCostInBytes
  // spark.sql.files.minPartitionNum
  val minPartitionNum = sparkSession.sessionState.conf.filesMinPartitionNum
    .getOrElse(sparkSession.sparkContext.defaultParallelism)
  // 计算目标分片大小
  val totalBytes = selectedPartitions.flatMap(_.files.map(_.getLen + openCostInBytes)).sum
  val bytesPerCore = totalBytes / minPartitionNum

  Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))
}

def getFilePartitions(
    sparkSession: SparkSession,
    partitionedFiles: Seq[PartitionedFile],
    maxSplitBytes: Long): Seq[FilePartition] = {
  val partitions = new ArrayBuffer[FilePartition]
  val currentFiles = new ArrayBuffer[PartitionedFile]
  var currentSize = 0L

  /** Close the current partition and move to the next. */
  def closePartition(): Unit = {
    if (currentFiles.nonEmpty) {
      // Copy to a new Array.
      val newPartition = FilePartition(partitions.size, currentFiles.toArray)
      partitions += newPartition
    }
    currentFiles.clear()
    currentSize = 0
  }

  val openCostInBytes = sparkSession.sessionState.conf.filesOpenCostInBytes
  // 使用"Next Fit Decreasing"将文件分配到分区
  partitionedFiles.foreach { file =>
    // maxSplitBytes控制分区大小，达到指定阈值，则创建分区
    // 注意，没有计入文件开销
    if (currentSize + file.length > maxSplitBytes) {
      closePartition()
    }
    // Add the given file to the current partition.
    // 注意，当前文件在添加到当前分区时考虑了文件开销
    currentSize += file.length + openCostInBytes
    currentFiles += file
  }
  closePartition()
  partitions.toSeq
}