package com.airbnb.di.hive.replication.primitives;

import com.airbnb.di.common.DistCpException;
import com.airbnb.di.hive.common.HiveMetastoreClient;
import com.airbnb.di.hive.common.HiveMetastoreException;
import com.airbnb.di.hive.common.HiveObjectSpec;
import com.airbnb.di.hive.replication.DirectoryCopier;
import com.airbnb.di.hive.replication.ReplicationUtils;
import com.airbnb.di.hive.replication.RunInfo;
import com.airbnb.di.hive.replication.configuration.Cluster;
import com.airbnb.di.hive.replication.configuration.DestinationObjectFactory;
import com.airbnb.di.hive.replication.configuration.ObjectConflictHandler;
import com.airbnb.di.multiprocessing.Lock;
import com.airbnb.di.multiprocessing.LockSet;
import com.airbnb.di.multiprocessing.ParallelJobExecutor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.util.StringUtils;

import java.io.IOException;
import java.util.Optional;

public class RenameTableTask implements ReplicationTask {
  private static final Log LOG = LogFactory.getLog(RenameTableTask.class);

  private Configuration conf;
  private DestinationObjectFactory destObjectFactory;
  private ObjectConflictHandler objectConflictHandler;
  private Cluster srcCluster;
  private Cluster destCluster;
  private HiveObjectSpec renameFromSpec;
  private HiveObjectSpec renameToSpec;
  private Optional<Path> renameFromPath;
  private Optional<Path> renameToPath;
  private Optional<String> renameFromTableTdlt;

  private ParallelJobExecutor copyPartitionsExecutor;
  private DirectoryCopier directoryCopier;

  /**
   * TODO.
   *
   * @param conf TODO
   * @param srcCluster TODO
   * @param destCluster TODO
   * @param destObjectFactory TODO
   * @param objectConflictHandler TODO
   * @param renameFromSpec TODO
   * @param renameToSpec TODO
   * @param renameFromPath TODO
   * @param renameToPath TODO
   * @param renameFromTableTldt TODO
   * @param copyPartitionsExecutor TODO
   * @param directoryCopier TODO
   */
  public RenameTableTask(
      Configuration conf,
      Cluster srcCluster,
      Cluster destCluster,
      DestinationObjectFactory destObjectFactory,
      ObjectConflictHandler objectConflictHandler,
      HiveObjectSpec renameFromSpec,
      HiveObjectSpec renameToSpec,
      Optional<Path> renameFromPath,
      Optional<Path> renameToPath,
      Optional<String> renameFromTableTldt,
      ParallelJobExecutor copyPartitionsExecutor,
      DirectoryCopier directoryCopier) {
    this.conf = conf;
    this.destObjectFactory = destObjectFactory;
    this.objectConflictHandler = objectConflictHandler;
    this.srcCluster = srcCluster;
    this.destCluster = destCluster;
    this.renameFromSpec = renameFromSpec;
    this.renameToSpec = renameToSpec;
    this.renameFromPath = renameFromPath;
    this.renameToPath = renameToPath;
    this.renameFromTableTdlt = renameFromTableTldt;

    this.copyPartitionsExecutor = copyPartitionsExecutor;
    this.directoryCopier = directoryCopier;
  }

  enum HandleRenameAction {
    RENAME_TABLE, COPY_TABLE, NO_OP
  }

