You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by si...@apache.org on 2024/07/16 17:52:33 UTC
(hudi) branch master updated: [HUDI-7905] Use cluster action for clustering pending instants (#11553)
This is an automated email from the ASF dual-hosted git repository.
sivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 6456d779600 [HUDI-7905] Use cluster action for clustering pending instants (#11553)
6456d779600 is described below
commit 6456d77960067284c4bf900d0c7bb21fb57ed654
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Tue Jul 16 23:22:27 2024 +0530
[HUDI-7905] Use cluster action for clustering pending instants (#11553)
- Currently, we use replacecommit action in timeline for clustering, insert overwrite and delete partition operation. But since clustering is a table service, many times we are interested in pending clustering plans, we had to go over all pending replace commits in timeline, deser the plan to deduce clustering ones. In this patch, we are introducing a new action named "CLUSTERING" which will be used for requested and inflight commit files in timeline for clustering operation.
---
.../hudi/cli/commands/ArchivedCommitsCommand.java | 1 +
.../hudi/cli/integ/ITTestClusteringCommand.java | 4 +-
.../hudi/client/BaseHoodieTableServiceClient.java | 45 +++------
.../apache/hudi/client/BaseHoodieWriteClient.java | 7 +-
.../client/transaction/ConcurrentOperation.java | 3 +
.../PreferWriterConflictResolutionStrategy.java | 3 +-
...urrentFileWritesConflictResolutionStrategy.java | 7 +-
.../utils/LegacyArchivedMetaEntryReader.java | 2 +
.../apache/hudi/client/utils/TransactionUtils.java | 2 +-
.../org/apache/hudi/metrics/HoodieMetrics.java | 8 +-
.../java/org/apache/hudi/table/HoodieTable.java | 4 +-
.../cluster/ClusteringPlanActionExecutor.java | 4 +-
.../util/ConsistentHashingUpdateStrategyUtils.java | 4 +-
.../action/commit/BaseCommitActionExecutor.java | 5 +-
.../restore/CopyOnWriteRestoreActionExecutor.java | 3 +-
.../restore/MergeOnReadRestoreActionExecutor.java | 1 +
.../rollback/BaseRollbackActionExecutor.java | 12 +--
.../rollback/ListingBasedRollbackStrategy.java | 1 +
.../action/rollback/RestorePlanActionExecutor.java | 2 +-
.../org/apache/hudi/table/marker/WriteMarkers.java | 8 +-
.../org/apache/hudi/table/repair/RepairUtils.java | 2 +
.../TestConflictResolutionStrategyUtil.java | 46 +++++++--
...TestPreferWriterConflictResolutionStrategy.java | 11 +-
...urrentFileWritesConflictResolutionStrategy.java | 23 ++---
.../common/testutils/HoodieMetadataTestTable.java | 10 ++
.../hudi/client/HoodieFlinkTableServiceClient.java | 12 ++-
.../hudi/client/HoodieJavaTableServiceClient.java | 3 +-
.../JavaExecuteClusteringCommitActionExecutor.java | 4 +-
.../hudi/client/SparkRDDTableServiceClient.java | 3 +-
.../bucket/HoodieSparkConsistentBucketIndex.java | 5 +-
...SparkExecuteClusteringCommitActionExecutor.java | 4 +-
.../TestHoodieClientOnCopyOnWriteStorage.java | 11 +-
.../apache/hudi/io/TestHoodieTimelineArchiver.java | 4 +-
.../java/org/apache/hudi/table/TestCleaner.java | 14 +--
.../table/action/commit/TestUpsertPartitioner.java | 2 +-
.../TestCopyOnWriteRollbackActionExecutor.java | 2 +-
.../config/HoodieTableServiceManagerConfig.java | 8 +-
.../org/apache/hudi/common/model/ActionType.java | 2 +-
.../apache/hudi/common/model/TableServiceType.java | 2 +-
.../hudi/common/table/cdc/HoodieCDCExtractor.java | 3 +-
.../table/timeline/HoodieActiveTimeline.java | 47 ++++++++-
.../table/timeline/HoodieArchivedTimeline.java | 2 +-
.../table/timeline/HoodieDefaultTimeline.java | 64 ++++++++----
.../hudi/common/table/timeline/HoodieInstant.java | 7 ++
.../hudi/common/table/timeline/HoodieTimeline.java | 38 ++++++-
.../table/timeline/MetadataConversionUtils.java | 13 ++-
.../hudi/common/table/timeline/TimelineUtils.java | 22 ++--
.../table/view/AbstractTableFileSystemView.java | 2 +-
.../IncrementalTimelineSyncFileSystemView.java | 6 +-
.../apache/hudi/common/util/ClusteringUtils.java | 112 ++++++++++++++++++---
.../org/apache/hudi/common/util/CommitUtils.java | 2 +-
.../org/apache/hudi/common/util/MarkerUtils.java | 2 +-
.../hudi/common/testutils/FileCreateUtils.java | 18 ++++
.../common/testutils/HoodieTestDataGenerator.java | 10 +-
.../hudi/sink/clustering/ClusteringCommitSink.java | 15 ++-
.../sink/clustering/ClusteringPlanOperator.java | 5 +-
.../clustering/ClusteringPlanSourceFunction.java | 2 +-
.../sink/clustering/HoodieFlinkClusteringJob.java | 12 +--
.../java/org/apache/hudi/util/ClusteringUtil.java | 9 +-
.../ITTestFlinkConsistentHashingClustering.java | 2 +-
.../sink/cluster/ITTestHoodieFlinkClustering.java | 28 +++---
.../hudi/source/TestIncrementalInputSplits.java | 16 +--
.../org/apache/hudi/utils/TestClusteringUtil.java | 6 +-
.../hudi/common/table/TestTimelineUtils.java | 45 ++++++---
.../table/timeline/TestHoodieActiveTimeline.java | 15 ++-
.../view/TestHoodieTableFSViewWithClustering.java | 8 +-
.../table/view/TestHoodieTableFileSystemView.java | 30 +++---
.../hudi/common/testutils/HoodieTestTable.java | 44 +++++++-
.../hudi/common/util/TestClusteringUtils.java | 48 ++++-----
.../apache/hudi/common/util/TestCommitUtils.java | 5 +-
.../hudi/common/util/TestCompactionUtils.java | 3 +-
.../org/apache/hudi/HoodieStreamingSink.scala | 2 +-
.../org/apache/hudi/HoodieDataSourceHelpers.java | 6 +-
.../RunRollbackInflightTableServiceProcedure.scala | 8 +-
.../procedures/ShowClusteringProcedure.scala | 2 +-
.../ShowCommitExtraMetadataProcedure.scala | 6 +-
.../procedures/ShowCommitFilesProcedure.scala | 5 +-
.../procedures/ShowCommitPartitionsProcedure.scala | 5 +-
.../procedures/ShowCommitWriteStatsProcedure.scala | 5 +-
.../functional/TestSparkSortAndSizeClustering.java | 2 +-
.../apache/hudi/functional/TestCOWDataSource.scala | 2 +-
.../hudi/functional/TestHoodieActiveTimeline.scala | 31 +++++-
.../hudi/procedure/TestClusteringProcedure.scala | 28 +++---
.../hudi/procedure/TestCompactionProcedure.scala | 2 +-
...tRunRollbackInflightTableServiceProcedure.scala | 4 +-
.../hudi/utilities/streamer/HoodieStreamer.java | 2 +-
.../deltastreamer/HoodieDeltaStreamerTestBase.java | 22 ++--
.../deltastreamer/TestHoodieDeltaStreamer.java | 14 +--
.../utilities/sources/TestHoodieIncrSource.java | 2 +-
89 files changed, 720 insertions(+), 363 deletions(-)
diff --git a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
index 85497927803..eeb12365f10 100644
--- a/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
+++ b/hudi-cli/src/main/java/org/apache/hudi/cli/commands/ArchivedCommitsCommand.java
@@ -241,6 +241,7 @@ public class ArchivedCommitsCommand {
case HoodieTimeline.COMPACTION_ACTION:
return commitDetail(record, "hoodieCompactionMetadata", skipMetadata);
case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION:
return commitDetail(record, "hoodieReplaceCommitMetadata", skipMetadata);
default: {
throw new HoodieException("Unexpected action type: " + actionType);
diff --git a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
index 3385bbd06ba..ce44debb506 100644
--- a/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
+++ b/hudi-cli/src/test/java/org/apache/hudi/cli/integ/ITTestClusteringCommand.java
@@ -98,7 +98,7 @@ public class ITTestClusteringCommand extends HoodieCLIIntegrationTestBase {
// there is 1 requested clustering
HoodieActiveTimeline timeline = HoodieCLI.getTableMetaClient().getActiveTimeline();
- assertEquals(1, timeline.filterPendingReplaceTimeline().countInstants());
+ assertEquals(1, timeline.filterPendingClusteringTimeline().countInstants());
}
/**
@@ -115,7 +115,7 @@ public class ITTestClusteringCommand extends HoodieCLIIntegrationTestBase {
// get clustering instance
HoodieActiveTimeline timeline = HoodieCLI.getTableMetaClient().getActiveTimeline();
Option<String> instanceOpt =
- timeline.filterPendingReplaceTimeline().firstInstant().map(HoodieInstant::getTimestamp);
+ timeline.filterPendingClusteringTimeline().firstInstant().map(HoodieInstant::getTimestamp);
assertTrue(instanceOpt.isPresent(), "Must have pending clustering.");
final String instance = instanceOpt.get();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
index b282d8af7f6..81d2d75ea84 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
@@ -447,11 +447,11 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
*/
public HoodieWriteMetadata<O> cluster(String clusteringInstant, boolean shouldComplete) {
HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
- HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringInstant);
- if (pendingClusteringTimeline.containsInstant(inflightInstant)) {
- if (pendingClusteringTimeline.isPendingClusterInstant(inflightInstant.getTimestamp())) {
- table.rollbackInflightClustering(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false));
+ HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceOrClusteringTimeline();
+ Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(clusteringInstant, table.getActiveTimeline());
+ if (inflightInstantOpt.isPresent()) {
+ if (pendingClusteringTimeline.isPendingClusteringInstant(inflightInstantOpt.get().getTimestamp())) {
+ table.rollbackInflightClustering(inflightInstantOpt.get(), commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false));
table.getMetaClient().reloadActiveTimeline();
} else {
throw new HoodieClusteringException("Non clustering replace-commit inflight at timestamp " + clusteringInstant);
@@ -482,10 +482,9 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
public boolean purgePendingClustering(String clusteringInstant) {
HoodieTable<?, I, ?, T> table = createTable(config, context.getStorageConf().unwrapAs(Configuration.class));
- HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingReplaceTimeline();
- HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringInstant);
- if (pendingClusteringTimeline.containsInstant(inflightInstant)) {
- table.rollbackInflightClustering(inflightInstant, commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), true);
+ Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(clusteringInstant, table.getActiveTimeline());
+ if (inflightInstantOpt.isPresent()) {
+ table.rollbackInflightClustering(inflightInstantOpt.get(), commitToRollback -> getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false), true);
table.getMetaClient().reloadActiveTimeline();
return true;
}
@@ -515,7 +514,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
Option<HoodieData<WriteStatus>> writeStatuses) {
List<HoodieWriteStat> writeStats = metadata.getWriteStats();
handleWriteErrors(writeStats, TableServiceType.CLUSTER);
- final HoodieInstant clusteringInstant = HoodieTimeline.getReplaceCommitInflightInstant(clusteringCommitTime);
+ final HoodieInstant clusteringInstant = ClusteringUtils.getInflightClusteringInstant(clusteringCommitTime, table.getActiveTimeline()).get();
try {
this.txnManager.beginTransaction(Option.of(clusteringInstant), Option.empty());
@@ -531,10 +530,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
LOG.info("Committing Clustering {}", clusteringCommitTime);
LOG.debug("Clustering {} finished with result {}", clusteringCommitTime, metadata);
- table.getActiveTimeline().transitionReplaceInflightToComplete(
- false,
- clusteringInstant,
- serializeCommitMetadata(metadata));
+ ClusteringUtils.transitionClusteringOrReplaceInflightToComplete(false, clusteringInstant, serializeCommitMetadata(metadata), table.getActiveTimeline());
} catch (Exception e) {
throw new HoodieClusteringException("unable to transition clustering inflight to complete: " + clusteringCommitTime, e);
} finally {
@@ -546,7 +542,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
if (clusteringTimer != null) {
long durationInMs = metrics.getDurationInMs(clusteringTimer.stop());
HoodieActiveTimeline.parseDateFromInstantTimeSafely(clusteringCommitTime).ifPresent(parsedInstant ->
- metrics.updateCommitMetrics(parsedInstant.getTime(), durationInMs, metadata, HoodieActiveTimeline.REPLACE_COMMIT_ACTION)
+ metrics.updateCommitMetrics(parsedInstant.getTime(), durationInMs, metadata, HoodieActiveTimeline.CLUSTERING_ACTION)
);
}
LOG.info("Clustering successfully on commit {}", clusteringCommitTime);
@@ -695,7 +691,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
}
private void inlineClustering(HoodieTable table, Option<Map<String, String>> extraMetadata) {
- if (shouldDelegateToTableServiceManager(config, ActionType.replacecommit)) {
+ if (shouldDelegateToTableServiceManager(config, ActionType.clustering)) {
scheduleClustering(extraMetadata);
} else {
runAnyPendingClustering(table);
@@ -714,7 +710,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
}
protected void runAnyPendingClustering(HoodieTable table) {
- table.getActiveTimeline().filterPendingReplaceTimeline().getInstants().forEach(instant -> {
+ table.getActiveTimeline().filterPendingReplaceOrClusteringTimeline().getInstants().forEach(instant -> {
Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlan = ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant);
if (instantPlan.isPresent()) {
LOG.info("Running pending clustering at instant {}", instantPlan.get().getLeft());
@@ -821,16 +817,8 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
* @return
*/
private HoodieTimeline getInflightTimelineExcludeCompactionAndClustering(HoodieTableMetaClient metaClient) {
- HoodieTimeline inflightTimelineWithReplaceCommit = metaClient.getCommitsTimeline().filterPendingExcludingCompaction();
- HoodieTimeline inflightTimelineExcludeClusteringCommit = inflightTimelineWithReplaceCommit.filter(instant -> {
- if (instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlan = ClusteringUtils.getClusteringPlan(metaClient, instant);
- return !instantPlan.isPresent();
- } else {
- return true;
- }
- });
- return inflightTimelineExcludeClusteringCommit;
+ HoodieTimeline inflightTimelineExcludingCompaction = metaClient.getCommitsTimeline().filterPendingExcludingCompaction();
+ return inflightTimelineExcludingCompaction.filter(instant -> !ClusteringUtils.isClusteringInstant(inflightTimelineExcludingCompaction, instant));
}
protected Option<HoodiePendingRollbackInfo> getPendingRollbackInfo(HoodieTableMetaClient metaClient, String commitToRollback) {
@@ -880,8 +868,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O> extends BaseHoodieCl
String instantToRollback = rollbackPlan.getInstantToRollback().getCommitTime();
if (ignoreCompactionAndClusteringInstants) {
if (!HoodieTimeline.COMPACTION_ACTION.equals(action)) {
- boolean isClustering = HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action)
- && ClusteringUtils.getClusteringPlan(metaClient, new HoodieInstant(true, action, instantToRollback)).isPresent();
+ boolean isClustering = ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline(), new HoodieInstant(true, action, instantToRollback));
if (!isClustering) {
infoMap.putIfAbsent(instantToRollback, Option.of(new HoodiePendingRollbackInfo(rollbackInstant, rollbackPlan)));
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
index e1d0524ce6f..e2c7799afe8 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java
@@ -50,6 +50,7 @@ import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
@@ -928,8 +929,8 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
this.heartbeatClient.start(instantTime);
}
- if (actionType.equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
- metaClient.getActiveTimeline().createRequestedReplaceCommit(instantTime, actionType);
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(actionType)) {
+ metaClient.getActiveTimeline().createRequestedCommitWithReplaceMetadata(instantTime, actionType);
} else {
metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(HoodieInstant.State.REQUESTED, actionType,
instantTime));
@@ -1013,7 +1014,7 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient
* @return Collection of WriteStatus to inspect errors and counts
*/
public HoodieWriteMetadata<O> cluster(String clusteringInstantTime) {
- if (shouldDelegateToTableServiceManager(config, ActionType.replacecommit)) {
+ if (shouldDelegateToTableServiceManager(config, ActionType.clustering)) {
throw new UnsupportedOperationException("Clustering should be delegated to table service manager instead of direct run.");
}
return cluster(clusteringInstantTime, true);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
index cba0f516e88..2e22c59238e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/ConcurrentOperation.java
@@ -37,6 +37,7 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
@@ -122,6 +123,7 @@ public class ConcurrentOperation {
this.operationType = WriteOperationType.fromValue(this.metadataWrapper.getMetadataFromTimeline().getHoodieCommitMetadata().getOperationType());
break;
case REPLACE_COMMIT_ACTION:
+ case CLUSTERING_ACTION:
if (instant.isCompleted()) {
this.mutatedPartitionAndFileIds = getPartitionAndFileIdWithoutSuffixFromSpecificRecord(
this.metadataWrapper.getMetadataFromTimeline().getHoodieReplaceCommitMetadata().getPartitionToWriteStats());
@@ -164,6 +166,7 @@ public class ConcurrentOperation {
case COMMIT_ACTION:
case DELTA_COMMIT_ACTION:
case REPLACE_COMMIT_ACTION:
+ case CLUSTERING_ACTION:
case LOG_COMPACTION_ACTION:
this.mutatedPartitionAndFileIds = CommitUtils.getPartitionAndFileIdWithoutSuffix(this.metadataWrapper.getCommitMetadata().getPartitionToWriteStats());
this.operationType = this.metadataWrapper.getCommitMetadata().getOperationType();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/PreferWriterConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/PreferWriterConflictResolutionStrategy.java
index 62fbf64a7f9..3a013b6e78d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/PreferWriterConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/PreferWriterConflictResolutionStrategy.java
@@ -54,8 +54,7 @@ public class PreferWriterConflictResolutionStrategy
public Stream<HoodieInstant> getCandidateInstants(HoodieTableMetaClient metaClient, HoodieInstant currentInstant,
Option<HoodieInstant> lastSuccessfulInstant) {
HoodieActiveTimeline activeTimeline = metaClient.reloadActiveTimeline();
- if ((REPLACE_COMMIT_ACTION.equals(currentInstant.getAction())
- && ClusteringUtils.isClusteringInstant(activeTimeline, currentInstant))
+ if (ClusteringUtils.isClusteringInstant(activeTimeline, currentInstant)
|| COMPACTION_ACTION.equals(currentInstant.getAction())) {
return getCandidateInstantsForTableServicesCommits(activeTimeline, currentInstant);
} else {
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
index ce16e14af22..c162aee380a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/SimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -24,7 +24,6 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
-import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.exception.HoodieWriteConflictException;
@@ -38,9 +37,6 @@ import java.util.HashSet;
import java.util.Set;
import java.util.stream.Stream;
-import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
-import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
-
/**
* This class is a basic implementation of a conflict resolution strategy for concurrent writes {@link ConflictResolutionStrategy}.
*/
@@ -65,9 +61,8 @@ public class SimpleConcurrentFileWritesConflictResolutionStrategy
.getInstantsAsStream();
Stream<HoodieInstant> compactionAndClusteringPendingTimeline = activeTimeline
- .getTimelineOfActions(CollectionUtils.createSet(REPLACE_COMMIT_ACTION, COMPACTION_ACTION))
+ .filterPendingReplaceClusteringAndCompactionTimeline()
.findInstantsAfter(currentInstant.getTimestamp())
- .filterInflightsAndRequested()
.getInstantsAsStream();
return Stream.concat(completedCommitsInstantStream, compactionAndClusteringPendingTimeline);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LegacyArchivedMetaEntryReader.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LegacyArchivedMetaEntryReader.java
index 26ccc2a456e..9c092e2fecc 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LegacyArchivedMetaEntryReader.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/LegacyArchivedMetaEntryReader.java
@@ -122,6 +122,8 @@ public class LegacyArchivedMetaEntryReader {
return Option.of("hoodieCompactionPlan");
case HoodieTimeline.REPLACE_COMMIT_ACTION:
return Option.of("hoodieReplaceCommitMetadata");
+ case HoodieTimeline.CLUSTERING_ACTION:
+ return Option.of("hoodieReplaceCommitMetadata");
case HoodieTimeline.INDEXING_ACTION:
return Option.of("hoodieIndexCommitMetadata");
default:
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
index 0277492ca09..7c82682f850 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/utils/TransactionUtils.java
@@ -136,7 +136,7 @@ public class TransactionUtils {
public static Set<String> getInflightAndRequestedInstants(HoodieTableMetaClient metaClient) {
// collect InflightAndRequest instants for deltaCommit/commit/compaction/clustering
Set<String> timelineActions = CollectionUtils
- .createImmutableSet(HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.COMMIT_ACTION);
+ .createImmutableSet(HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION, HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.COMMIT_ACTION);
return metaClient
.getActiveTimeline()
.getTimelineOfActions(timelineActions)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
index 49c02afa326..15a39249d21 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/metrics/HoodieMetrics.java
@@ -81,7 +81,7 @@ public class HoodieMetrics {
public String commitTimerName = null;
public String logCompactionTimerName = null;
public String deltaCommitTimerName = null;
- public String replaceCommitTimerName = null;
+ public String clusterCommitTimerName = null;
public String finalizeTimerName = null;
public String compactionTimerName = null;
public String indexTimerName = null;
@@ -120,7 +120,7 @@ public class HoodieMetrics {
this.archiveTimerName = getMetricsName(TIMER_ACTION, ARCHIVE_ACTION);
this.commitTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.COMMIT_ACTION);
this.deltaCommitTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION);
- this.replaceCommitTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION);
+ this.clusterCommitTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.CLUSTERING_ACTION);
this.finalizeTimerName = getMetricsName(TIMER_ACTION, FINALIZE_ACTION);
this.compactionTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.COMPACTION_ACTION);
this.logCompactionTimerName = getMetricsName(TIMER_ACTION, HoodieTimeline.LOG_COMPACTION_ACTION);
@@ -165,7 +165,7 @@ public class HoodieMetrics {
public Timer.Context getClusteringCtx() {
if (config.isMetricsOn() && clusteringTimer == null) {
- clusteringTimer = createTimer(replaceCommitTimerName);
+ clusteringTimer = createTimer(clusterCommitTimerName);
}
return clusteringTimer == null ? null : clusteringTimer.time();
}
@@ -371,7 +371,7 @@ public class HoodieMetrics {
}
public void updateClusteringFileCreationMetrics(long durationInMs) {
- reportMetrics(HoodieTimeline.REPLACE_COMMIT_ACTION, "fileCreationTime", durationInMs);
+ reportMetrics(HoodieTimeline.CLUSTERING_ACTION, "fileCreationTime", durationInMs);
}
/**
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index da0ddff032e..ef14396bf9e 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -652,7 +652,9 @@ public abstract class HoodieTable<T, I, K, O> implements Serializable {
*/
public void rollbackInflightClustering(HoodieInstant inflightInstant,
Function<String, Option<HoodiePendingRollbackInfo>> getPendingRollbackInstantFunc, boolean deleteInstants) {
- ValidationUtils.checkArgument(inflightInstant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION));
+ ValidationUtils.checkArgument(inflightInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)
+ || inflightInstant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION),
+ String.format("Expected replace or clustering action instant but got %s", inflightInstant));
rollbackInflightInstant(inflightInstant, getPendingRollbackInstantFunc);
if (deleteInstants) {
// above rollback would still keep requested in the timeline. so, lets delete it if if are looking to purge the pending clustering fully.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java
index 6cb8f023ba8..e0f50046769 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/ClusteringPlanActionExecutor.java
@@ -91,14 +91,14 @@ public class ClusteringPlanActionExecutor<T, I, K, O> extends BaseActionExecutor
Option<HoodieClusteringPlan> planOption = createClusteringPlan();
if (planOption.isPresent()) {
HoodieInstant clusteringInstant =
- new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime);
+ new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, instantTime);
try {
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setOperationType(WriteOperationType.CLUSTER.name())
.setExtraMetadata(extraMetadata.orElse(Collections.emptyMap()))
.setClusteringPlan(planOption.get())
.build();
- table.getActiveTimeline().saveToPendingReplaceCommit(clusteringInstant,
+ table.getActiveTimeline().saveToPendingClusterCommit(clusteringInstant,
TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
} catch (IOException ioe) {
throw new HoodieIOException("Exception scheduling clustering", ioe);
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/util/ConsistentHashingUpdateStrategyUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/util/ConsistentHashingUpdateStrategyUtils.java
index 9e50c6056a4..bb1b3124c95 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/util/ConsistentHashingUpdateStrategyUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/util/ConsistentHashingUpdateStrategyUtils.java
@@ -23,7 +23,6 @@ import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.common.model.ConsistentHashingNode;
import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
@@ -58,7 +57,8 @@ public class ConsistentHashingUpdateStrategyUtils {
public static Map<String, Pair<String, ConsistentBucketIdentifier>> constructPartitionToIdentifier(Set<String> partitions, HoodieTable table) {
// Read all pending/ongoing clustering plans
List<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPairs =
- table.getMetaClient().getActiveTimeline().filterInflightsAndRequested().filter(instant -> instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)).getInstantsAsStream()
+ table.getMetaClient().getActiveTimeline()
+ .filterPendingReplaceOrClusteringTimeline().getInstantsAsStream()
.map(instant -> ClusteringUtils.getClusteringPlan(table.getMetaClient(), instant))
.flatMap(o -> o.isPresent() ? Stream.of(o.get()) : Stream.empty())
.collect(Collectors.toList());
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
index 47ad5f07d83..91cd6f883b0 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/commit/BaseCommitActionExecutor.java
@@ -36,7 +36,6 @@ import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieInstant.State;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
@@ -265,9 +264,9 @@ public abstract class BaseCommitActionExecutor<T, I, K, O, R>
protected HoodieWriteMetadata<HoodieData<WriteStatus>> executeClustering(HoodieClusteringPlan clusteringPlan) {
context.setJobStatus(this.getClass().getSimpleName(), "Clustering records for " + config.getTableName());
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(instantTime);
+ HoodieInstant instant = ClusteringUtils.getRequestedClusteringInstant(instantTime, table.getActiveTimeline()).get();
// Mark instant as clustering inflight
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ ClusteringUtils.transitionClusteringOrReplaceRequestedToInflight(instant, Option.empty(), table.getActiveTimeline());
table.getMetaClient().reloadActiveTimeline();
// Disable auto commit. Strategy is only expected to write data in new files.
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/CopyOnWriteRestoreActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/CopyOnWriteRestoreActionExecutor.java
index 0cc0960bddc..366351fad7d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/CopyOnWriteRestoreActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/CopyOnWriteRestoreActionExecutor.java
@@ -23,6 +23,7 @@ import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieRollbackException;
import org.apache.hudi.table.HoodieTable;
@@ -41,7 +42,7 @@ public class CopyOnWriteRestoreActionExecutor<T, I, K, O>
@Override
protected HoodieRollbackMetadata rollbackInstant(HoodieInstant instantToRollback) {
if (!instantToRollback.getAction().equals(HoodieTimeline.COMMIT_ACTION)
- && !instantToRollback.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+ && !ClusteringUtils.isClusteringOrReplaceCommitAction(instantToRollback.getAction())) {
throw new HoodieRollbackException("Unsupported action in rollback instant:" + instantToRollback);
}
table.getMetaClient().reloadActiveTimeline();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/MergeOnReadRestoreActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/MergeOnReadRestoreActionExecutor.java
index d4d4fac16af..7a7dc78355a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/MergeOnReadRestoreActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/restore/MergeOnReadRestoreActionExecutor.java
@@ -41,6 +41,7 @@ public class MergeOnReadRestoreActionExecutor<T, I, K, O>
case HoodieTimeline.DELTA_COMMIT_ACTION:
case HoodieTimeline.COMPACTION_ACTION:
case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION:
// TODO : Get file status and create a rollback stat and file
// TODO : Delete the .aux files along with the instant file, okay for now since the archival process will
// delete these files when it does not see a corresponding instant file under .hoodie
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
index 1ab943fea88..5e02f20c86a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/BaseRollbackActionExecutor.java
@@ -184,12 +184,9 @@ public abstract class BaseRollbackActionExecutor<T, I, K, O> extends BaseActionE
}
}
- List<String> inflights = inflightAndRequestedCommitTimeline.getInstantsAsStream().filter(instant -> {
- if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
- return true;
- }
- return !ClusteringUtils.isClusteringInstant(table.getActiveTimeline(), instant);
- }).map(HoodieInstant::getTimestamp)
+ List<String> inflights = inflightAndRequestedCommitTimeline.getInstantsAsStream()
+ .filter(instant -> !ClusteringUtils.isClusteringInstant(table.getActiveTimeline(), instant))
+ .map(HoodieInstant::getTimestamp)
.collect(Collectors.toList());
if ((instantTimeToRollback != null) && !inflights.isEmpty()
&& (inflights.indexOf(instantTimeToRollback) != inflights.size() - 1)) {
@@ -211,8 +208,7 @@ public abstract class BaseRollbackActionExecutor<T, I, K, O> extends BaseActionE
final boolean isPendingCompaction = Objects.equals(HoodieTimeline.COMPACTION_ACTION, instantToRollback.getAction())
&& !instantToRollback.isCompleted();
- final boolean isPendingClustering = Objects.equals(HoodieTimeline.REPLACE_COMMIT_ACTION, instantToRollback.getAction())
- && !instantToRollback.isCompleted() && ClusteringUtils.getClusteringPlan(table.getMetaClient(), instantToRollback).isPresent();
+ final boolean isPendingClustering = !instantToRollback.isCompleted() && ClusteringUtils.isClusteringInstant(table.getMetaClient().getActiveTimeline(), instantToRollback);
validateSavepointRollbacks();
if (!isPendingCompaction && !isPendingClustering) {
validateRollbackCommitSequence();
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
index a6698cea4ff..5f9b3dcfb22 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackStrategy.java
@@ -128,6 +128,7 @@ public class ListingBasedRollbackStrategy implements BaseRollbackPlanActionExecu
switch (action) {
case HoodieTimeline.COMMIT_ACTION:
case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION:
hoodieRollbackRequests.addAll(getHoodieRollbackRequests(partitionPath, filesToDelete.get()));
break;
case HoodieTimeline.COMPACTION_ACTION:
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RestorePlanActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RestorePlanActionExecutor.java
index 2f9e96859ff..d2983a9aa3a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RestorePlanActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/RestorePlanActionExecutor.java
@@ -69,7 +69,7 @@ public class RestorePlanActionExecutor<T, I, K, O> extends BaseActionExecutor<T,
try {
// Get all the commits on the timeline after the provided commit time
// rollback pending clustering instants first before other instants (See HUDI-3362)
- List<HoodieInstant> pendingClusteringInstantsToRollback = table.getActiveTimeline().filterPendingReplaceTimeline()
+ List<HoodieInstant> pendingClusteringInstantsToRollback = table.getActiveTimeline().filterPendingReplaceOrClusteringTimeline()
// filter only clustering related replacecommits (Not insert_overwrite related commits)
.filter(instant -> ClusteringUtils.isClusteringInstant(table.getActiveTimeline(), instant))
.getReverseOrderedInstants()
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
index 65131116dad..08954cbb0f2 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/WriteMarkers.java
@@ -83,10 +83,10 @@ public abstract class WriteMarkers implements Serializable {
String fileId, HoodieActiveTimeline activeTimeline) {
if (writeConfig.getWriteConcurrencyMode().isOptimisticConcurrencyControl() && writeConfig.isEarlyConflictDetectionEnable()) {
HoodieTimeline pendingCompactionTimeline = activeTimeline.filterPendingCompactionTimeline();
- HoodieTimeline pendingReplaceTimeline = activeTimeline.filterPendingReplaceTimeline();
+ HoodieTimeline pendingReplaceTimeline = activeTimeline.filterPendingReplaceOrClusteringTimeline();
// TODO If current is compact or clustering then create marker directly without early conflict detection.
// Need to support early conflict detection between table service and common writers.
- // ok to use filterPendingReplaceTimeline().containsInstant because early conflict detection is not relevant for insert overwrite as well
+ // ok to use filterPendingReplaceOrClusteringTimeline().containsInstant because early conflict detection is not relevant for insert overwrite as well
if (pendingCompactionTimeline.containsInstant(instantTime) || pendingReplaceTimeline.containsInstant(instantTime)) {
return create(partitionPath, fileName, type, false);
}
@@ -124,10 +124,10 @@ public abstract class WriteMarkers implements Serializable {
if (writeConfig.isEarlyConflictDetectionEnable()
&& writeConfig.getWriteConcurrencyMode().isOptimisticConcurrencyControl()) {
HoodieTimeline pendingCompactionTimeline = activeTimeline.filterPendingCompactionTimeline();
- HoodieTimeline pendingReplaceTimeline = activeTimeline.filterPendingReplaceTimeline();
+ HoodieTimeline pendingReplaceTimeline = activeTimeline.filterPendingReplaceOrClusteringTimeline();
// TODO If current is compact or clustering then create marker directly without early conflict detection.
// Need to support early conflict detection between table service and common writers.
- // ok to use filterPendingReplaceTimeline().containsInstant because early conflict detection is not relevant for insert overwrite as well
+ // ok to use filterPendingReplaceOrClusteringTimeline().containsInstant because early conflict detection is not relevant for insert overwrite as well
if (pendingCompactionTimeline.containsInstant(instantTime) || pendingReplaceTimeline.containsInstant(instantTime)) {
return create(partitionPath, fileName, type, true);
}
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
index 672f358e6a4..17def321f4a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/repair/RepairUtils.java
@@ -44,6 +44,7 @@ import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
@@ -98,6 +99,7 @@ public final class RepairUtils {
return Option.of(commitMetadata.getPartitionToWriteStats().values().stream().flatMap(List::stream)
.map(HoodieWriteStat::getPath).collect(Collectors.toSet()));
case REPLACE_COMMIT_ACTION:
+ case CLUSTERING_ACTION:
final HoodieReplaceCommitMetadata replaceCommitMetadata =
HoodieReplaceCommitMetadata.fromBytes(
timeline.getInstantDetails(instant).get(), HoodieReplaceCommitMetadata.class);
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
index f78ff39ca85..d6988cae0fd 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestConflictResolutionStrategyUtil.java
@@ -109,7 +109,7 @@ public class TestConflictResolutionStrategyUtil {
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
- public static void createReplaceRequested(String instantTime, HoodieTableMetaClient metaClient) throws Exception {
+ public static void createClusterRequested(String instantTime, HoodieTableMetaClient metaClient) throws Exception {
String fileId1 = "file-1";
String fileId2 = "file-2";
@@ -126,11 +126,11 @@ public class TestConflictResolutionStrategyUtil {
requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
HoodieTestTable.of(metaClient)
- .addRequestedReplace(instantTime, Option.of(requestedReplaceMetadata))
+ .addRequestedCluster(instantTime, requestedReplaceMetadata)
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
- public static void createReplaceInflight(String instantTime, HoodieTableMetaClient metaClient) throws Exception {
+ public static void createClusterInflight(String instantTime, HoodieTableMetaClient metaClient) throws Exception {
String fileId1 = "file-1";
String fileId2 = "file-2";
@@ -140,7 +140,7 @@ public class TestConflictResolutionStrategyUtil {
writeStat.setFileId("file-1");
inflightReplaceMetadata.addWriteStat(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, writeStat);
HoodieTestTable.of(metaClient)
- .addInflightReplace(instantTime, Option.of(inflightReplaceMetadata))
+ .addInflightCluster(instantTime, Option.of(inflightReplaceMetadata))
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
@@ -174,7 +174,37 @@ public class TestConflictResolutionStrategyUtil {
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
- public static void createPendingReplace(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) throws Exception {
+ public static void createCluster(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) throws Exception {
+ String fileId1 = "file-1";
+ String fileId2 = "file-2";
+
+ // create replace instant to mark fileId1 as deleted
+ HoodieReplaceCommitMetadata replaceMetadata = new HoodieReplaceCommitMetadata();
+ Map<String, List<String>> partitionFileIds = new HashMap<>();
+ partitionFileIds.put(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, Arrays.asList(fileId2));
+ replaceMetadata.setPartitionToReplaceFileIds(partitionFileIds);
+ HoodieWriteStat writeStat = new HoodieWriteStat();
+ writeStat.setFileId("file-1");
+ replaceMetadata.addWriteStat(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, writeStat);
+ replaceMetadata.setOperationType(writeOperationType);
+ // create replace instant to mark fileId1 as deleted
+ HoodieRequestedReplaceMetadata requestedReplaceMetadata = new HoodieRequestedReplaceMetadata();
+ requestedReplaceMetadata.setOperationType(WriteOperationType.CLUSTER.name());
+ HoodieClusteringPlan clusteringPlan = new HoodieClusteringPlan();
+ HoodieClusteringGroup clusteringGroup = new HoodieClusteringGroup();
+ HoodieSliceInfo sliceInfo = new HoodieSliceInfo();
+ sliceInfo.setFileId(fileId1);
+ sliceInfo.setPartitionPath(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH);
+ clusteringGroup.setSlices(Arrays.asList(sliceInfo));
+ clusteringPlan.setInputGroups(Arrays.asList(clusteringGroup));
+ requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
+ requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
+ HoodieTestTable.of(metaClient)
+ .addCluster(instantTime, requestedReplaceMetadata, Option.empty(), replaceMetadata)
+ .withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
+ }
+
+ public static void createPendingCluster(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) throws Exception {
String fileId1 = "file-1";
String fileId2 = "file-2";
// create replace instant to mark fileId2 as deleted
@@ -190,7 +220,7 @@ public class TestConflictResolutionStrategyUtil {
requestedReplaceMetadata.setClusteringPlan(clusteringPlan);
requestedReplaceMetadata.setVersion(TimelineLayoutVersion.CURR_VERSION);
HoodieTestTable.of(metaClient)
- .addPendingReplace(instantTime, Option.of(requestedReplaceMetadata), Option.empty())
+ .addPendingCluster(instantTime, requestedReplaceMetadata, Option.empty())
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
@@ -261,13 +291,13 @@ public class TestConflictResolutionStrategyUtil {
.withBaseFilesInPartition(HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH, fileId1, fileId2);
}
- public static void createReplaceInflight(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) throws Exception {
+ public static void createClusterInflight(String instantTime, WriteOperationType writeOperationType, HoodieTableMetaClient metaClient) throws Exception {
Option<HoodieCommitMetadata> inflightReplaceMetadata = Option.empty();
if (WriteOperationType.INSERT_OVERWRITE.equals(writeOperationType)) {
inflightReplaceMetadata = Option.of(createReplaceCommitMetadata(WriteOperationType.INSERT_OVERWRITE));
}
HoodieTestTable.of(metaClient)
- .addInflightReplace(instantTime, inflightReplaceMetadata);
+ .addInflightCluster(instantTime, inflightReplaceMetadata);
}
private static HoodieReplaceCommitMetadata createReplaceCommitMetadata(WriteOperationType writeOperationType) {
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestPreferWriterConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestPreferWriterConflictResolutionStrategy.java
index 7cc109bb3bb..4b1440904a1 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestPreferWriterConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestPreferWriterConflictResolutionStrategy.java
@@ -35,14 +35,15 @@ import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCluster;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommit;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommitMetadata;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompaction;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompactionRequested;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createInflightCommit;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplace;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplaceInflight;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplaceRequested;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createClusterInflight;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createClusterRequested;
public class TestPreferWriterConflictResolutionStrategy extends HoodieCommonTestHarness {
@@ -171,8 +172,8 @@ public class TestPreferWriterConflictResolutionStrategy extends HoodieCommonTest
createInflightCommit(currentWriterInstant, metaClient);
// clustering 1 gets scheduled
String newInstantTime = metaClient.createNewInstantTime();
- createReplaceRequested(newInstantTime, metaClient);
- createReplaceInflight(newInstantTime, WriteOperationType.CLUSTER, metaClient);
+ createClusterRequested(newInstantTime, metaClient);
+ createClusterInflight(newInstantTime, WriteOperationType.CLUSTER, metaClient);
Option<HoodieInstant> currentInstant = Option.of(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
PreferWriterConflictResolutionStrategy strategy = new PreferWriterConflictResolutionStrategy();
@@ -202,7 +203,7 @@ public class TestPreferWriterConflictResolutionStrategy extends HoodieCommonTest
Thread.sleep(1000);
// clustering writer starts and complete before ingestion commit.
String replaceWriterInstant = metaClient.createNewInstantTime();
- createReplace(replaceWriterInstant, WriteOperationType.CLUSTER, metaClient);
+ createCluster(replaceWriterInstant, WriteOperationType.CLUSTER, metaClient);
Option<HoodieInstant> currentInstant = Option.of(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
PreferWriterConflictResolutionStrategy strategy = new PreferWriterConflictResolutionStrategy();
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
index 18a1e5212c4..d2ee5d67285 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestSimpleConcurrentFileWritesConflictResolutionStrategy.java
@@ -40,19 +40,18 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCluster;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createClusterRequested;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommit;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCommitMetadata;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompaction;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompactionRequested;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompleteCommit;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompleteCompaction;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createCompleteReplace;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createInflightCommit;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createPendingCompaction;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createPendingReplace;
+import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createPendingCluster;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplace;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplaceInflight;
-import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createReplaceRequested;
import static org.apache.hudi.client.transaction.TestConflictResolutionStrategyUtil.createRequestedCommit;
public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends HoodieCommonTestHarness {
@@ -128,7 +127,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
@Test
public void testConcurrentWritesWithReplaceInflightCommit() throws Exception {
- createReplaceInflight(metaClient.createNewInstantTime(), metaClient);
+ TestConflictResolutionStrategyUtil.createClusterInflight(metaClient.createNewInstantTime(), metaClient);
HoodieActiveTimeline timeline = metaClient.getActiveTimeline();
Option<HoodieInstant> lastSuccessfulInstant = Option.empty();
@@ -139,7 +138,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
// writer 2 starts and finishes
String newInstantTime = metaClient.createNewInstantTime();
- createReplaceInflight(newInstantTime, metaClient);
+ TestConflictResolutionStrategyUtil.createClusterInflight(newInstantTime, metaClient);
SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new SimpleConcurrentFileWritesConflictResolutionStrategy();
HoodieCommitMetadata currentMetadata = createCommitMetadata(currentWriterInstant);
@@ -149,7 +148,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
Collectors.toList());
// writer 1 conflicts with writer 2
- Assertions.assertTrue(candidateInstants.size() == 1);
+ Assertions.assertEquals(1, candidateInstants.size());
ConcurrentOperation thatCommitOperation = new ConcurrentOperation(candidateInstants.get(0), metaClient);
ConcurrentOperation thisCommitOperation = new ConcurrentOperation(currentInstant.get(), currentMetadata);
Assertions.assertTrue(strategy.hasConflict(thisCommitOperation, thatCommitOperation));
@@ -262,7 +261,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
createInflightCommit(currentWriterInstant, metaClient);
// clustering 1 gets scheduled
String newInstantTime = metaClient.createNewInstantTime();
- createReplaceRequested(newInstantTime, metaClient);
+ createClusterRequested(newInstantTime, metaClient);
Option<HoodieInstant> currentInstant = Option.of(new HoodieInstant(State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new SimpleConcurrentFileWritesConflictResolutionStrategy();
@@ -271,7 +270,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
List<HoodieInstant> candidateInstants = strategy.getCandidateInstants(metaClient, currentInstant.get(), lastSuccessfulInstant).collect(
Collectors.toList());
// writer 1 conflicts with scheduled compaction plan 1
- Assertions.assertTrue(candidateInstants.size() == 1);
+ Assertions.assertEquals(1, candidateInstants.size());
ConcurrentOperation thatCommitOperation = new ConcurrentOperation(candidateInstants.get(0), metaClient);
ConcurrentOperation thisCommitOperation = new ConcurrentOperation(currentInstant.get(), currentMetadata);
Assertions.assertTrue(strategy.hasConflict(thisCommitOperation, thatCommitOperation));
@@ -294,7 +293,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
createInflightCommit(currentWriterInstant, metaClient);
// cluster 1 gets scheduled and finishes
String newInstantTime = metaClient.createNewInstantTime();
- createReplace(newInstantTime, WriteOperationType.CLUSTER, metaClient);
+ createCluster(newInstantTime, WriteOperationType.CLUSTER, metaClient);
Option<HoodieInstant> currentInstant = Option.of(new HoodieInstant(State.INFLIGHT, HoodieTimeline.COMMIT_ACTION, currentWriterInstant));
SimpleConcurrentFileWritesConflictResolutionStrategy strategy = new SimpleConcurrentFileWritesConflictResolutionStrategy();
@@ -352,7 +351,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
public void testConcurrentWritesWithPendingInstants() throws Exception {
// step1: create a pending replace/commit/compact instant: C1,C11,C12
String newInstantTimeC1 = metaClient.createNewInstantTime();
- createPendingReplace(newInstantTimeC1, WriteOperationType.CLUSTER, metaClient);
+ createPendingCluster(newInstantTimeC1, WriteOperationType.CLUSTER, metaClient);
String newCompactionInstantTimeC11 = metaClient.createNewInstantTime();
createPendingCompaction(newCompactionInstantTimeC11, metaClient);
@@ -375,7 +374,7 @@ public class TestSimpleConcurrentFileWritesConflictResolutionStrategy extends Ho
Set<String> pendingInstant = TransactionUtils.getInflightAndRequestedInstants(metaClient);
pendingInstant.remove(currentWriterInstant);
// step5: finished pending cluster/compaction/commit operation
- createCompleteReplace(newInstantTimeC1, WriteOperationType.CLUSTER, metaClient);
+ createCluster(newInstantTimeC1, WriteOperationType.CLUSTER, metaClient);
createCompleteCompaction(newCompactionInstantTimeC11, metaClient);
createCompleteCommit(newCommitInstantTimeC12, metaClient);
createCompleteCommit(commitC4, metaClient);
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/common/testutils/HoodieMetadataTestTable.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/common/testutils/HoodieMetadataTestTable.java
index 57b987d46b1..54ed0ff9063 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/common/testutils/HoodieMetadataTestTable.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/common/testutils/HoodieMetadataTestTable.java
@@ -181,6 +181,16 @@ public class HoodieMetadataTestTable extends HoodieTestTable {
return this;
}
+ @Override
+ public HoodieTestTable addCluster(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata, Option<HoodieCommitMetadata> inflightReplaceMetadata,
+ HoodieReplaceCommitMetadata completeReplaceMetadata) throws Exception {
+ super.addCluster(instantTime, requestedReplaceMetadata, inflightReplaceMetadata, completeReplaceMetadata);
+ if (writer != null) {
+ writer.updateFromWriteStatuses(completeReplaceMetadata, context.get().emptyHoodieData(), instantTime);
+ }
+ return this;
+ }
+
@Override
public void close() throws Exception {
if (writer != null) {
diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
index e9bf583b431..a8eed1e90e6 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
@@ -32,6 +32,7 @@ import org.apache.hudi.common.model.TableServiceType;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieClusteringException;
@@ -115,7 +116,7 @@ public class HoodieFlinkTableServiceClient<T> extends BaseHoodieTableServiceClie
String clusteringCommitTime,
Option<HoodieData<WriteStatus>> writeStatuses) {
this.context.setJobStatus(this.getClass().getSimpleName(), "Collect clustering write status and commit clustering");
- HoodieInstant clusteringInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringCommitTime);
+ HoodieInstant clusteringInstant = ClusteringUtils.getInflightClusteringInstant(clusteringCommitTime, table.getActiveTimeline()).get();
List<HoodieWriteStat> writeStats = metadata.getPartitionToWriteStats().entrySet().stream().flatMap(e ->
e.getValue().stream()).collect(Collectors.toList());
if (writeStats.stream().mapToLong(HoodieWriteStat::getTotalWriteErrors).sum() > 0) {
@@ -137,10 +138,11 @@ public class HoodieFlinkTableServiceClient<T> extends BaseHoodieTableServiceClie
writeTableMetadata(table, clusteringCommitTime, metadata, writeStatuses.orElseGet(context::emptyHoodieData));
LOG.info("Committing Clustering {} finished with result {}.", clusteringCommitTime, metadata);
- table.getActiveTimeline().transitionReplaceInflightToComplete(
+ ClusteringUtils.transitionClusteringOrReplaceInflightToComplete(
false,
- HoodieTimeline.getReplaceCommitInflightInstant(clusteringCommitTime),
- serializeCommitMetadata(metadata));
+ clusteringInstant,
+ serializeCommitMetadata(metadata),
+ table.getActiveTimeline());
} catch (IOException e) {
throw new HoodieClusteringException(
"Failed to commit " + table.getMetaClient().getBasePath() + " at time " + clusteringCommitTime, e);
@@ -154,7 +156,7 @@ public class HoodieFlinkTableServiceClient<T> extends BaseHoodieTableServiceClie
long durationInMs = metrics.getDurationInMs(clusteringTimer.stop());
try {
metrics.updateCommitMetrics(HoodieActiveTimeline.parseDateFromInstantTime(clusteringCommitTime).getTime(),
- durationInMs, metadata, HoodieActiveTimeline.REPLACE_COMMIT_ACTION);
+ durationInMs, metadata, HoodieActiveTimeline.CLUSTERING_ACTION);
} catch (ParseException e) {
throw new HoodieCommitException("Commit time is not of valid format. Failed to commit compaction "
+ config.getBasePath() + " at time " + clusteringCommitTime, e);
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
index 29be463ae5b..67030d14ff6 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaTableServiceClient.java
@@ -25,7 +25,6 @@ import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.data.HoodieListData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -51,7 +50,7 @@ public class HoodieJavaTableServiceClient<T> extends BaseHoodieTableServiceClien
protected void validateClusteringCommit(HoodieWriteMetadata<List<WriteStatus>> clusteringMetadata, String clusteringCommitTime, HoodieTable table) {
if (clusteringMetadata.getWriteStatuses().isEmpty()) {
HoodieClusteringPlan clusteringPlan = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(clusteringCommitTime))
+ table.getMetaClient(), ClusteringUtils.getInflightClusteringInstant(clusteringCommitTime, table.getActiveTimeline()).get())
.map(Pair::getRight).orElseThrow(() -> new HoodieClusteringException(
"Unable to read clustering plan for instant: " + clusteringCommitTime));
throw new HoodieClusteringException("Clustering plan produced 0 WriteStatus for " + clusteringCommitTime
diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/cluster/JavaExecuteClusteringCommitActionExecutor.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/cluster/JavaExecuteClusteringCommitActionExecutor.java
index 717f8446a2b..9d61eb8359d 100644
--- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/cluster/JavaExecuteClusteringCommitActionExecutor.java
+++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/cluster/JavaExecuteClusteringCommitActionExecutor.java
@@ -46,7 +46,7 @@ public class JavaExecuteClusteringCommitActionExecutor<T>
String instantTime) {
super(context, config, table, instantTime, WriteOperationType.CLUSTER);
this.clusteringPlan = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime))
+ table.getMetaClient(), ClusteringUtils.getRequestedClusteringInstant(instantTime, table.getActiveTimeline()).get())
.map(Pair::getRight).orElseThrow(() -> new HoodieClusteringException(
"Unable to read clustering plan for instant: " + instantTime));
}
@@ -60,6 +60,6 @@ public class JavaExecuteClusteringCommitActionExecutor<T>
@Override
protected String getCommitActionType() {
- return HoodieTimeline.REPLACE_COMMIT_ACTION;
+ return HoodieTimeline.CLUSTERING_ACTION;
}
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDTableServiceClient.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDTableServiceClient.java
index 98914be7496..34f4fdde13c 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDTableServiceClient.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/SparkRDDTableServiceClient.java
@@ -25,7 +25,6 @@ import org.apache.hudi.client.utils.SparkReleaseResources;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.HoodieRecord;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -50,7 +49,7 @@ public class SparkRDDTableServiceClient<T> extends BaseHoodieTableServiceClient<
protected void validateClusteringCommit(HoodieWriteMetadata<JavaRDD<WriteStatus>> clusteringMetadata, String clusteringCommitTime, HoodieTable table) {
if (clusteringMetadata.getWriteStatuses().isEmpty()) {
HoodieClusteringPlan clusteringPlan = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(clusteringCommitTime))
+ table.getMetaClient(), ClusteringUtils.getInflightClusteringInstant(clusteringCommitTime, table.getActiveTimeline()).get())
.map(Pair::getRight).orElseThrow(() -> new HoodieClusteringException(
"Unable to read clustering plan for instant: " + clusteringCommitTime));
throw new HoodieClusteringException("Clustering plan produced 0 WriteStatus for " + clusteringCommitTime
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java
index b49599b4c51..1bd74393540 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/index/bucket/HoodieSparkConsistentBucketIndex.java
@@ -26,7 +26,6 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.ConsistentHashingNode;
import org.apache.hudi.common.model.HoodieConsistentHashingMetadata;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ValidationUtils;
@@ -66,13 +65,13 @@ public class HoodieSparkConsistentBucketIndex extends HoodieConsistentBucketInde
throws HoodieIndexException {
HoodieInstant instant = hoodieTable.getMetaClient().getActiveTimeline().findInstantsAfterOrEquals(instantTime, 1).firstInstant().get();
ValidationUtils.checkState(instant.getTimestamp().equals(instantTime), "Cannot get the same instant, instantTime: " + instantTime);
- if (!instant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)) {
+ if (!ClusteringUtils.isClusteringOrReplaceCommitAction(instant.getAction())) {
return writeStatuses;
}
// Double-check if it is a clustering operation by trying to obtain the clustering plan
Option<Pair<HoodieInstant, HoodieClusteringPlan>> instantPlanPair =
- ClusteringUtils.getClusteringPlan(hoodieTable.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime));
+ ClusteringUtils.getClusteringPlan(hoodieTable.getMetaClient(), instant);
if (!instantPlanPair.isPresent()) {
return writeStatuses;
}
diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/cluster/SparkExecuteClusteringCommitActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/cluster/SparkExecuteClusteringCommitActionExecutor.java
index ee61b57a109..288b0d24de9 100644
--- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/cluster/SparkExecuteClusteringCommitActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/cluster/SparkExecuteClusteringCommitActionExecutor.java
@@ -42,7 +42,7 @@ public class SparkExecuteClusteringCommitActionExecutor<T>
String instantTime) {
super(context, config, table, instantTime, WriteOperationType.CLUSTER);
this.clusteringPlan = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), HoodieTimeline.getReplaceCommitRequestedInstant(instantTime))
+ table.getMetaClient(), ClusteringUtils.getRequestedClusteringInstant(instantTime, table.getActiveTimeline()).get())
.map(Pair::getRight).orElseThrow(() -> new HoodieClusteringException(
"Unable to read clustering plan for instant: " + instantTime));
}
@@ -54,6 +54,6 @@ public class SparkExecuteClusteringCommitActionExecutor<T>
@Override
protected String getCommitActionType() {
- return HoodieTimeline.REPLACE_COMMIT_ACTION;
+ return HoodieTimeline.CLUSTERING_ACTION;
}
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 04feef2926a..81c508f5d3c 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -122,6 +122,7 @@ import static org.apache.hudi.common.config.LockConfiguration.FILESYSTEM_LOCK_PA
import static org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy.EAGER;
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.INFLIGHT;
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion.VERSION_0;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.DEFAULT_FIRST_PARTITION_PATH;
@@ -710,7 +711,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
List<List<FileSlice>> firstInsertFileSlicesList = table.getFileSystemView().getAllFileGroups(testPartitionPath)
.map(fileGroup -> fileGroup.getAllFileSlices().collect(Collectors.toList())).collect(Collectors.toList());
List<FileSlice>[] fileSlices = (List<FileSlice>[]) firstInsertFileSlicesList.toArray(new List[0]);
- createRequestedReplaceInstant(this.metaClient, "002", fileSlices);
+ createRequestedClusterInstant(this.metaClient, "002", fileSlices);
// 3. insert one record with no updating reject exception, and not merge the small file, just generate a new file group
insertBatchRecords(client, "003", 1, 1, 1, SparkRDDWriteClient::upsert);
@@ -1087,7 +1088,7 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setClusteringPlan(clusteringPlan).setOperationType(WriteOperationType.CLUSTER.name()).build();
- FileCreateUtils.createRequestedReplaceCommit(metaClient.getBasePath().toString(), pendingClusteringInstant.getTimestamp(), Option.of(requestedReplaceMetadata));
+ FileCreateUtils.createRequestedClusterCommit(metaClient.getBasePath().toString(), pendingClusteringInstant.getTimestamp(), requestedReplaceMetadata);
// trigger clustering again. no new rollback instants should be generated.
try {
@@ -1630,14 +1631,14 @@ public class TestHoodieClientOnCopyOnWriteStorage extends HoodieClientTestBase {
assertThrows(HoodieWriteConflictException.class, () -> client.commit(inflightCommit, ingestionResult));
}
- protected HoodieInstant createRequestedReplaceInstant(HoodieTableMetaClient metaClient, String clusterTime, List<FileSlice>[] fileSlices) throws IOException {
+ protected HoodieInstant createRequestedClusterInstant(HoodieTableMetaClient metaClient, String clusterTime, List<FileSlice>[] fileSlices) throws IOException {
HoodieClusteringPlan clusteringPlan =
ClusteringUtils.createClusteringPlan(EXECUTION_STRATEGY_CLASS_NAME.defaultValue(), STRATEGY_PARAMS, fileSlices, Collections.emptyMap());
- HoodieInstant clusteringInstant = new HoodieInstant(REQUESTED, REPLACE_COMMIT_ACTION, clusterTime);
+ HoodieInstant clusteringInstant = new HoodieInstant(REQUESTED, CLUSTERING_ACTION, clusterTime);
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setClusteringPlan(clusteringPlan).setOperationType(WriteOperationType.CLUSTER.name()).build();
- metaClient.getActiveTimeline().saveToPendingReplaceCommit(clusteringInstant, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
+ metaClient.getActiveTimeline().saveToPendingClusterCommit(clusteringInstant, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
return clusteringInstant;
}
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
index 400c9ac43c4..cb4a871cab1 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
@@ -811,7 +811,7 @@ public class TestHoodieTimelineArchiver extends HoodieSparkClientTestHarness {
@ValueSource(booleans = {true, false})
public void testPendingClusteringWillBlockArchival(boolean enableMetadata) throws Exception {
HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(enableMetadata, 4, 5, 2);
- HoodieTestDataGenerator.createPendingReplaceFile(basePath, "00000000", storageConf);
+ HoodieTestDataGenerator.createPendingClusterFile(basePath, "00000000", storageConf);
for (int i = 1; i < 8; i++) {
testTable.doWriteOperation("0000000" + i, WriteOperationType.UPSERT, Arrays.asList("p1", "p2"), Arrays.asList("p1", "p2"), 2);
// archival
@@ -1604,7 +1604,7 @@ public class TestHoodieTimelineArchiver extends HoodieSparkClientTestHarness {
public void testPendingClusteringAfterArchiveCommit(boolean enableMetadata) throws Exception {
HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(enableMetadata, 4, 5, 2);
// timeline:0000000(completed)->00000001(completed)->00000002(replace&inflight)->00000003(completed)->...->00000007(completed)
- HoodieTestDataGenerator.createPendingReplaceFile(basePath, "00000002", storageConf);
+ HoodieTestDataGenerator.createPendingClusterFile(basePath, "00000002", storageConf);
for (int i = 1; i < 8; i++) {
if (i != 2) {
testTable.doWriteOperation("0000000" + i, WriteOperationType.CLUSTER, Arrays.asList("p1", "p2"), Arrays.asList("p1", "p2"), 2);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java
index fd74d9f652b..16e3ba650c2 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestCleaner.java
@@ -645,7 +645,7 @@ public class TestCleaner extends HoodieCleanerTestBase {
String file2P0C1 = partitionAndFileId002.get(p0);
Pair<HoodieRequestedReplaceMetadata, HoodieReplaceCommitMetadata> replaceMetadata =
generateReplaceCommitMetadata("00000000000002", p0, file1P0C0, file2P0C1);
- testTable.addReplaceCommit("00000000000002", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000002", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// run cleaner
List<HoodieCleanStat> hoodieCleanStatsTwo = runCleanerWithInstantFormat(config, true);
@@ -659,7 +659,7 @@ public class TestCleaner extends HoodieCleanerTestBase {
Map<String, String> partitionAndFileId003 = testTable.forReplaceCommit("00000000000003").getFileIdsWithBaseFilesInPartitions(p1);
String file3P1C2 = partitionAndFileId003.get(p1);
replaceMetadata = generateReplaceCommitMetadata("00000000000003", p1, file1P1C0, file3P1C2);
- testTable.addReplaceCommit("00000000000003", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000003", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// run cleaner
List<HoodieCleanStat> hoodieCleanStatsThree = runCleanerWithInstantFormat(config, true);
@@ -674,7 +674,7 @@ public class TestCleaner extends HoodieCleanerTestBase {
Map<String, String> partitionAndFileId004 = testTable.forReplaceCommit("00000000000004").getFileIdsWithBaseFilesInPartitions(p0);
String file4P0C3 = partitionAndFileId004.get(p0);
replaceMetadata = generateReplaceCommitMetadata("00000000000004", p0, file2P0C1, file4P0C3);
- testTable.addReplaceCommit("00000000000004", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000004", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// run cleaner
List<HoodieCleanStat> hoodieCleanStatsFour = runCleaner(config, 5, true);
@@ -690,7 +690,7 @@ public class TestCleaner extends HoodieCleanerTestBase {
Map<String, String> partitionAndFileId005 = testTable.forReplaceCommit("00000000000006").getFileIdsWithBaseFilesInPartitions(p1);
String file4P1C4 = partitionAndFileId005.get(p1);
replaceMetadata = generateReplaceCommitMetadata("00000000000006", p0, file3P1C2, file4P1C4);
- testTable.addReplaceCommit("00000000000006", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000006", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
List<HoodieCleanStat> hoodieCleanStatsFive = runCleaner(config, 7, true);
assertTrue(testTable.baseFileExists(p0, "00000000000004", file4P0C3));
@@ -1066,21 +1066,21 @@ public class TestCleaner extends HoodieCleanerTestBase {
String file2P0C1 = partitionAndFileId002.get(p0);
Pair<HoodieRequestedReplaceMetadata, HoodieReplaceCommitMetadata> replaceMetadata =
generateReplaceCommitMetadata("00000000000002", p0, file1P0C0, file2P0C1);
- testTable.addReplaceCommit("00000000000002", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000002", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// make next replacecommit, with 1 clustering operation. Replace data in p1. No change to p0
// notice that clustering generates empty inflight commit files
Map<String, String> partitionAndFileId003 = testTable.forReplaceCommit("00000000000003").getFileIdsWithBaseFilesInPartitions(p1);
String file3P1C2 = partitionAndFileId003.get(p1);
replaceMetadata = generateReplaceCommitMetadata("00000000000003", p1, file1P1C0, file3P1C2);
- testTable.addReplaceCommit("00000000000003", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000003", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// make next replacecommit, with 1 clustering operation. Replace data in p0 again
// notice that clustering generates empty inflight commit files
Map<String, String> partitionAndFileId004 = testTable.forReplaceCommit("00000000000004").getFileIdsWithBaseFilesInPartitions(p0);
String file4P0C3 = partitionAndFileId004.get(p0);
replaceMetadata = generateReplaceCommitMetadata("00000000000004", p0, file2P0C1, file4P0C3);
- testTable.addReplaceCommit("00000000000004", Option.of(replaceMetadata.getKey()), Option.empty(), replaceMetadata.getValue());
+ testTable.addCluster("00000000000004", replaceMetadata.getKey(), Option.empty(), replaceMetadata.getValue());
// run cleaner with failures
List<HoodieCleanStat> hoodieCleanStats = runCleaner(config, true, simulateMetadataFailure, 5, true);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
index 5c35685b7bd..f4f45990689 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestUpsertPartitioner.java
@@ -370,7 +370,7 @@ public class TestUpsertPartitioner extends HoodieClientTestBase {
// create requested replace commit
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setClusteringPlan(clusteringPlan).setOperationType(WriteOperationType.CLUSTER.name()).build();
- FileCreateUtils.createRequestedReplaceCommit(basePath,"002", Option.of(requestedReplaceMetadata));
+ FileCreateUtils.createRequestedClusterCommit(basePath,"002", requestedReplaceMetadata);
// create file slice 003
FileCreateUtils.createBaseFile(basePath, testPartitionPath, "003", "3", 1);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
index 4fd43a22057..2fb21b00cc0 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/rollback/TestCopyOnWriteRollbackActionExecutor.java
@@ -474,7 +474,7 @@ public class TestCopyOnWriteRollbackActionExecutor extends HoodieClientRollbackT
assertThrows(HoodieRollbackException.class, copyOnWriteRollbackActionExecutor::execute);
// Schedule rollback for incomplete clustering instant.
- needRollBackInstant = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstant1);
+ needRollBackInstant = new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, clusteringInstant1);
rollbackInstant = writeClient.createNewInstantTime();
copyOnWriteRollbackPlanActionExecutor =
new BaseRollbackPlanActionExecutor(context, table.getConfig(), table, rollbackInstant, needRollBackInstant, false,
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieTableServiceManagerConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieTableServiceManagerConfig.java
index 0da588fb54f..8e59ce1efa0 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieTableServiceManagerConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieTableServiceManagerConfig.java
@@ -50,7 +50,7 @@ public class HoodieTableServiceManagerConfig extends HoodieConfig {
public static final ConfigProperty<String> TABLE_SERVICE_MANAGER_ACTIONS = ConfigProperty
.key(TABLE_SERVICE_MANAGER_PREFIX + ".actions")
- .noDefaultValue()
+ .defaultValue("")
.markAdvanced()
.sinceVersion("0.13.0")
.withDocumentation("The actions deployed on table service manager, such as compaction or clean.");
@@ -171,7 +171,11 @@ public class HoodieTableServiceManagerConfig extends HoodieConfig {
}
public boolean isEnabledAndActionSupported(ActionType actionType) {
- return isTableServiceManagerEnabled() && getTableServiceManagerActions().contains(actionType.name());
+ boolean isActionSupported = getTableServiceManagerActions().contains(actionType.name());
+ if (actionType.equals(ActionType.clustering)) {
+ isActionSupported = isActionSupported || getTableServiceManagerActions().contains(ActionType.replacecommit.name());
+ }
+ return isTableServiceManagerEnabled() && isActionSupported;
}
public static class Builder {
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/ActionType.java b/hudi-common/src/main/java/org/apache/hudi/common/model/ActionType.java
index eb7e578522e..ea0ab7f3333 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/model/ActionType.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/model/ActionType.java
@@ -22,5 +22,5 @@ package org.apache.hudi.common.model;
* The supported action types.
*/
public enum ActionType {
- commit, savepoint, compaction, clean, rollback, replacecommit, deltacommit, logcompaction
+ commit, savepoint, compaction, clean, rollback, replacecommit, deltacommit, logcompaction, clustering
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/model/TableServiceType.java b/hudi-common/src/main/java/org/apache/hudi/common/model/TableServiceType.java
index 2aa5d08f06a..4aefc370eef 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/model/TableServiceType.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/model/TableServiceType.java
@@ -36,7 +36,7 @@ public enum TableServiceType {
case CLEAN:
return HoodieTimeline.CLEAN_ACTION;
case CLUSTER:
- return HoodieTimeline.REPLACE_COMMIT_ACTION;
+ return HoodieTimeline.CLUSTERING_ACTION;
case LOG_COMPACT:
return HoodieTimeline.LOG_COMPACTION_ACTION;
default:
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java
index 28de790050a..3b15373044a 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/cdc/HoodieCDCExtractor.java
@@ -60,6 +60,7 @@ import static org.apache.hudi.common.table.cdc.HoodieCDCInferenceCase.BASE_FILE_
import static org.apache.hudi.common.table.cdc.HoodieCDCInferenceCase.BASE_FILE_INSERT;
import static org.apache.hudi.common.table.cdc.HoodieCDCInferenceCase.LOG_FILE;
import static org.apache.hudi.common.table.cdc.HoodieCDCInferenceCase.REPLACE_COMMIT;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
@@ -214,7 +215,7 @@ public class HoodieCDCExtractor {
*/
private void initInstantAndCommitMetadata() {
try {
- Set<String> requiredActions = new HashSet<>(Arrays.asList(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION));
+ Set<String> requiredActions = new HashSet<>(Arrays.asList(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION));
HoodieActiveTimeline activeTimeLine = metaClient.getActiveTimeline();
this.commits = activeTimeLine.getInstantsAsStream()
.filter(instant ->
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
index 935bcd91e2a..d77af71d0f4 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java
@@ -74,7 +74,8 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
ROLLBACK_EXTENSION, REQUESTED_ROLLBACK_EXTENSION, INFLIGHT_ROLLBACK_EXTENSION,
REQUESTED_REPLACE_COMMIT_EXTENSION, INFLIGHT_REPLACE_COMMIT_EXTENSION, REPLACE_COMMIT_EXTENSION,
REQUESTED_INDEX_COMMIT_EXTENSION, INFLIGHT_INDEX_COMMIT_EXTENSION, INDEX_COMMIT_EXTENSION,
- REQUESTED_SAVE_SCHEMA_ACTION_EXTENSION, INFLIGHT_SAVE_SCHEMA_ACTION_EXTENSION, SAVE_SCHEMA_ACTION_EXTENSION));
+ REQUESTED_SAVE_SCHEMA_ACTION_EXTENSION, INFLIGHT_SAVE_SCHEMA_ACTION_EXTENSION, SAVE_SCHEMA_ACTION_EXTENSION,
+ REQUESTED_CLUSTERING_COMMIT_EXTENSION, INFLIGHT_CLUSTERING_COMMIT_EXTENSION));
public static final Set<String> NOT_PARSABLE_TIMESTAMPS = new HashSet<String>(3) {{
add(HoodieTimeline.INIT_INSTANT_TS);
@@ -223,7 +224,7 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
createFileInMetaPath(instant.getFileName(), Option.empty(), false);
}
- public void createRequestedReplaceCommit(String instantTime, String actionType) {
+ public void createRequestedCommitWithReplaceMetadata(String instantTime, String actionType) {
try {
HoodieInstant instant = new HoodieInstant(State.REQUESTED, actionType, instantTime);
LOG.info("Creating a new instant " + instant);
@@ -628,6 +629,22 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
return inflightInstant;
}
+ /**
+ * Transition cluster requested file to cluster inflight.
+ *
+ * @param requestedInstant Requested instant
+ * @param data Extra Metadata
+ * @return inflight instant
+ */
+ public HoodieInstant transitionClusterRequestedToInflight(HoodieInstant requestedInstant, Option<byte[]> data) {
+ ValidationUtils.checkArgument(requestedInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION));
+ ValidationUtils.checkArgument(requestedInstant.isRequested());
+ HoodieInstant inflightInstant = new HoodieInstant(State.INFLIGHT, CLUSTERING_ACTION, requestedInstant.getTimestamp());
+ // Then write to timeline
+ transitionPendingState(requestedInstant, inflightInstant, data);
+ return inflightInstant;
+ }
+
/**
* Transition replace inflight to Committed.
*
@@ -646,6 +663,24 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
return commitInstant;
}
+ /**
+ * Transition cluster inflight to replace committed.
+ *
+ * @param shouldLock Whether to hold the lock when performing transition
+ * @param inflightInstant Inflight instant
+ * @param data Extra Metadata
+ * @return commit instant
+ */
+ public HoodieInstant transitionClusterInflightToComplete(boolean shouldLock,
+ HoodieInstant inflightInstant, Option<byte[]> data) {
+ ValidationUtils.checkArgument(inflightInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION));
+ ValidationUtils.checkArgument(inflightInstant.isInflight());
+ HoodieInstant commitInstant = new HoodieInstant(State.COMPLETED, REPLACE_COMMIT_ACTION, inflightInstant.getTimestamp());
+ // Then write to timeline
+ transitionStateToComplete(shouldLock, inflightInstant, commitInstant, data);
+ return commitInstant;
+ }
+
private void transitionPendingState(HoodieInstant fromInstant, HoodieInstant toInstant, Option<byte[]> data) {
transitionPendingState(fromInstant, toInstant, data, false);
}
@@ -794,6 +829,14 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline {
createFileInMetaPath(instant.getFileName(), content, false);
}
+ /**
+ * Saves content for requested CLUSTER instant.
+ */
+ public void saveToPendingClusterCommit(HoodieInstant instant, Option<byte[]> content) {
+ ValidationUtils.checkArgument(instant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION));
+ createFileInMetaPath(instant.getFileName(), content, false);
+ }
+
public void saveToCleanRequested(HoodieInstant instant, Option<byte[]> content) {
ValidationUtils.checkArgument(instant.getAction().equals(HoodieTimeline.CLEAN_ACTION));
ValidationUtils.checkArgument(instant.getState().equals(State.REQUESTED));
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
index 1e6050d60f6..17a308c9f12 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieArchivedTimeline.java
@@ -293,7 +293,7 @@ public class HoodieArchivedTimeline extends HoodieDefaultTimeline {
@Override
public HoodieDefaultTimeline getWriteTimeline() {
// filter in-memory instants
- Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
+ Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION);
return new HoodieDefaultTimeline(getInstantsAsStream().filter(i ->
readCommits.containsKey(i.getTimestamp()))
.filter(s -> validActions.contains(s.getAction())), details);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
index a497eb17d86..bc6dd6094c5 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
@@ -63,8 +63,8 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
private List<HoodieInstant> instants;
// for efficient #contains queries.
private transient volatile Set<String> instantTimeSet;
- // for efficient #isPendingClusterInstant queries
- private transient volatile Set<String> pendingReplaceClusteringInstants;
+ // for efficient #isPendingClusteringInstant queries
+ private transient volatile Set<String> pendingClusteringInstants;
// for efficient #isBeforeTimelineStarts check.
private transient volatile Option<HoodieInstant> firstNonSavepointCommit;
private String timelineHash;
@@ -160,7 +160,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public HoodieDefaultTimeline getWriteTimeline() {
- Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION);
+ Set<String> validActions = CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION);
return new HoodieDefaultTimeline(getInstantsAsStream().filter(s -> validActions.contains(s.getAction())), details);
}
@@ -186,6 +186,27 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
s -> s.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION) && !s.isCompleted()), details);
}
+ @Override
+ public HoodieTimeline filterPendingClusteringTimeline() {
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(
+ s -> s.getAction().equals(HoodieTimeline.CLUSTERING_ACTION) && !s.isCompleted()), details);
+ }
+
+ @Override
+ public HoodieTimeline filterPendingReplaceOrClusteringTimeline() {
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(
+ s -> (s.getAction().equals(HoodieTimeline.CLUSTERING_ACTION) || s.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION))
+ && !s.isCompleted()), details);
+ }
+
+ @Override
+ public HoodieTimeline filterPendingReplaceClusteringAndCompactionTimeline() {
+ return new HoodieDefaultTimeline(getInstantsAsStream().filter(
+ s -> !s.isCompleted() && (s.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)
+ || s.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION)
+ || s.getAction().equals(HoodieTimeline.COMPACTION_ACTION))), details);
+ }
+
@Override
public HoodieTimeline filterPendingRollbackTimeline() {
return new HoodieDefaultTimeline(getInstantsAsStream().filter(
@@ -309,14 +330,14 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
* Get all instants (commits, delta commits) that produce new data, in the active timeline.
*/
public HoodieTimeline getCommitsTimeline() {
- return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION));
+ return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION));
}
/**
* Get all instants (commits, delta commits, replace, compaction) that produce new data or merge file, in the active timeline.
*/
public HoodieTimeline getCommitsAndCompactionTimeline() {
- return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, COMPACTION_ACTION));
+ return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION, COMPACTION_ACTION));
}
/**
@@ -325,8 +346,8 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
*/
public HoodieTimeline getAllCommitsTimeline() {
return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION,
- CLEAN_ACTION, COMPACTION_ACTION, SAVEPOINT_ACTION, ROLLBACK_ACTION, REPLACE_COMMIT_ACTION, INDEXING_ACTION,
- LOG_COMPACTION_ACTION));
+ CLEAN_ACTION, COMPACTION_ACTION, SAVEPOINT_ACTION, ROLLBACK_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION,
+ INDEXING_ACTION, LOG_COMPACTION_ACTION));
}
/**
@@ -334,7 +355,7 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
*/
public HoodieTimeline getCommitAndReplaceTimeline() {
//TODO: Make sure this change does not break existing functionality.
- return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION));
+ return getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION));
}
/**
@@ -519,7 +540,8 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
@Override
public Option<HoodieInstant> getLastClusteringInstant() {
- return Option.fromJavaOptional(getCommitsTimeline().filter(s -> s.getAction().equalsIgnoreCase(HoodieTimeline.REPLACE_COMMIT_ACTION))
+ return Option.fromJavaOptional(getCommitsTimeline().filter(s -> s.getAction().equalsIgnoreCase(HoodieTimeline.REPLACE_COMMIT_ACTION)
+ || s.getAction().equalsIgnoreCase(HoodieTimeline.CLUSTERING_ACTION))
.getReverseOrderedInstants()
.filter(i -> ClusteringUtils.isClusteringInstant(this, i))
.findFirst());
@@ -536,19 +558,19 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
}
private Option<HoodieInstant> getLastOrFirstPendingClusterInstant(boolean isLast) {
- HoodieTimeline replaceTimeline = filterPendingReplaceTimeline();
- Stream<HoodieInstant> replaceStream;
+ HoodieTimeline pendingClusteringTimeline = filterPendingReplaceOrClusteringTimeline();
+ Stream<HoodieInstant> clusterStream;
if (isLast) {
- replaceStream = replaceTimeline.getReverseOrderedInstants();
+ clusterStream = pendingClusteringTimeline.getReverseOrderedInstants();
} else {
- replaceStream = replaceTimeline.getInstantsAsStream();
+ clusterStream = pendingClusteringTimeline.getInstantsAsStream();
}
- return Option.fromJavaOptional(replaceStream
+ return Option.fromJavaOptional(clusterStream
.filter(i -> ClusteringUtils.isClusteringInstant(this, i)).findFirst());
}
@Override
- public boolean isPendingClusterInstant(String instantTime) {
+ public boolean isPendingClusteringInstant(String instantTime) {
return getOrCreatePendingClusteringInstantSet().contains(instantTime);
}
@@ -579,24 +601,24 @@ public class HoodieDefaultTimeline implements HoodieTimeline {
}
private Set<String> getOrCreatePendingClusteringInstantSet() {
- if (this.pendingReplaceClusteringInstants == null) {
+ if (this.pendingClusteringInstants == null) {
synchronized (this) {
- if (this.pendingReplaceClusteringInstants == null) {
- List<HoodieInstant> pendingReplaceInstants = getCommitsTimeline().filterPendingReplaceTimeline().getInstants();
+ if (this.pendingClusteringInstants == null) {
+ List<HoodieInstant> pendingClusterInstants = getCommitsTimeline().filterPendingReplaceOrClusteringTimeline().getInstants();
// Validate that there are no instants with same timestamp
- pendingReplaceInstants.stream().collect(Collectors.groupingBy(HoodieInstant::getTimestamp)).forEach((timestamp, instants) -> {
+ pendingClusterInstants.stream().collect(Collectors.groupingBy(HoodieInstant::getTimestamp)).forEach((timestamp, instants) -> {
if (instants.size() > 1) {
throw new IllegalStateException("Multiple instants with same timestamp: " + timestamp + " instants: " + instants);
}
});
// Filter replace commits down to those that are due to clustering
- this.pendingReplaceClusteringInstants = pendingReplaceInstants.stream()
+ this.pendingClusteringInstants = pendingClusterInstants.stream()
.filter(instant -> ClusteringUtils.isClusteringInstant(this, instant))
.map(HoodieInstant::getTimestamp).collect(Collectors.toSet());
}
}
}
- return this.pendingReplaceClusteringInstants;
+ return this.pendingClusteringInstants;
}
/**
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
index 8ec5b77398a..05d34241ec1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieInstant.java
@@ -191,6 +191,7 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> {
Map<String, String> comparableMap = new HashMap<>();
comparableMap.put(HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.COMMIT_ACTION);
comparableMap.put(HoodieTimeline.LOG_COMPACTION_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION);
+ comparableMap.put(HoodieTimeline.CLUSTERING_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION);
return comparableMap;
}
@@ -245,6 +246,12 @@ public class HoodieInstant implements Serializable, Comparable<HoodieInstant> {
} else if (isRequested()) {
return HoodieTimeline.makeRequestedReplaceFileName(timestamp);
}
+ } else if (HoodieTimeline.CLUSTERING_ACTION.equals(action)) {
+ if (isInflight()) {
+ return HoodieTimeline.makeInflightClusteringFileName(timestamp);
+ } else if (isRequested()) {
+ return HoodieTimeline.makeRequestedClusteringFileName(timestamp);
+ }
} else if (HoodieTimeline.INDEXING_ACTION.equals(action)) {
if (isInflight()) {
return HoodieTimeline.makeInflightIndexFileName(timestamp);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
index 9553fc3d323..4ec700fe577 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -50,6 +50,7 @@ public interface HoodieTimeline extends Serializable {
String ROLLBACK_ACTION = "rollback";
String SAVEPOINT_ACTION = "savepoint";
String REPLACE_COMMIT_ACTION = "replacecommit";
+ String CLUSTERING_ACTION = "clustering";
String INFLIGHT_EXTENSION = ".inflight";
// With Async Compaction, compaction instant can be in 3 states :
// (compaction-requested), (compaction-inflight), (completed)
@@ -63,7 +64,7 @@ public interface HoodieTimeline extends Serializable {
String SCHEMA_COMMIT_ACTION = "schemacommit";
String[] VALID_ACTIONS_IN_TIMELINE = {COMMIT_ACTION, DELTA_COMMIT_ACTION,
CLEAN_ACTION, SAVEPOINT_ACTION, RESTORE_ACTION, ROLLBACK_ACTION,
- COMPACTION_ACTION, REPLACE_COMMIT_ACTION, INDEXING_ACTION};
+ COMPACTION_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION, INDEXING_ACTION};
String COMMIT_EXTENSION = "." + COMMIT_ACTION;
String DELTA_COMMIT_EXTENSION = "." + DELTA_COMMIT_ACTION;
@@ -90,6 +91,8 @@ public interface HoodieTimeline extends Serializable {
String INFLIGHT_REPLACE_COMMIT_EXTENSION = "." + REPLACE_COMMIT_ACTION + INFLIGHT_EXTENSION;
String REQUESTED_REPLACE_COMMIT_EXTENSION = "." + REPLACE_COMMIT_ACTION + REQUESTED_EXTENSION;
String REPLACE_COMMIT_EXTENSION = "." + REPLACE_COMMIT_ACTION;
+ String INFLIGHT_CLUSTERING_COMMIT_EXTENSION = "." + CLUSTERING_ACTION + INFLIGHT_EXTENSION;
+ String REQUESTED_CLUSTERING_COMMIT_EXTENSION = "." + CLUSTERING_ACTION + REQUESTED_EXTENSION;
String INFLIGHT_INDEX_COMMIT_EXTENSION = "." + INDEXING_ACTION + INFLIGHT_EXTENSION;
String REQUESTED_INDEX_COMMIT_EXTENSION = "." + INDEXING_ACTION + REQUESTED_EXTENSION;
String INDEX_COMMIT_EXTENSION = "." + INDEXING_ACTION;
@@ -221,6 +224,21 @@ public interface HoodieTimeline extends Serializable {
*/
HoodieTimeline filterPendingReplaceTimeline();
+ /**
+ * Filter this timeline to just include requested and inflight cluster commit instants.
+ */
+ HoodieTimeline filterPendingClusteringTimeline();
+
+ /**
+ * Filter this timeline to just include requested and inflight cluster or replace commit instants.
+ */
+ HoodieTimeline filterPendingReplaceOrClusteringTimeline();
+
+ /**
+ * Filter this timeline to just include requested and inflight cluster, replace commit or compaction instants.
+ */
+ HoodieTimeline filterPendingReplaceClusteringAndCompactionTimeline();
+
/**
* Filter this timeline to include pending rollbacks.
*/
@@ -415,7 +433,7 @@ public interface HoodieTimeline extends Serializable {
/**
* return true if instant is a pending clustering commit, otherwise false
*/
- public boolean isPendingClusterInstant(String instantTime);
+ public boolean isPendingClusteringInstant(String instantTime);
/**
* Read the completed instant details.
@@ -513,6 +531,14 @@ public interface HoodieTimeline extends Serializable {
return new HoodieInstant(State.INFLIGHT, REPLACE_COMMIT_ACTION, timestamp);
}
+ static HoodieInstant getClusteringCommitRequestedInstant(final String timestamp) {
+ return new HoodieInstant(State.REQUESTED, CLUSTERING_ACTION, timestamp);
+ }
+
+ static HoodieInstant getClusteringCommitInflightInstant(final String timestamp) {
+ return new HoodieInstant(State.INFLIGHT, CLUSTERING_ACTION, timestamp);
+ }
+
static HoodieInstant getRollbackRequestedInstant(HoodieInstant instant) {
return instant.isRequested() ? instant : HoodieTimeline.getRequestedInstant(instant);
}
@@ -648,6 +674,14 @@ public interface HoodieTimeline extends Serializable {
return StringUtils.join(instant, HoodieTimeline.REQUESTED_REPLACE_COMMIT_EXTENSION);
}
+ static String makeRequestedClusteringFileName(String instant) {
+ return StringUtils.join(instant, HoodieTimeline.REQUESTED_CLUSTERING_COMMIT_EXTENSION);
+ }
+
+ static String makeInflightClusteringFileName(String instant) {
+ return StringUtils.join(instant, HoodieTimeline.INFLIGHT_CLUSTERING_COMMIT_EXTENSION);
+ }
+
static String makeDeltaFileName(String instantTime) {
return instantTime + HoodieTimeline.DELTA_COMMIT_EXTENSION;
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
index fb7710eee5c..7bd4068dab3 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/MetadataConversionUtils.java
@@ -87,7 +87,8 @@ public class MetadataConversionUtils {
archivedMetaWrapper.setActionType(ActionType.deltacommit.name());
break;
}
- case HoodieTimeline.REPLACE_COMMIT_ACTION: {
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION: {
if (hoodieInstant.isCompleted()) {
HoodieReplaceCommitMetadata replaceCommitMetadata = HoodieReplaceCommitMetadata.fromBytes(instantDetails.get(), HoodieReplaceCommitMetadata.class);
archivedMetaWrapper.setHoodieReplaceCommitMetadata(convertReplaceCommitMetadata(replaceCommitMetadata));
@@ -107,7 +108,8 @@ public class MetadataConversionUtils {
archivedMetaWrapper.setHoodieRequestedReplaceMetadata(requestedReplaceMetadata.get());
}
}
- archivedMetaWrapper.setActionType(ActionType.replacecommit.name());
+ archivedMetaWrapper.setActionType(
+ hoodieInstant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION) ? ActionType.replacecommit.name() : ActionType.clustering.name());
break;
}
case HoodieTimeline.ROLLBACK_ACTION: {
@@ -157,7 +159,8 @@ public class MetadataConversionUtils {
activeAction.getCleanPlan(metaClient).ifPresent(plan -> lsmTimelineInstant.setPlan(ByteBuffer.wrap(plan)));
break;
}
- case HoodieTimeline.REPLACE_COMMIT_ACTION: {
+ case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case HoodieTimeline.CLUSTERING_ACTION: {
// we may have cases with empty HoodieRequestedReplaceMetadata e.g. insert_overwrite_table or insert_overwrite
// without clustering. However, we should revisit the requested commit file standardization
activeAction.getRequestedCommitMetadata(metaClient).ifPresent(metadata -> lsmTimelineInstant.setPlan(ByteBuffer.wrap(metadata)));
@@ -203,6 +206,10 @@ public class MetadataConversionUtils {
archivedMetaWrapper.setActionType(ActionType.replacecommit.name());
break;
}
+ case HoodieTimeline.CLUSTERING_ACTION: {
+ archivedMetaWrapper.setActionType(ActionType.clustering.name());
+ break;
+ }
case HoodieTimeline.ROLLBACK_ACTION: {
archivedMetaWrapper.setActionType(ActionType.rollback.name());
break;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
index 89aaf65c90b..3d8d31193ef 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimelineUtils.java
@@ -51,6 +51,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.hudi.common.config.HoodieCommonConfig.INCREMENTAL_READ_HANDLE_HOLLOW_COMMIT;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.GREATER_THAN;
@@ -249,19 +250,20 @@ public class TimelineUtils {
}
}
- public static boolean isClusteringCommit(HoodieTableMetaClient metaClient, HoodieInstant instant) {
+ public static boolean isClusteringCommit(HoodieTableMetaClient metaClient, HoodieInstant completedInstant) {
+ ValidationUtils.checkArgument(completedInstant.isCompleted(), "The instant should be completed for this API");
try {
- if (REPLACE_COMMIT_ACTION.equals(instant.getAction())) {
+ if (REPLACE_COMMIT_ACTION.equals(completedInstant.getAction())) {
// replacecommit is used for multiple operations: insert_overwrite/cluster etc.
// Check operation type to see if this instant is related to clustering.
HoodieReplaceCommitMetadata replaceMetadata = HoodieReplaceCommitMetadata.fromBytes(
- metaClient.getActiveTimeline().getInstantDetails(instant).get(), HoodieReplaceCommitMetadata.class);
+ metaClient.getActiveTimeline().getInstantDetails(completedInstant).get(), HoodieReplaceCommitMetadata.class);
return WriteOperationType.CLUSTER.equals(replaceMetadata.getOperationType());
}
return false;
} catch (IOException e) {
- throw new HoodieIOException("Unable to read instant information: " + instant + " for " + metaClient.getBasePath(), e);
+ throw new HoodieIOException("Unable to read instant information: " + completedInstant + " for " + metaClient.getBasePath(), e);
}
}
@@ -317,7 +319,7 @@ public class TimelineUtils {
HoodieInstant instant,
HoodieTimeline timeline) throws IOException {
byte[] data = timeline.getInstantDetails(instant).get();
- if (instant.getAction().equals(REPLACE_COMMIT_ACTION)) {
+ if (instant.getAction().equals(REPLACE_COMMIT_ACTION) || instant.getAction().equals(CLUSTERING_ACTION)) {
return HoodieReplaceCommitMetadata.fromBytes(data, HoodieReplaceCommitMetadata.class);
} else {
return HoodieCommitMetadata.fromBytes(data, HoodieCommitMetadata.class);
@@ -346,7 +348,7 @@ public class TimelineUtils {
Option<HoodieInstant> earliestCommit = shouldArchiveBeyondSavepoint
? dataTableActiveTimeline.getTimelineOfActions(
CollectionUtils.createSet(
- COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, SAVEPOINT_ACTION))
+ COMMIT_ACTION, DELTA_COMMIT_ACTION, REPLACE_COMMIT_ACTION, CLUSTERING_ACTION, SAVEPOINT_ACTION))
.getFirstNonSavepointCommit()
: dataTableActiveTimeline.getCommitsTimeline().firstInstant();
// This is for all instants which are in-flight
@@ -375,9 +377,7 @@ public class TimelineUtils {
public static void validateTimestampAsOf(HoodieTableMetaClient metaClient, String timestampAsOf) {
Option<HoodieInstant> firstIncompleteCommit = metaClient.getCommitsTimeline()
.filterInflightsAndRequested()
- .filter(instant ->
- !HoodieTimeline.REPLACE_COMMIT_ACTION.equals(instant.getAction())
- || !ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent())
+ .filter(instant -> !ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline(), instant))
.firstInstant();
if (firstIncompleteCommit.isPresent()) {
@@ -428,9 +428,7 @@ public class TimelineUtils {
Option<HoodieInstant> firstIncompleteCommit = metaClient.getCommitsTimeline()
.filterInflightsAndRequested()
- .filter(instant ->
- !HoodieTimeline.REPLACE_COMMIT_ACTION.equals(instant.getAction())
- || !ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent())
+ .filter(instant -> !ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline(), instant))
.firstInstant();
boolean noHollowCommit = firstIncompleteCommit
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
index 6965f33aa80..0c72359b110 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/AbstractTableFileSystemView.java
@@ -552,7 +552,7 @@ public abstract class AbstractTableFileSystemView implements SyncableFileSystemV
* @param baseFile base File
*/
protected boolean isBaseFileDueToPendingClustering(HoodieBaseFile baseFile) {
- return metaClient.getActiveTimeline().isPendingClusterInstant(baseFile.getCommitTime());
+ return metaClient.getActiveTimeline().isPendingClusteringInstant(baseFile.getCommitTime());
}
/**
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
index eca3ac65323..36ab0edf4fb 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/view/IncrementalTimelineSyncFileSystemView.java
@@ -20,6 +20,7 @@ package org.apache.hudi.common.table.view;
import org.apache.hudi.avro.model.HoodieCleanMetadata;
import org.apache.hudi.avro.model.HoodieCompactionPlan;
+import org.apache.hudi.avro.model.HoodieInstantInfo;
import org.apache.hudi.avro.model.HoodieRestoreMetadata;
import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.common.fs.FSUtils;
@@ -38,6 +39,7 @@ import org.apache.hudi.common.table.timeline.TimelineDiffHelper;
import org.apache.hudi.common.table.timeline.TimelineDiffHelper.TimelineDiffResult;
import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
import org.apache.hudi.common.util.CleanerUtils;
+import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CompactionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -306,8 +308,8 @@ public abstract class IncrementalTimelineSyncFileSystemView extends AbstractTabl
if (metadata.getRestoreInstantInfo() != null) {
Set<String> rolledbackInstants = metadata.getRestoreInstantInfo().stream()
- .filter(instantInfo -> HoodieTimeline.REPLACE_COMMIT_ACTION.equals(instantInfo.getAction()))
- .map(instantInfo -> instantInfo.getCommitTime()).collect(Collectors.toSet());
+ .filter(instantInfo -> ClusteringUtils.isClusteringOrReplaceCommitAction(instantInfo.getAction()))
+ .map(HoodieInstantInfo::getCommitTime).collect(Collectors.toSet());
removeReplacedFileIdsAtInstants(rolledbackInstants);
}
LOG.info("Done Syncing restore instant (" + instant + ")");
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
index 64eb27453b3..4b4dddccd9b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
@@ -43,6 +43,7 @@ import org.apache.hudi.exception.HoodieIOException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.AbstractMap;
import java.util.Arrays;
@@ -68,44 +69,123 @@ public class ClusteringUtils {
*/
public static Stream<Pair<HoodieInstant, HoodieClusteringPlan>> getAllPendingClusteringPlans(
HoodieTableMetaClient metaClient) {
- List<HoodieInstant> pendingReplaceInstants =
- metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
- return pendingReplaceInstants.stream().map(instant -> getClusteringPlan(metaClient, instant))
+ List<HoodieInstant> pendingClusterInstants =
+ metaClient.getActiveTimeline().filterPendingReplaceOrClusteringTimeline().getInstants();
+ return pendingClusterInstants.stream().map(instant -> getClusteringPlan(metaClient, instant))
.filter(Option::isPresent).map(Option::get);
}
+ /**
+ * Returns the pending clustering instant. This can be older pending replace commit or a new
+ * clustering inflight commit. After HUDI-7905, all the requested and inflight clustering instants
+ * use clustering action instead of replacecommit.
+ */
+ public static Option<HoodieInstant> getInflightClusteringInstant(String timestamp, HoodieActiveTimeline activeTimeline) {
+ HoodieTimeline pendingReplaceOrClusterTimeline = activeTimeline.filterPendingReplaceOrClusteringTimeline();
+ HoodieInstant inflightInstant = HoodieTimeline.getClusteringCommitInflightInstant(timestamp);
+ if (pendingReplaceOrClusterTimeline.containsInstant(inflightInstant)) {
+ return Option.of(inflightInstant);
+ }
+ inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(timestamp);
+ return Option.ofNullable(pendingReplaceOrClusterTimeline.containsInstant(inflightInstant) ? inflightInstant : null);
+ }
+
+ /**
+ * Returns the pending clustering instant. This can be older requested replace commit or a new
+ * clustering requested commit. After HUDI-7905, all the requested and inflight clustering instants
+ * use clustering action instead of replacecommit.
+ */
+ public static Option<HoodieInstant> getRequestedClusteringInstant(String timestamp, HoodieActiveTimeline activeTimeline) {
+ HoodieTimeline pendingReplaceOrClusterTimeline = activeTimeline.filterPendingReplaceOrClusteringTimeline();
+ HoodieInstant requestedInstant = HoodieTimeline.getClusteringCommitRequestedInstant(timestamp);
+ if (pendingReplaceOrClusterTimeline.containsInstant(requestedInstant)) {
+ return Option.of(requestedInstant);
+ }
+ requestedInstant = HoodieTimeline.getReplaceCommitRequestedInstant(timestamp);
+ return Option.ofNullable(pendingReplaceOrClusterTimeline.containsInstant(requestedInstant) ? requestedInstant : null);
+ }
+
+ /**
+ * Transitions the provided clustering instant fron inflight to complete based on the clustering
+ * action type. After HUDI-7905, the new clustering commits are written with clustering action.
+ */
+ public static void transitionClusteringOrReplaceInflightToComplete(boolean shouldLock, HoodieInstant clusteringInstant,
+ Option<byte[]> commitMetadata, HoodieActiveTimeline activeTimeline) {
+ if (clusteringInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)) {
+ activeTimeline.transitionClusterInflightToComplete(shouldLock, clusteringInstant, commitMetadata);
+ } else {
+ activeTimeline.transitionReplaceInflightToComplete(shouldLock, clusteringInstant, commitMetadata);
+ }
+ }
+
+ /**
+ * Transitions the provided clustering instant fron requested to inflight based on the clustering
+ * action type. After HUDI-7905, the new clustering commits are written with clustering action.
+ */
+ public static void transitionClusteringOrReplaceRequestedToInflight(HoodieInstant requestedClusteringInstant, Option<byte[]> data,
+ HoodieActiveTimeline activeTimeline) {
+ if (requestedClusteringInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)) {
+ activeTimeline.transitionClusterRequestedToInflight(requestedClusteringInstant, data);
+ } else {
+ activeTimeline.transitionReplaceRequestedToInflight(requestedClusteringInstant, data);
+ }
+ }
+
/**
* Checks if the requested, inflight, or completed instant of replacecommit action
* is a clustering operation, by checking whether the requested instant contains
* a clustering plan.
*
+ * @param actionType Action type
+ * @return whether the action type is a clustering or replace commit action type
+ */
+ public static boolean isClusteringOrReplaceCommitAction(String actionType) {
+ return actionType.equals(HoodieTimeline.CLUSTERING_ACTION) || actionType.equals(HoodieTimeline.REPLACE_COMMIT_ACTION);
+ }
+
+ /**
+ * Checks if the action type is a clustering or replace commit action type.
+ *
* @param timeline Hudi timeline.
* @param replaceInstant the instant of replacecommit action to check.
* @return whether the instant is a clustering operation.
*/
public static boolean isClusteringInstant(HoodieTimeline timeline, HoodieInstant replaceInstant) {
- return getClusteringPlan(timeline, replaceInstant).isPresent();
+ return replaceInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)
+ || (replaceInstant.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION) && getClusteringPlan(timeline, replaceInstant).isPresent());
}
/**
* Get requested replace metadata from timeline.
*
* @param timeline used to get the bytes stored in the requested replace instant in the timeline
- * @param pendingReplaceInstant can be in any state, because it will always be converted to requested state
+ * @param pendingReplaceOrClusterInstant can be in any state, because it will always be converted to requested state
* @return option of the replace metadata if present, else empty
* @throws IOException
*/
- private static Option<HoodieRequestedReplaceMetadata> getRequestedReplaceMetadata(HoodieTimeline timeline, HoodieInstant pendingReplaceInstant) throws IOException {
- final HoodieInstant requestedInstant;
- if (!pendingReplaceInstant.isRequested()) {
+ private static Option<HoodieRequestedReplaceMetadata> getRequestedReplaceMetadata(HoodieTimeline timeline, HoodieInstant pendingReplaceOrClusterInstant) throws IOException {
+ HoodieInstant requestedInstant;
+ if (pendingReplaceOrClusterInstant.isInflight()) {
// inflight replacecommit files don't have clustering plan.
// This is because replacecommit inflight can have workload profile for 'insert_overwrite'.
// Get the plan from corresponding requested instant.
- requestedInstant = HoodieTimeline.getReplaceCommitRequestedInstant(pendingReplaceInstant.getTimestamp());
+ requestedInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, pendingReplaceOrClusterInstant.getAction(), pendingReplaceOrClusterInstant.getTimestamp());
+ } else if (pendingReplaceOrClusterInstant.isRequested()) {
+ requestedInstant = pendingReplaceOrClusterInstant;
} else {
- requestedInstant = pendingReplaceInstant;
+ requestedInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, pendingReplaceOrClusterInstant.getTimestamp());
+ }
+ Option<byte[]> content = Option.empty();
+ try {
+ content = timeline.getInstantDetails(requestedInstant);
+ } catch (HoodieIOException e) {
+ if (e.getCause() instanceof FileNotFoundException && pendingReplaceOrClusterInstant.isCompleted()) {
+ // For clustering instants, completed instant is also a replace commit instant. For input replace commit instant,
+ // it is not known whether requested instant is CLUSTER or REPLACE_COMMIT_ACTION. So we need to query both.
+ requestedInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, pendingReplaceOrClusterInstant.getTimestamp());
+ content = timeline.getInstantDetails(requestedInstant);
+ }
}
- Option<byte[]> content = timeline.getInstantDetails(requestedInstant);
if (!content.isPresent() || content.get().length == 0) {
// few operations create requested file without any content. Assume these are not clustering
return Option.empty();
@@ -244,7 +324,7 @@ public class ClusteringUtils {
}
public static List<HoodieInstant> getPendingClusteringInstantTimes(HoodieTableMetaClient metaClient) {
- return metaClient.getActiveTimeline().filterPendingReplaceTimeline().getInstantsAsStream()
+ return metaClient.getActiveTimeline().filterPendingReplaceOrClusteringTimeline().getInstantsAsStream()
.filter(instant -> isClusteringInstant(metaClient.getActiveTimeline(), instant))
.collect(Collectors.toList());
}
@@ -260,8 +340,8 @@ public class ClusteringUtils {
public static Option<HoodieInstant> getEarliestInstantToRetainForClustering(
HoodieActiveTimeline activeTimeline, HoodieTableMetaClient metaClient) throws IOException {
Option<HoodieInstant> oldestInstantToRetain = Option.empty();
- HoodieTimeline replaceTimeline = activeTimeline.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.REPLACE_COMMIT_ACTION));
- if (!replaceTimeline.empty()) {
+ HoodieTimeline replaceOrClusterTimeline = activeTimeline.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION));
+ if (!replaceOrClusterTimeline.empty()) {
Option<HoodieInstant> cleanInstantOpt =
activeTimeline.getCleanerTimeline().filterCompletedInstants().lastInstant();
if (cleanInstantOpt.isPresent()) {
@@ -286,14 +366,14 @@ public class ClusteringUtils {
retainLowerBound = cleanInstant.getTimestamp();
}
- oldestInstantToRetain = replaceTimeline.filter(instant ->
+ oldestInstantToRetain = replaceOrClusterTimeline.filter(instant ->
HoodieTimeline.compareTimestamps(
instant.getTimestamp(),
HoodieTimeline.GREATER_THAN_OR_EQUALS,
retainLowerBound))
.firstInstant();
} else {
- oldestInstantToRetain = replaceTimeline.firstInstant();
+ oldestInstantToRetain = replaceOrClusterTimeline.firstInstant();
}
}
return oldestInstantToRetain;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
index 782506e15f8..30398af1f0f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java
@@ -107,7 +107,7 @@ public class CommitUtils {
String commitActionType,
WriteOperationType operationType) {
final HoodieCommitMetadata commitMetadata;
- if (HoodieTimeline.REPLACE_COMMIT_ACTION.equals(commitActionType)) {
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(commitActionType)) {
HoodieReplaceCommitMetadata replaceMetadata = new HoodieReplaceCommitMetadata();
replaceMetadata.setPartitionToReplaceFileIds(partitionToReplaceFileIds);
commitMetadata = replaceMetadata;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/MarkerUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/MarkerUtils.java
index 3b1270069c3..aa775a36ed1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/MarkerUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/MarkerUtils.java
@@ -307,7 +307,7 @@ public class MarkerUtils {
String instantTime = markerDirToInstantTime(instantPath);
return instantTime.compareToIgnoreCase(currentInstantTime) < 0
&& !activeTimeline.filterPendingCompactionTimeline().containsInstant(instantTime)
- && !activeTimeline.filterPendingReplaceTimeline().containsInstant(instantTime);
+ && !activeTimeline.filterPendingReplaceOrClusteringTimeline().containsInstant(instantTime);
}).filter(instantPath -> {
try {
return !isHeartbeatExpired(markerDirToInstantTime(instantPath),
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
index 3ad3ce54265..fc36397d996 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileCreateUtils.java
@@ -277,6 +277,24 @@ public class FileCreateUtils {
serializeCommitMetadata(metadata).get());
}
+ public static void createRequestedClusterCommit(String basePath, String instantTime,
+ HoodieRequestedReplaceMetadata requestedReplaceMetadata)
+ throws IOException {
+ createMetaFile(basePath, instantTime, HoodieTimeline.REQUESTED_CLUSTERING_COMMIT_EXTENSION,
+ serializeRequestedReplaceMetadata(requestedReplaceMetadata).get());
+ }
+
+ public static void createInflightClusterCommit(String basePath, String instantTime,
+ Option<HoodieCommitMetadata> inflightReplaceMetadata)
+ throws IOException {
+ if (inflightReplaceMetadata.isPresent()) {
+ createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_CLUSTERING_COMMIT_EXTENSION,
+ serializeCommitMetadata(inflightReplaceMetadata.get()).get());
+ } else {
+ createMetaFile(basePath, instantTime, HoodieTimeline.INFLIGHT_CLUSTERING_COMMIT_EXTENSION);
+ }
+ }
+
public static void createRequestedReplaceCommit(String basePath, String instantTime,
Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata)
throws IOException {
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
index 80d22a279ec..54b9b316bac 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
@@ -615,15 +615,15 @@ public class HoodieTestDataGenerator implements AutoCloseable {
createEmptyFile(basePath, commitFile, configuration);
}
- private static void createPendingReplaceFile(String basePath, String instantTime, StorageConfiguration<?> configuration, HoodieCommitMetadata commitMetadata) {
- Arrays.asList(HoodieTimeline.makeInflightReplaceFileName(instantTime),
- HoodieTimeline.makeRequestedReplaceFileName(instantTime))
+ private static void createPendingClusterFile(String basePath, String instantTime, StorageConfiguration<?> configuration, HoodieCommitMetadata commitMetadata) {
+ Arrays.asList(HoodieTimeline.makeInflightClusteringFileName(instantTime),
+ HoodieTimeline.makeRequestedClusteringFileName(instantTime))
.forEach(f -> createMetadataFile(f, basePath, configuration, commitMetadata));
}
- public static void createPendingReplaceFile(String basePath, String instantTime, StorageConfiguration<?> configuration) {
+ public static void createPendingClusterFile(String basePath, String instantTime, StorageConfiguration<?> configuration) {
HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
- createPendingReplaceFile(basePath, instantTime, configuration, commitMetadata);
+ createPendingClusterFile(basePath, instantTime, configuration, commitMetadata);
}
public static void createEmptyCleanRequestedFile(String basePath, String instantTime, StorageConfiguration<?> configuration)
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
index 75f025687e4..adaaa394481 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringCommitSink.java
@@ -26,7 +26,7 @@ import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.TableServiceType;
import org.apache.hudi.common.model.WriteOperationType;
-import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.CommitUtils;
@@ -127,14 +127,21 @@ public class ClusteringCommitSink extends CleanFunction<ClusteringCommitEvent> {
private void commitIfNecessary(String instant, Collection<ClusteringCommitEvent> events) {
HoodieClusteringPlan clusteringPlan = clusteringPlanCache.computeIfAbsent(instant, k -> {
try {
- Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- this.writeClient.getHoodieTable().getMetaClient(), HoodieTimeline.getReplaceCommitInflightInstant(instant));
- return clusteringPlanOption.get().getRight();
+ HoodieTableMetaClient metaClient = this.writeClient.getHoodieTable().getMetaClient();
+ return ClusteringUtils.getInflightClusteringInstant(instant, metaClient.getActiveTimeline())
+ .flatMap(pendingInstant -> ClusteringUtils.getClusteringPlan(
+ metaClient, pendingInstant))
+ .map(Pair::getRight)
+ .orElse(null);
} catch (Exception e) {
throw new HoodieException(e);
}
});
+ if (clusteringPlan == null) {
+ return;
+ }
+
boolean isReady = clusteringPlan.getInputGroups().size() == events.size();
if (!isReady) {
return;
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
index 327d688f951..8dfe83302d4 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanOperator.java
@@ -22,7 +22,6 @@ import org.apache.hudi.avro.model.HoodieClusteringGroup;
import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.common.model.ClusteringGroupInfo;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -119,7 +118,7 @@ public class ClusteringPlanOperator extends AbstractStreamOperator<ClusteringPla
// generate clustering plan
// should support configurable commit metadata
- HoodieInstant clusteringInstant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
+ HoodieInstant clusteringInstant = firstRequested.get();
Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
table.getMetaClient(), clusteringInstant);
@@ -137,7 +136,7 @@ public class ClusteringPlanOperator extends AbstractStreamOperator<ClusteringPla
LOG.info("Empty clustering plan for instant " + clusteringInstantTime);
} else {
// Mark instant as clustering inflight
- table.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringInstant, Option.empty());
+ ClusteringUtils.transitionClusteringOrReplaceRequestedToInflight(clusteringInstant, Option.empty(), table.getActiveTimeline());
table.getMetaClient().reloadActiveTimeline();
for (HoodieClusteringGroup clusteringGroup : clusteringPlan.getInputGroups()) {
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanSourceFunction.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanSourceFunction.java
index 21a7213c434..e99bebb9d66 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanSourceFunction.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringPlanSourceFunction.java
@@ -75,7 +75,7 @@ public class ClusteringPlanSourceFunction extends AbstractRichFunction implement
@Override
public void run(SourceContext<ClusteringPlanEvent> sourceContext) throws Exception {
- boolean isPending = StreamerUtil.createMetaClient(conf).getActiveTimeline().isPendingClusterInstant(clusteringInstantTime);
+ boolean isPending = StreamerUtil.createMetaClient(conf).getActiveTimeline().isPendingClusteringInstant(clusteringInstantTime);
if (isPending) {
for (HoodieClusteringGroup clusteringGroup : clusteringPlan.getInputGroups()) {
LOG.info("Execute clustering plan for instant {} as {} file slices", clusteringInstantTime, clusteringGroup.getSlices().size());
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
index 0966f6995bd..e0828a0431e 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/HoodieFlinkClusteringJob.java
@@ -23,7 +23,6 @@ import org.apache.hudi.avro.model.HoodieClusteringPlan;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
@@ -279,11 +278,10 @@ public class HoodieFlinkClusteringJob {
CompactionUtil.isLIFO(cfg.clusteringSeq) ? instants.get(instants.size() - 1) : instants.get(0);
}
- HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(
- clusteringInstant.getTimestamp());
- if (table.getMetaClient().getActiveTimeline().containsInstant(inflightInstant)) {
+ Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(clusteringInstant.getTimestamp(), table.getActiveTimeline());
+ if (inflightInstantOpt.isPresent()) {
LOG.info("Rollback inflight clustering instant: [" + clusteringInstant + "]");
- table.rollbackInflightClustering(inflightInstant,
+ table.rollbackInflightClustering(inflightInstantOpt.get(),
commitToRollback -> writeClient.getTableServiceClient().getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false));
table.getMetaClient().reloadActiveTimeline();
}
@@ -308,7 +306,7 @@ public class HoodieFlinkClusteringJob {
return;
}
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstant.getTimestamp());
+ HoodieInstant instant = ClusteringUtils.getRequestedClusteringInstant(clusteringInstant.getTimestamp(), table.getActiveTimeline()).get();
int inputGroupSize = clusteringPlan.getInputGroups().size();
@@ -318,7 +316,7 @@ public class HoodieFlinkClusteringJob {
: Math.min(conf.getInteger(FlinkOptions.CLUSTERING_TASKS), inputGroupSize);
// Mark instant as clustering inflight
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ ClusteringUtils.transitionClusteringOrReplaceRequestedToInflight(instant, Option.empty(), table.getActiveTimeline());
final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
diff --git a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
index 0350fb41f7b..1c988e6d303 100644
--- a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClusteringUtil.java
@@ -20,8 +20,8 @@ package org.apache.hudi.util;
import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.client.clustering.plan.strategy.FlinkConsistentBucketClusteringPlanStrategy;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
-import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.configuration.FlinkOptions;
@@ -104,10 +104,11 @@ public class ClusteringUtil {
* @param instantTime The instant time
*/
public static void rollbackClustering(HoodieFlinkTable<?> table, HoodieFlinkWriteClient<?> writeClient, String instantTime) {
- HoodieInstant inflightInstant = HoodieTimeline.getReplaceCommitInflightInstant(instantTime);
- if (table.getMetaClient().reloadActiveTimeline().isPendingClusterInstant(instantTime)) {
+ HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline();
+ Option<HoodieInstant> inflightInstantOpt = ClusteringUtils.getInflightClusteringInstant(instantTime, activeTimeline);
+ if (inflightInstantOpt.isPresent() && ClusteringUtils.isClusteringInstant(activeTimeline, inflightInstantOpt.get())) {
LOG.warn("Rollback failed clustering instant: [" + instantTime + "]");
- table.rollbackInflightClustering(inflightInstant,
+ table.rollbackInflightClustering(inflightInstantOpt.get(),
commitToRollback -> writeClient.getTableServiceClient().getPendingRollbackInfo(table.getMetaClient(), commitToRollback, false));
}
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
index a00cceaabf0..cfe7886af07 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestFlinkConsistentHashingClustering.java
@@ -114,7 +114,7 @@ public class ITTestFlinkConsistentHashingClustering {
HoodieFlinkTable<?> table = writeClient.getHoodieTable();
table.getMetaClient().reloadActiveTimeline();
Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlanOption = ClusteringUtils.getClusteringPlan(
- table.getMetaClient(), table.getMetaClient().getActiveTimeline().filterPendingReplaceTimeline().lastInstant().get());
+ table.getMetaClient(), table.getMetaClient().getActiveTimeline().filterPendingClusteringTimeline().lastInstant().get());
return clusteringPlanOption.get().getRight();
}
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
index dff930e3fbb..2736cdd119c 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/cluster/ITTestHoodieFlinkClustering.java
@@ -164,7 +164,7 @@ public class ITTestHoodieFlinkClustering {
// fetch the instant based on the configured execution sequence
table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingClusteringTimeline()
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
// generate clustering plan
@@ -175,8 +175,8 @@ public class ITTestHoodieFlinkClustering {
HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
// Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ HoodieInstant instant = HoodieTimeline.getClusteringCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionClusterRequestedToInflight(instant, Option.empty());
final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
@@ -373,7 +373,7 @@ public class ITTestHoodieFlinkClustering {
// fetch the instant based on the configured execution sequence
table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingClusteringTimeline()
.filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
// generate clustering plan
@@ -384,8 +384,8 @@ public class ITTestHoodieFlinkClustering {
HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
// Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(firstClusteringInstant);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ HoodieInstant instant = HoodieTimeline.getClusteringCommitRequestedInstant(firstClusteringInstant);
+ table.getActiveTimeline().transitionClusterRequestedToInflight(instant, Option.empty());
final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
@@ -419,14 +419,14 @@ public class ITTestHoodieFlinkClustering {
// wait for the asynchronous commit to finish
TimeUnit.SECONDS.sleep(3);
- // archive the first commit, retain the second commit before the inflight replacecommit
+ // archive the first commit, retain the second commit before the inflight cluster commit
writeClient.archive();
scheduled = writeClient.scheduleClusteringAtInstant(writeClient.createNewInstantTime(), Option.empty());
assertTrue(scheduled, "The clustering plan should be scheduled");
table.getMetaClient().reloadActiveTimeline();
- timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ timeline = table.getActiveTimeline().filterPendingClusteringTimeline()
.filter(i -> i.getState() == HoodieInstant.State.REQUESTED);
HoodieInstant secondClusteringInstant = timeline.lastInstant().get();
@@ -570,7 +570,7 @@ public class ITTestHoodieFlinkClustering {
// fetch the instant based on the configured execution sequence
table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingClusteringTimeline()
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
// generate clustering plan
@@ -581,8 +581,8 @@ public class ITTestHoodieFlinkClustering {
HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
// Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ HoodieInstant instant = HoodieTimeline.getClusteringCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionClusterRequestedToInflight(instant, Option.empty());
DataStream<ClusteringCommitEvent> dataStream = env.addSource(new ClusteringPlanSourceFunction(clusteringInstantTime, clusteringPlan, conf))
.name("clustering_source")
@@ -691,7 +691,7 @@ public class ITTestHoodieFlinkClustering {
// fetch the instant based on the configured execution sequence
table.getMetaClient().reloadActiveTimeline();
- HoodieTimeline timeline = table.getActiveTimeline().filterPendingReplaceTimeline()
+ HoodieTimeline timeline = table.getActiveTimeline().filterPendingClusteringTimeline()
.filter(instant -> instant.getState() == HoodieInstant.State.REQUESTED);
// generate clustering plan
@@ -702,8 +702,8 @@ public class ITTestHoodieFlinkClustering {
HoodieClusteringPlan clusteringPlan = clusteringPlanOption.get().getRight();
// Mark instant as clustering inflight
- HoodieInstant instant = HoodieTimeline.getReplaceCommitRequestedInstant(clusteringInstantTime);
- table.getActiveTimeline().transitionReplaceRequestedToInflight(instant, Option.empty());
+ HoodieInstant instant = HoodieTimeline.getClusteringCommitRequestedInstant(clusteringInstantTime);
+ table.getActiveTimeline().transitionClusterRequestedToInflight(instant, Option.empty());
final Schema tableAvroSchema = StreamerUtil.getTableAvroSchema(table.getMetaClient(), false);
final DataType rowDataType = AvroSchemaConverter.convertToDataType(tableAvroSchema);
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java
index edbb8341433..6910cbd5cbf 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/source/TestIncrementalInputSplits.java
@@ -154,9 +154,9 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness {
HoodieInstant commit2 = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.DELTA_COMMIT_ACTION, "2");
// commit3: clustering
timelineMOR.createCompleteInstant(commit2);
- HoodieInstant commit3 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, "3");
+ HoodieInstant commit3 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, "3");
timelineMOR.createNewInstant(commit3);
- commit3 = timelineMOR.transitionReplaceRequestedToInflight(commit3, Option.empty());
+ commit3 = timelineMOR.transitionClusterRequestedToInflight(commit3, Option.empty());
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(
new ArrayList<>(),
new HashMap<>(),
@@ -164,8 +164,8 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness {
WriteOperationType.CLUSTER,
"",
HoodieTimeline.REPLACE_COMMIT_ACTION);
- timelineMOR.transitionReplaceInflightToComplete(true,
- HoodieTimeline.getReplaceCommitInflightInstant(commit3.getTimestamp()),
+ timelineMOR.transitionClusterInflightToComplete(true,
+ HoodieTimeline.getClusteringCommitInflightInstant(commit3.getTimestamp()),
serializeCommitMetadata(commitMetadata));
// commit4: insert overwrite
HoodieInstant commit4 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, "4");
@@ -235,9 +235,9 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness {
HoodieInstant commit2 = new HoodieInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "2");
// commit3: clustering
timelineCOW.createCompleteInstant(commit2);
- HoodieInstant commit3 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, "3");
+ HoodieInstant commit3 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, "3");
timelineCOW.createNewInstant(commit3);
- commit3 = timelineCOW.transitionReplaceRequestedToInflight(commit3, Option.empty());
+ commit3 = timelineCOW.transitionClusterRequestedToInflight(commit3, Option.empty());
HoodieCommitMetadata commitMetadata = CommitUtils.buildMetadata(
new ArrayList<>(),
new HashMap<>(),
@@ -245,8 +245,8 @@ public class TestIncrementalInputSplits extends HoodieCommonTestHarness {
WriteOperationType.CLUSTER,
"",
HoodieTimeline.REPLACE_COMMIT_ACTION);
- timelineCOW.transitionReplaceInflightToComplete(true,
- HoodieTimeline.getReplaceCommitInflightInstant(commit3.getTimestamp()),
+ timelineCOW.transitionClusterInflightToComplete(true,
+ HoodieTimeline.getClusteringCommitInflightInstant(commit3.getTimestamp()),
serializeCommitMetadata(commitMetadata));
// commit4: insert overwrite
HoodieInstant commit4 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, "4");
diff --git a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
index 2abd51e3a51..38ae95a9d2b 100644
--- a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
+++ b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestClusteringUtil.java
@@ -135,11 +135,11 @@ public class TestClusteringUtil {
plan, Collections.emptyMap(), 1);
String instantTime = table.getMetaClient().createNewInstantTime();
HoodieInstant clusteringInstant =
- new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime);
+ new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, instantTime);
try {
- metaClient.getActiveTimeline().saveToPendingReplaceCommit(clusteringInstant,
+ metaClient.getActiveTimeline().saveToPendingClusterCommit(clusteringInstant,
TimelineMetadataUtils.serializeRequestedReplaceMetadata(metadata));
- table.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringInstant, Option.empty());
+ table.getActiveTimeline().transitionClusterRequestedToInflight(clusteringInstant, Option.empty());
} catch (IOException ioe) {
throw new HoodieIOException("Exception scheduling clustering", ioe);
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
index 83cf0957d7f..f82f627323c 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
@@ -49,6 +49,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.nio.file.Paths;
@@ -66,6 +67,7 @@ import static org.apache.hudi.common.table.timeline.HoodieInstant.State.COMPLETE
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.INFLIGHT;
import static org.apache.hudi.common.table.timeline.HoodieInstant.State.REQUESTED;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLEAN_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION;
@@ -104,8 +106,9 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
cleanMetaClient();
}
- @Test
- public void testGetPartitionsWithReplaceCommits() throws IOException {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testGetPartitionsWithReplaceOrClusterCommits(boolean withReplace) throws IOException {
HoodieActiveTimeline activeTimeline = metaClient.getActiveTimeline();
HoodieTimeline activeCommitTimeline = activeTimeline.getCommitAndReplaceTimeline();
assertTrue(activeCommitTimeline.empty());
@@ -113,12 +116,18 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
String ts1 = "1";
String replacePartition = "2021/01/01";
String newFilePartition = "2021/01/02";
- HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, ts1);
+ HoodieInstant instant1 = new HoodieInstant(true, withReplace ? HoodieTimeline.REPLACE_COMMIT_ACTION : HoodieTimeline.CLUSTERING_ACTION, ts1);
activeTimeline.createNewInstant(instant1);
// create replace metadata only with replaced file Ids (no new files created)
- activeTimeline.saveAsComplete(instant1,
- Option.of(getReplaceCommitMetadata(basePath, ts1, replacePartition, 2,
- newFilePartition, 0, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ if (withReplace) {
+ activeTimeline.saveAsComplete(instant1,
+ Option.of(getReplaceCommitMetadata(basePath, ts1, replacePartition, 2,
+ newFilePartition, 0, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ } else {
+ activeTimeline.transitionClusterInflightToComplete(true, instant1,
+ Option.of(getReplaceCommitMetadata(basePath, ts1, replacePartition, 2,
+ newFilePartition, 0, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ }
metaClient.reloadActiveTimeline();
List<String> partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("0", 10));
@@ -126,12 +135,18 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
assertEquals(replacePartition, partitions.get(0));
String ts2 = "2";
- HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, ts2);
+ HoodieInstant instant2 = new HoodieInstant(true, withReplace ? HoodieTimeline.REPLACE_COMMIT_ACTION : HoodieTimeline.CLUSTERING_ACTION, ts2);
activeTimeline.createNewInstant(instant2);
// create replace metadata only with replaced file Ids (no new files created)
- activeTimeline.saveAsComplete(instant2,
- Option.of(getReplaceCommitMetadata(basePath, ts2, replacePartition, 0,
- newFilePartition, 3, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ if (withReplace) {
+ activeTimeline.saveAsComplete(instant2,
+ Option.of(getReplaceCommitMetadata(basePath, ts2, replacePartition, 0,
+ newFilePartition, 3, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ } else {
+ activeTimeline.transitionClusterInflightToComplete(true, instant2,
+ Option.of(getReplaceCommitMetadata(basePath, ts2, replacePartition, 0,
+ newFilePartition, 3, Collections.emptyMap(), WriteOperationType.CLUSTER)));
+ }
metaClient.reloadActiveTimeline();
partitions = TimelineUtils.getAffectedPartitions(metaClient.getActiveTimeline().findInstantsAfter("1", 10));
assertEquals(1, partitions.size());
@@ -262,11 +277,11 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
// verify adding clustering commit doesn't change behavior of getExtraMetadataFromLatest
String ts2 = "2";
- HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, ts2);
+ HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, ts2);
activeTimeline.createNewInstant(instant2);
String newValueForMetadata = "newValue2";
extraMetadata.put(extraMetadataKey, newValueForMetadata);
- activeTimeline.saveAsComplete(instant2,
+ activeTimeline.transitionClusterInflightToComplete(true, instant2,
Option.of(getReplaceCommitMetadata(basePath, ts2, "p2", 0,
"p2", 3, extraMetadata, WriteOperationType.CLUSTER)));
metaClient.reloadActiveTimeline();
@@ -417,7 +432,8 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
new HoodieInstant(COMPLETED, CLEAN_ACTION, "002"),
new HoodieInstant(REQUESTED, CLEAN_ACTION, "003"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "010"),
- new HoodieInstant(COMPLETED, REPLACE_COMMIT_ACTION, "011"))), false));
+ new HoodieInstant(COMPLETED, REPLACE_COMMIT_ACTION, "011"),
+ new HoodieInstant(REQUESTED, CLUSTERING_ACTION, "012"))), false));
// No inflight instants
assertEquals(
@@ -429,7 +445,8 @@ public class TestTimelineUtils extends HoodieCommonTestHarness {
new HoodieInstant(COMPLETED, CLEAN_ACTION, "002"),
new HoodieInstant(COMPLETED, CLEAN_ACTION, "003"),
new HoodieInstant(COMPLETED, COMMIT_ACTION, "010"),
- new HoodieInstant(COMPLETED, REPLACE_COMMIT_ACTION, "011"))), false));
+ new HoodieInstant(COMPLETED, REPLACE_COMMIT_ACTION, "011"),
+ new HoodieInstant(REQUESTED, CLUSTERING_ACTION, "012"))), false));
// Rollbacks only
assertEquals(
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
index e8dd23f00e8..da0f8c021ee 100755
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
@@ -321,7 +321,8 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
new HoodieInstant(State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "03"),
new HoodieInstant(State.COMPLETED, HoodieTimeline.SAVEPOINT_ACTION, "03"),
new HoodieInstant(State.COMPLETED, HoodieTimeline.COMMIT_ACTION, "05"),
- new HoodieInstant(State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, "06")
+ new HoodieInstant(State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, "06"),
+ new HoodieInstant(State.INFLIGHT, HoodieTimeline.CLUSTERING_ACTION, "07")
).collect(Collectors.toList()));
assertTrue(timeline.isBeforeTimelineStarts("00"));
assertTrue(timeline.isBeforeTimelineStarts("01"));
@@ -352,10 +353,10 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
// Test that various types of getXXX operations from HoodieActiveTimeline
// return the correct set of Instant
checkTimeline.accept(timeline.getCommitsTimeline(), CollectionUtils.createSet(
- HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION));
+ HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION));
checkTimeline.accept(timeline.getWriteTimeline(), CollectionUtils.createSet(
- HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION));
- checkTimeline.accept(timeline.getCommitAndReplaceTimeline(), CollectionUtils.createSet(HoodieTimeline.COMMIT_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION));
+ HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.COMPACTION_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION));
+ checkTimeline.accept(timeline.getCommitAndReplaceTimeline(), CollectionUtils.createSet(HoodieTimeline.COMMIT_ACTION, HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION));
checkTimeline.accept(timeline.getDeltaCommitTimeline(), Collections.singleton(HoodieTimeline.DELTA_COMMIT_ACTION));
checkTimeline.accept(timeline.getCleanerTimeline(), Collections.singleton(HoodieTimeline.CLEAN_ACTION));
checkTimeline.accept(timeline.getRollbackTimeline(), Collections.singleton(HoodieTimeline.ROLLBACK_ACTION));
@@ -363,7 +364,7 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
checkTimeline.accept(timeline.getSavePointTimeline(), Collections.singleton(HoodieTimeline.SAVEPOINT_ACTION));
checkTimeline.accept(timeline.getAllCommitsTimeline(), CollectionUtils.createSet(
HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION, HoodieTimeline.CLEAN_ACTION, HoodieTimeline.COMPACTION_ACTION,
- HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.SAVEPOINT_ACTION, HoodieTimeline.ROLLBACK_ACTION, HoodieTimeline.INDEXING_ACTION));
+ HoodieTimeline.REPLACE_COMMIT_ACTION, HoodieTimeline.CLUSTERING_ACTION, HoodieTimeline.SAVEPOINT_ACTION, HoodieTimeline.ROLLBACK_ACTION, HoodieTimeline.INDEXING_ACTION));
// Get some random Instants
Random rand = new Random();
@@ -740,6 +741,10 @@ public class TestHoodieActiveTimeline extends HoodieCommonTestHarness {
if (state == State.COMPLETED && action.equals(HoodieTimeline.COMPACTION_ACTION)) {
action = HoodieTimeline.COMMIT_ACTION;
}
+ // Cluster complete is called replacecommit complete
+ if (state == State.COMPLETED && action.equals(HoodieTimeline.CLUSTERING_ACTION)) {
+ action = HoodieTimeline.REPLACE_COMMIT_ACTION;
+ }
allInstants.add(new HoodieInstant(state, action, String.format("%03d", instantTime++)));
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
index ea15e70407b..4ff5ce2a888 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
@@ -145,7 +145,7 @@ public class TestHoodieTableFSViewWithClustering extends HoodieCommonTestHarness
CommitUtils.buildMetadata(Collections.emptyList(), partitionToReplaceFileIds, Option.empty(), WriteOperationType.INSERT_OVERWRITE, "", HoodieTimeline.REPLACE_COMMIT_ACTION);
HoodieActiveTimeline commitTimeline = metaClient.getActiveTimeline();
- HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, commitTime1);
+ HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, commitTime1);
saveAsComplete(
commitTimeline,
instant1,
@@ -187,7 +187,11 @@ public class TestHoodieTableFSViewWithClustering extends HoodieCommonTestHarness
HoodieInstant requested = new HoodieInstant(HoodieInstant.State.REQUESTED, inflight.getAction(), inflight.getTimestamp());
timeline.createNewInstant(requested);
timeline.transitionRequestedToInflight(requested, Option.empty());
- timeline.saveAsComplete(inflight, data);
+ if (inflight.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)) {
+ timeline.transitionClusterInflightToComplete(true, inflight, data);
+ } else {
+ timeline.saveAsComplete(inflight, data);
+ }
}
}
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
index 1eb41270419..bb8e0ba59a7 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
@@ -182,18 +182,18 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
HoodieInstant instant1 = new HoodieInstant(true, HoodieTimeline.COMMIT_ACTION, instantTime1);
HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.COMMIT_ACTION, instantTime2);
HoodieInstant clusteringInstant3 =
- new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstantTime3);
+ new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, clusteringInstantTime3);
HoodieInstant clusteringInstant4 =
- new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstantTime4);
+ new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, clusteringInstantTime4);
HoodieCommitMetadata commitMetadata =
CommitUtils.buildMetadata(Collections.emptyList(), partitionToReplaceFileIds,
Option.empty(), WriteOperationType.CLUSTER, "", HoodieTimeline.REPLACE_COMMIT_ACTION);
saveAsComplete(commitTimeline, instant1, Option.empty());
saveAsComplete(commitTimeline, instant2, Option.empty());
- saveAsComplete(commitTimeline, clusteringInstant3,
+ saveAsCompleteCluster(commitTimeline, clusteringInstant3,
serializeCommitMetadata((HoodieReplaceCommitMetadata) commitMetadata));
- saveAsComplete(
+ saveAsCompleteCluster(
commitTimeline,
clusteringInstant4,
serializeCommitMetadata((HoodieReplaceCommitMetadata) commitMetadata));
@@ -207,13 +207,13 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
HoodieTimeline.REPLACE_COMMIT_ACTION);
boolean deleteReplaceCommit = metaClient.getStorage().deleteDirectory(completeInstantPath);
- boolean deleteReplaceCommitRequested = new File(
- this.basePath + "/.hoodie/" + clusteringInstantTime3 + ".replacecommit.requested").delete();
- boolean deleteReplaceCommitInflight = new File(
- this.basePath + "/.hoodie/" + clusteringInstantTime3 + ".replacecommit.inflight").delete();
+ boolean deleteClusterCommitRequested = new File(
+ this.basePath + "/.hoodie/" + clusteringInstantTime3 + ".clustering.requested").delete();
+ boolean deleteClusterCommitInflight = new File(
+ this.basePath + "/.hoodie/" + clusteringInstantTime3 + ".clustering.inflight").delete();
// confirm deleted
- assertTrue(deleteReplaceCommit && deleteReplaceCommitInflight && deleteReplaceCommitRequested);
+ assertTrue(deleteReplaceCommit && deleteClusterCommitInflight && deleteClusterCommitRequested);
assertDoesNotThrow(() -> fsView.close());
}
@@ -1861,7 +1861,7 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
}
private void saveAsCompleteCluster(HoodieActiveTimeline timeline, HoodieInstant inflight, Option<byte[]> data) {
- assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION, inflight.getAction());
+ assertEquals(HoodieTimeline.CLUSTERING_ACTION, inflight.getAction());
HoodieInstant clusteringInstant = new HoodieInstant(State.REQUESTED, inflight.getAction(), inflight.getTimestamp());
HoodieClusteringPlan plan = new HoodieClusteringPlan();
plan.setExtraMetadata(new HashMap<>());
@@ -1875,13 +1875,13 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
.setExtraMetadata(Collections.emptyMap())
.setClusteringPlan(plan)
.build();
- timeline.saveToPendingReplaceCommit(clusteringInstant,
+ timeline.saveToPendingClusterCommit(clusteringInstant,
TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
} catch (IOException ioe) {
throw new HoodieIOException("Exception scheduling clustering", ioe);
}
timeline.transitionRequestedToInflight(clusteringInstant, Option.empty());
- timeline.saveAsComplete(inflight, data);
+ timeline.transitionClusterInflightToComplete(true, inflight, data);
}
@Test
@@ -2091,10 +2091,10 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
fileSliceGroups, Collections.emptyMap());
String clusterTime = "2";
- HoodieInstant instant2 = new HoodieInstant(State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, clusterTime);
+ HoodieInstant instant2 = new HoodieInstant(State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, clusterTime);
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setClusteringPlan(plan).setOperationType(WriteOperationType.CLUSTER.name()).build();
- metaClient.getActiveTimeline().saveToPendingReplaceCommit(instant2, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
+ metaClient.getActiveTimeline().saveToPendingClusterCommit(instant2, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
//make sure view doesn't include fileId1
refreshFsView();
@@ -2205,7 +2205,7 @@ public class TestHoodieTableFileSystemView extends HoodieCommonTestHarness {
String fileName3 = FSUtils.makeBaseFileName(commitTime2, TEST_WRITE_TOKEN, fileId3, BASE_FILE_EXTENSION);
new File(basePath + "/" + partitionPath + "/" + fileName3).createNewFile();
- HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.REPLACE_COMMIT_ACTION, commitTime2);
+ HoodieInstant instant2 = new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, commitTime2);
Map<String, List<String>> partitionToReplaceFileIds = new HashMap<>();
List<String> replacedFileIds = new ArrayList<>();
replacedFileIds.add(fileId1);
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
index e7e442f3ed6..14c00111d21 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
@@ -105,12 +105,14 @@ import static org.apache.hudi.common.model.WriteOperationType.CLUSTER;
import static org.apache.hudi.common.model.WriteOperationType.COMPACT;
import static org.apache.hudi.common.model.WriteOperationType.UPSERT;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLEAN_ACTION;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
import static org.apache.hudi.common.testutils.FileCreateUtils.baseFileName;
import static org.apache.hudi.common.testutils.FileCreateUtils.createCleanFile;
import static org.apache.hudi.common.testutils.FileCreateUtils.createCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createDeltaCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCleanFile;
+import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightClusterCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightCompaction;
import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightDeltaCommit;
@@ -120,6 +122,7 @@ import static org.apache.hudi.common.testutils.FileCreateUtils.createInflightSav
import static org.apache.hudi.common.testutils.FileCreateUtils.createMarkerFile;
import static org.apache.hudi.common.testutils.FileCreateUtils.createReplaceCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCleanFile;
+import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedClusterCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedDeltaCommit;
import static org.apache.hudi.common.testutils.FileCreateUtils.createRequestedReplaceCommit;
@@ -338,6 +341,37 @@ public class HoodieTestTable implements AutoCloseable {
return this;
}
+ public HoodieTestTable addPendingCluster(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata, Option<HoodieCommitMetadata> inflightReplaceMetadata) throws Exception {
+ createRequestedClusterCommit(basePath, instantTime, requestedReplaceMetadata);
+ createInflightClusterCommit(basePath, instantTime, inflightReplaceMetadata);
+ currentInstantTime = instantTime;
+ return this;
+ }
+
+ public HoodieTestTable addRequestedCluster(String instantTime, HoodieRequestedReplaceMetadata requestedReplaceMetadata) throws Exception {
+ createRequestedClusterCommit(basePath, instantTime, requestedReplaceMetadata);
+ currentInstantTime = instantTime;
+ return this;
+ }
+
+ public HoodieTestTable addInflightCluster(String instantTime, Option<HoodieCommitMetadata> inflightReplaceMetadata) throws Exception {
+ createInflightClusterCommit(basePath, instantTime, inflightReplaceMetadata);
+ currentInstantTime = instantTime;
+ return this;
+ }
+
+ public HoodieTestTable addCluster(
+ String instantTime,
+ HoodieRequestedReplaceMetadata requestedReplaceMetadata,
+ Option<HoodieCommitMetadata> inflightReplaceMetadata,
+ HoodieReplaceCommitMetadata completeReplaceMetadata) throws Exception {
+ createRequestedClusterCommit(basePath, instantTime, requestedReplaceMetadata);
+ createInflightClusterCommit(basePath, instantTime, inflightReplaceMetadata);
+ createReplaceCommit(basePath, instantTime, completeReplaceMetadata);
+ currentInstantTime = instantTime;
+ return this;
+ }
+
public HoodieTestTable addRequestedReplace(String instantTime, Option<HoodieRequestedReplaceMetadata> requestedReplaceMetadata) throws Exception {
createRequestedReplaceCommit(basePath, instantTime, requestedReplaceMetadata);
currentInstantTime = instantTime;
@@ -1012,7 +1046,12 @@ public class HoodieTestTable implements AutoCloseable {
HoodieReplaceCommitMetadata replaceMetadata =
(HoodieReplaceCommitMetadata) buildMetadata(writeStats, partitionToReplaceFileIds, Option.empty(), CLUSTER, PHONY_TABLE_SCHEMA,
REPLACE_COMMIT_ACTION);
- addReplaceCommit(commitTime, Option.empty(), Option.empty(), replaceMetadata);
+ HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
+ .setOperationType(WriteOperationType.CLUSTER.name())
+ .setExtraMetadata(Collections.emptyMap())
+ .setClusteringPlan(new HoodieClusteringPlan())
+ .build();
+ addCluster(commitTime, requestedReplaceMetadata, Option.empty(), replaceMetadata);
return replaceMetadata;
}
@@ -1196,7 +1235,8 @@ public class HoodieTestTable implements AutoCloseable {
private Option<HoodieCommitMetadata> getCommitMeta(HoodieInstant hoodieInstant) throws IOException {
switch (hoodieInstant.getAction()) {
- case HoodieTimeline.REPLACE_COMMIT_ACTION:
+ case REPLACE_COMMIT_ACTION:
+ case CLUSTERING_ACTION:
HoodieReplaceCommitMetadata replaceCommitMetadata = HoodieReplaceCommitMetadata
.fromBytes(metaClient.getActiveTimeline().getInstantDetails(hoodieInstant).get(), HoodieReplaceCommitMetadata.class);
return Option.of(replaceCommitMetadata);
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
index 72630787c13..7ac82c8e7ae 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
@@ -76,7 +76,7 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
fileIds1.add(UUID.randomUUID().toString());
fileIds1.add(UUID.randomUUID().toString());
String clusterTime1 = "1";
- createRequestedReplaceInstant(partitionPath1, clusterTime1, fileIds1);
+ createRequestedClusterInstant(partitionPath1, clusterTime1, fileIds1);
List<String> fileIds2 = new ArrayList<>();
fileIds2.add(UUID.randomUUID().toString());
@@ -87,7 +87,7 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
fileIds3.add(UUID.randomUUID().toString());
String clusterTime = "2";
- createRequestedReplaceInstant(partitionPath1, clusterTime, fileIds2, fileIds3);
+ createRequestedClusterInstant(partitionPath1, clusterTime, fileIds2, fileIds3);
// create replace.requested without clustering plan. this instant should be ignored by ClusteringUtils
createRequestedReplaceInstantNotClustering("3");
@@ -96,7 +96,7 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, "4"));
metaClient.reloadActiveTimeline();
- assertEquals(4, metaClient.getActiveTimeline().filterPendingReplaceTimeline().countInstants());
+ assertEquals(2, metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
Map<HoodieFileGroupId, HoodieInstant> fileGroupToInstantMap =
ClusteringUtils.getAllFileGroupsInPendingClusteringPlans(metaClient);
@@ -109,13 +109,13 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
assertEquals("2", lastPendingClustering.get().getTimestamp());
//check that it still gets picked if it is inflight
- HoodieInstant inflight = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(lastPendingClustering.get(), Option.empty());
+ HoodieInstant inflight = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(lastPendingClustering.get(), Option.empty());
assertEquals(HoodieInstant.State.INFLIGHT, inflight.getState());
lastPendingClustering = metaClient.reloadActiveTimeline().getLastPendingClusterInstant();
assertEquals("2", lastPendingClustering.get().getTimestamp());
//now that it is complete, the first instant should be picked
- HoodieInstant complete = metaClient.getActiveTimeline().transitionReplaceInflightToComplete(false, inflight, Option.empty());
+ HoodieInstant complete = metaClient.getActiveTimeline().transitionClusterInflightToComplete(false, inflight, Option.empty());
assertEquals(HoodieInstant.State.COMPLETED, complete.getState());
lastPendingClustering = metaClient.reloadActiveTimeline().getLastPendingClusterInstant();
assertEquals("1", lastPendingClustering.get().getTimestamp());
@@ -131,7 +131,7 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
fileIds1.add(UUID.randomUUID().toString());
fileIds1.add(UUID.randomUUID().toString());
String clusterTime1 = "1";
- HoodieInstant requestedInstant = createRequestedReplaceInstant(partitionPath1, clusterTime1, fileIds1);
+ HoodieInstant requestedInstant = createRequestedClusterInstant(partitionPath1, clusterTime1, fileIds1);
HoodieInstant inflightInstant = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant, Option.empty());
assertTrue(ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline(), requestedInstant));
HoodieClusteringPlan requestedClusteringPlan = ClusteringUtils.getClusteringPlan(metaClient, requestedInstant).get().getRight();
@@ -146,24 +146,24 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
List<String> fileIds1 = new ArrayList<>();
fileIds1.add(UUID.randomUUID().toString());
String clusterTime1 = "1";
- HoodieInstant requestedInstant1 = createRequestedReplaceInstant(partitionPath1, clusterTime1, fileIds1);
- HoodieInstant inflightInstant1 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant1, Option.empty());
- metaClient.getActiveTimeline().transitionReplaceInflightToComplete(true, inflightInstant1, Option.empty());
+ HoodieInstant requestedInstant1 = createRequestedClusterInstant(partitionPath1, clusterTime1, fileIds1);
+ HoodieInstant inflightInstant1 = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(requestedInstant1, Option.empty());
+ metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant1, Option.empty());
List<String> fileIds2 = new ArrayList<>();
fileIds2.add(UUID.randomUUID().toString());
fileIds2.add(UUID.randomUUID().toString());
String clusterTime2 = "2";
- HoodieInstant requestedInstant2 = createRequestedReplaceInstant(partitionPath1, clusterTime2, fileIds2);
- HoodieInstant inflightInstant2 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant2, Option.empty());
- metaClient.getActiveTimeline().transitionReplaceInflightToComplete(true, inflightInstant2, Option.empty());
+ HoodieInstant requestedInstant2 = createRequestedClusterInstant(partitionPath1, clusterTime2, fileIds2);
+ HoodieInstant inflightInstant2 = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(requestedInstant2, Option.empty());
+ metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant2, Option.empty());
List<String> fileIds3 = new ArrayList<>();
fileIds3.add(UUID.randomUUID().toString());
fileIds3.add(UUID.randomUUID().toString());
fileIds3.add(UUID.randomUUID().toString());
String clusterTime3 = "3";
- HoodieInstant requestedInstant3 = createRequestedReplaceInstant(partitionPath1, clusterTime3, fileIds3);
- HoodieInstant inflightInstant3 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant3, Option.empty());
- HoodieInstant completedInstant3 = metaClient.getActiveTimeline().transitionReplaceInflightToComplete(true, inflightInstant3, Option.empty());
+ HoodieInstant requestedInstant3 = createRequestedClusterInstant(partitionPath1, clusterTime3, fileIds3);
+ HoodieInstant inflightInstant3 = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(requestedInstant3, Option.empty());
+ HoodieInstant completedInstant3 = metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant3, Option.empty());
metaClient.reloadActiveTimeline();
Option<HoodieInstant> actual = ClusteringUtils.getEarliestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
assertTrue(actual.isPresent());
@@ -199,9 +199,9 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
List<String> fileIds1 = new ArrayList<>();
fileIds1.add(UUID.randomUUID().toString());
String clusterTime1 = "1";
- HoodieInstant requestedInstant1 = createRequestedReplaceInstant(partitionPath1, clusterTime1, fileIds1);
- HoodieInstant inflightInstant1 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant1, Option.empty());
- metaClient.getActiveTimeline().transitionReplaceInflightToComplete(true, inflightInstant1, Option.empty());
+ HoodieInstant requestedInstant1 = createRequestedClusterInstant(partitionPath1, clusterTime1, fileIds1);
+ HoodieInstant inflightInstant1 = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(requestedInstant1, Option.empty());
+ metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant1, Option.empty());
String cleanTime1 = "2";
HoodieInstant requestedInstant2 = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLEAN_ACTION, cleanTime1);
@@ -220,9 +220,9 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
fileIds2.add(UUID.randomUUID().toString());
fileIds2.add(UUID.randomUUID().toString());
String clusterTime2 = "3";
- HoodieInstant requestedInstant3 = createRequestedReplaceInstant(partitionPath1, clusterTime2, fileIds2);
- HoodieInstant inflightInstant3 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant3, Option.empty());
- metaClient.getActiveTimeline().transitionReplaceInflightToComplete(true, inflightInstant3, Option.empty());
+ HoodieInstant requestedInstant3 = createRequestedClusterInstant(partitionPath1, clusterTime2, fileIds2);
+ HoodieInstant inflightInstant3 = metaClient.getActiveTimeline().transitionClusterRequestedToInflight(requestedInstant3, Option.empty());
+ metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant3, Option.empty());
metaClient.reloadActiveTimeline();
Option<HoodieInstant> actual = ClusteringUtils.getEarliestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
@@ -245,7 +245,7 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
return newRequestedInstant;
}
- private HoodieInstant createRequestedReplaceInstant(String partitionPath1, String clusterTime, List<String>... fileIds) throws IOException {
+ private HoodieInstant createRequestedClusterInstant(String partitionPath1, String clusterTime, List<String>... fileIds) throws IOException {
List<FileSlice>[] fileSliceGroups = new List[fileIds.length];
for (int i = 0; i < fileIds.length; i++) {
fileSliceGroups[i] = fileIds[i].stream().map(fileId -> generateFileSlice(partitionPath1, fileId, "0")).collect(Collectors.toList());
@@ -254,10 +254,10 @@ public class TestClusteringUtils extends HoodieCommonTestHarness {
HoodieClusteringPlan clusteringPlan =
ClusteringUtils.createClusteringPlan(CLUSTERING_STRATEGY_CLASS, STRATEGY_PARAMS, fileSliceGroups, Collections.emptyMap());
- HoodieInstant clusteringInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, clusterTime);
+ HoodieInstant clusteringInstant = new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, clusterTime);
HoodieRequestedReplaceMetadata requestedReplaceMetadata = HoodieRequestedReplaceMetadata.newBuilder()
.setClusteringPlan(clusteringPlan).setOperationType(WriteOperationType.CLUSTER.name()).build();
- metaClient.getActiveTimeline().saveToPendingReplaceCommit(clusteringInstant, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
+ metaClient.getActiveTimeline().saveToPendingClusterCommit(clusteringInstant, TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata));
return clusteringInstant;
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
index 7deed9551f4..7421d9e0c2c 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
@@ -44,6 +44,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.hudi.common.table.timeline.HoodieTimeline.CLUSTERING_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION;
import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION;
import static org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_SCHEMA;
@@ -210,8 +211,8 @@ public class TestCommitUtils {
.setExtraMetadata(Collections.emptyMap())
.setClusteringPlan(new HoodieClusteringPlan())
.build();
- timeline.saveToPendingReplaceCommit(
- new HoodieInstant(HoodieInstant.State.REQUESTED, REPLACE_COMMIT_ACTION, ts),
+ timeline.saveToPendingClusterCommit(
+ new HoodieInstant(HoodieInstant.State.REQUESTED, CLUSTERING_ACTION, ts),
TimelineMetadataUtils.serializeRequestedReplaceMetadata(requestedReplaceMetadata)
);
}
diff --git a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
index a7c7164505d..715abe789e2 100644
--- a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
@@ -325,7 +325,8 @@ public class TestCompactionUtils extends HoodieCommonTestHarness {
new HoodieInstant(false, HoodieTimeline.REPLACE_COMMIT_ACTION, "06"),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "07"),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "08"),
- new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, "09")).collect(Collectors.toList()));
+ new HoodieInstant(true, HoodieTimeline.DELTA_COMMIT_ACTION, "09"),
+ new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, "10")).collect(Collectors.toList()));
actual =
CompactionUtils.getDeltaCommitsSinceLatestCompaction(timeline).get();
diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
index 95d8714e05c..1aaef8f55af 100644
--- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
+++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
@@ -163,7 +163,7 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
if (clusteringInstant.isPresent) {
asyncClusteringService.enqueuePendingAsyncServiceInstant(new HoodieInstant(
- State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstant.get()
+ State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, clusteringInstant.get()
))
}
Success((true, commitOps, compactionInstantOps))
diff --git a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
index 69c8f618cab..cd0f146db78 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
+++ b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDataSourceHelpers.java
@@ -130,9 +130,11 @@ public class HoodieDataSourceHelpers {
HoodieTableMetaClient metaClient = HoodieTableMetaClient.builder()
.setConf(HadoopFSUtils.getStorageConfWithCopy(fs.getConf()))
.setBasePath(basePath).setLoadActiveTimelineOnLoad(true).build();
- HoodieInstant hoodieInstant = HoodieTimeline.getReplaceCommitRequestedInstant(instantTime);
+ Option<HoodieInstant> hoodieInstant = metaClient.getActiveTimeline().filter(instant -> instant.getTimestamp().equals(instantTime)
+ && ClusteringUtils.isClusteringOrReplaceCommitAction(instant.getAction()))
+ .firstInstant();
Option<Pair<HoodieInstant, HoodieClusteringPlan>> clusteringPlan =
- ClusteringUtils.getClusteringPlan(metaClient, hoodieInstant);
+ hoodieInstant.flatMap(instant -> ClusteringUtils.getClusteringPlan(metaClient, instant));
if (clusteringPlan.isPresent()) {
return Option.of(clusteringPlan.get().getValue());
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunRollbackInflightTableServiceProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunRollbackInflightTableServiceProcedure.scala
index 28e0d03b85d..029b7546e16 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunRollbackInflightTableServiceProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/RunRollbackInflightTableServiceProcedure.scala
@@ -22,7 +22,7 @@ import org.apache.hudi.client.SparkRDDWriteClient
import org.apache.hudi.common.HoodiePendingRollbackInfo
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
-import org.apache.hudi.common.util.Option
+import org.apache.hudi.common.util.{ClusteringUtils, Option}
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.table.HoodieSparkTable
import org.apache.spark.internal.Logging
@@ -78,9 +78,9 @@ class RunRollbackInflightTableServiceProcedure extends BaseProcedure
var isClustering: Boolean = true
var instant: HoodieInstant = null
val pendingCompactionInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.COMPACTION_ACTION, pendingInstant)
- val pendingClusteringInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, pendingInstant)
+ val pendingClusteringInstant = ClusteringUtils.getInflightClusteringInstant(pendingInstant, metaClient.getActiveTimeline)
val timeline = metaClient.getActiveTimeline.getWriteTimeline
- if (!timeline.containsInstant(pendingCompactionInstant) && !timeline.containsInstant(pendingClusteringInstant)) {
+ if (!timeline.containsInstant(pendingCompactionInstant) && !pendingClusteringInstant.isPresent) {
throw new RuntimeException(s"there is no pending instant : [$pendingClusteringInstant | $pendingCompactionInstant]")
} else if (timeline.containsInstant(pendingCompactionInstant)) {
isClustering = false
@@ -88,7 +88,7 @@ class RunRollbackInflightTableServiceProcedure extends BaseProcedure
logInfo(s"compaction instant to rollback : ${instant}")
} else {
isClustering = true
- instant = pendingClusteringInstant
+ instant = pendingClusteringInstant.get()
logInfo(s"clustering instant to rollback : ${instant}")
}
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala
index 2f8a92e5ee6..060d6cb48aa 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowClusteringProcedure.scala
@@ -59,7 +59,7 @@ class ShowClusteringProcedure extends BaseProcedure with ProcedureBuilder with S
val basePath: String = getBasePath(tableName, tablePath)
val metaClient = createMetaClient(jsc, basePath)
val clusteringInstants = metaClient.getActiveTimeline.getInstants.iterator().asScala
- .filter(p => p.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION)
+ .filter(p => ClusteringUtils.isClusteringOrReplaceCommitAction(p.getAction))
.toSeq
.sortBy(f => f.getTimestamp)
.reverse
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitExtraMetadataProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitExtraMetadataProcedure.scala
index f438fc22755..1fa4d566917 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitExtraMetadataProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitExtraMetadataProcedure.scala
@@ -20,14 +20,13 @@ package org.apache.spark.sql.hudi.command.procedures
import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieReplaceCommitMetadata}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.exception.HoodieException
-
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
import java.util
import java.util.function.Supplier
-
import scala.collection.JavaConverters._
class ShowCommitExtraMetadataProcedure() extends BaseProcedure with ProcedureBuilder {
@@ -109,6 +108,7 @@ class ShowCommitExtraMetadataProcedure() extends BaseProcedure with ProcedureBui
val instants: util.List[HoodieInstant] = util.Arrays.asList(
new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime),
+ new HoodieInstant(false, HoodieTimeline.CLUSTERING_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, instantTime))
val hoodieInstant: Option[HoodieInstant] = instants.asScala.find((i: HoodieInstant) => timeline.containsInstant(i))
@@ -117,7 +117,7 @@ class ShowCommitExtraMetadataProcedure() extends BaseProcedure with ProcedureBui
private def getHoodieCommitMetadata(timeline: HoodieTimeline, hoodieInstant: Option[HoodieInstant]): Option[HoodieCommitMetadata] = {
if (hoodieInstant.isDefined) {
- if (hoodieInstant.get.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION) {
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(hoodieInstant.get.getAction)) {
Option(HoodieReplaceCommitMetadata.fromBytes(timeline.getInstantDetails(hoodieInstant.get).get,
classOf[HoodieReplaceCommitMetadata]))
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitFilesProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitFilesProcedure.scala
index 1801ed647f4..f2ecbbc7a9e 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitFilesProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitFilesProcedure.scala
@@ -20,8 +20,8 @@ package org.apache.spark.sql.hudi.command.procedures
import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieReplaceCommitMetadata, HoodieWriteStat}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.exception.HoodieException
-
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
@@ -93,6 +93,7 @@ class ShowCommitFilesProcedure() extends BaseProcedure with ProcedureBuilder {
val instants: util.List[HoodieInstant] = util.Arrays.asList(
new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime),
+ new HoodieInstant(false, HoodieTimeline.CLUSTERING_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, instantTime))
val hoodieInstant: Option[HoodieInstant] = instants.asScala.find((i: HoodieInstant) => timeline.containsInstant(i))
@@ -101,7 +102,7 @@ class ShowCommitFilesProcedure() extends BaseProcedure with ProcedureBuilder {
private def getHoodieCommitMetadata(timeline: HoodieTimeline, hoodieInstant: Option[HoodieInstant]): Option[HoodieCommitMetadata] = {
if (hoodieInstant.isDefined) {
- if (hoodieInstant.get.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION) {
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(hoodieInstant.get.getAction)) {
Option(HoodieReplaceCommitMetadata.fromBytes(timeline.getInstantDetails(hoodieInstant.get).get,
classOf[HoodieReplaceCommitMetadata]))
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitPartitionsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitPartitionsProcedure.scala
index b7180810e4c..307ddabef99 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitPartitionsProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitPartitionsProcedure.scala
@@ -20,8 +20,8 @@ package org.apache.spark.sql.hudi.command.procedures
import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieReplaceCommitMetadata, HoodieWriteStat}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.exception.HoodieException
-
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
@@ -108,6 +108,7 @@ class ShowCommitPartitionsProcedure() extends BaseProcedure with ProcedureBuilde
val instants: util.List[HoodieInstant] = util.Arrays.asList(
new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime),
+ new HoodieInstant(false, HoodieTimeline.CLUSTERING_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, instantTime))
val hoodieInstant: Option[HoodieInstant] = instants.asScala.find((i: HoodieInstant) => timeline.containsInstant(i))
@@ -116,7 +117,7 @@ class ShowCommitPartitionsProcedure() extends BaseProcedure with ProcedureBuilde
private def getHoodieCommitMetadata(timeline: HoodieTimeline, hoodieInstant: Option[HoodieInstant]): Option[HoodieCommitMetadata] = {
if (hoodieInstant.isDefined) {
- if (hoodieInstant.get.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION) {
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(hoodieInstant.get.getAction)) {
Option(HoodieReplaceCommitMetadata.fromBytes(timeline.getInstantDetails(hoodieInstant.get).get,
classOf[HoodieReplaceCommitMetadata]))
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitWriteStatsProcedure.scala b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitWriteStatsProcedure.scala
index f5f58966f73..1bc17803d63 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitWriteStatsProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/ShowCommitWriteStatsProcedure.scala
@@ -20,8 +20,8 @@ package org.apache.spark.sql.hudi.command.procedures
import org.apache.hudi.HoodieCLIUtils
import org.apache.hudi.common.model.{HoodieCommitMetadata, HoodieReplaceCommitMetadata}
import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
+import org.apache.hudi.common.util.ClusteringUtils
import org.apache.hudi.exception.HoodieException
-
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{DataTypes, Metadata, StructField, StructType}
@@ -85,6 +85,7 @@ class ShowCommitWriteStatsProcedure() extends BaseProcedure with ProcedureBuilde
val instants: util.List[HoodieInstant] = util.Arrays.asList(
new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.REPLACE_COMMIT_ACTION, instantTime),
+ new HoodieInstant(false, HoodieTimeline.CLUSTERING_ACTION, instantTime),
new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, instantTime))
val hoodieInstant: Option[HoodieInstant] = instants.asScala.find((i: HoodieInstant) => timeline.containsInstant(i))
@@ -93,7 +94,7 @@ class ShowCommitWriteStatsProcedure() extends BaseProcedure with ProcedureBuilde
private def getHoodieCommitMetadata(timeline: HoodieTimeline, hoodieInstant: Option[HoodieInstant]): Option[HoodieCommitMetadata] = {
if (hoodieInstant.isDefined) {
- if (hoodieInstant.get.getAction == HoodieTimeline.REPLACE_COMMIT_ACTION) {
+ if (ClusteringUtils.isClusteringOrReplaceCommitAction(hoodieInstant.get.getAction)) {
Option(HoodieReplaceCommitMetadata.fromBytes(timeline.getInstantDetails(hoodieInstant.get).get,
classOf[HoodieReplaceCommitMetadata]))
} else {
diff --git a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkSortAndSizeClustering.java b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkSortAndSizeClustering.java
index d785eacb385..e748c7ace53 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkSortAndSizeClustering.java
+++ b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkSortAndSizeClustering.java
@@ -112,7 +112,7 @@ public class TestSparkSortAndSizeClustering extends HoodieSparkClientTestHarness
String clusteringTime = (String) writeClient.scheduleClustering(Option.empty()).get();
HoodieClusteringPlan plan = ClusteringUtils.getClusteringPlan(
- metaClient, HoodieTimeline.getReplaceCommitRequestedInstant(clusteringTime)).map(Pair::getRight).get();
+ metaClient, HoodieTimeline.getClusteringCommitRequestedInstant(clusteringTime)).map(Pair::getRight).get();
List<HoodieClusteringGroup> inputGroups = plan.getInputGroups();
Assertions.assertEquals(1, inputGroups.size(), "Clustering plan will contain 1 input group");
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
index 0c7f578107a..36e2ac07b3f 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestCOWDataSource.scala
@@ -1857,7 +1857,7 @@ class TestCOWDataSource extends HoodieSparkClientTestBase with ScalaAssertionSup
// Last instant is clustering
assertTrue(TimelineUtils.getCommitMetadata(lastInstant, metaClient.getActiveTimeline)
.getOperationType.equals(WriteOperationType.CLUSTER))
- assertTrue(ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline, lastInstant))
+ assertTrue(ClusteringUtils.isClusteringInstant(metaClient.getActiveTimeline, new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, lastInstant.getTimestamp)))
lastClustering = lastInstant
assertEquals(
lastClustering,
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestHoodieActiveTimeline.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestHoodieActiveTimeline.scala
index 2f83eeca449..4bf643bb239 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestHoodieActiveTimeline.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestHoodieActiveTimeline.scala
@@ -19,16 +19,18 @@ package org.apache.hudi.functional
import org.apache.hudi.common.model.HoodieFileFormat
import org.apache.hudi.common.table.HoodieTableMetaClient
+import org.apache.hudi.common.table.timeline.{HoodieInstant, HoodieTimeline}
import org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings
import org.apache.hudi.config.HoodieWriteConfig
+import org.apache.hudi.exception.HoodieIOException
import org.apache.hudi.testutils.HoodieSparkClientTestBase
import org.apache.hudi.{DataSourceWriteOptions, HoodieDataSourceHelpers}
-
import org.apache.spark.sql._
-import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.{assertEquals, assertNotNull, assertTrue, fail}
import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
import org.slf4j.LoggerFactory
+import java.io.FileNotFoundException
import scala.collection.JavaConverters._
/**
@@ -228,4 +230,29 @@ class TestHoodieActiveTimeline extends HoodieSparkClientTestBase {
assert(relativePath4.contains(commit4Time))
assert(relativePath4.contains(HoodieFileFormat.HOODIE_LOG.getFileExtension))
}
+
+ @Test
+ def testGetInstantDetails(): Unit = {
+ // First Operation:
+ val records1 = recordsToStrings(dataGen.generateInserts("001", 100)).asScala.toList
+ val inputDF1 = spark.read.json(spark.sparkContext.parallelize(records1, 2))
+ inputDF1.write.format("org.apache.hudi")
+ .options(commonOpts)
+ .option("hoodie.compact.inline", "false")
+ .option(DataSourceWriteOptions.OPERATION.key, DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL)
+ .option(DataSourceWriteOptions.TABLE_TYPE.key, DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL)
+ .mode(SaveMode.Overwrite)
+ .save(basePath)
+ val metaClient: HoodieTableMetaClient = createMetaClient(basePath)
+ val activeTimeline = metaClient.getActiveTimeline
+ assertNotNull(activeTimeline.getInstantDetails(activeTimeline.lastInstant().get()))
+ try {
+ activeTimeline.getInstantDetails(new HoodieInstant(true, HoodieTimeline.CLUSTERING_ACTION, metaClient.createNewInstantTime()))
+ } catch {
+ // org.apache.hudi.common.util.ClusteringUtils.getRequestedReplaceMetadata depends upon this behaviour
+ // where FileNotFoundException is the cause of exception thrown by the API getInstantDetails
+ case e: HoodieIOException => assertTrue(classOf[FileNotFoundException].equals(e.getCause.getClass))
+ case _ => fail("Should have failed with FileNotFoundException")
+ }
+ }
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
index 55355ae0b04..da080d23c45 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestClusteringProcedure.scala
@@ -446,7 +446,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
val conf = new Configuration
val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
- val instants = metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.iterator().asScala.map(_.getTimestamp).toSeq
+ val instants = metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.iterator().asScala.map(_.getTimestamp).toSeq
assert(2 == instants.size)
checkExceptionContain(
@@ -454,14 +454,14 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
)("specific 000000 instants is not exist")
metaClient.reloadActiveTimeline()
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(2 == metaClient.getActiveTimeline.filterPendingReplaceTimeline.getInstants.size())
+ assert(2 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
writeRecords(2, 4, 0, basePath, Map("hoodie.avro.schema.validate" -> "false"))
// specific instants will not schedule new cluster plan
spark.sql(s"call run_clustering(table => '$tableName', instants => '${instants.mkString(",")}')")
metaClient.reloadActiveTimeline()
assert(2 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline.getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
// test with operator schedule
checkExceptionContain(
@@ -476,11 +476,11 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
// test with operator execute
spark.sql(s"call run_clustering(table => '$tableName', op => 'schedule')")
metaClient.reloadActiveTimeline()
- val instants2 = metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.iterator().asScala.map(_.getTimestamp).toSeq
+ val instants2 = metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.iterator().asScala.map(_.getTimestamp).toSeq
spark.sql(s"call run_clustering(table => '$tableName', instants => '${instants2.mkString(",")}', op => 'execute')")
metaClient.reloadActiveTimeline()
assert(3 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline.getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
}
}
@@ -511,27 +511,27 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
val conf = new Configuration
val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(metaClient.getActiveTimeline.filterPendingReplaceTimeline().empty())
+ assert(metaClient.getActiveTimeline.filterPendingClusteringTimeline().empty())
spark.sql(s"call run_clustering(table => '$tableName', op => 'schedule')")
metaClient.reloadActiveTimeline()
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(1 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(1 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
spark.sql(s"call run_clustering(table => '$tableName', op => 'execute')")
metaClient.reloadActiveTimeline()
assert(1 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
spark.sql(s"call run_clustering(table => '$tableName')")
metaClient.reloadActiveTimeline()
assert(2 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
spark.sql(s"call run_clustering(table => '$tableName')")
metaClient.reloadActiveTimeline()
assert(3 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
checkExceptionContain(s"call run_clustering(table => '$tableName', op => 'null')")("Invalid value")
}
@@ -746,7 +746,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
val conf = new Configuration
val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(metaClient.getActiveTimeline.filterPendingReplaceTimeline().empty())
+ assert(metaClient.getActiveTimeline.filterPendingClusteringTimeline().empty())
writeRecords(2, 4, 0, basePath, Map("hoodie.avro.schema.validate" -> "false"))
spark.sql(s"call run_clustering(table => '$tableName', op => 'schedule')")
@@ -756,12 +756,12 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
metaClient.reloadActiveTimeline()
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(2 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(2 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
spark.sql(s"call run_clustering(table => '$tableName', op => 'execute')")
metaClient.reloadActiveTimeline()
assert(2 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(0 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(0 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
writeRecords(2, 4, 0, basePath, Map("hoodie.avro.schema.validate" -> "false"))
spark.sql(s"call run_clustering(table => '$tableName', op => 'schedule')")
@@ -772,7 +772,7 @@ class TestClusteringProcedure extends HoodieSparkProcedureTestBase {
spark.sql(s"call run_clustering(table => '$tableName', op => 'execute', limit => 1)")
metaClient.reloadActiveTimeline()
assert(3 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(1 == metaClient.getActiveTimeline.filterPendingReplaceTimeline().getInstants.size())
+ assert(1 == metaClient.getActiveTimeline.filterPendingClusteringTimeline().getInstants.size())
}
}
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
index 05fcc3f2a87..470d49c03b1 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestCompactionProcedure.scala
@@ -361,7 +361,7 @@ class TestCompactionProcedure extends HoodieSparkProcedureTestBase {
val metaClient = HoodieTestUtils.createMetaClient(new HadoopStorageConfiguration(conf), basePath)
assert(0 == metaClient.getActiveTimeline.getCompletedReplaceTimeline.getInstants.size())
- assert(metaClient.getActiveTimeline.filterPendingReplaceTimeline().empty())
+ assert(metaClient.getActiveTimeline.filterPendingClusteringTimeline().empty())
spark.sql(s"insert into $tableName values(1, 'a1', 10, 1000)")
spark.sql(s"update $tableName set name = 'a2' where id = 1")
diff --git a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunRollbackInflightTableServiceProcedure.scala b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunRollbackInflightTableServiceProcedure.scala
index b2bb7e00daf..e2923f3be86 100644
--- a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunRollbackInflightTableServiceProcedure.scala
+++ b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestRunRollbackInflightTableServiceProcedure.scala
@@ -60,9 +60,9 @@ class TestRunRollbackInflightTableServiceProcedure extends HoodieSparkProcedureT
spark.sql(s"call run_rollback_inflight_tableservice(table => '$tableName', pending_instant => '$clusteringInstantTime')")
Assertions.assertTrue(!metaClient.reloadActiveTimeline().getInstants
- .contains(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstantTime)))
+ .contains(new HoodieInstant(HoodieInstant.State.INFLIGHT, HoodieTimeline.CLUSTERING_ACTION, clusteringInstantTime)))
Assertions.assertTrue(metaClient.reloadActiveTimeline().getInstants
- .contains(new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstantTime)))
+ .contains(new HoodieInstant(HoodieInstant.State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, clusteringInstantTime)))
}}
}
diff --git a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
index 27db59ab7cd..e10582f52a8 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
@@ -801,7 +801,7 @@ public class HoodieStreamer implements Serializable {
Option<String> clusteringInstant = streamSync.getClusteringInstantOpt();
if (clusteringInstant.isPresent()) {
LOG.info("Scheduled async clustering for instant: " + clusteringInstant.get());
- asyncClusteringService.get().enqueuePendingAsyncServiceInstant(new HoodieInstant(State.REQUESTED, HoodieTimeline.REPLACE_COMMIT_ACTION, clusteringInstant.get()));
+ asyncClusteringService.get().enqueuePendingAsyncServiceInstant(new HoodieInstant(State.REQUESTED, HoodieTimeline.CLUSTERING_ACTION, clusteringInstant.get()));
asyncClusteringService.get().waitTillPendingAsyncServiceInstantsReducesTo(cfg.maxPendingClustering);
if (asyncClusteringService.get().hasError()) {
error = true;
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
index de154577921..55c9507ed0c 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/HoodieDeltaStreamerTestBase.java
@@ -477,8 +477,8 @@ public class HoodieDeltaStreamerTestBase extends UtilitiesTestBase {
addCommitToTimeline(metaClient, WriteOperationType.UPSERT, HoodieTimeline.COMMIT_ACTION, extraMetadata);
}
- static void addReplaceCommitToTimeline(HoodieTableMetaClient metaClient, Map<String, String> extraMetadata) throws IOException {
- addCommitToTimeline(metaClient, WriteOperationType.CLUSTER, HoodieTimeline.REPLACE_COMMIT_ACTION, extraMetadata);
+ static void addClusterCommitToTimeline(HoodieTableMetaClient metaClient, Map<String, String> extraMetadata) throws IOException {
+ addCommitToTimeline(metaClient, WriteOperationType.CLUSTER, HoodieTimeline.CLUSTERING_ACTION, extraMetadata);
}
static void addCommitToTimeline(HoodieTableMetaClient metaClient, WriteOperationType writeOperationType, String commitActiontype,
@@ -488,10 +488,16 @@ public class HoodieDeltaStreamerTestBase extends UtilitiesTestBase {
extraMetadata.forEach((k, v) -> commitMetadata.getExtraMetadata().put(k, v));
String commitTime = metaClient.createNewInstantTime();
metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(HoodieInstant.State.REQUESTED, commitActiontype, commitTime));
- metaClient.getActiveTimeline().createNewInstant(new HoodieInstant(HoodieInstant.State.INFLIGHT, commitActiontype, commitTime));
- metaClient.getActiveTimeline().saveAsComplete(
- new HoodieInstant(HoodieInstant.State.INFLIGHT, commitActiontype, commitTime),
- TimelineMetadataUtils.serializeCommitMetadata(commitMetadata));
+ HoodieInstant inflightInstant = new HoodieInstant(HoodieInstant.State.INFLIGHT, commitActiontype, commitTime);
+ metaClient.getActiveTimeline().createNewInstant(inflightInstant);
+ if (commitActiontype.equals(HoodieTimeline.CLUSTERING_ACTION)) {
+ metaClient.getActiveTimeline().transitionClusterInflightToComplete(true, inflightInstant,
+ TimelineMetadataUtils.serializeCommitMetadata(commitMetadata));
+ } else {
+ metaClient.getActiveTimeline().saveAsComplete(
+ new HoodieInstant(HoodieInstant.State.INFLIGHT, commitActiontype, commitTime),
+ TimelineMetadataUtils.serializeCommitMetadata(commitMetadata));
+ }
}
void assertRecordCount(long expected, String tablePath, SQLContext sqlContext) {
@@ -735,9 +741,9 @@ public class HoodieDeltaStreamerTestBase extends UtilitiesTestBase {
assertEquals(0, numDeltaCommits, "Got=" + numDeltaCommits + ", exp =" + 0);
}
- static void assertAtLeastNReplaceRequests(int minExpected, String tablePath) {
+ static void assertAtLeastNClusterRequests(int minExpected, String tablePath) {
HoodieTableMetaClient meta = createMetaClient(storage.getConf(), tablePath);
- HoodieTimeline timeline = meta.getActiveTimeline().filterPendingReplaceTimeline();
+ HoodieTimeline timeline = meta.getActiveTimeline().filterPendingClusteringTimeline();
LOG.info("Timeline Instants=" + meta.getActiveTimeline().getInstants());
int numDeltaCommits = timeline.countInstants();
assertTrue(minExpected <= numDeltaCommits, "Got=" + numDeltaCommits + ", exp >=" + minExpected);
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
index ffc34e51c8c..fc78f7f0939 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieDeltaStreamer.java
@@ -813,9 +813,9 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
HoodieClusteringJob clusteringJob = initialHoodieClusteringJob(tableBasePath, null, false, "schedule");
clusteringJob.cluster(0);
HoodieTableMetaClient meta = HoodieTestUtils.createMetaClient(storage, tableBasePath);
- List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
+ List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingClusteringTimeline().getInstants();
HoodieInstant clusteringRequest = hoodieClusteringInstants.get(0);
- meta.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringRequest, Option.empty());
+ meta.getActiveTimeline().transitionClusterRequestedToInflight(clusteringRequest, Option.empty());
// do another ingestion with inline clustering enabled
cfg.configs.addAll(getTableServicesConfigs(totalRecords, "false", "true", "2", "", ""));
@@ -1236,9 +1236,9 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
// convert clustering request into inflight, Simulate the last clustering failed scenario
HoodieTableMetaClient meta = HoodieTestUtils.createMetaClient(storage, tableBasePath);
- List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingReplaceTimeline().getInstants();
+ List<HoodieInstant> hoodieClusteringInstants = meta.getActiveTimeline().filterPendingClusteringTimeline().getInstants();
HoodieInstant clusteringRequest = hoodieClusteringInstants.get(0);
- HoodieInstant hoodieInflightInstant = meta.getActiveTimeline().transitionReplaceRequestedToInflight(clusteringRequest, Option.empty());
+ HoodieInstant hoodieInflightInstant = meta.getActiveTimeline().transitionClusterRequestedToInflight(clusteringRequest, Option.empty());
// trigger a scheduleAndExecute clustering job
// when retryFailedClustering true => will rollback and re-execute failed clustering plan with same instant timestamp.
@@ -1289,7 +1289,7 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
return true;
}
case UtilHelpers.SCHEDULE: {
- TestHelpers.assertAtLeastNReplaceRequests(2, tableBasePath);
+ TestHelpers.assertAtLeastNClusterRequests(2, tableBasePath);
TestHelpers.assertNoReplaceCommits(tableBasePath);
return true;
}
@@ -2648,8 +2648,8 @@ public class TestHoodieDeltaStreamer extends HoodieDeltaStreamerTestBase {
assertEquals(testDeltaSync.getLatestCommitMetadataWithValidCheckpointInfo(metaClient.getActiveTimeline()
.getCommitsTimeline()).get().getMetadata(HoodieDeltaStreamer.CHECKPOINT_KEY), "def");
- // add a replace commit which does not have CHECKPOINT_KEY. Deltastreamer should be able to go back and pick the right checkpoint.
- addReplaceCommitToTimeline(metaClient, Collections.emptyMap());
+ // add a cluster commit which does not have CHECKPOINT_KEY. Deltastreamer should be able to go back and pick the right checkpoint.
+ addClusterCommitToTimeline(metaClient, Collections.emptyMap());
metaClient.reloadActiveTimeline();
assertEquals(testDeltaSync.getLatestCommitMetadataWithValidCheckpointInfo(metaClient.getActiveTimeline()
.getCommitsTimeline()).get().getMetadata(HoodieDeltaStreamer.CHECKPOINT_KEY), "def");
diff --git a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
index eb5574a3f64..c2775e4982f 100644
--- a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
+++ b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestHoodieIncrSource.java
@@ -282,7 +282,7 @@ public class TestHoodieIncrSource extends SparkClientFunctionalTestHarness {
String latestCommitTimestamp = dataBatches.get(dataBatches.size() - 1).getKey();
// Pending clustering exists
Option<HoodieInstant> clusteringInstant =
- metaClient.getActiveTimeline().filterPendingReplaceTimeline()
+ metaClient.getActiveTimeline().filterPendingClusteringTimeline()
.filter(instant -> ClusteringUtils.getClusteringPlan(metaClient, instant).isPresent())
.firstInstant();
assertTrue(clusteringInstant.isPresent());