amogh-jahagirdar commented on a change in pull request #4071: URL: https://github.com/apache/iceberg/pull/4071#discussion_r830669999
########## File path: core/src/main/java/org/apache/iceberg/UpdateSnapshotReferencesOperation.java ########## @@ -0,0 +1,166 @@ +/* + * 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.iceberg; + +import java.util.Map; +import java.util.Set; + +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.util.Tasks; + +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS; +import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES; +import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS; +import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT; + +class UpdateSnapshotReferencesOperation implements PendingUpdate<Map<String, SnapshotRef>> { + + private final TableOperations ops; + private TableMetadata base; + private final Map<String, SnapshotRef> refsToUpdate; + private final Set<String> refsToRemove; + + UpdateSnapshotReferencesOperation(TableOperations ops) { + this.ops = ops; + this.base = ops.current(); + this.refsToUpdate = Maps.newHashMap(); + this.refsToRemove = Sets.newHashSet(); + } + + @Override + public Map<String, SnapshotRef> apply() { + this.base = ops.refresh(); + Map<String, SnapshotRef> refs = Maps.newHashMap(); + TableMetadata.Builder updatedBuilder = TableMetadata.buildFrom(base); + for (String refName : refsToRemove) { + SnapshotRef ref = base.ref(refName); + ValidationException.check(ref != null, "Cannot remove nonexistent snapshot ref %s", refName); + if (ref.isBranch()) { + updatedBuilder.removeBranch(refName); + } + else { + updatedBuilder.removeTag(refName); + } + } + for (Map.Entry<String, SnapshotRef> refEntry : refsToUpdate.entrySet()) { + String name = refEntry.getKey(); + SnapshotRef snapshotRef = refEntry.getValue(); + updatedBuilder.createSnapshotRef(name, snapshotRef); + } + return refs; + } + + @Override + public void commit() { + Tasks.foreach(ops) + .retry(base.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT)) + .exponentialBackoff( + base.propertyAsInt(COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT), + base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT), + 2.0 /* exponential */) + .onlyRetryOn(CommitFailedException.class) + .run(taskOps -> { + Map<String, SnapshotRef> refs = apply(); + TableMetadata.Builder updatedMetadata = TableMetadata.buildFrom(base); + updatedMetadata. + taskOps.commit(base, updated); + }); + } + + public UpdateSnapshotReferencesOperation createBranch(String name, long snapshotId) { Review comment: @rdblue I was still working on changes and iteratively pushing to my branch (sorry for the confusion), I think this was reviewed when it wasn't yet complete on my side. Let me know what you think on the latest ! For supporting transaction.updateSnapshotReferencesOperation().createBranch(...).setMinSnapshotsToKeep() to avoid the error, in my latest change when we look for refs in setMinSnapshotsToKeep we always search for the "current" ref. Where current is either the ref in the refsToUpdate (which if someone did a createBranch, it would be in there) or we then fall back to the ref. validateAndGetLatestRef is the helper method to accomplish this. So effectively it's the same as operating on a copy of the refs. Currently, in apply we do the diff to keep track of the metadata changes but it is implemented slightly differently than what you mentioned (the diffing logic is contained within the operation itself). So we will go through the build methods to create/remove the relevant references. At that point, we do the build of the new metadata. If this is successful, this implies apply can be successful, and we get the diff of the references from before -> after to surface as the changes on the PendingUpdate. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
