hudi-client/src/main/java/org/apache/hudi/table/action/rollback/RollbackUtils.java
/**
   * Generate all rollback requests that we need to perform for rolling back this action without actually performing rolling back for MOR table type.
   *
   * @param instantToRollback Instant to Rollback
   * @param table instance of {@link HoodieTable} to use.
   * @param jsc instance of {@link JavaSparkContext} to use.
   * @return list of rollback requests
   */
Generate all rollback requests that we need to perform for rolling back this action without actually performing rolling back for MOR table type.
@param instantToRollback Instant to Rollback
@param table instance of  HoodieTable to use.
@param jsc instance of  JavaSparkContext to use.
@return list of rollback requests
Generate all rollback requests that we need to perform for rolling back this action without actually performing rolling back for MOR table type . @param instantToRollback Instant to Rollback @param table instance of HoodieTable to use . @param jsc instance of JavaSparkContext to use . @return list of rollback requests
public static List<ListingBasedRollbackRequest> generateRollbackRequestsUsingFileListingMOR(HoodieInstant instantToRollback, HoodieTable table, JavaSparkContext jsc) throws IOException {
    String commit = instantToRollback.getTimestamp();
    HoodieWriteConfig config = table.getConfig();
    List<String> partitions = FSUtils.getAllPartitionPaths(table.getMetaClient().getFs(), table.getMetaClient().getBasePath(),
        config.shouldAssumeDatePartitioning());
    int sparkPartitions = Math.max(Math.min(partitions.size(), config.getRollbackParallelism()), 1);
    jsc.setJobGroup(RollbackUtils.class.getSimpleName(), "Generate all rollback requests");
    return jsc.parallelize(partitions, Math.min(partitions.size(), sparkPartitions)).flatMap(partitionPath -> {
      HoodieActiveTimeline activeTimeline = table.getMetaClient().reloadActiveTimeline();
      List<ListingBasedRollbackRequest> partitionRollbackRequests = new ArrayList<>();
      switch (instantToRollback.getAction()) {
        case HoodieTimeline.COMMIT_ACTION:
          LOG.info("Rolling back commit action.");
          partitionRollbackRequests.add(
              ListingBasedRollbackRequest.createRollbackRequestWithDeleteDataAndLogFilesAction(partitionPath));
          break;
        case HoodieTimeline.COMPACTION_ACTION:
          boolean higherDeltaCommits =
              !activeTimeline.getDeltaCommitTimeline().filterCompletedInstants().findInstantsAfter(commit, 1).empty();
          if (higherDeltaCommits) {
            LOG.info("Rolling back compaction. There are higher delta commits. So only deleting data files");
            partitionRollbackRequests.add(
                ListingBasedRollbackRequest.createRollbackRequestWithDeleteDataFilesOnlyAction(partitionPath));
          } else {
            LOG.info("Rolling back compaction plan. There are NO higher delta commits. So deleting both data and"
                + " log files");
            partitionRollbackRequests.add(
                ListingBasedRollbackRequest.createRollbackRequestWithDeleteDataAndLogFilesAction(partitionPath));
          }
          break;
        case HoodieTimeline.DELTA_COMMIT_ACTION:
          -
          -
          try {
            HoodieCommitMetadata commitMetadata = HoodieCommitMetadata.fromBytes(
                table.getMetaClient().getCommitTimeline()
                    .getInstantDetails(new HoodieInstant(true, instantToRollback.getAction(), instantToRollback.getTimestamp()))
                    .get(),
                HoodieCommitMetadata.class);
            partitionRollbackRequests.add(
                ListingBasedRollbackRequest.createRollbackRequestWithDeleteDataAndLogFilesAction(partitionPath));
            if (commitMetadata.getPartitionToWriteStats().containsKey(partitionPath)) {
              partitionRollbackRequests
                  .addAll(generateAppendRollbackBlocksAction(partitionPath, instantToRollback, commitMetadata, table));
            }
            break;
          } catch (IOException io) {
            throw new HoodieIOException("Failed to collect rollback actions for commit " + commit, io);
          }
        default:
          break;
      }
      return partitionRollbackRequests.iterator();
    }).filter(Objects::nonNull).collect();
  }
