vinothchandar commented on a change in pull request #942: [HUDI-137] Fix state
transitions for Hudi cleaning action
URL: https://github.com/apache/incubator-hudi/pull/942#discussion_r337004958
##########
File path:
hudi-client/src/main/java/org/apache/hudi/table/HoodieCopyOnWriteTable.java
##########
@@ -281,26 +282,82 @@ protected HoodieMergeHandle getUpdateHandle(String
commitTime, String fileId,
}
/**
- * Performs cleaning of partition paths according to cleaning policy and
returns the number of
- * files cleaned. Handles skews in partitions to clean by making files to
clean as the unit of
- * task distribution.
- *
- * @throws IllegalArgumentException if unknown cleaning policy is provided
+ * Generates List of files to be cleaned
+ * @param jsc
+ * @return
*/
- @Override
- public List<HoodieCleanStat> clean(JavaSparkContext jsc) {
+ public HoodieCleanerPlan scheduleClean(JavaSparkContext jsc) {
try {
FileSystem fs = getMetaClient().getFs();
List<String> partitionsToClean = FSUtils
.getAllPartitionPaths(fs, getMetaClient().getBasePath(),
config.shouldAssumeDatePartitioning());
- logger.info("Partitions to clean up : " + partitionsToClean + ", with
policy " + config
- .getCleanerPolicy());
if (partitionsToClean.isEmpty()) {
- logger.info("Nothing to clean here mom. It is already clean");
- return Collections.emptyList();
+ logger.info("Nothing to clean here. It is already clean");
+ return
HoodieCleanerPlan.newBuilder().setPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name()).build();
}
- return cleanPartitionPaths(partitionsToClean, jsc);
+ logger.info("Total Partitions to clean : " + partitionsToClean.size() +
", with policy " + config
+ .getCleanerPolicy());
+ int cleanerParallelism = Math.min(partitionsToClean.size(),
config.getCleanerParallelism());
+ logger.info("Using cleanerParallelism: " + cleanerParallelism);
+ HoodieCleanHelper cleaner = new HoodieCleanHelper(this, config);
+ Option<HoodieInstant> earliestInstant =
cleaner.getEarliestCommitToRetain();
+
+ Map<String, List<String>> cleanOps = jsc.parallelize(partitionsToClean,
cleanerParallelism).map(
Review comment:
Since we are finishing up any inflight/requested cleans first before getting
here, this list to clean here would do mostly pick only actual new files to
clean ?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