  @Override
  public RunInfo runTask() throws DistCpException, HiveMetastoreException, IOException {
    LOG.debug("Renaming " + renameFromSpec + " to " + renameToSpec);

    HiveMetastoreClient destMs = destCluster.getMetastoreClient();
    HiveMetastoreClient srcMs = srcCluster.getMetastoreClient();

    Table freshSrcRenameToTable =
        srcMs.getTable(renameToSpec.getDbName(), renameToSpec.getTableName());

    Table freshDestRenameToTable =
        destMs.getTable(renameToSpec.getDbName(), renameToSpec.getTableName());

    // Get a fresh copy of the metadata from the dest Hive metastore
    Table freshDestTable =
        destMs.getTable(renameFromSpec.getDbName(), renameFromSpec.getTableName());

    if (!renameFromTableTdlt.isPresent()) {
      LOG.error(
          "For safety, not completing rename task since source " + " object TLDT is missing!");
      return new RunInfo(RunInfo.RunStatus.NOT_COMPLETABLE, 0);
    }

    String expectedTldt = renameFromTableTdlt.get();

    HandleRenameAction renameAction = null;
    if (ReplicationUtils.transientLastDdlTimesMatch(freshSrcRenameToTable,
        freshDestRenameToTable)) {
      LOG.debug(
          "Rename to table exists on destination and has a " + "matching TLDT. Not doing anything");
      renameAction = HandleRenameAction.NO_OP;

    } else if (freshDestRenameToTable != null) {
      LOG.debug("Rename to table already exists on destination, but "
          + "doesn't have a matching TLDT. Copying instead...");
      renameAction = HandleRenameAction.COPY_TABLE;
    } else if (freshDestTable == null) {
      LOG.warn(StringUtils.format("Destination rename from table %s " + "doesn't exist. "
          + "Copying %s to destination instead.", renameFromSpec, renameToSpec));
      renameAction = HandleRenameAction.COPY_TABLE;
    } else if (!ReplicationUtils.transientLastDdlTimesMatch(expectedTldt, freshDestTable)) {
      LOG.warn(StringUtils.format(
          "Destination table %s doesn't have " + "the expected modified time", renameFromSpec));
      LOG.debug("Renamed from source table with a TLDT: " + expectedTldt);
      LOG.debug("Table on destination: " + freshDestTable);
      LOG.debug(String.format("Copying %s to destination instead", renameToSpec));
      renameAction = HandleRenameAction.COPY_TABLE;
    } else {
      LOG.debug(String.format("Destination table (%s) matches " + "expected TLDT(%s) - will rename",
          renameFromSpec, ReplicationUtils.getTldt(freshDestTable)));
      renameAction = HandleRenameAction.RENAME_TABLE;
    }

    switch (renameAction) {
      case NO_OP:
        return new RunInfo(RunInfo.RunStatus.SUCCESSFUL, 0);

      case RENAME_TABLE:
        LOG.debug(StringUtils.format("Renaming %s to %s", renameFromSpec, renameToSpec));
        Table newTableOnDestination = new Table(freshDestTable);
        newTableOnDestination.setDbName(renameToSpec.getDbName());
        newTableOnDestination.setTableName(renameToSpec.getTableName());
        destMs.alterTable(renameFromSpec.getDbName(), renameFromSpec.getTableName(),
            newTableOnDestination);
        LOG.debug(StringUtils.format("Renamed %s to %s", renameFromSpec, renameToSpec));
      // After a rename, the table should be re-copied to get the
      // correct modified time changes. With a proper rename, this
      // should be a mostly no-op. Fall through to the next case.
      // fallthrough

      case COPY_TABLE:
        CopyCompleteTableTask task =
            new CopyCompleteTableTask(conf, destObjectFactory, objectConflictHandler, srcCluster,
                destCluster, renameToSpec, renameToPath, copyPartitionsExecutor, directoryCopier);
        return task.runTask();

      default:
        throw new RuntimeException("Unhandled case: " + renameAction);
    }
  }

  @Override
  public LockSet getRequiredLocks() {
    LockSet lockSet = new LockSet();
    lockSet.add(new Lock(Lock.Type.EXCLUSIVE, renameFromSpec.toString()));
    lockSet.add(new Lock(Lock.Type.EXCLUSIVE, renameToSpec.toString()));
    return lockSet;
  }
}

@@ -6323,95 +6323,8 @@
 ;%0A  
-      LOG.debug(StringUtils.format(%22Renamed %25s to %25s%22, renameFromSpec, renameToSpec));%0A
     
@@ -6385,16 +6385,18 @@
 get the%0A
+  
       //
@@ -6454,16 +6454,18 @@
 e, this%0A
+  
       //
@@ -6494,61 +6494,162 @@
 -op.
- Fall through to the next case.%0A      // fallthrough%0A
+%0A        LOG.debug(StringUtils.format(%22Renamed %25s to %25s%22, renameFromSpec, renameToSpec));%0A      // fallthrough to run the table copy as per above comment.
 %0A   

Fix comment in RenameTableTask