This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch release-1.2.1 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 796104c079b58490bab14e584eb5bc32a5492eaa 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]> (cherry picked from commit e2cf7216fddf37f49cfdf10861be7aaa31e3feb5) --- .../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();
