codope commented on code in PR #9345:
URL: https://github.com/apache/hudi/pull/9345#discussion_r1284982434
##########
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:
Please try `INMEMORY` index type. I just did an upsert with quickstart data
and in-memory index as below, and it generates only log files.
```
df.write.format("hudi").
option("hoodie.insert.shuffle.parallelism", "1").
option("hoodie.upsert.shuffle.parallelism", "1").
option(TABLE_TYPE_OPT_KEY,"MERGE_ON_READ").
option(PRECOMBINE_FIELD_OPT_KEY, "ts").
option(RECORDKEY_FIELD_OPT_KEY, "uuid").
option(PARTITIONPATH_FIELD_OPT_KEY, "partitionpath").
option(TABLE_NAME, tableName).
option(OPERATION_OPT_KEY, "upsert").
option("hoodie.index.type", "INMEMORY").
mode(Append).
save(basePath)
```
--
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]
