This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 e2cf7216fddf fix(spark): reject INSERT_OVERWRITE when overlapping with
pending clustering (#18829)
e2cf7216fddf is described below
commit e2cf7216fddf37f49cfdf10861be7aaa31e3feb5
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Fri Jun 26 16:50:15 2026 -0700
fix(spark): reject INSERT_OVERWRITE when overlapping with pending
clustering (#18829)
fix(spark): detect INSERT_OVERWRITE / pending-clustering overlap in update
strategy
Clustering update strategies only saw record-level updates and missed file
groups being wholesale replaced by INSERT_OVERWRITE /
INSERT_OVERWRITE_TABLE.
Under the default SparkRejectUpdateStrategy the overwrite silently won the
race against pending clustering instead of being rejected.
Plumb fileGroupsToBeReplaced through UpdateStrategy (3-arg ctor kept for
back-compat; loader falls back on NoSuchMethodException). Replace
executors
populate the set; SparkRejectUpdateStrategy unions it with explicit
updates
before the overlap check. Extended to the bulk-insert overwrite path via
BaseDatasetBulkInsertCommitActionExecutor.
Closes #19074
---------
Co-authored-by: Claude Opus 4.7 (1M context) <[email protected]>
---
.../action/cluster/strategy/UpdateStrategy.java | 17 +-
.../update/strategy/BaseSparkUpdateStrategy.java | 13 +-
.../update/strategy/SparkAllowUpdateStrategy.java | 11 +-
...arkConsistentBucketDuplicateUpdateStrategy.java | 8 +-
.../update/strategy/SparkRejectUpdateStrategy.java | 17 +-
.../commit/BaseSparkCommitActionExecutor.java | 53 +-
.../SparkInsertOverwriteCommitActionExecutor.java | 28 +-
...rkInsertOverwriteTableCommitActionExecutor.java | 22 +
.../commit/TestInsertOverwriteWithClustering.java | 834 +++++++++++++++++++++
.../strategy/ConsistentBucketUpdateStrategy.java | 2 +-
.../FlinkConsistentBucketUpdateStrategy.java | 2 +-
.../BaseDatasetBulkInsertCommitActionExecutor.java | 89 +++
...setBulkInsertOverwriteCommitActionExecutor.java | 44 ++
...lkInsertOverwriteTableCommitActionExecutor.java | 11 +
14 files changed, 1123 insertions(+), 28 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/UpdateStrategy.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/UpdateStrategy.java
index 1c61db4b572e..281aa97acfa5 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/UpdateStrategy.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/cluster/strategy/UpdateStrategy.java
@@ -24,6 +24,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.table.HoodieTable;
import java.io.Serializable;
+import java.util.Collections;
import java.util.Set;
/**
@@ -34,11 +35,25 @@ public abstract class UpdateStrategy<T, I> implements
Serializable {
protected final transient HoodieEngineContext engineContext;
protected HoodieTable table;
protected Set<HoodieFileGroupId> fileGroupsInPendingClustering;
+ protected Set<HoodieFileGroupId> fileGroupsToBeReplaced;
- public UpdateStrategy(HoodieEngineContext engineContext, HoodieTable table,
Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+ public UpdateStrategy(HoodieEngineContext engineContext, HoodieTable table,
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced) {
this.engineContext = engineContext;
this.table = table;
this.fileGroupsInPendingClustering = fileGroupsInPendingClustering;
+ this.fileGroupsToBeReplaced = fileGroupsToBeReplaced;
+ }
+
+ /**
+ * Backward-compatible 3-arg constructor for custom {@code
hoodie.clustering.updates.strategy}
+ * classes that pre-date the addition of {@code fileGroupsToBeReplaced}.
Delegates to the 4-arg
+ * form with an empty replaced set so the existing reflection lookup keeps
working.
+ */
+ public UpdateStrategy(HoodieEngineContext engineContext, HoodieTable table,
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
+ this(engineContext, table, fileGroupsInPendingClustering,
Collections.emptySet());
}
/**
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/BaseSparkUpdateStrategy.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/BaseSparkUpdateStrategy.java
index 751e2a2858bc..3ff660f7dcb1 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/BaseSparkUpdateStrategy.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/BaseSparkUpdateStrategy.java
@@ -25,7 +25,7 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
-import java.util.List;
+import java.util.HashSet;
import java.util.Set;
/**
@@ -35,8 +35,9 @@ import java.util.Set;
public abstract class BaseSparkUpdateStrategy<T> extends UpdateStrategy<T,
HoodieData<HoodieRecord<T>>> {
public BaseSparkUpdateStrategy(HoodieEngineContext engineContext,
HoodieTable table,
- Set<HoodieFileGroupId>
fileGroupsInPendingClustering) {
- super(engineContext, table, fileGroupsInPendingClustering);
+ Set<HoodieFileGroupId>
fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId>
fileGroupsToBeReplaced) {
+ super(engineContext, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
}
/**
@@ -44,9 +45,9 @@ public abstract class BaseSparkUpdateStrategy<T> extends
UpdateStrategy<T, Hoodi
* @param inputRecords the records to write, tagged with target file id
* @return the records matched file group ids
*/
- protected List<HoodieFileGroupId>
getGroupIdsWithUpdate(HoodieData<HoodieRecord<T>> inputRecords) {
- return inputRecords
+ protected Set<HoodieFileGroupId>
getGroupIdsWithUpdate(HoodieData<HoodieRecord<T>> inputRecords) {
+ return new HashSet<>(inputRecords
.filter(record -> record.getCurrentLocation() != null)
- .map(record -> new HoodieFileGroupId(record.getPartitionPath(),
record.getCurrentLocation().getFileId())).distinct().collectAsList();
+ .map(record -> new HoodieFileGroupId(record.getPartitionPath(),
record.getCurrentLocation().getFileId())).distinct().collectAsList());
}
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkAllowUpdateStrategy.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkAllowUpdateStrategy.java
index 7de85ae97787..7da558a5d568 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkAllowUpdateStrategy.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkAllowUpdateStrategy.java
@@ -25,7 +25,6 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.table.HoodieTable;
-import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@@ -35,13 +34,17 @@ import java.util.stream.Collectors;
public class SparkAllowUpdateStrategy<T> extends BaseSparkUpdateStrategy<T> {
public SparkAllowUpdateStrategy(
- HoodieEngineContext engineContext, HoodieTable table,
Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
- super(engineContext, table, fileGroupsInPendingClustering);
+ HoodieEngineContext engineContext, HoodieTable table,
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced) {
+ super(engineContext, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
}
@Override
public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>>
handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
- List<HoodieFileGroupId> fileGroupIdsWithRecordUpdate =
getGroupIdsWithUpdate(taggedRecordsRDD);
+ // TODO: also consider fileGroupsToBeReplaced so INSERT_OVERWRITE
overlapping with pending
+ // clustering can be rejected/handled here for users who set
SparkAllowUpdateStrategy.
+ Set<HoodieFileGroupId> fileGroupIdsWithRecordUpdate =
getGroupIdsWithUpdate(taggedRecordsRDD);
Set<HoodieFileGroupId> fileGroupIdsWithUpdatesAndPendingClustering =
fileGroupIdsWithRecordUpdate.stream()
.filter(fileGroupsInPendingClustering::contains)
.collect(Collectors.toSet());
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java
index eb4b308a668a..d63fe860cdac 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkConsistentBucketDuplicateUpdateStrategy.java
@@ -49,12 +49,16 @@ import static
org.apache.hudi.index.HoodieIndexUtils.tagAsNewRecordIfNeeded;
*/
public class SparkConsistentBucketDuplicateUpdateStrategy<T> extends
UpdateStrategy<T, HoodieData<HoodieRecord<T>>> {
- public SparkConsistentBucketDuplicateUpdateStrategy(HoodieEngineContext
engineContext, HoodieTable table, Set<HoodieFileGroupId>
fileGroupsInPendingClustering) {
- super(engineContext, table, fileGroupsInPendingClustering);
+ public SparkConsistentBucketDuplicateUpdateStrategy(HoodieEngineContext
engineContext, HoodieTable table,
+ Set<HoodieFileGroupId>
fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId>
fileGroupsToBeReplaced) {
+ super(engineContext, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
}
@Override
public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>>
handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
+ // TODO: also consider fileGroupsToBeReplaced so INSERT_OVERWRITE
overlapping with pending
+ // clustering is handled here for the consistent-bucket duplicate-update
strategy.
if (fileGroupsInPendingClustering.isEmpty()) {
return Pair.of(taggedRecordsRDD, Collections.emptySet());
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkRejectUpdateStrategy.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkRejectUpdateStrategy.java
index 8f943d92fd7d..0e9874705968 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkRejectUpdateStrategy.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/update/strategy/SparkRejectUpdateStrategy.java
@@ -29,7 +29,6 @@ import org.apache.hudi.table.HoodieTable;
import lombok.extern.slf4j.Slf4j;
import java.util.Collections;
-import java.util.List;
import java.util.Set;
/**
@@ -39,18 +38,22 @@ import java.util.Set;
@Slf4j
public class SparkRejectUpdateStrategy<T> extends BaseSparkUpdateStrategy<T> {
- public SparkRejectUpdateStrategy(HoodieEngineContext engineContext,
HoodieTable table, Set<HoodieFileGroupId> fileGroupsInPendingClustering) {
- super(engineContext, table, fileGroupsInPendingClustering);
+ public SparkRejectUpdateStrategy(HoodieEngineContext engineContext,
HoodieTable table,
+ Set<HoodieFileGroupId>
fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId>
fileGroupsToBeReplaced) {
+ super(engineContext, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
}
@Override
public Pair<HoodieData<HoodieRecord<T>>, Set<HoodieFileGroupId>>
handleUpdate(HoodieData<HoodieRecord<T>> taggedRecordsRDD) {
- List<HoodieFileGroupId> fileGroupIdsWithRecordUpdate =
getGroupIdsWithUpdate(taggedRecordsRDD);
- fileGroupIdsWithRecordUpdate.forEach(fileGroupIdWithRecordUpdate -> {
- if (fileGroupsInPendingClustering.contains(fileGroupIdWithRecordUpdate))
{
+ Set<HoodieFileGroupId> allAffectedFileGroups =
getGroupIdsWithUpdate(taggedRecordsRDD);
+ // also treat replaced file groups as potential conflict targets
+ allAffectedFileGroups.addAll(fileGroupsToBeReplaced);
+ allAffectedFileGroups.forEach(affectedFileGroup -> {
+ if (fileGroupsInPendingClustering.contains(affectedFileGroup)) {
String msg = String.format("Not allowed to update the clustering file
group %s. "
+ "For pending clustering operations, we are not going to
support update for now.",
- fileGroupIdWithRecordUpdate.toString());
+ affectedFileGroup.toString());
log.error(msg);
throw new HoodieClusteringUpdateException(msg);
}
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
index e79ee136e98f..ed0dcaf321be 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/BaseSparkCommitActionExecutor.java
@@ -44,6 +44,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.execution.SparkLazyInsertIterable;
import org.apache.hudi.index.HoodieIndex;
@@ -121,9 +122,11 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
return inputRecords;
}
- UpdateStrategy<T, HoodieData<HoodieRecord<T>>> updateStrategy =
(UpdateStrategy<T, HoodieData<HoodieRecord<T>>>) ReflectionUtils
- .loadClass(config.getClusteringUpdatesStrategyClass(), new Class<?>[]
{HoodieEngineContext.class, HoodieTable.class, Set.class},
- this.context, table, fileGroupsInPendingClustering);
+ // Get file groups that will be replaced by this operation (for
INSERT_OVERWRITE, etc.)
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced =
getFileGroupsBeingReplaced(inputRecords);
+
+ UpdateStrategy<T, HoodieData<HoodieRecord<T>>> updateStrategy =
+ loadClusteringUpdateStrategy(fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
// For SparkAllowUpdateStrategy with rollback pending clustering as false,
need not handle
// the file group intersection between current ingestion and pending
clustering file groups.
// This will be handled at the conflict resolution strategy.
@@ -163,6 +166,50 @@ public abstract class BaseSparkCommitActionExecutor<T>
extends
return recordsAndPendingClusteringFileGroups.getLeft();
}
+ /**
+ * Get the file groups that will be replaced by the current operation.
+ * This is relevant for INSERT_OVERWRITE, INSERT_OVERWRITE_TABLE, and
DELETE_PARTITION operations.
+ *
+ * @param inputRecords the input records for the operation
+ * @return set of file group IDs that will be replaced
+ */
+ protected Set<HoodieFileGroupId>
getFileGroupsBeingReplaced(HoodieData<HoodieRecord<T>> inputRecords) {
+ // Default implementation returns empty set. Subclasses should override as
needed.
+ return Collections.emptySet();
+ }
+
+ /**
+ * Loads {@code hoodie.clustering.updates.strategy} via reflection,
preferring the new 4-arg
+ * constructor (with {@code fileGroupsToBeReplaced}) and falling back to the
legacy 3-arg
+ * constructor for custom strategies that pre-date this PR.
+ */
+ @SuppressWarnings("unchecked")
+ private UpdateStrategy<T, HoodieData<HoodieRecord<T>>>
loadClusteringUpdateStrategy(
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced) {
+ String strategyClass = config.getClusteringUpdatesStrategyClass();
+ try {
+ return (UpdateStrategy<T, HoodieData<HoodieRecord<T>>>)
ReflectionUtils.loadClass(
+ strategyClass,
+ new Class<?>[] {HoodieEngineContext.class, HoodieTable.class,
Set.class, Set.class},
+ this.context, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
+ } catch (HoodieException ex) {
+ if (!(ex.getCause() instanceof NoSuchMethodException)) {
+ throw ex;
+ }
+ // Legacy custom strategies only have the 3-arg constructor.
INSERT_OVERWRITE overlap with
+ // pending clustering will not be detected for these classes (they never
see
+ // fileGroupsToBeReplaced); recommend bumping to the 4-arg constructor.
+ log.warn("Clustering update strategy {} is missing the 4-arg constructor
with "
+ + "fileGroupsToBeReplaced; falling back to the 3-arg constructor.
INSERT_OVERWRITE "
+ + "overlap with pending clustering will not be detected for this
strategy.", strategyClass);
+ return (UpdateStrategy<T, HoodieData<HoodieRecord<T>>>)
ReflectionUtils.loadClass(
+ strategyClass,
+ new Class<?>[] {HoodieEngineContext.class, HoodieTable.class,
Set.class},
+ this.context, table, fileGroupsInPendingClustering);
+ }
+ }
+
@Override
public HoodieWriteMetadata<HoodieData<WriteStatus>>
execute(HoodieData<HoodieRecord<T>> inputRecords) {
return this.execute(inputRecords, Option.empty());
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteCommitActionExecutor.java
index 6ac976f2e544..0b699ae540e8 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteCommitActionExecutor.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
@@ -41,6 +42,7 @@ import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
public class SparkInsertOverwriteCommitActionExecutor<T>
@@ -82,10 +84,10 @@ public class SparkInsertOverwriteCommitActionExecutor<T>
@Override
protected Map<String, List<String>>
getPartitionToReplacedFileIds(HoodieWriteMetadata<HoodieData<WriteStatus>>
writeMetadata) {
- String staticOverwritePartition =
config.getStringOrDefault(HoodieInternalConfig.STATIC_OVERWRITE_PARTITION_PATHS);
- if (StringUtils.nonEmpty(staticOverwritePartition)) {
+ String staticOverwritePartitionPaths =
config.getStringOrDefault(HoodieInternalConfig.STATIC_OVERWRITE_PARTITION_PATHS);
+ if (StringUtils.nonEmpty(staticOverwritePartitionPaths)) {
// static insert overwrite partitions
- List<String> partitionPaths =
Arrays.asList(staticOverwritePartition.split(","));
+ List<String> partitionPaths =
Arrays.asList(staticOverwritePartitionPaths.split(","));
context.setJobStatus(this.getClass().getSimpleName(), "Getting
ExistingFileIds of matching static partitions");
return
HoodieJavaPairRDD.getJavaPairRDD(context.parallelize(partitionPaths,
partitionPaths.size()).mapToPair(
partitionPath -> Pair.of(partitionPath,
getAllExistingFileIds(partitionPath)))).collectAsMap();
@@ -101,6 +103,26 @@ public class SparkInsertOverwriteCommitActionExecutor<T>
return
table.getSliceView().getLatestFileSlices(partitionPath).map(FileSlice::getFileId).distinct().collect(Collectors.toList());
}
+ @Override
+ protected Set<HoodieFileGroupId>
getFileGroupsBeingReplaced(HoodieData<HoodieRecord<T>> inputRecords) {
+ String staticOverwritePartitionPaths =
config.getStringOrDefault(HoodieInternalConfig.STATIC_OVERWRITE_PARTITION_PATHS);
+ List<String> partitionPaths;
+
+ if (StringUtils.nonEmpty(staticOverwritePartitionPaths)) {
+ // Static insert overwrite: use the configured partitions
+ partitionPaths = Arrays.asList(staticOverwritePartitionPaths.split(","));
+ } else {
+ // Dynamic insert overwrite: determine partitions from input records
+ partitionPaths =
inputRecords.map(HoodieRecord::getPartitionPath).distinct().collectAsList();
+ }
+
+ // Get all file groups in the partitions to be overwritten
+ return partitionPaths.stream()
+ .flatMap(partitionPath -> getAllExistingFileIds(partitionPath).stream()
+ .map(fileId -> new HoodieFileGroupId(partitionPath, fileId)))
+ .collect(Collectors.toSet());
+ }
+
@Override
protected Iterator<List<WriteStatus>> handleInsertPartition(String
instantTime, Integer partition, Iterator recordItr,
Broadcast<SparkBucketInfoGetter> bucketInfoGetter) {
BucketInfo binfo = bucketInfoGetter.getValue().getBucketInfo(partition);
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteTableCommitActionExecutor.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteTableCommitActionExecutor.java
index d300ea683a90..67f1ab352073 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteTableCommitActionExecutor.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/SparkInsertOverwriteTableCommitActionExecutor.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.common.model.HoodieFileGroupId;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.util.collection.Pair;
@@ -31,8 +32,10 @@ import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
public class SparkInsertOverwriteTableCommitActionExecutor<T>
extends SparkInsertOverwriteCommitActionExecutor<T> {
@@ -53,4 +56,23 @@ public class SparkInsertOverwriteTableCommitActionExecutor<T>
return
HoodieJavaPairRDD.getJavaPairRDD(context.parallelize(partitionPaths,
partitionPaths.size()).mapToPair(
partitionPath -> Pair.of(partitionPath,
getAllExistingFileIds(partitionPath)))).collectAsMap();
}
+
+ @Override
+ protected Set<HoodieFileGroupId>
getFileGroupsBeingReplaced(HoodieData<HoodieRecord<T>> inputRecords) {
+ // INSERT_OVERWRITE_TABLE replaces every file group across every
partition, not just the
+ // partitions present in the input records. Enumerate all partitions in
parallel via the
+ // engine context (matches the parallelization in
getPartitionToReplacedFileIds above and
+ // avoids a sequential driver-side walk for tables with many partitions
whose file system
+ // view isn't fully cached).
+ List<String> partitionPaths = FSUtils.getAllPartitionPaths(context,
table.getMetaClient(), config.getMetadataConfig());
+ if (partitionPaths == null || partitionPaths.isEmpty()) {
+ return Collections.emptySet();
+ }
+ context.setJobStatus(this.getClass().getSimpleName(), "Resolving file
groups being replaced across all partitions");
+ return new HashSet<>(context.parallelize(partitionPaths,
partitionPaths.size())
+ .flatMap(partitionPath ->
table.getSliceView().getLatestFileSlices(partitionPath)
+ .map(fileSlice -> new HoodieFileGroupId(partitionPath,
fileSlice.getFileId()))
+ .iterator())
+ .collectAsList());
+ }
}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestInsertOverwriteWithClustering.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestInsertOverwriteWithClustering.java
new file mode 100644
index 000000000000..955a8160c264
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/commit/TestInsertOverwriteWithClustering.java
@@ -0,0 +1,834 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.action.commit;
+
+import org.apache.hudi.client.HoodieWriteResult;
+import org.apache.hudi.client.SparkRDDWriteClient;
+import org.apache.hudi.client.WriteClientTestUtils;
+import org.apache.hudi.client.WriteStatus;
+import
org.apache.hudi.client.clustering.plan.strategy.SparkSingleFileSortPlanStrategy;
+import
org.apache.hudi.client.clustering.run.strategy.SparkSingleFileSortExecutionStrategy;
+import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieReplaceCommitMetadata;
+import org.apache.hudi.common.model.HoodieTableType;
+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.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.util.ClusteringUtils;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieClusteringConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.data.HoodieJavaRDD;
+import org.apache.hudi.exception.HoodieUpsertException;
+import org.apache.hudi.table.HoodieSparkTable;
+import org.apache.hudi.table.HoodieTable;
+import org.apache.hudi.testutils.HoodieClientTestBase;
+
+import org.apache.spark.api.java.JavaRDD;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static
org.apache.hudi.common.testutils.HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for INSERT_OVERWRITE, INSERT_OVERWRITE_TABLE, and DELETE_PARTITION
operations
+ * when there are pending clustering operations on the file groups being
replaced.
+ */
+public class TestInsertOverwriteWithClustering extends HoodieClientTestBase {
+
+ private HoodieTestDataGenerator dataGen;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ initPath();
+ initSparkContexts();
+ initTestDataGenerator();
+ initMetaClient(HoodieTableType.COPY_ON_WRITE);
+ dataGen = new HoodieTestDataGenerator();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ cleanupResources();
+ }
+
+ private HoodieClusteringConfig.Builder baseClusteringConfigBuilder(boolean
rollbackPendingClustering) {
+ return HoodieClusteringConfig.newBuilder()
+
.withClusteringPlanStrategyClass(SparkSingleFileSortPlanStrategy.class.getName())
+
.withClusteringExecutionStrategyClass(SparkSingleFileSortExecutionStrategy.class.getName())
+ .withClusteringMaxNumGroups(10)
+ .withRollbackPendingClustering(rollbackPendingClustering);
+ }
+
+ private HoodieWriteConfig.Builder getConfigBuilder(boolean
rollbackPendingClustering) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .withBulkInsertParallelism(2)
+ .withFinalizeWriteParallelism(2)
+ .withDeleteParallelism(2)
+ .withRollbackParallelism(2)
+
.withClusteringConfig(baseClusteringConfigBuilder(rollbackPendingClustering).build());
+ }
+
+ private HoodieWriteConfig.Builder
getConfigBuilderWithPartitionFilter(boolean rollbackPendingClustering, String
partitionFilter) {
+ return HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(TRIP_EXAMPLE_SCHEMA)
+ .withParallelism(2, 2)
+ .withBulkInsertParallelism(2)
+ .withFinalizeWriteParallelism(2)
+ .withDeleteParallelism(2)
+ .withRollbackParallelism(2)
+
.withClusteringConfig(baseClusteringConfigBuilder(rollbackPendingClustering)
+ .withClusteringPartitionSelected(partitionFilter)
+ .build());
+ }
+
+ /**
+ * Test that INSERT_OVERWRITE operation throws an exception when file groups
+ * to be replaced conflict with pending clustering.
+ */
+ @Test
+ public void testStaticInsertOverwriteWithPendingClusteringRejectsUpdate()
throws Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Initial insert to create some data
+ String instant1 = nextInstant();
+ List<HoodieRecord> records1 = dataGen.generateInserts(instant1, 100);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ // Get partition path from the first record
+ String partitionPath = records1.get(0).getPartitionPath();
+
+ // Step 2: Schedule clustering for the partition
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline = metaClient.getActiveTimeline()
+ .filterPendingClusteringTimeline();
+ assertEquals(1, pendingReplaceTimeline.countInstants());
+
+ // Get file groups involved in pending clustering
+ Set<HoodieFileGroupId> pendingClusteringFileGroups = ClusteringUtils
+ .getAllFileGroupsInPendingClusteringPlans(metaClient).keySet();
+ assertFalse(pendingClusteringFileGroups.isEmpty());
+
+ // Step 3: Perform static INSERT_OVERWRITE on the same partition
+ // This should throw HoodieUpsertException because file groups to be
replaced
+ // conflict with pending clustering
+ String instant3 = nextInstant();
+ List<HoodieRecord> records3 =
dataGen.generateInsertsForPartition(instant3, 50, partitionPath);
+ JavaRDD<HoodieRecord> writeRecords3 = jsc.parallelize(records3, 2);
+
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieUpsertException upsertException =
assertThrows(HoodieUpsertException.class, () ->
+ commitInsertOverwrite(client, writeRecords3, instant3)
+ );
+ assertTrue(upsertException.getCause().getMessage().contains("Not allowed
to update the clustering file group"));
+
+ // Verify clustering is still pending (NOT rolled back)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertEquals(1, pendingReplaceTimeline.countInstants(),
+ "Pending clustering should remain pending after failed
INSERT_OVERWRITE");
+ assertTrue(pendingReplaceTimeline.containsInstant(clusteringInstant));
+
+ // Verify the INSERT_OVERWRITE did NOT complete
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertFalse(completedReplaceTimeline.containsInstant(instant3));
+ }
+
+ /**
+ * Test that dynamic INSERT_OVERWRITE_TABLE operation gets aborted when it
overlaps w/ pending clustering.
+ */
+ @Test
+ public void testDynamicInsertOverwriteWithPendingClustering() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(true).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Initial insert to create data in multiple partitions
+ String instant1 = nextInstant();
+ List<HoodieRecord> records1 = dataGen.generateInserts(instant1, 100);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ // Get partitions that have data
+ Set<String> partitionsWithData = records1.stream()
+ .map(HoodieRecord::getPartitionPath)
+ .collect(Collectors.toSet());
+
+ // Step 2: Schedule clustering
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ // Get file groups involved in pending clustering
+ Set<HoodieFileGroupId> pendingClusteringFileGroups = ClusteringUtils
+ .getAllFileGroupsInPendingClusteringPlans(metaClient).keySet();
+ assertFalse(pendingClusteringFileGroups.isEmpty());
+
+ // Step 3: Perform dynamic INSERT_OVERWRITE_TABLE on overlapping partitions
+ String instant3 = nextInstant();
+ String targetPartition = partitionsWithData.iterator().next();
+ List<HoodieRecord> records3 =
dataGen.generateInsertsForPartition(instant3, 50, targetPartition);
+ JavaRDD<HoodieRecord> writeRecords3 = jsc.parallelize(records3, 2);
+ HoodieUpsertException upsertException =
assertThrows(HoodieUpsertException.class, () ->
+ commitInsertOverwrite(client, writeRecords3, instant3)
+ );
+ assertTrue(upsertException.getCause().getMessage().contains("Not allowed
to update the clustering file group"));
+
+ // Verify clustering was never rolled back.
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertTrue(pendingReplaceTimeline.containsInstant(clusteringInstant),
+ "Pending clustering should not be rolled back");
+
+ // Verify the INSERT_OVERWRITE did NOT complete
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertFalse(completedReplaceTimeline.containsInstant(instant3));
+ }
+
+ /**
+ * Test that DELETE_PARTITION operation succeeds when pending clustering
does not overlap.
+ */
+ @Test
+ public void testDeletePartitionWithNonOverlappingPendingClustering() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilderWithPartitionFilter(false,
"partition1").build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: insert into partition1
+ String instant1 = nextInstant();
+ List<HoodieRecord> records1 =
dataGen.generateInsertsForPartition(instant1, 100, "partition1");
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ String instant2 = nextInstant();
+ List<HoodieRecord> records2 =
dataGen.generateInsertsForPartition(instant2, 100, "partition2");
+ JavaRDD<HoodieRecord> writeRecords2 = jsc.parallelize(records2, 2);
+ commitInsert(client, writeRecords2, instant2);
+
+ // Step 3: Schedule clustering for the partition1
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ // Get file groups involved in pending clustering
+ Set<HoodieFileGroupId> pendingClusteringFileGroups = ClusteringUtils
+ .getAllFileGroupsInPendingClusteringPlans(metaClient).keySet();
+ assertFalse(pendingClusteringFileGroups.isEmpty());
+
+ client.close();
+ SparkRDDWriteClient client2 = getHoodieWriteClient(config);
+
+ // Step 3: Delete the partition2
+ String instant3 = nextInstant();
+ commitDeletePartitions(client2, Arrays.asList("partition2"), instant3);
+
+ // Verify clustering is still pending (NOT rolled back)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertTrue(pendingReplaceTimeline.containsInstant(clusteringInstant),
+ "Pending clustering should remain pending after successful
DELETE_PARTITION");
+
+ // Verify the DELETE_PARTITION is completed
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertTrue(completedReplaceTimeline.containsInstant(instant3));
+ }
+
+ /**
+ * Test getFileGroupsBeingReplaced method in
SparkInsertOverwriteCommitActionExecutor
+ * for static INSERT_OVERWRITE scenario.
+ */
+ @Test
+ public void testGetFileGroupsBeingReplacedForStaticOverwrite() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(true).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Initial insert to create some data
+ String instant1 = nextInstant();
+ String partitionPath = "2023/01/01";
+ List<HoodieRecord> records1 =
dataGen.generateInsertsForPartition(instant1, 100, partitionPath);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ // Step 2: Get the file groups in the partition
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ List<String> existingFileIds = table.getSliceView()
+ .getLatestFileSlices(partitionPath)
+ .map(fileSlice -> fileSlice.getFileId())
+ .collect(Collectors.toList());
+ assertFalse(existingFileIds.isEmpty(), "Should have at least one file
group");
+
+ // Step 3: Create INSERT_OVERWRITE executor and test
getFileGroupsBeingReplaced
+ String instant2 = nextInstant();
+ List<HoodieRecord> records = dataGen.generateInsertsForPartition(instant2,
10, partitionPath);
+ HoodieData<HoodieRecord> inputRecords =
HoodieJavaRDD.of(jsc.parallelize(records, 1));
+
+ SparkInsertOverwriteCommitActionExecutor executor =
+ new SparkInsertOverwriteCommitActionExecutor(
+ context, config, table, instant2, inputRecords);
+
+ // Invoke the method - it's protected so we test it indirectly through the
workflow
+ Set<HoodieFileGroupId> fileGroupsBeingReplaced =
executor.getFileGroupsBeingReplaced(inputRecords);
+
+ // Verify that the file groups in the partition are identified as being
replaced
+ assertEquals(existingFileIds.size(), fileGroupsBeingReplaced.size(),
+ "Should identify all existing file groups in the partition as being
replaced");
+ assertTrue(fileGroupsBeingReplaced.stream()
+ .allMatch(fg -> fg.getPartitionPath().equals(partitionPath)),
+ "All identified file groups should be in the target partition");
+ }
+
+ /**
+ * Test that INSERT_OVERWRITE on a non-overlapping partition succeeds
+ * even when there is pending clustering on a different partition.
+ */
+ @Test
+ public void
testInsertOverwriteNonOverlappingPartitionWithPendingClustering() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(true)
+ .withClusteringConfig(HoodieClusteringConfig.newBuilder()
+
.withClusteringPlanStrategyClass(SparkSingleFileSortPlanStrategy.class.getName())
+
.withClusteringExecutionStrategyClass(SparkSingleFileSortExecutionStrategy.class.getName())
+ .withClusteringMaxNumGroups(10)
+ .withRollbackPendingClustering(true)
+ .withClusteringPartitionSelected("partition1")
+ .build())
+ .build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into partition1
+ String instant1 = nextInstant();
+ List<HoodieRecord> records1 =
dataGen.generateInsertsForPartition(instant1, 100, "partition1");
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(records1, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ String instant2 = nextInstant();
+ List<HoodieRecord> records2 =
dataGen.generateInsertsForPartition(instant1, 100, "partition2");
+ JavaRDD<HoodieRecord> writeRecords2 = jsc.parallelize(records2, 2);
+ commitInsert(client, writeRecords2, instant2);
+
+ // Step 2: Schedule clustering for partition1
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ // Step 3: Perform INSERT_OVERWRITE on partition2 (non-overlapping)
+ String instant3 = nextInstant();
+ List<HoodieRecord> records3 =
dataGen.generateInsertsForPartition(instant3, 50, "partition2");
+ JavaRDD<HoodieRecord> writeRecords3 = jsc.parallelize(records3, 2);
+ commitInsertOverwrite(client, writeRecords3, instant3);
+
+ // Verify clustering was NOT rolled back (no overlap)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertEquals(1, pendingReplaceTimeline.countInstants(),
+ "Pending clustering should NOT be rolled back for non-overlapping
partition");
+
+ // Verify the INSERT_OVERWRITE completed successfully
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertTrue(completedReplaceTimeline.containsInstant(instant3));
+ }
+
+ /**
+ * Test dynamic INSERT_OVERWRITE that determines partitions from input
records.
+ * If input records target a partition with pending clustering, it should
detect the conflict and abort.
+ */
+ @Test
+ public void testDynamicInsertOverwriteDetectsOverlap() throws Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into multiple partitions
+ String instant1 = nextInstant();
+ List<HoodieRecord> partition1Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/01");
+ List<HoodieRecord> partition2Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/02");
+ List<HoodieRecord> partition3Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/03");
+ List<HoodieRecord> allRecords = new ArrayList<>();
+ allRecords.addAll(partition1Records);
+ allRecords.addAll(partition2Records);
+ allRecords.addAll(partition3Records);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(allRecords, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ // Step 2: Schedule clustering
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ // Get partitions in clustering
+ Set<HoodieFileGroupId> clusteringFileGroups = ClusteringUtils
+ .getAllFileGroupsInPendingClusteringPlans(metaClient).keySet();
+ Set<String> clusteringPartitions = clusteringFileGroups.stream()
+ .map(HoodieFileGroupId::getPartitionPath)
+ .collect(Collectors.toSet());
+ assertFalse(clusteringPartitions.isEmpty());
+
+ // Step 3: Perform dynamic INSERT_OVERWRITE_TABLE with records in
overlapping partition
+ // This should fail
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ String instant3 = nextInstant();
+ String targetPartition = clusteringPartitions.iterator().next();
+ List<HoodieRecord> records3 =
dataGen.generateInsertsForPartition(instant3, 50, targetPartition);
+ JavaRDD<HoodieRecord> writeRecords3 = jsc.parallelize(records3, 2);
+ HoodieUpsertException upsertException =
assertThrows(HoodieUpsertException.class, () ->
+ commitInsertOverwrite(client, writeRecords3, instant3)
+ );
+ assertTrue(upsertException.getCause().getMessage().contains("Not allowed
to update the clustering file group"));
+
+ // Verify clustering is still pending (NOT rolled back)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertEquals(1, pendingReplaceTimeline.countInstants(),
+ "Pending clustering should remain pending after failed
INSERT_OVERWRITE_TABLE");
+ assertTrue(pendingReplaceTimeline.containsInstant(clusteringInstant));
+
+ // Verify the INSERT_OVERWRITE_TABLE did NOT complete
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertFalse(completedReplaceTimeline.containsInstant(instant3));
+ }
+
+ /**
+ * Test multiple concurrent INSERT_OVERWRITE operations on different
partitions
+ * with one partition having pending clustering.
+ */
+ @Test
+ public void testMultipleInsertOverwriteWithSelectiveOverlap() throws
Exception {
+ HoodieWriteConfig config =
getConfigBuilderWithPartitionFilter(false,"2023/01/01,2023/01/02").build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into three partitions
+ String instant1 = nextInstant();
+ List<HoodieRecord> partition1Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/01");
+ List<HoodieRecord> partition2Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/02");
+ List<HoodieRecord> partition3Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/03");
+ List<HoodieRecord> allRecords = new ArrayList<>();
+ allRecords.addAll(partition1Records);
+ allRecords.addAll(partition2Records);
+ allRecords.addAll(partition3Records);
+ JavaRDD<HoodieRecord> writeRecords1 = jsc.parallelize(allRecords, 2);
+ commitInsert(client, writeRecords1, instant1);
+
+ // Step 2: Schedule clustering (will cluster partition1 and partition2)
+ Option<String> clusteringInstantOpt =
client.scheduleClustering(Option.empty());
+ assertTrue(clusteringInstantOpt.isPresent(), "Expected clustering to be
scheduled but returned empty");
+ String clusteringInstant = clusteringInstantOpt.get();
+
+ // Verify clustering is pending
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants());
+
+ client.close();
+ SparkRDDWriteClient client2 = getHoodieWriteClient(config);
+
+ // Step 3: Perform INSERT_OVERWRITE on partition3 (no overlap) - should
succeed without rollback
+ String instant3 = nextInstant();
+ List<HoodieRecord> records3 =
dataGen.generateInsertsForPartition(instant3, 50, "2023/01/03");
+ JavaRDD<HoodieRecord> writeRecords3 = jsc.parallelize(records3, 2);
+ commitInsertOverwrite(client2, writeRecords3, instant3);
+
+ // Verify clustering is still pending (no rollback for non-overlapping
partition)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ assertEquals(1,
metaClient.getActiveTimeline().filterPendingClusteringTimeline().countInstants(),
+ "Clustering should still be pending after INSERT_OVERWRITE on
non-overlapping partition");
+
+ client2.close();
+ SparkRDDWriteClient client3 = getHoodieWriteClient(config);
+ // Step 4: Now perform INSERT_OVERWRITE on partition1 (with overlap) -
should fail
+ String instant4 = nextInstant();
+ List<HoodieRecord> records4 =
dataGen.generateInsertsForPartition(instant4, 50, "2023/01/01");
+ JavaRDD<HoodieRecord> writeRecords4 = jsc.parallelize(records4, 2);
+ HoodieUpsertException upsertException =
assertThrows(HoodieUpsertException.class, () ->
+ commitInsertOverwrite(client3, writeRecords4, instant4)
+ );
+ assertTrue(upsertException.getCause().getMessage().contains("Not allowed
to update the clustering file group"));
+
+ // Verify clustering is still pending (NOT rolled back)
+ metaClient = HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTimeline pendingReplaceTimeline =
metaClient.getActiveTimeline().filterPendingClusteringTimeline();
+ assertTrue(pendingReplaceTimeline.containsInstant(clusteringInstant),
"Pending clustering should remain pending after failed INSERT_OVERWRITE");
+
+ // Verify the INSERT_OVERWRITE did NOT complete
+ HoodieTimeline completedReplaceTimeline = metaClient.getCommitTimeline()
+ .filterCompletedInstants();
+ assertFalse(completedReplaceTimeline.containsInstant(instant4));
+ }
+
+ /**
+ * Test getPartitionToReplacedFileIds for static INSERT_OVERWRITE.
+ * Static overwrite uses configured partition paths, not input records.
+ */
+ @Test
+ public void testGetPartitionToReplacedFileIdsForStaticOverwrite() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into multiple partitions
+ String instant1 = nextInstant();
+ List<HoodieRecord> partition1Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/01");
+ List<HoodieRecord> partition2Records =
dataGen.generateInsertsForPartition(instant1, 100, "2023/01/02");
+ List<HoodieRecord> allRecords = new ArrayList<>();
+ allRecords.addAll(partition1Records);
+ allRecords.addAll(partition2Records);
+ commitInsert(client, jsc.parallelize(allRecords, 2), instant1);
+
+ // Insert more data to create multiple file groups per partition
+ String instant2 = nextInstant();
+ List<HoodieRecord> moreRecords1 =
dataGen.generateInsertsForPartition(instant2, 50, "2023/01/01");
+ List<HoodieRecord> moreRecords2 =
dataGen.generateInsertsForPartition(instant2, 50, "2023/01/02");
+ List<HoodieRecord> moreRecords = new ArrayList<>();
+ moreRecords.addAll(moreRecords1);
+ moreRecords.addAll(moreRecords2);
+ commitInsert(client, jsc.parallelize(moreRecords, 2), instant2);
+
+ // Get existing file IDs before overwrite
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ List<String> partition1FileIds = table.getSliceView()
+ .getLatestFileSlices("2023/01/01")
+ .map(slice -> slice.getFileId())
+ .collect(Collectors.toList());
+ List<String> partition2FileIds = table.getSliceView()
+ .getLatestFileSlices("2023/01/02")
+ .map(slice -> slice.getFileId())
+ .collect(Collectors.toList());
+
+ assertFalse(partition1FileIds.isEmpty(), "Partition 2023/01/01 should have
file groups");
+ assertFalse(partition2FileIds.isEmpty(), "Partition 2023/01/02 should have
file groups");
+
+ // Step 2: Perform static INSERT_OVERWRITE on partition 2023/01/01 only
+ String instant3 = nextInstant();
+ List<HoodieRecord> overwriteRecords =
dataGen.generateInsertsForPartition(instant3, 30, "2023/01/01");
+ commitInsertOverwrite(client, jsc.parallelize(overwriteRecords, 2),
instant3);
+
+ // Step 3: Verify replaced file IDs - should only include partition
2023/01/01
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant instant3Instant =
metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(i -> i.requestedTime().equals(instant3)).firstInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
metaClient.getActiveTimeline()
+ .readReplaceCommitMetadata(instant3Instant);
+
+ Map<String, List<String>> partitionToReplacedFileIds =
replaceMetadata.getPartitionToReplaceFileIds();
+
+ // Verify partition 2023/01/01 is in replaced file IDs
+ assertTrue(partitionToReplacedFileIds.containsKey("2023/01/01"),
+ "Partition 2023/01/01 should be in replaced file IDs");
+
+ // Verify all file IDs from partition 2023/01/01 are marked as replaced
+ List<String> replacedFileIds =
partitionToReplacedFileIds.get("2023/01/01");
+ assertEquals(partition1FileIds.size(), replacedFileIds.size(),
+ "All file IDs from partition 2023/01/01 should be marked as replaced");
+ assertTrue(replacedFileIds.containsAll(partition1FileIds),
+ "Replaced file IDs should match original file IDs in partition
2023/01/01");
+
+ // Verify partition 2023/01/02 is NOT in replaced file IDs (was not
overwritten)
+ assertFalse(partitionToReplacedFileIds.containsKey("2023/01/02"),
+ "Partition 2023/01/02 should NOT be in replaced file IDs");
+ }
+
+ /**
+ * Test getPartitionToReplacedFileIds for dynamic INSERT_OVERWRITE.
+ * Dynamic overwrite determines partitions from input records.
+ */
+ @Test
+ public void testGetPartitionToReplacedFileIdsForDynamicOverwrite() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into three partitions
+ String instant1 = nextInstant();
+ List<HoodieRecord> partition1Records =
dataGen.generateInsertsForPartition(instant1, 50, "2023/01/01");
+ List<HoodieRecord> partition2Records =
dataGen.generateInsertsForPartition(instant1, 50, "2023/01/02");
+ List<HoodieRecord> partition3Records =
dataGen.generateInsertsForPartition(instant1, 50, "2023/01/03");
+ List<HoodieRecord> allRecords = new ArrayList<>();
+ allRecords.addAll(partition1Records);
+ allRecords.addAll(partition2Records);
+ allRecords.addAll(partition3Records);
+ commitInsert(client, jsc.parallelize(allRecords, 2), instant1);
+
+ // Get existing file IDs
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ List<String> partition1FileIds = table.getSliceView()
+ .getLatestFileSlices("2023/01/01")
+ .map(slice -> slice.getFileId())
+ .collect(Collectors.toList());
+ List<String> partition2FileIds = table.getSliceView()
+ .getLatestFileSlices("2023/01/02")
+ .map(slice -> slice.getFileId())
+ .collect(Collectors.toList());
+
+ // Step 2: Perform dynamic INSERT_OVERWRITE with records for partitions
01/01 and 01/02
+ String instant2 = nextInstant();
+ List<HoodieRecord> overwriteRecords = new ArrayList<>();
+ overwriteRecords.addAll(dataGen.generateInsertsForPartition(instant2, 30,
"2023/01/01"));
+ overwriteRecords.addAll(dataGen.generateInsertsForPartition(instant2, 30,
"2023/01/02"));
+ commitInsertOverwrite(client, jsc.parallelize(overwriteRecords, 2),
instant2);
+
+ // Step 3: Verify replaced file IDs include both partitions
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant instant2Instant =
metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(i -> i.requestedTime().equals(instant2)).firstInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
metaClient.getActiveTimeline()
+ .readReplaceCommitMetadata(instant2Instant);
+
+ Map<String, List<String>> partitionToReplacedFileIds =
replaceMetadata.getPartitionToReplaceFileIds();
+
+ // Verify both partitions are in replaced file IDs
+ assertTrue(partitionToReplacedFileIds.containsKey("2023/01/01"),
+ "Partition 2023/01/01 should be in replaced file IDs");
+ assertTrue(partitionToReplacedFileIds.containsKey("2023/01/02"),
+ "Partition 2023/01/02 should be in replaced file IDs");
+
+ // Verify file IDs match
+ assertEquals(partition1FileIds.size(),
partitionToReplacedFileIds.get("2023/01/01").size());
+ assertEquals(partition2FileIds.size(),
partitionToReplacedFileIds.get("2023/01/02").size());
+
+ // Verify partition 2023/01/03 is NOT in replaced file IDs
+ assertFalse(partitionToReplacedFileIds.containsKey("2023/01/03"),
+ "Partition 2023/01/03 should NOT be in replaced file IDs");
+ }
+
+ /**
+ * Test getPartitionToReplacedFileIds when overwriting a partition with
multiple file groups.
+ */
+ @Test
+ public void testGetPartitionToReplacedFileIdsWithMultipleFileGroups() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Create multiple file groups in a single partition through
multiple inserts
+ String partitionPath = "2023/01/01";
+ for (int i = 1; i <= 3; i++) {
+ String instant = nextInstant();
+ List<HoodieRecord> records =
dataGen.generateInsertsForPartition(instant, 50, partitionPath);
+ commitBulkInsert(client, jsc.parallelize(records, 2), instant);
+ }
+
+ // Get all file IDs in the partition
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ List<String> existingFileIds = table.getSliceView()
+ .getLatestFileSlices(partitionPath)
+ .map(slice -> slice.getFileId())
+ .distinct()
+ .collect(Collectors.toList());
+
+ assertTrue(existingFileIds.size() >= 2,
+ "Should have multiple file groups in partition from multiple inserts");
+
+ // Step 2: Perform INSERT_OVERWRITE
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ String overwriteInstant = nextInstant();
+ List<HoodieRecord> overwriteRecords =
dataGen.generateInsertsForPartition(overwriteInstant, 100, partitionPath);
+ commitInsertOverwrite(client, jsc.parallelize(overwriteRecords, 2),
overwriteInstant);
+
+ // Step 3: Verify all file groups are marked as replaced
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant overwriteInstantObj =
metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(i ->
i.requestedTime().equals(overwriteInstant)).firstInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
metaClient.getActiveTimeline()
+ .readReplaceCommitMetadata(overwriteInstantObj);
+
+ Map<String, List<String>> partitionToReplacedFileIds =
replaceMetadata.getPartitionToReplaceFileIds();
+
+ assertTrue(partitionToReplacedFileIds.containsKey(partitionPath));
+ List<String> replacedFileIds =
partitionToReplacedFileIds.get(partitionPath);
+
+ // All existing file IDs should be marked as replaced
+ assertEquals(existingFileIds.size(), replacedFileIds.size(),
+ "All file groups should be marked as replaced");
+ assertTrue(replacedFileIds.containsAll(existingFileIds),
+ "Replaced file IDs should include all original file groups");
+ }
+
+ /**
+ * Test getPartitionToReplacedFileIds when overwriting an empty partition.
+ */
+ @Test
+ public void testGetPartitionToReplacedFileIdsForEmptyPartition() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data into partition1
+ String instant1 = nextInstant();
+ String partition1 = "2023/01/01";
+ List<HoodieRecord> records1 =
dataGen.generateInsertsForPartition(instant1, 100, partition1);
+ commitInsert(client, jsc.parallelize(records1, 2), instant1);
+
+ // Step 2: Perform INSERT_OVERWRITE on a different partition (empty
partition2)
+ String instant2 = nextInstant();
+ String partition2 = "2023/01/02";
+ List<HoodieRecord> overwriteRecords =
dataGen.generateInsertsForPartition(instant2, 50, partition2);
+ commitInsertOverwrite(client, jsc.parallelize(overwriteRecords, 2),
instant2);
+
+ // Step 3: Verify partition2 is in replaced file IDs even though it was
empty
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieInstant instant2Instant =
metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(i -> i.requestedTime().equals(instant2)).firstInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
metaClient.getActiveTimeline()
+ .readReplaceCommitMetadata(instant2Instant);
+ Map<String, List<String>> partitionToReplacedFileIds =
replaceMetadata.getPartitionToReplaceFileIds();
+
+ // partition2 should be present with empty list (no existing files to
replace)
+ assertTrue(partitionToReplacedFileIds.containsKey(partition2),
+ "Empty partition should still be in replaced file IDs map");
+ assertTrue(partitionToReplacedFileIds.get(partition2).isEmpty(),
+ "Empty partition should have empty list of replaced file IDs");
+
+ // partition1 should NOT be in replaced file IDs
+ assertFalse(partitionToReplacedFileIds.containsKey(partition1),
+ "Non-overwritten partition should not be in replaced file IDs");
+ }
+
+ /**
+ * Test getPartitionToReplacedFileIds validates actual file IDs, not just
counts.
+ * Ensures the specific file IDs returned match the file groups that existed.
+ */
+ @Test
+ public void testGetPartitionToReplacedFileIdsValidatesActualFileIds() throws
Exception {
+ HoodieWriteConfig config = getConfigBuilder(false).build();
+ SparkRDDWriteClient client = getHoodieWriteClient(config);
+
+ // Step 1: Insert data
+ String instant1 = nextInstant();
+ String partitionPath = "2023/01/01";
+ List<HoodieRecord> records = dataGen.generateInsertsForPartition(instant1,
100, partitionPath);
+ commitInsert(client, jsc.parallelize(records, 2), instant1);
+
+ // Step 2: Insert more data to create additional file groups
+ String instant2 = nextInstant();
+ List<HoodieRecord> moreRecords =
dataGen.generateInsertsForPartition(instant2, 50, partitionPath);
+ commitInsert(client, jsc.parallelize(moreRecords, 2), instant2);
+
+ // Capture exact file IDs before overwrite
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.reload(this.metaClient);
+ HoodieTable table = HoodieSparkTable.create(config, context, metaClient);
+ Set<String> expectedFileIds = table.getSliceView()
+ .getLatestFileSlices(partitionPath)
+ .map(slice -> slice.getFileId())
+ .collect(Collectors.toSet());
+
+ assertFalse(expectedFileIds.isEmpty(), "Should have file groups before
overwrite");
+
+ // Step 3: Perform INSERT_OVERWRITE
+ String instant3 = nextInstant();
+ List<HoodieRecord> overwriteRecords =
dataGen.generateInsertsForPartition(instant3, 75, partitionPath);
+ commitInsertOverwrite(client, jsc.parallelize(overwriteRecords, 2),
instant3);
+
+ // Step 4: Validate exact file IDs in replaced list
+ metaClient = HoodieTableMetaClient.reload(metaClient);
+ HoodieInstant instant3Instant =
metaClient.getActiveTimeline().filterCompletedInstants()
+ .filter(i -> i.requestedTime().equals(instant3)).firstInstant().get();
+ HoodieReplaceCommitMetadata replaceMetadata =
metaClient.getActiveTimeline()
+ .readReplaceCommitMetadata(instant3Instant);
+ Map<String, List<String>> partitionToReplacedFileIds =
replaceMetadata.getPartitionToReplaceFileIds();
+
+ Set<String> actualReplacedFileIds = new
HashSet<>(partitionToReplacedFileIds.get(partitionPath));
+
+ // Validate exact file IDs, not just counts
+ assertEquals(expectedFileIds, actualReplacedFileIds,
+ "Replaced file IDs should exactly match the file groups that existed
before overwrite");
+ }
+
+ // Hudi instant times are millisecond-resolution; sleep briefly between
generations so
+ // back-to-back instants in a single test method are strictly ordered.
+ private String nextInstant() throws InterruptedException {
+ Thread.sleep(2);
+ return WriteClientTestUtils.createNewInstantTime();
+ }
+
+ private JavaRDD<WriteStatus> commitInsert(SparkRDDWriteClient client,
JavaRDD<HoodieRecord> records, String instantTime) {
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses = client.insert(records, instantTime);
+ List<WriteStatus> statusList = writeStatuses.collect();
+ client.commit(instantTime, jsc.parallelize(statusList, 1));
+ return writeStatuses;
+ }
+
+ private JavaRDD<WriteStatus> commitBulkInsert(SparkRDDWriteClient client,
JavaRDD<HoodieRecord> records, String instantTime) {
+ WriteClientTestUtils.startCommitWithTime(client, instantTime);
+ JavaRDD<WriteStatus> writeStatuses = client.bulkInsert(records,
instantTime);
+ List<WriteStatus> statusList = writeStatuses.collect();
+ client.commit(instantTime, jsc.parallelize(statusList, 1));
+ return writeStatuses;
+ }
+
+ private HoodieWriteResult commitInsertOverwrite(SparkRDDWriteClient client,
JavaRDD<HoodieRecord> records, String instantTime) {
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
HoodieTimeline.REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result = client.insertOverwrite(records, instantTime);
+ List<WriteStatus> statusList = result.getWriteStatuses().collect();
+ client.commit(instantTime, jsc.parallelize(statusList, 1), Option.empty(),
HoodieTimeline.REPLACE_COMMIT_ACTION,
+ result.getPartitionToReplaceFileIds());
+ return result;
+ }
+
+ private HoodieWriteResult commitDeletePartitions(SparkRDDWriteClient client,
List<String> partitions, String instantTime) {
+ WriteClientTestUtils.startCommitWithTime(client, instantTime,
HoodieTimeline.REPLACE_COMMIT_ACTION);
+ HoodieWriteResult result = client.deletePartitions(partitions,
instantTime);
+ List<WriteStatus> statusList = result.getWriteStatuses().collect();
+ client.commit(instantTime, jsc.parallelize(statusList, 1), Option.empty(),
HoodieTimeline.REPLACE_COMMIT_ACTION,
+ result.getPartitionToReplaceFileIds());
+ return result;
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/ConsistentBucketUpdateStrategy.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/ConsistentBucketUpdateStrategy.java
index c7ea6d8747f6..3f4515e1b2f1 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/ConsistentBucketUpdateStrategy.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/ConsistentBucketUpdateStrategy.java
@@ -62,7 +62,7 @@ public class ConsistentBucketUpdateStrategy<T> extends
UpdateStrategy<T, List<Bu
public ConsistentBucketUpdateStrategy(
HoodieFlinkWriteClient writeClient, List<String> indexKeyFields) {
- super(writeClient.getEngineContext(), writeClient.getHoodieTable(),
Collections.emptySet());
+ super(writeClient.getEngineContext(), writeClient.getHoodieTable(),
Collections.emptySet(), Collections.emptySet());
this.indexKeyFields = indexKeyFields;
this.partitionToIdentifier = new HashMap<>();
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/FlinkConsistentBucketUpdateStrategy.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/FlinkConsistentBucketUpdateStrategy.java
index b6159f411bd2..9272653096da 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/FlinkConsistentBucketUpdateStrategy.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/update/strategy/FlinkConsistentBucketUpdateStrategy.java
@@ -62,7 +62,7 @@ public class FlinkConsistentBucketUpdateStrategy<T extends
HoodieRecordPayload>
private String lastRefreshInstant = HoodieTimeline.INIT_INSTANT_TS;
public FlinkConsistentBucketUpdateStrategy(HoodieFlinkWriteClient
writeClient, List<String> indexKeyFields) {
- super(writeClient.getEngineContext(), writeClient.getHoodieTable(),
Collections.emptySet());
+ super(writeClient.getEngineContext(), writeClient.getHoodieTable(),
Collections.emptySet(), Collections.emptySet());
this.indexKeyFields = indexKeyFields;
this.partitionToIdentifier = new HashMap<>();
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
index 75b5e6637f9a..8b6466d5117d 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/BaseDatasetBulkInsertCommitActionExecutor.java
@@ -24,12 +24,18 @@ import org.apache.hudi.HoodieDatasetBulkInsertHelper;
import org.apache.hudi.client.HoodieWriteResult;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
+import
org.apache.hudi.client.clustering.update.strategy.SparkAllowUpdateStrategy;
import org.apache.hudi.common.data.HoodieData;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.CommitUtils;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaRDD;
import org.apache.hudi.exception.HoodieException;
@@ -41,8 +47,10 @@ import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.BulkInsertPartitioner;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.table.action.HoodieWriteMetadata;
+import org.apache.hudi.table.action.cluster.strategy.UpdateStrategy;
import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
@@ -51,9 +59,12 @@ import java.io.Serializable;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
import static
org.apache.hudi.config.HoodieWriteConfig.WRITE_STATUS_STORAGE_LEVEL_VALUE;
+@Slf4j
public abstract class BaseDatasetBulkInsertCommitActionExecutor implements
Serializable {
protected final transient HoodieWriteConfig writeConfig;
@@ -109,6 +120,11 @@ public abstract class
BaseDatasetBulkInsertCommitActionExecutor implements Seria
BulkInsertPartitioner<Dataset<Row>> bulkInsertPartitionerRows =
getPartitioner(populateMetaFields, isTablePartitioned);
Dataset<Row> hoodieDF =
HoodieDatasetBulkInsertHelper.prepareForBulkInsert(records, writeConfig,
table.getMetaClient().getTableConfig(), bulkInsertPartitionerRows, instantTime);
+ // Reject INSERT_OVERWRITE / INSERT_OVERWRITE_TABLE against partitions
with pending
+ // clustering before any writes materialize. Subclasses override
getFileGroupsBeingReplaced;
+ // default is a no-op (empty set) which preserves the non-overwrite paths.
+ rejectIfOverlappingPendingClustering(hoodieDF);
+
HoodieWriteMetadata<JavaRDD<WriteStatus>> result =
buildHoodieWriteMetadata(doExecute(hoodieDF,
bulkInsertPartitionerRows.arePartitionRecordsSorted()));
afterExecute(result);
@@ -141,4 +157,77 @@ public abstract class
BaseDatasetBulkInsertCommitActionExecutor implements Seria
}
protected abstract Map<String, List<String>>
getPartitionToReplacedFileIds(HoodieData<WriteStatus> writeStatuses);
+
+ /**
+ * Returns the file groups this operation will replace. Default is empty
(non-overwrite paths).
+ * Bulk-insert overwrite executors override this so the
overlap-with-pending-clustering check
+ * can fire before any writes materialize.
+ *
+ * @param preparedRecords the dataset after {@code
HoodieDatasetBulkInsertHelper.prepareForBulkInsert}
+ * has populated the {@code _hoodie_partition_path}
meta field, so dynamic
+ * partition resolution can read it.
+ */
+ protected Set<HoodieFileGroupId> getFileGroupsBeingReplaced(Dataset<Row>
preparedRecords) {
+ return Collections.emptySet();
+ }
+
+ /**
+ * Mirrors {@code BaseSparkCommitActionExecutor#clusteringHandleUpdate} for
the bulk-insert row
+ * path: if any of the file groups this operation will replace are in
pending clustering, route
+ * through the configured {@code hoodie.clustering.updates.strategy}. With
the default
+ * {@code SparkRejectUpdateStrategy} this throws {@code
HoodieClusteringUpdateException}; with
+ * {@code SparkAllowUpdateStrategy} (and {@code
!isRollbackPendingClustering()}) the overlap is
+ * deferred to the conflict-resolution strategy, matching the existing
Spark-side behavior.
+ */
+ protected void rejectIfOverlappingPendingClustering(Dataset<Row>
preparedRecords) {
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering =
table.getFileSystemView()
+
.getFileGroupsInPendingClustering().map(Pair::getKey).collect(Collectors.toSet());
+ if (fileGroupsInPendingClustering.isEmpty()) {
+ return;
+ }
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced =
getFileGroupsBeingReplaced(preparedRecords);
+ if (fileGroupsToBeReplaced.isEmpty()) {
+ return;
+ }
+
+ HoodieEngineContext engineContext = writeClient.getEngineContext();
+ UpdateStrategy<HoodieRecord, HoodieData<HoodieRecord>> updateStrategy =
loadClusteringUpdateStrategy(
+ engineContext, fileGroupsInPendingClustering, fileGroupsToBeReplaced);
+ if (updateStrategy instanceof SparkAllowUpdateStrategy &&
!writeConfig.isRollbackPendingClustering()) {
+ return;
+ }
+ // handleUpdate consumes only fileGroupsToBeReplaced on this path (no
tagged records to
+ // inspect for the bulk-insert overwrite case), so pass an empty
HoodieData.
+ updateStrategy.handleUpdate(engineContext.emptyHoodieData());
+ }
+
+ /**
+ * Loads {@code hoodie.clustering.updates.strategy} via reflection,
preferring the 4-arg
+ * constructor (with {@code fileGroupsToBeReplaced}) and falling back to the
legacy 3-arg
+ * constructor for custom strategies that pre-date this PR.
+ */
+ @SuppressWarnings("unchecked")
+ private UpdateStrategy<HoodieRecord, HoodieData<HoodieRecord>>
loadClusteringUpdateStrategy(
+ HoodieEngineContext engineContext,
+ Set<HoodieFileGroupId> fileGroupsInPendingClustering,
+ Set<HoodieFileGroupId> fileGroupsToBeReplaced) {
+ String strategyClass = writeConfig.getClusteringUpdatesStrategyClass();
+ try {
+ return (UpdateStrategy<HoodieRecord, HoodieData<HoodieRecord>>)
ReflectionUtils.loadClass(
+ strategyClass,
+ new Class<?>[] {HoodieEngineContext.class, HoodieTable.class,
Set.class, Set.class},
+ engineContext, table, fileGroupsInPendingClustering,
fileGroupsToBeReplaced);
+ } catch (HoodieException ex) {
+ if (!(ex.getCause() instanceof NoSuchMethodException)) {
+ throw ex;
+ }
+ log.warn("Clustering update strategy {} is missing the 4-arg constructor
with "
+ + "fileGroupsToBeReplaced; falling back to the 3-arg constructor.
INSERT_OVERWRITE "
+ + "overlap with pending clustering will not be detected for this
strategy.", strategyClass);
+ return (UpdateStrategy<HoodieRecord, HoodieData<HoodieRecord>>)
ReflectionUtils.loadClass(
+ strategyClass,
+ new Class<?>[] {HoodieEngineContext.class, HoodieTable.class,
Set.class},
+ engineContext, table, fileGroupsInPendingClustering);
+ }
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteCommitActionExecutor.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteCommitActionExecutor.java
index deaeac4df45f..397407adb6b3 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteCommitActionExecutor.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteCommitActionExecutor.java
@@ -23,6 +23,8 @@ import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.WriteStatus;
import org.apache.hudi.common.data.HoodieData;
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.model.HoodieFileGroupId;
+import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.util.Option;
@@ -33,12 +35,14 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
public class DatasetBulkInsertOverwriteCommitActionExecutor extends
BaseDatasetBulkInsertCommitActionExecutor {
@@ -56,6 +60,46 @@ public class DatasetBulkInsertOverwriteCommitActionExecutor
extends BaseDatasetB
.bulkInsert(records, instantTime, table, writeConfig,
arePartitionRecordsSorted, false));
}
+ /**
+ * For INSERT_OVERWRITE: enumerate latest file groups in the targeted
partitions so the caller
+ * can reject overlap with pending clustering before the bulk-insert
materializes. Mirrors
+ * {@code
SparkInsertOverwriteCommitActionExecutor#getFileGroupsBeingReplaced}; called by
the
+ * base class on the prepared dataset (after {@code prepareForBulkInsert}
populates the
+ * {@code _hoodie_partition_path} meta field), so dynamic partition
resolution can read it.
+ */
+ @Override
+ protected Set<HoodieFileGroupId> getFileGroupsBeingReplaced(Dataset<Row>
preparedRecords) {
+ List<String> partitionPaths = resolveTargetPartitions(preparedRecords);
+ if (partitionPaths.isEmpty()) {
+ return Collections.emptySet();
+ }
+ return partitionPaths.stream()
+ .flatMap(partitionPath ->
table.getSliceView().getLatestFileSlices(partitionPath)
+ .map(FileSlice::getFileGroupId))
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * Resolves the partition paths this overwrite will replace. Subclasses
override for the
+ * table-wide variant (enumerate every partition).
+ */
+ protected List<String> resolveTargetPartitions(Dataset<Row> preparedRecords)
{
+ if (!table.isPartitioned()) {
+ return Collections.singletonList(StringUtils.EMPTY_STRING);
+ }
+ String staticOverwritePartitionPaths =
writeConfig.getStringOrDefault(HoodieInternalConfig.STATIC_OVERWRITE_PARTITION_PATHS);
+ if (StringUtils.nonEmpty(staticOverwritePartitionPaths)) {
+ return Arrays.asList(staticOverwritePartitionPaths.split(","));
+ }
+ // Dynamic partition path: read the populated _hoodie_partition_path meta
field. The base
+ // class invokes this hook after
HoodieDatasetBulkInsertHelper.prepareForBulkInsert, so the
+ // field is guaranteed to be present and populated by the configured key
generator.
+ return preparedRecords.select(HoodieRecord.PARTITION_PATH_METADATA_FIELD)
+ .distinct()
+ .as(Encoders.STRING())
+ .collectAsList();
+ }
+
@Override
public WriteOperationType getWriteOperationType() {
return WriteOperationType.INSERT_OVERWRITE;
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteTableCommitActionExecutor.java
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteTableCommitActionExecutor.java
index d35325360745..3dc1df5dc270 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteTableCommitActionExecutor.java
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/java/org/apache/hudi/commit/DatasetBulkInsertOverwriteTableCommitActionExecutor.java
@@ -28,6 +28,9 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.data.HoodieJavaPairRDD;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -44,6 +47,14 @@ public class
DatasetBulkInsertOverwriteTableCommitActionExecutor extends Dataset
return WriteOperationType.INSERT_OVERWRITE_TABLE;
}
+ @Override
+ protected List<String> resolveTargetPartitions(Dataset<Row> preparedRecords)
{
+ // Table-wide overwrite replaces every file group in every partition;
enumerate them all.
+ List<String> partitionPaths =
FSUtils.getAllPartitionPaths(writeClient.getEngineContext(),
+ table.getMetaClient(), writeConfig.getMetadataConfig());
+ return partitionPaths == null ? Collections.emptyList() : partitionPaths;
+ }
+
@Override
protected Map<String, List<String>>
getPartitionToReplacedFileIds(HoodieData<WriteStatus> writeStatuses) {
HoodieEngineContext context = writeClient.getEngineContext();
