the-other-tim-brown commented on code in PR #13340:
URL: https://github.com/apache/hudi/pull/13340#discussion_r2110745648
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java:
##########
@@ -666,71 +632,64 @@ protected void runTableServicesInline(HoodieTable table,
HoodieCommitMetadata me
*
* @param extraMetadata Metadata to pass onto the scheduled service
instant
* @param tableServiceType Type of table service to schedule
- * @return
+ * @return the requested instant time if the service was scheduled
*/
- public Option<String> scheduleTableService(String instantTime,
Option<Map<String, String>> extraMetadata,
+ public Option<String> scheduleTableService(Option<Map<String, String>>
extraMetadata,
TableServiceType
tableServiceType) {
- // A lock is required to guard against race conditions between an ongoing
writer and scheduling a table service.
- HoodieTableConfig tableConfig =
HoodieTableConfig.loadFromHoodieProps(storage, config.getBasePath());
- InstantGenerator instantGenerator =
TimelineLayout.fromVersion(tableConfig.getTableVersion().getTimelineLayoutVersion()).getInstantGenerator();
- final Option<HoodieInstant> inflightInstant =
Option.of(instantGenerator.createNewInstant(HoodieInstant.State.REQUESTED,
- tableServiceType.getAction(), instantTime));
- try {
- this.txnManager.beginTransaction(inflightInstant, Option.empty());
- LOG.info("Scheduling table service {} for table {}", tableServiceType,
config.getBasePath());
- return scheduleTableServiceInternal(instantTime, extraMetadata,
tableServiceType);
- } finally {
- this.txnManager.endTransaction(inflightInstant);
- }
+ return scheduleTableServiceInternal(Option.empty(), extraMetadata,
tableServiceType);
}
- protected Option<String> scheduleTableServiceInternal(String instantTime,
Option<Map<String, String>> extraMetadata,
- TableServiceType
tableServiceType) {
+ Option<String> scheduleTableServiceInternal(Option<String>
providedInstantTime, Option<Map<String, String>> extraMetadata,
+ TableServiceType
tableServiceType) {
if (!tableServicesEnabled(config)) {
return Option.empty();
}
+ if (tableServiceType == TableServiceType.ARCHIVE) {
+ LOG.info("Scheduling archiving is not supported. Skipping.");
+ return Option.empty();
+ }
+ if (tableServiceType == TableServiceType.CLEAN) {
+ // Cleaning acts on historical commits and is handled differently
+ return scheduleCleaning(createTable(config, storageConf),
providedInstantTime);
+ }
+ txnManager.beginTransaction(Option.empty(), Option.empty());
+ try {
+ Option<String> option;
+ HoodieTable<?, ?, ?, ?> table = createTable(config, storageConf);
+ String instantTime = providedInstantTime.orElseGet(() ->
createNewInstantTime(false));
- Option<String> option = Option.empty();
- HoodieTable<?, ?, ?, ?> table = createTable(config, storageConf);
+ switch (tableServiceType) {
+ case CLUSTER:
+ LOG.info("Scheduling clustering at instant time: {} for table {}",
instantTime, config.getBasePath());
+ Option<HoodieClusteringPlan> clusteringPlan = table
+ .scheduleClustering(context, instantTime, extraMetadata);
+ option = clusteringPlan.isPresent() ? Option.of(instantTime) :
Option.empty();
Review Comment:
yes, will clean this up here and the other places in this older code
--
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]