public static List < ListingBasedRollbackRequest > generateRollbackRequestsUsingFileListingMOR ( HoodieInstant instantToRollback , HoodieTable table , JavaSparkContext jsc ) throws IOException { String commit = instantToRollback . getTimestamp ( ) ; HoodieWriteConfig config = table . getConfig ( ) ; List < String > partitions = FSUtils . getAllPartitionPaths ( table . getMetaClient ( ) . getFs ( ) , table . getMetaClient ( ) . getBasePath ( ) , config . shouldAssumeDatePartitioning ( ) ) ; int sparkPartitions = Math . max ( Math . min ( partitions . size ( ) , config . getRollbackParallelism ( ) ) , 1 ) ; jsc . setJobGroup ( RollbackUtils . class . getSimpleName ( ) , "Generate all rollback requests" ) ; return jsc . parallelize ( partitions , Math . min ( partitions . size ( ) , sparkPartitions ) ) . flatMap ( partitionPath -> { HoodieActiveTimeline activeTimeline = table . getMetaClient ( ) . reloadActiveTimeline ( ) ; List < ListingBasedRollbackRequest > partitionRollbackRequests = new ArrayList < > ( ) ; switch ( instantToRollback . getAction ( ) ) { case HoodieTimeline . COMMIT_ACTION : LOG . info ( "Rolling back commit action." ) ; partitionRollbackRequests . add ( ListingBasedRollbackRequest . createRollbackRequestWithDeleteDataAndLogFilesAction ( partitionPath ) ) ; break ; case HoodieTimeline . COMPACTION_ACTION : boolean higherDeltaCommits = ! activeTimeline . getDeltaCommitTimeline ( ) . filterCompletedInstants ( ) . findInstantsAfter ( commit , 1 ) . empty ( ) ; if ( higherDeltaCommits ) { LOG . info ( "Rolling back compaction. There are higher delta commits. So only deleting data files" ) ; partitionRollbackRequests . add ( ListingBasedRollbackRequest . createRollbackRequestWithDeleteDataFilesOnlyAction ( partitionPath ) ) ; } else { LOG . info ( "Rolling back compaction plan. There are NO higher delta commits. So deleting both data and" + " log files" ) ; partitionRollbackRequests . add ( ListingBasedRollbackRequest . createRollbackRequestWithDeleteDataAndLogFilesAction ( partitionPath ) ) ; } break ; case HoodieTimeline . DELTA_COMMIT_ACTION : try { HoodieCommitMetadata commitMetadata = HoodieCommitMetadata . fromBytes ( table . getMetaClient ( ) . getCommitTimeline ( ) . getInstantDetails ( new HoodieInstant ( true , instantToRollback . getAction ( ) , instantToRollback . getTimestamp ( ) ) ) . get ( ) , HoodieCommitMetadata . class ) ; partitionRollbackRequests . add ( ListingBasedRollbackRequest . createRollbackRequestWithDeleteDataAndLogFilesAction ( partitionPath ) ) ; if ( commitMetadata . getPartitionToWriteStats ( ) . containsKey ( partitionPath ) ) { partitionRollbackRequests . addAll ( generateAppendRollbackBlocksAction ( partitionPath , instantToRollback , commitMetadata , table ) ) ; } break ; } catch ( IOException io ) { throw new HoodieIOException ( "Failed to collect rollback actions for commit " + commit , io ) ; } default : break ; } return partitionRollbackRequests . iterator ( ) ; } ) . filter ( Objects :: nonNull ) . collect ( ) ; }
Generate all rollback requests that we need to perform for rolling back this action without actually performing rolling back for MOR table type.
Generate all rollback requests that we need to perform for rolling back this action without actually performing rolling back for MOR table type .
// If there is no delta commit present after the current commit (if compaction), no action, else we // need to make sure that a compaction commit rollback also deletes any log files written as part of the // succeeding deltacommit. // Rollback of a compaction action with no higher deltacommit means that the compaction is scheduled // and has not yet finished. In this scenario we should delete only the newly created parquet files // and not corresponding base commit log files created with this as baseCommit since updates would // have been written to the log files. // No deltacommits present after this compaction commit (inflight or requested). In this case, we // can also delete any log files that were created with this compaction commit as base // commit. // -------------------------------------------------------------------------------------------------- // (A) The following cases are possible if index.canIndexLogFiles and/or index.isGlobal // -------------------------------------------------------------------------------------------------- // (A.1) Failed first commit - Inserts were written to log files and HoodieWriteStat has no entries. In // this scenario we would want to delete these log files. // (A.2) Failed recurring commit - Inserts/Updates written to log files. In this scenario, // HoodieWriteStat will have the baseCommitTime for the first log file written, add rollback blocks. // (A.3) Rollback triggered for first commit - Inserts were written to the log files but the commit is // being reverted. In this scenario, HoodieWriteStat will be `null` for the attribute prevCommitTime and // and hence will end up deleting these log files. This is done so there are no orphan log files // lying around. // (A.4) Rollback triggered for recurring commits - Inserts/Updates are being rolled back, the actions // taken in this scenario is a combination of (A.2) and (A.3) // --------------------------------------------------------------------------------------------------- // (B) The following cases are possible if !index.canIndexLogFiles and/or !index.isGlobal // --------------------------------------------------------------------------------------------------- // (B.1) Failed first commit - Inserts were written to parquet files and HoodieWriteStat has no entries. // In this scenario, we delete all the parquet files written for the failed commit. // (B.2) Failed recurring commits - Inserts were written to parquet files and updates to log files. In // this scenario, perform (A.1) and for updates written to log files, write rollback blocks. // (B.3) Rollback triggered for first commit - Same as (B.1) // (B.4) Rollback triggered for recurring commits - Same as (B.2) plus we need to delete the log files // as well if the base parquet file gets deleted. // In case all data was inserts and the commit failed, delete the file belonging to that commit // We do not know fileIds for inserts (first inserts are either log files or parquet files), // delete all files for the corresponding failed commit, if present (same as COW) // append rollback blocks for updates