lokeshj1703 commented on code in PR #9345:
URL: https://github.com/apache/hudi/pull/9345#discussion_r1284805566
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieFileIndex.scala:
##########
@@ -122,81 +125,120 @@ case class HoodieFileIndex(spark: SparkSession,
* @return list of PartitionDirectory containing partition to base files
mapping
*/
override def listFiles(partitionFilters: Seq[Expression], dataFilters:
Seq[Expression]): Seq[PartitionDirectory] = {
+ // Prune the partition path by the partition filters
+ // NOTE: Non-partitioned tables are assumed to consist from a single
partition
+ // encompassing the whole table
+ val partitionsAndFileSlices =
getFileSlicesForPrunedPartitions(partitionFilters)
+ val listedPartitions = filterFileSlices(dataFilters,
partitionsAndFileSlices).map {
+ case (partition, fileSlices) =>
+ val allCandidateFiles: Seq[FileStatus] = fileSlices.flatMap(fs => {
+ val baseFileStatusOpt =
getBaseFileStatus(Option.apply(fs.getBaseFile.orElse(null)))
+ val logFilesStatus = if (includeLogFiles) {
+
fs.getLogFiles.map[FileStatus](JFunction.toJavaFunction[HoodieLogFile,
FileStatus](lf => lf.getFileStatus))
+ } else {
+ java.util.stream.Stream.empty()
+ }
+ val files =
logFilesStatus.collect(Collectors.toList[FileStatus]).asScala
+ baseFileStatusOpt.foreach(f => files.append(f))
+ files
+ })
+
+ PartitionDirectory(InternalRow.fromSeq(partition.get.values),
allCandidateFiles)
+ }
+
+ hasPushedDownPartitionPredicates = true
+
+ if (shouldReadAsPartitionedTable()) {
+ listedPartitions
+ } else {
+ Seq(PartitionDirectory(InternalRow.empty,
listedPartitions.flatMap(_.files)))
+ }
+ }
+
+ def filterFileSlices(dataFilters: Seq[Expression], partitionAndFileSlices:
Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])])
+ : Seq[(Option[BaseHoodieTableFileIndex.PartitionPath], Seq[FileSlice])] = {
// Look up candidate files names in the col-stats index, if all of the
following conditions are true
// - Data-skipping is enabled
// - Col-Stats Index is present
// - List of predicates (filters) is present
val candidateFilesNamesOpt: Option[Set[String]] =
- lookupCandidateFilesInMetadataTable(dataFilters) match {
- case Success(opt) => opt
- case Failure(e) =>
- logError("Failed to lookup candidate files in File Index", e)
-
- spark.sqlContext.getConf(DataSkippingFailureMode.configName,
DataSkippingFailureMode.Fallback.value) match {
- case DataSkippingFailureMode.Fallback.value => Option.empty
- case DataSkippingFailureMode.Strict.value => throw new
HoodieException(e);
- }
- }
+ lookupCandidateFilesInMetadataTable(dataFilters) match {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
