bvaradar 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_r340197742
##########
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:
Yes
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
