[HUDI-6288] Create IngestionPrimaryWriterBasedConflictResolutionStrategy to prioritize ingestion writers over other writers #8832
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| /* | ||
| * 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.common.table.HoodieTableMetaClient; | ||
| import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; | ||
| import org.apache.hudi.common.table.timeline.HoodieInstant; | ||
| import org.apache.hudi.common.util.ClusteringUtils; | ||
| import org.apache.hudi.common.util.CollectionUtils; | ||
| import org.apache.hudi.common.util.Option; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.Comparator; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMMIT_ACTION; | ||
| import static org.apache.hudi.common.table.timeline.HoodieTimeline.COMPACTION_ACTION; | ||
| import static org.apache.hudi.common.table.timeline.HoodieTimeline.DELTA_COMMIT_ACTION; | ||
| import static org.apache.hudi.common.table.timeline.HoodieTimeline.REPLACE_COMMIT_ACTION; | ||
|
|
||
| /** | ||
| * This class extends the base implementation of conflict resolution strategy. | ||
| * It gives preference to ingestion writers compared to table services. | ||
| */ | ||
| public class IngestionPrimaryWriterBasedConflictResolutionStrategy | ||
| extends SimpleConcurrentFileWritesConflictResolutionStrategy { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(IngestionPrimaryWriterBasedConflictResolutionStrategy.class); | ||
|
|
||
| /** | ||
| * For tableservices like replacecommit and compaction commits this method also returns ingestion inflight commits. | ||
| */ | ||
| @Override | ||
| public Stream<HoodieInstant> getCandidateInstants(HoodieTableMetaClient metaClient, HoodieInstant currentInstant, | ||
| Option<HoodieInstant> lastSuccessfulInstant) { | ||
| HoodieActiveTimeline activeTimeline = metaClient.reloadActiveTimeline(); | ||
| if ((REPLACE_COMMIT_ACTION.equals(currentInstant.getAction()) | ||
| && ClusteringUtils.isClusteringCommit(metaClient, currentInstant)) | ||
| || COMPACTION_ACTION.equals(currentInstant.getAction())) { | ||
| return getCandidateInstantsForTableServicesCommits(activeTimeline, currentInstant); | ||
| } else { | ||
| return getCandidateInstantsForNonTableServicesCommits(activeTimeline, currentInstant); | ||
| } | ||
| } | ||
|
|
||
| private Stream<HoodieInstant> getCandidateInstantsForNonTableServicesCommits(HoodieActiveTimeline activeTimeline, HoodieInstant currentInstant) { | ||
|
|
||
| // To findout which instants are conflicting, we apply the following logic | ||
| // Get all the completed instants timeline only for commits that have happened | ||
| // since the last successful write based on the transition times. | ||
| // We need to check for write conflicts since they may have mutated the same files | ||
| // that are being newly created by the current write. | ||
| List<HoodieInstant> completedCommitsInstants = activeTimeline | ||
|
Contributor
There was a problem hiding this comment. lets avoid code duplication. we can move code from within SimpleConcurrentFileWritesConflictResolutionStrategy and make them protected and re-use them here.
Contributor
Author
There was a problem hiding this comment. Created a follow task https://issues.apache.org/jira/browse/HUDI-6353 |
||
| .getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION, COMPACTION_ACTION, DELTA_COMMIT_ACTION)) | ||
| .filterCompletedInstants() | ||
| .findInstantsModifiedAfterByStateTransitionTime(currentInstant.getTimestamp()) | ||
| .getInstantsOrderedByStateTransitionTime() | ||
|
Contributor
There was a problem hiding this comment. Using state transition is driving by a config if I am not wrong.
Contributor
There was a problem hiding this comment. It's okay we add some apis on the timeline, the timeline api is kind of internal, users should not be exposed to/aware of these changes. Anyway we need some
Contributor
Author
There was a problem hiding this comment. Yeah, completion time ordering API will be ideal to use. For now rewording the findInstantsModifiedAfter method to findInstantsModifiedAfterByStateTransitionTime. |
||
| .collect(Collectors.toList()); | ||
| LOG.info(String.format("Instants that may have conflict with %s are %s", currentInstant, completedCommitsInstants)); | ||
| return completedCommitsInstants.stream(); | ||
| } | ||
|
|
||
| /** | ||
| * To find which instants are conflicting, we apply the following logic | ||
| * Get both completed instants and ingestion inflight commits that have happened since the last successful write. | ||
| * We need to check for write conflicts since they may have mutated the same files | ||
| * that are being newly created by the current write. | ||
| */ | ||
| private Stream<HoodieInstant> getCandidateInstantsForTableServicesCommits(HoodieActiveTimeline activeTimeline, HoodieInstant currentInstant) { | ||
| // Fetch list of completed commits. | ||
| Stream<HoodieInstant> completedCommitsStream = | ||
| activeTimeline | ||
| .getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, REPLACE_COMMIT_ACTION, COMPACTION_ACTION, DELTA_COMMIT_ACTION)) | ||
| .filterCompletedInstants() | ||
| .findInstantsModifiedAfterByStateTransitionTime(currentInstant.getTimestamp()) | ||
| .getInstantsAsStream(); | ||
|
|
||
| // Fetch list of ingestion inflight commits. | ||
| Stream<HoodieInstant> inflightIngestionCommitsStream = | ||
|
Contributor
There was a problem hiding this comment. @suryaprasanna @danny0405 One reminder, not all inflight commits have metadata, for example, bulk insert's inflight metadata is empty,
Flink even not save metadata in inflight
Contributor
Author
There was a problem hiding this comment. @Zouxxyy I idea behind creating IngestionPrimaryWriterBasedConflictResolutionStrategy, is to give preference for Ingestion writers over clustering writers.
Contributor
There was a problem hiding this comment.
Thanks for the reply, what I mean is now we use the metadata of pending ingestion instant for conflict detection here, but not all inflight commits have metadata, e.g. the .inflght file written by flink is empty. And can you help review this #9220 which improve the current stragies, thanks
Contributor
Author
There was a problem hiding this comment. Ok. I am not much familiar with Flink, but on the Spark engine the inflight commits that are doing upsert operations should have partition to fileIds(that needs to be modified) present in the .inflight file, this is actually done after the WorkloadProfile object is loaded. So, conflict resolution will only see those file ids that are getting modified not new ones. During conflict resolution that is all we care, since there won't be any overlap between new files. Sure, I will review the RFC. |
||
| activeTimeline | ||
| .getTimelineOfActions(CollectionUtils.createSet(COMMIT_ACTION, DELTA_COMMIT_ACTION)) | ||
| .filterInflights() | ||
| .getInstantsAsStream(); | ||
|
|
||
| // Merge and sort the instants and return. | ||
| List<HoodieInstant> instantsToConsider = Stream.concat(completedCommitsStream, inflightIngestionCommitsStream) | ||
| .sorted(Comparator.comparing(o -> o.getStateTransitionTime())) | ||
| .collect(Collectors.toList()); | ||
| LOG.info(String.format("Instants that may have conflict with %s are %s", currentInstant, instantsToConsider)); | ||
| return instantsToConsider.stream(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isPreCommitRequired() { | ||
| return true; | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
this is a public interface. there are chances that someone outside could have implemented their own resolution strategy. so, lets deprecate this and introduce a new one.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
but curious to know why this change though ?
from the impl, I see that we are reloading the active timeline. so, why can't we do that at the caller before calling this method ?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Infact resolveWriteConflictIfAny already had arguments whether to reload the active timeline or not. So, can't we leverage that?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It is hard code to false now, kind of think it is hard to maintain for the invoker to know whether to refresh these timelines.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.