yihua commented on a change in pull request #3664: URL: https://github.com/apache/hudi/pull/3664#discussion_r711206432
########## File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/rollback/ListingBasedRollbackHelper.java ########## @@ -0,0 +1,249 @@ +/* + * 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.rollback; + +import org.apache.hudi.common.HoodieRollbackStat; +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.common.model.HoodieLogFile; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.log.HoodieLogFormat; +import org.apache.hudi.common.table.log.block.HoodieCommandBlock; +import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType; +import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.util.collection.ImmutablePair; +import org.apache.hudi.common.util.collection.Pair; +import org.apache.hudi.config.HoodieWriteConfig; +import org.apache.hudi.exception.HoodieIOException; +import org.apache.hudi.exception.HoodieRollbackException; + +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.PathFilter; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; + +/** + * Performs Rollback of Hoodie Tables. + */ +public class ListingBasedRollbackHelper implements Serializable { + private static final Logger LOG = LogManager.getLogger(ListingBasedRollbackHelper.class); + + private final HoodieTableMetaClient metaClient; + private final HoodieWriteConfig config; + + public ListingBasedRollbackHelper(HoodieTableMetaClient metaClient, HoodieWriteConfig config) { + this.metaClient = metaClient; + this.config = config; + } + + /** + * Performs all rollback actions that we have collected in parallel. + */ + public List<HoodieRollbackStat> performRollback(HoodieEngineContext context, HoodieInstant instantToRollback, + List<ListingBasedRollbackRequest> rollbackRequests) { + int parallelism = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1); + context.setJobStatus(this.getClass().getSimpleName(), "Perform rollback actions"); + return context.mapToPairAndReduceByKey(rollbackRequests, + rollbackRequest -> maybeDeleteAndCollectStats(rollbackRequest, instantToRollback, true), + RollbackUtils::mergeRollbackStat, + parallelism); + } + + /** + * Collect all file info that needs to be rollbacked. + */ + public List<HoodieRollbackStat> collectRollbackStats(HoodieEngineContext context, HoodieInstant instantToRollback, + List<ListingBasedRollbackRequest> rollbackRequests) { + int parallelism = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1); + context.setJobStatus(this.getClass().getSimpleName(), "Collect rollback stats for upgrade/downgrade"); + return context.mapToPairAndReduceByKey(rollbackRequests, + rollbackRequest -> maybeDeleteAndCollectStats(rollbackRequest, instantToRollback, false), + RollbackUtils::mergeRollbackStat, + parallelism); Review comment: @vinothchandar in the existing Flink/Spark client (like below), we collect the rollback stats without merging them based on the partition path. @nsivabalan suggested that there should be a merge logic here as well. Just to make sure this change is okay since it's changing the existing logic. ``` int sparkPartitions = Math.max(Math.min(rollbackRequests.size(), config.getRollbackParallelism()), 1); context.setJobStatus(this.getClass().getSimpleName(), "Collect rollback stats for upgrade/downgrade"); JavaPairRDD<String, HoodieRollbackStat> partitionPathRollbackStatsPairRDD = maybeDeleteAndCollectStats(context, instantToRollback, rollbackRequests, sparkPartitions, false); return partitionPathRollbackStatsPairRDD.map(Tuple2::_2).collect(); ``` -- 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]
