nsivabalan commented on code in PR #13316: URL: https://github.com/apache/hudi/pull/13316#discussion_r2101431783
########## hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/TableServicePlanConflictDetection.java: ########## @@ -0,0 +1,106 @@ +/* + * 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.client.transaction; + +import org.apache.hudi.avro.model.HoodieClusteringPlan; +import org.apache.hudi.avro.model.HoodieCompactionPlan; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.exception.HoodieIOException; + +import java.io.IOException; +import java.util.stream.Stream; + +/** + * Handles conflict detection for the plans of various table services. + * Cleaning and archiving are not included in this class as they do not require conflict detection. + */ +public class TableServicePlanConflictDetection { + private final ConflictResolutionStrategy conflictResolutionStrategy; + private final HoodieTableMetaClient metaClient; + private final String lastKnownCompletionTime; + + /** + * Constructors a new instance to detect conflicts for a table service plan. + * Note that this should be used within a transaction to prevent race conditions. The meta client used must be constructed or reloaded within the transaction. + * @param conflictResolutionStrategy The strategy to use for conflict resolution for the writer. + * @param metaClient The meta client for the table. This must be constructed or reloaded during the transaction. + * @param lastKnownCompletionTime The last completion time of an instant on the timeline when the planner started. + */ + public TableServicePlanConflictDetection(ConflictResolutionStrategy conflictResolutionStrategy, HoodieTableMetaClient metaClient, String lastKnownCompletionTime) { + this.conflictResolutionStrategy = conflictResolutionStrategy; + this.metaClient = metaClient; + this.lastKnownCompletionTime = lastKnownCompletionTime; + } + + /** + * Checks if there are any potential conflicts with actions that have happened since the planner start instant. + * The compaction requires a check for conflicts with any commits for ingesting new data since the compaction plan is expected to contain all the log files for the slice it is compacting. + * It also requires checks for new plans for clustering or potentially another compaction on the timeline to avoid multiple services operating on the same file groups. + * @param compactionPlan the requested plan + * @param requestedInstantTime the requested instant time that will be used when persisting the plan to the timeline + * @return true if there is a conflict with the requested plan, false otherwise + */ + public boolean hasConflict(HoodieCompactionPlan compactionPlan, String requestedInstantTime) { + ConcurrentOperation compactionOperation = new ConcurrentOperation(compactionPlan, requestedInstantTime); + return hasConflict(compactionOperation); + } + + /** + * Checks if there are any potential conflicts with actions that have happened since the planner start instant. + * The clustering requires a check for conflicts with any commits for ingesting new data since the clustering plan will also contain log files, + * and all the log files for the slice it is clustering must be in the plan. + * It also requires checks for new plans for compaction or potentially another clustering on the timeline to avoid multiple services operating on the same file groups. + * @param clusteringPlan the requested plan + * @param requestedInstantTime the requested instant time that will be used when persisting the plan to the timeline + * @return true if there is a conflict with the requested plan, false otherwise + */ + public boolean hasConflict(HoodieClusteringPlan clusteringPlan, String requestedInstantTime) { + ConcurrentOperation clusteringOperation = new ConcurrentOperation(clusteringPlan, requestedInstantTime); + return hasConflict(clusteringOperation); + } + + private boolean hasConflict(ConcurrentOperation plannedOperation) { + return generateCandidateInstants().anyMatch(instant -> { + try { + ConcurrentOperation otherOperation = new ConcurrentOperation(instant, metaClient); + return conflictResolutionStrategy.hasConflict(plannedOperation, otherOperation); Review Comment: but here is my understanding. we need to distinguish if its an inflight ingestion commit or completed. a. completed ingestion commit. and compaction plan that just got generated is going through conflict resolution. if there are overlapping file groups, we are failing here. bcoz, the cur plan generated would not have included log files produced by an inglith ingestion commit. b. inflight ingestion commit. and compaction plan that just got generated is going through conflict resolution. here we let the compaction plan go ahead. since compaction plan conflict resolution should be happening within a lock and once the resolution succeeds, compaction.requested will be added to the timeline. and hence the ingestion commit's completion time will definitely be > the compaction instant time. in this case, lets also see how does conflict resolution of entire write will happen for ingestion commit (I am not talking about TableServicePlanConflictDetection). - Guess we might need a fix here. currently(as per master), if the ingestion commits's start time is < compaction instant time, we fail the resolution step. if its >, we let is pass. we might need to make changes to this once we have this patch landed. essentially, ingestion commits can never conflict w/ a compaction commit during the final conflict resolution. But I will let Tim also go over this and see if my understanding is right. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
