lokeshj1703 commented on code in PR #9345:
URL: https://github.com/apache/hudi/pull/9345#discussion_r1284805813
##########
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 {
+ 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);
+ }
+ }
logDebug(s"Overlapping candidate files from Column Stats Index:
${candidateFilesNamesOpt.getOrElse(Set.empty)}")
- var totalFileSize = 0
- var candidateFileSize = 0
-
- // 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 prunedPartitions = listMatchingPartitionPaths(partitionFilters)
- val listedPartitions = getInputFileSlices(prunedPartitions:
_*).asScala.toSeq.map {
- case (partition, fileSlices) =>
- val baseFileStatuses: Seq[FileStatus] = getBaseFileStatus(fileSlices
- .asScala
- .map(fs => fs.getBaseFile.orElse(null))
- .filter(_ != null))
+ var totalFileSliceSize = 0
+ var candidateFileSliceSize = 0
+ val listedPartitions = partitionAndFileSlices.map {
+ case (partitionOpt, fileSlices) =>
// Filter in candidate files based on the col-stats index lookup
- val candidateFiles = baseFileStatuses.filter(fs =>
- // NOTE: This predicate is true when {@code Option} is empty
- candidateFilesNamesOpt.forall(_.contains(fs.getPath.getName)))
+ val candidateFileSlices: Seq[FileSlice] = {
+ fileSlices.filter(fs => {
+ val baseFileStatusOpt =
getBaseFileStatus(Option.apply(fs.getBaseFile.orElse(null)))
+ val logFiles =
fs.getLogFiles.collect(Collectors.toSet[HoodieLogFile]).asScala.toSet[HoodieLogFile]
+ // NOTE: This predicate is true when {@code Option} is empty
+ if (candidateFilesNamesOpt.forall(files =>
baseFileStatusOpt.exists(f => files.contains(f.getPath.getName))
+ || files.intersect(logFiles.map(f =>
f.getPath.getName)).nonEmpty)) {
+ true
+ } else {
+ false
+ }
Review Comment:
I tried with bucket index but it does not create only log files initially. I
can only see parquet files getting created and then subsequent inserts create
log files. Included the test still.
##########
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 {
+ 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);
+ }
+ }
logDebug(s"Overlapping candidate files from Column Stats Index:
${candidateFilesNamesOpt.getOrElse(Set.empty)}")
- var totalFileSize = 0
- var candidateFileSize = 0
-
- // 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 prunedPartitions = listMatchingPartitionPaths(partitionFilters)
- val listedPartitions = getInputFileSlices(prunedPartitions:
_*).asScala.toSeq.map {
- case (partition, fileSlices) =>
- val baseFileStatuses: Seq[FileStatus] = getBaseFileStatus(fileSlices
- .asScala
- .map(fs => fs.getBaseFile.orElse(null))
- .filter(_ != null))
+ var totalFileSliceSize = 0
+ var candidateFileSliceSize = 0
+ val listedPartitions = partitionAndFileSlices.map {
+ case (partitionOpt, fileSlices) =>
// Filter in candidate files based on the col-stats index lookup
- val candidateFiles = baseFileStatuses.filter(fs =>
- // NOTE: This predicate is true when {@code Option} is empty
- candidateFilesNamesOpt.forall(_.contains(fs.getPath.getName)))
+ val candidateFileSlices: Seq[FileSlice] = {
+ fileSlices.filter(fs => {
+ val baseFileStatusOpt =
getBaseFileStatus(Option.apply(fs.getBaseFile.orElse(null)))
+ val logFiles =
fs.getLogFiles.collect(Collectors.toSet[HoodieLogFile]).asScala.toSet[HoodieLogFile]
+ // NOTE: This predicate is true when {@code Option} is empty
+ if (candidateFilesNamesOpt.forall(files =>
baseFileStatusOpt.exists(f => files.contains(f.getPath.getName))
+ || files.intersect(logFiles.map(f =>
f.getPath.getName)).nonEmpty)) {
+ true
+ } else {
+ false
+ }
+ })
+ }
- totalFileSize += baseFileStatuses.size
- candidateFileSize += candidateFiles.size
- PartitionDirectory(InternalRow.fromSeq(partition.values),
candidateFiles)
+ totalFileSliceSize += fileSlices.size
+ candidateFileSliceSize += candidateFileSlices.size
+ (partitionOpt, candidateFileSlices)
}
val skippingRatio =
if (!areAllFileSlicesCached) -1
- else if (allFiles.nonEmpty && totalFileSize > 0) (totalFileSize -
candidateFileSize) / totalFileSize.toDouble
+ else if (allFiles.nonEmpty && totalFileSliceSize > 0)
(totalFileSliceSize - candidateFileSliceSize) / totalFileSliceSize.toDouble
Review Comment:
The calculation is considering file slices size for calculating
skippingRatio.
Yes, allFiles needs to be fixed. Fixed it.
--
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]
