/*
 * 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.hadoop.hive.metastore;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;

import com.google.common.collect.Multimap;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.common.AcidMetaDataFile.DataFormat;
import org.apache.hadoop.hive.common.repl.ReplConst;
import org.apache.hadoop.hive.common.TableName;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.events.AlterPartitionEvent;
import org.apache.hadoop.hive.metastore.events.AlterTableEvent;
import org.apache.hadoop.hive.metastore.messaging.EventMessage;
import org.apache.hadoop.hive.metastore.model.MTable;
import org.apache.hadoop.hive.metastore.utils.FileUtils;
import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils;
import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsDesc;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.txn.TxnUtils;
import org.apache.hadoop.hive.metastore.api.InvalidInputException;
import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;

import javax.jdo.Constants;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;
import java.util.LinkedList;
import java.util.Optional;

import static org.apache.hadoop.hive.metastore.HMSHandler.addTruncateBaseFile;
import static org.apache.hadoop.hive.metastore.HiveMetaHook.ALTERLOCATION;
import static org.apache.hadoop.hive.metastore.HiveMetaHook.ALTER_TABLE_OPERATION_TYPE;
import static org.apache.hadoop.hive.metastore.HiveMetaStoreClient.RENAME_PARTITION_MAKE_COPY;
import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier;

/**
 * Hive specific implementation of alter
 */
public class HiveAlterHandler implements AlterHandler {

  protected Configuration conf;
  private static final Logger LOG = LoggerFactory.getLogger(HiveAlterHandler.class
      .getName());

  // hiveConf, getConf and setConf are in this class because AlterHandler extends Configurable.
  // Always use the configuration from HMS Handler.  Making AlterHandler not extend Configurable
  // is not in the scope of the fix for HIVE-17942.
  @Override
  public Configuration getConf() {
    return conf;
  }

  @Override
  @SuppressWarnings("nls")
  public void setConf(Configuration conf) {
    this.conf = conf;
  }

  @Override
  public void alterTable(RawStore msdb, Warehouse wh, String catName, String dbname,
      String name, Table newt, EnvironmentContext environmentContext,
      IHMSHandler handler, String writeIdList)
          throws InvalidOperationException, MetaException {
    catName = normalizeIdentifier(catName);
    name = normalizeIdentifier(name);
    dbname = normalizeIdentifier(dbname);

    final boolean cascade;
    final boolean replDataLocationChanged;
    final boolean isReplicated;
    if ((environmentContext != null) && environmentContext.isSetProperties()) {
      cascade = StatsSetupConst.TRUE.equals(environmentContext.getProperties().get(StatsSetupConst.CASCADE));
      replDataLocationChanged = ReplConst.TRUE.equals(environmentContext.getProperties().get(ReplConst.REPL_DATA_LOCATION_CHANGED));
    } else {
      cascade = false;
      replDataLocationChanged = false;
    }

    if (newt == null) {
      throw new InvalidOperationException("New table is null");
    }

    String newTblName = newt.getTableName().toLowerCase();
    String newDbName = newt.getDbName().toLowerCase();

    if (!MetaStoreUtils.validateName(newTblName, handler.getConf())) {
      throw new InvalidOperationException(newTblName + " is not a valid object name");
    }
    String validate = MetaStoreServerUtils.validateTblColumns(newt.getSd().getCols());
    if (validate != null) {
      throw new InvalidOperationException("Invalid column " + validate);
    }

    // Validate bucketedColumns in new table
    List<String> bucketColumns = MetaStoreServerUtils.validateBucketColumns(newt.getSd());
    if (CollectionUtils.isNotEmpty(bucketColumns)) {
      String errMsg = "Bucket columns - " + bucketColumns.toString() + " doesn't match with any table columns";
      LOG.error(errMsg);
      throw new InvalidOperationException(errMsg);
    }

    Path srcPath = null;
    FileSystem srcFs;
    Path destPath = null;
    FileSystem destFs = null;

    boolean success = false;
    boolean dataWasMoved = false;
    boolean isPartitionedTable = false;

    Database olddb = null;
    Table oldt = null;

    List<TransactionalMetaStoreEventListener> transactionalListeners = handler.getTransactionalListeners();
    List<MetaStoreEventListener> listeners = handler.getListeners();
    Map<String, String> txnAlterTableEventResponses = Collections.emptyMap();

    try {
      boolean rename = false;
      List<Partition> parts;

      // Switching tables between catalogs is not allowed.
      if (!catName.equalsIgnoreCase(newt.getCatName())) {
        throw new InvalidOperationException("Tables cannot be moved between catalogs, old catalog" +
            catName + ", new catalog " + newt.getCatName());
      }

      // check if table with the new name already exists
      if (!newTblName.equals(name) || !newDbName.equals(dbname)) {
        if (msdb.getTable(catName, newDbName, newTblName, null) != null) {
          throw new InvalidOperationException("new table " + newDbName
              + "." + newTblName + " already exists");
        }
        rename = true;
      }

      String expectedKey = environmentContext != null && environmentContext.getProperties() != null ?
              environmentContext.getProperties().get(hive_metastoreConstants.EXPECTED_PARAMETER_KEY) : null;
      String expectedValue = environmentContext != null && environmentContext.getProperties() != null ?
              environmentContext.getProperties().get(hive_metastoreConstants.EXPECTED_PARAMETER_VALUE) : null;

      if (expectedKey != null) {
        // If we have to check the expected state of the table we have to prevent nonrepeatable reads.
        msdb.openTransaction(Constants.TX_REPEATABLE_READ);
      } else {
        msdb.openTransaction();
      }
      // get old table
      // Note: we don't verify stats here; it's done below in alterTableUpdateTableColumnStats.
      olddb = msdb.getDatabase(catName, dbname);
      oldt = msdb.getTable(catName, dbname, name, null);
      if (oldt == null) {
        throw new InvalidOperationException("table " +
            TableName.getQualified(catName, dbname, name) + " doesn't exist");
      }

      if (expectedKey != null && expectedValue != null
              && !expectedValue.equals(oldt.getParameters().get(expectedKey))) {
        throw new MetaException("The table has been modified. The parameter value for key '" + expectedKey + "' is '"
                + oldt.getParameters().get(expectedKey) + "'. The expected was value was '" + expectedValue + "'");
      }

      validateTableChangesOnReplSource(olddb, oldt, newt, environmentContext);

      // On a replica this alter table will be executed only if old and new both the databases are
      // available and being replicated into. Otherwise, it will be either create or drop of table.
      isReplicated = HMSHandler.isDbReplicationTarget(olddb);
      if (oldt.getPartitionKeysSize() != 0) {
        isPartitionedTable = true;
      }

      // Throws InvalidOperationException if the new column types are not
      // compatible with the current column types.
      DefaultIncompatibleTableChangeHandler.get()
          .allowChange(handler.getConf(), oldt, newt);

      //check that partition keys have not changed, except for virtual views
      //however, allow the partition comments to change
      boolean partKeysPartiallyEqual = checkPartialPartKeysEqual(oldt.getPartitionKeys(),
          newt.getPartitionKeys());

      if(!oldt.getTableType().equals(TableType.VIRTUAL_VIEW.toString())){
        Map<String, String> properties = environmentContext.getProperties();
        if (properties == null || (properties != null &&
            !Boolean.parseBoolean(properties.getOrDefault(HiveMetaHook.ALLOW_PARTITION_KEY_CHANGE,
                "false")))) {
          if (!partKeysPartiallyEqual) {
            throw new InvalidOperationException("partition keys can not be changed.");
          }
        }
      }

      // Two mutually exclusive flows possible.
      // i) Partition locations needs update if replDataLocationChanged is true which means table's
      // data location is changed with all partition sub-directories.
      // ii) Rename needs change the data location and move the data to the new location corresponding
      // to the new name if:
      // 1) the table is not a virtual view, and
      // 2) the table is not an external table, and
      // 3) the user didn't change the default location (or new location is empty), and
      // 4) the table was not initially created with a specified location
      boolean renamedManagedTable = rename && !oldt.getTableType().equals(TableType.VIRTUAL_VIEW.toString())
          && (oldt.getSd().getLocation().compareTo(newt.getSd().getLocation()) == 0
              || StringUtils.isEmpty(newt.getSd().getLocation()))
          && (!MetaStoreUtils.isExternalTable(oldt));

      Database db = msdb.getDatabase(catName, newDbName);

      boolean renamedTranslatedToExternalTable = rename && MetaStoreUtils.isTranslatedToExternalTable(oldt)
          && MetaStoreUtils.isTranslatedToExternalTable(newt);

      boolean isRenameIcebergTable =
          rename && HiveMetaHook.ICEBERG.equalsIgnoreCase(newt.getParameters().get(HiveMetaHook.TABLE_TYPE));

      List<ColumnStatistics> columnStatistics = getColumnStats(msdb, oldt);
      columnStatistics = deleteTableColumnStats(msdb, oldt, newt, columnStatistics);

      if (!isRenameIcebergTable &&
          (replDataLocationChanged || renamedManagedTable || renamedTranslatedToExternalTable)) {
        srcPath = new Path(oldt.getSd().getLocation());

        if (replDataLocationChanged) {
          // If data location is changed in replication flow, then new path was already set in
          // the newt. Also, it is as good as the data is moved and set dataWasMoved=true so that
          // location in partitions are also updated accordingly.
          // No need to validate if the destPath exists as in replication flow, data gets replicated
          // separately.
          destPath = new Path(newt.getSd().getLocation());
          dataWasMoved = true;
        } else {
          // Rename flow.
          // If a table was created in a user specified location using the DDL like
          // create table tbl ... location ...., it should be treated like an external table
          // in the table rename, its data location should not be changed. We can check
          // if the table directory was created directly under its database directory to tell
          // if it is such a table
          // Same applies to the ACID tables suffixed with the `txnId`, case with `lockless reads`.
          String oldtRelativePath = wh.getDatabaseManagedPath(olddb).toUri()
              .relativize(srcPath.toUri()).toString();
          boolean tableInSpecifiedLoc = !oldtRelativePath.equalsIgnoreCase(name)
                  && !oldtRelativePath.equalsIgnoreCase(name + Path.SEPARATOR);


          if (renamedTranslatedToExternalTable || !tableInSpecifiedLoc) {
            srcFs = wh.getFs(srcPath);

            // get new location
            assert(isReplicated == HMSHandler.isDbReplicationTarget(db));
            if (renamedTranslatedToExternalTable) {
              if (!tableInSpecifiedLoc) {
                destPath = new Path(newt.getSd().getLocation());
              } else {
                Path databasePath = constructRenamedPath(wh.getDatabaseExternalPath(db), srcPath);
                destPath = new Path(databasePath, newTblName);
                newt.getSd().setLocation(destPath.toString());
              }
            } else {
              Path databasePath = constructRenamedPath(wh.getDatabaseManagedPath(db), srcPath);
              destPath = new Path(databasePath, newTblName);
              newt.getSd().setLocation(destPath.toString());
            }

            destFs = wh.getFs(destPath);

            // check that destination does not exist otherwise we will be
            // overwriting data
            // check that src and dest are on the same file system
            if (!FileUtils.equalsFileSystem(srcFs, destFs)) {
              throw new InvalidOperationException("table new location " + destPath
                      + " is on a different file system than the old location "
                      + srcPath + ". This operation is not supported");
            }

            try {
              if (destFs.exists(destPath)) {
                throw new InvalidOperationException("New location for this table " +
                        TableName.getQualified(catName, newDbName, newTblName) +
                        " already exists : " + destPath);
              }
              // check that src exists and also checks permissions necessary, rename src to dest
              if (srcFs.exists(srcPath) && wh.renameDir(srcPath, destPath,
                      ReplChangeManager.shouldEnableCm(olddb, oldt))) {
                dataWasMoved = true;
              }
            } catch (IOException | MetaException e) {
              LOG.error("Alter Table operation for " + dbname + "." + name + " failed.", e);
              throw new InvalidOperationException("Alter Table operation for " + dbname + "." + name +
                      " failed to move data due to: '" + getSimpleMessage(e)
                      + "' See hive log file for details.");
            }

            if (!HiveMetaStore.isRenameAllowed(olddb, db)) {
              LOG.error("Alter Table operation for " + TableName.getQualified(catName, dbname, name) +
                      "to new table = " + TableName.getQualified(catName, newDbName, newTblName) + " failed ");
              throw new MetaException("Alter table not allowed for table " +
                      TableName.getQualified(catName, dbname, name) +
                      "to new table = " + TableName.getQualified(catName, newDbName, newTblName));
            }
          }
        }

        if (isPartitionedTable) {
          String oldTblLocPath = srcPath.toUri().getPath();
          String newTblLocPath = dataWasMoved ? destPath.toUri().getPath() : null;

          // also the location field in partition
          parts = msdb.getPartitions(catName, dbname, name, -1);
          Multimap<Partition, ColumnStatistics> columnStatsNeedUpdated = ArrayListMultimap.create();
          for (Partition part : parts) {
            String oldPartLoc = part.getSd().getLocation();
            if (dataWasMoved && oldPartLoc.contains(oldTblLocPath)) {
              URI oldUri = new Path(oldPartLoc).toUri();
              String newPath = oldUri.getPath().replace(oldTblLocPath, newTblLocPath);
              Path newPartLocPath = new Path(oldUri.getScheme(), oldUri.getAuthority(), newPath);
              part.getSd().setLocation(newPartLocPath.toString());
            }
            part.setDbName(newDbName);
            part.setTableName(newTblName);
            List<ColumnStatistics> multiColStats = updateOrGetPartitionColumnStats(msdb, catName, dbname, name,
                part.getValues(), part.getSd().getCols(), oldt, part, null, null);
            for (ColumnStatistics colStats : multiColStats) {
              columnStatsNeedUpdated.put(part, colStats);
            }
          }
          // Do not verify stats parameters on a partitioned table.
          msdb.alterTable(catName, dbname, name, newt, null);
          // alterPartition is only for changing the partition location in the table rename
          if (dataWasMoved) {

            int partsToProcess = parts.size();
            int partitionBatchSize = MetastoreConf.getIntVar(handler.getConf(),
                MetastoreConf.ConfVars.BATCH_RETRIEVE_MAX);
            int batchStart = 0;
            while (partsToProcess > 0) {
              int batchEnd = Math.min(batchStart + partitionBatchSize, parts.size());
              List<Partition> partBatch = parts.subList(batchStart, batchEnd);
              int partBatchSize = partBatch.size();
              partsToProcess -= partBatchSize;
              batchStart += partBatchSize;
              List<List<String>> partValues = new ArrayList<>(partBatchSize);
              for (Partition part : partBatch) {
                partValues.add(part.getValues());
              }
              msdb.alterPartitions(catName, newDbName, newTblName, partValues,
                  partBatch, newt.getWriteId(), writeIdList);
            }
          }
          Deadline.checkTimeout();
          Table table = msdb.getTable(catName, newDbName, newTblName);
          MTable mTable = msdb.ensureGetMTable(catName, newDbName, newTblName);
          for (Entry<Partition, ColumnStatistics> partColStats : columnStatsNeedUpdated.entries()) {
            ColumnStatistics newPartColStats = partColStats.getValue();
            newPartColStats.getStatsDesc().setDbName(newDbName);
            newPartColStats.getStatsDesc().setTableName(newTblName);
            msdb.updatePartitionColumnStatistics(table, mTable, newPartColStats, 
                partColStats.getKey().getValues(), writeIdList, newt.getWriteId());
          }
        } else {
          msdb.alterTable(catName, dbname, name, newt, writeIdList);
        }
      } else {
        // operations other than table rename
        if (MetaStoreServerUtils.requireCalStats(null, null, newt, environmentContext) &&
            !isPartitionedTable) {
          assert(isReplicated == HMSHandler.isDbReplicationTarget(db));
          // Update table stats. For partitioned table, we update stats in alterPartition()
          MetaStoreServerUtils.updateTableStatsSlow(db, newt, wh, false, true, environmentContext);
        }

        if (isPartitionedTable) {
          //Currently only column related changes can be cascaded in alter table
          boolean runPartitionMetadataUpdate =
              (cascade && !MetaStoreServerUtils.areSameColumns(oldt.getSd().getCols(), newt.getSd().getCols()));
          // we may skip the update entirely if there are only new columns added
          runPartitionMetadataUpdate |=
              !cascade && !MetaStoreServerUtils.arePrefixColumns(oldt.getSd().getCols(), newt.getSd().getCols());

          boolean retainOnColRemoval =
              MetastoreConf.getBoolVar(handler.getConf(), MetastoreConf.ConfVars.COLSTATS_RETAIN_ON_COLUMN_REMOVAL);

          if (runPartitionMetadataUpdate) {
            if (cascade || retainOnColRemoval) {
              parts = msdb.getPartitions(catName, dbname, name, -1);
              for (Partition part : parts) {
                Partition oldPart = new Partition(part);
                List<FieldSchema> oldCols = part.getSd().getCols();
                part.getSd().setCols(newt.getSd().getCols());
                List<ColumnStatistics> colStats = updateOrGetPartitionColumnStats(msdb, catName, dbname, name,
                    part.getValues(), oldCols, oldt, part, null, null);
                assert (colStats.isEmpty());
                Deadline.checkTimeout();
                if (cascade) {
                  msdb.alterPartition(
                    catName, dbname, name, part.getValues(), part, writeIdList);
                } else {
                  // update changed properties (stats)
                  oldPart.setParameters(part.getParameters());
                  msdb.alterPartition(catName, dbname, name, part.getValues(), oldPart, writeIdList);
                }
              }
            } else {
              // clear all column stats to prevent incorract behaviour in case same column is reintroduced
              TableName tableName = new TableName(catName, dbname, name);
              msdb.deleteAllPartitionColumnStatistics(tableName, writeIdList);
            }
            // Don't validate table-level stats for a partitoned table.
            msdb.alterTable(catName, dbname, name, newt, null);
          } else {
            LOG.warn("Alter table not cascaded to partitions.");
            msdb.alterTable(catName, dbname, name, newt, writeIdList);
          }
        } else {
          msdb.alterTable(catName, dbname, name, newt, writeIdList);
        }
      }

      //HIVE-26504: Table columns stats may exist even for partitioned tables, so it must be updated in all cases
      updateTableColumnStats(msdb, newt, writeIdList, columnStatistics);

      if (transactionalListeners != null && !transactionalListeners.isEmpty()) {
        txnAlterTableEventResponses = MetaStoreListenerNotifier.notifyEvent(transactionalListeners,
                  EventMessage.EventType.ALTER_TABLE,
                  new AlterTableEvent(oldt, newt, false, true,
                          newt.getWriteId(), handler, isReplicated),
                  environmentContext);
      }
      // commit the changes
      success = msdb.commitTransaction();
    } catch (InvalidObjectException e) {
      LOG.debug("Failed to get object from Metastore ", e);
      throw new InvalidOperationException(
          "Unable to change partition or table."
              + " Check metastore logs for detailed stack." + e.getMessage());
    } catch (InvalidInputException e) {
        LOG.debug("Accessing Metastore failed due to invalid input ", e);
        throw new InvalidOperationException(
            "Unable to change partition or table."
                + " Check metastore logs for detailed stack." + e.getMessage());
    } catch (NoSuchObjectException e) {
      LOG.debug("Object not found in metastore ", e);
      throw new InvalidOperationException(
          "Unable to change partition or table. Object " +  e.getMessage() + " does not exist."
              + " Check metastore logs for detailed stack.");
    } finally {
      if (success) {
        // Txn was committed successfully.
        // If data location is changed in replication flow, then need to delete the old path.
        if (replDataLocationChanged) {
          assert(olddb != null);
          assert(oldt != null);
          Path deleteOldDataLoc = new Path(oldt.getSd().getLocation());
          boolean isSkipTrash = MetaStoreUtils.isSkipTrash(oldt.getParameters());
          try {
            wh.deleteDir(deleteOldDataLoc, true, isSkipTrash,
                    ReplChangeManager.shouldEnableCm(olddb, oldt));
            LOG.info("Deleted the old data location: {} for the table: {}",
                    deleteOldDataLoc, dbname + "." + name);
          } catch (MetaException ex) {
            // Eat the exception as it doesn't affect the state of existing tables.
            // Expect, user to manually drop this path when exception and so logging a warning.
            LOG.warn("Unable to delete the old data location: {} for the table: {}",
                    deleteOldDataLoc, dbname + "." + name);
          }
        }
      } else {
        LOG.error("Failed to alter table " + TableName.getQualified(catName, dbname, name));
        msdb.rollbackTransaction();
        if (!replDataLocationChanged && dataWasMoved) {
          try {
            if (destFs.exists(destPath)) {
              if (!destFs.rename(destPath, srcPath)) {
                LOG.error("Failed to restore data from " + destPath + " to " + srcPath
                    + " in alter table failure. Manual restore is needed.");
              }
            }
          } catch (IOException e) {
            LOG.error("Failed to restore data from " + destPath + " to " + srcPath
                +  " in alter table failure. Manual restore is needed.");
          }
        }
      }
    }

    if (!listeners.isEmpty()) {
      // I don't think event notifications in case of failures are necessary, but other HMS operations
      // make this call whether the event failed or succeeded. To make this behavior consistent,
      // this call is made for failed events also.
      MetaStoreListenerNotifier.notifyEvent(listeners, EventMessage.EventType.ALTER_TABLE,
          new AlterTableEvent(oldt, newt, false, success, newt.getWriteId(), handler, isReplicated),
          environmentContext, txnAlterTableEventResponses, msdb);
    }
  }

  /**
   * MetaException that encapsulates error message from RemoteException from hadoop RPC which wrap
   * the stack trace into e.getMessage() which makes logs/stack traces confusing.
   * @param ex
   * @return
   */
  String getSimpleMessage(Exception ex) {
    if(ex instanceof MetaException) {
      String msg = ex.getMessage();
      if(msg == null || !msg.contains("\n")) {
        return msg;
      }
      return msg.substring(0, msg.indexOf('\n'));
    }
    return ex.getMessage();
  }

  @Override
  public Partition alterPartition(final RawStore msdb, Warehouse wh, final String dbname,
    final String name, final List<String> part_vals, final Partition new_part,
    EnvironmentContext environmentContext)
      throws InvalidOperationException, InvalidObjectException, AlreadyExistsException, MetaException, NoSuchObjectException {
    return alterPartition(msdb, wh, MetaStoreUtils.getDefaultCatalog(conf), dbname, name, part_vals, new_part,
        environmentContext, null, null);
  }

  @Override
  public Partition alterPartition(RawStore msdb, Warehouse wh, String catName, String dbname,
      String name, List<String> part_vals, final Partition new_part,
      EnvironmentContext environmentContext, IHMSHandler handler, String validWriteIds)
      throws InvalidOperationException, InvalidObjectException, AlreadyExistsException, MetaException, NoSuchObjectException {
    boolean success = false;
    Partition oldPart;
    List<TransactionalMetaStoreEventListener> transactionalListeners = null;
    if (handler != null) {
      transactionalListeners = handler.getTransactionalListeners();
    }

    // Set DDL time to now if not specified
    if (new_part.getParameters() == null ||
        new_part.getParameters().get(hive_metastoreConstants.DDL_TIME) == null ||
        Integer.parseInt(new_part.getParameters().get(hive_metastoreConstants.DDL_TIME)) == 0) {
      new_part.putToParameters(hive_metastoreConstants.DDL_TIME, Long.toString(System
          .currentTimeMillis() / 1000));
    }

    //alter partition
    if (part_vals == null || part_vals.size() == 0) {
      try {
        msdb.openTransaction();

        Table tbl = msdb.getTable(catName, dbname, name, null);
        if (tbl == null) {
          throw new InvalidObjectException(
              "Unable to alter partition because table or database does not exist.");
        }
        oldPart = msdb.getPartition(catName, dbname, name, new_part.getValues());
        if (MetaStoreServerUtils.requireCalStats(oldPart, new_part, tbl, environmentContext)) {
          // if stats are same, no need to update
          if (MetaStoreServerUtils.isFastStatsSame(oldPart, new_part)) {
            MetaStoreServerUtils.updateBasicState(environmentContext, new_part.getParameters());
          } else {
            MetaStoreServerUtils.updatePartitionStatsFast(
                new_part, tbl, wh, false, true, environmentContext, false);
          }
        }

        // PartitionView does not have SD. We do not need update its column stats
        if (oldPart.getSd() != null) {
          updateOrGetPartitionColumnStats(msdb, catName, dbname, name, new_part.getValues(),
              oldPart.getSd().getCols(), tbl, new_part, null, null);
        }
        Deadline.checkTimeout();
        msdb.alterPartition(
            catName, dbname, name, new_part.getValues(), new_part, validWriteIds);
        if (transactionalListeners != null && !transactionalListeners.isEmpty()) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners,
                                                EventMessage.EventType.ALTER_PARTITION,
                                                new AlterPartitionEvent(oldPart, new_part, tbl, false,
                                                        true, new_part.getWriteId(), handler),
                                                environmentContext);


        }
        success = msdb.commitTransaction();
      } catch (InvalidObjectException e) {
        LOG.warn("Alter failed", e);
        throw new InvalidOperationException("alter is not possible: " + e.getMessage());
      } catch (NoSuchObjectException e) {
        //old partition does not exist
        throw new InvalidOperationException("alter is not possible: " + e.getMessage());
      } finally {
        if(!success) {
          msdb.rollbackTransaction();
        }
      }
      return oldPart;
    }

    //rename partition
    String oldPartLoc;
    String newPartLoc;
    Path srcPath = null;
    Path destPath = null;
    FileSystem srcFs;
    FileSystem destFs = null;
    boolean dataWasMoved = false;
    Database db;
    try {
      msdb.openTransaction();
      Table tbl = msdb.getTable(catName, dbname, name, null);
      if (tbl == null) {
        throw new InvalidObjectException(
            "Unable to alter partition because table or database does not exist.");
      }
      MTable mTable = msdb.ensureGetMTable(catName, dbname, name);
      try {
        oldPart = msdb.getPartition(catName, dbname, name, part_vals);
      } catch (NoSuchObjectException e) {
        // this means there is no existing partition
        throw new InvalidObjectException(
            "Unable to rename partition because old partition does not exist");
      }

      Partition check_part;
      try {
        check_part = msdb.getPartition(catName, dbname, name, new_part.getValues());
      } catch(NoSuchObjectException e) {
        // this means there is no existing partition
        check_part = null;
      }

      if (check_part != null) {
        throw new AlreadyExistsException("Partition already exists:" + dbname + "." + name + "." +
            new_part.getValues());
      }

      // when renaming a partition, we should update
      // 1) partition SD Location
      // 2) partition column stats if there are any because of part_name field in HMS table PART_COL_STATS
      // 3) rename the partition directory if it is not an external table
      if (!tbl.getTableType().equals(TableType.EXTERNAL_TABLE.toString())) {
        // TODO: refactor this into a separate method after master merge, this one is too big.
        try {
          db = msdb.getDatabase(catName, dbname);

          // if tbl location is available use it
          // else derive the tbl location from database location
          destPath = wh.getPartitionPath(db, tbl, new_part.getValues());
          destPath = constructRenamedPath(destPath, new Path(new_part.getSd().getLocation()));
        } catch (NoSuchObjectException e) {
          LOG.debug("Didn't find object in metastore ", e);
          throw new InvalidOperationException(
            "Unable to change partition or table. Database " + dbname + " does not exist"
              + " Check metastore logs for detailed stack." + e.getMessage());
        }

        if (destPath != null) {
          newPartLoc = destPath.toString();
          oldPartLoc = oldPart.getSd().getLocation();
          LOG.info("srcPath:" + oldPartLoc);
          LOG.info("descPath:" + newPartLoc);
          srcPath = new Path(oldPartLoc);
          srcFs = wh.getFs(srcPath);
          destFs = wh.getFs(destPath);
          // check that src and dest are on the same file system
          if (!FileUtils.equalsFileSystem(srcFs, destFs)) {
            throw new InvalidOperationException("New table location " + destPath
              + " is on a different file system than the old location "
              + srcPath + ". This operation is not supported.");
          }

          try {
            if (srcFs.exists(srcPath)) {
              if (newPartLoc.compareTo(oldPartLoc) != 0 && destFs.exists(destPath)) {
                throw new InvalidOperationException("New location for this table "
                  + tbl.getDbName() + "." + tbl.getTableName()
                  + " already exists : " + destPath);
              }
              //if destPath's parent path doesn't exist, we should mkdir it
              Path destParentPath = destPath.getParent();
              if (!wh.mkdirs(destParentPath)) {
                  throw new MetaException("Unable to create path " + destParentPath);
              }

              boolean clonePart = Optional.ofNullable(environmentContext)
                  .map(EnvironmentContext::getProperties)
                  .map(prop -> prop.get(RENAME_PARTITION_MAKE_COPY))
                  .map(Boolean::parseBoolean)
                  .orElse(false);
              long writeId = new_part.getWriteId();

              if (writeId > 0 && clonePart) {
                LOG.debug("Making a copy of the partition directory: {} under a new location: {}", srcPath, destPath);

                if (!wh.copyDir(srcPath, destPath, ReplChangeManager.shouldEnableCm(db, tbl))) {
                  LOG.error("Copy failed for source: " + srcPath + " to destination: " + destPath);
                  throw new IOException("File copy failed.");
                }
                addTruncateBaseFile(srcPath, writeId, conf, DataFormat.DROPPED);
              } else {
                //rename the data directory
                wh.renameDir(srcPath, destPath, ReplChangeManager.shouldEnableCm(db, tbl));
              }
              LOG.info("Partition directory rename from " + srcPath + " to " + destPath + " done.");
              dataWasMoved = true;
            }
          } catch (IOException e) {
            LOG.error("Cannot rename partition directory from " + srcPath + " to " + destPath, e);
            throw new InvalidOperationException("Unable to access src or dest location for partition "
                + tbl.getDbName() + "." + tbl.getTableName() + " " + new_part.getValues());
          } catch (MetaException me) {
            LOG.error("Cannot rename partition directory from " + srcPath + " to " + destPath, me);
            throw me;
          }
          new_part.getSd().setLocation(newPartLoc);
        }
      } else {
        new_part.getSd().setLocation(oldPart.getSd().getLocation());
      }

      if (MetaStoreServerUtils.requireCalStats(oldPart, new_part, tbl, environmentContext)) {
        MetaStoreServerUtils.updatePartitionStatsFast(
            new_part, tbl, wh, false, true, environmentContext, false);
      }

      String newPartName = Warehouse.makePartName(tbl.getPartitionKeys(), new_part.getValues());
      List<ColumnStatistics> multiColumnStats = updateOrGetPartitionColumnStats(msdb, catName, dbname, name, oldPart.getValues(),
          oldPart.getSd().getCols(), tbl, new_part, null, null);
      msdb.alterPartition(catName, dbname, name, part_vals, new_part, validWriteIds);
      if (!multiColumnStats.isEmpty()) {
        for (ColumnStatistics cs : multiColumnStats) {
          cs.getStatsDesc().setPartName(newPartName);
          try {
            msdb.updatePartitionColumnStatistics(tbl, mTable, cs, new_part.getValues(),
                validWriteIds, new_part.getWriteId());
          } catch (InvalidInputException iie) {
            throw new InvalidOperationException("Unable to update partition stats in table rename." + iie);
          } catch (NoSuchObjectException nsoe) {
            // It is ok, ignore
          }
        }
      }

      if (transactionalListeners != null && !transactionalListeners.isEmpty()) {
        MetaStoreListenerNotifier.notifyEvent(transactionalListeners,
                                              EventMessage.EventType.ALTER_PARTITION,
                                              new AlterPartitionEvent(oldPart, new_part, tbl, false,
                                                      true, new_part.getWriteId(), handler),
                                              environmentContext);
      }

      success = msdb.commitTransaction();
    } finally {
      if (!success) {
        LOG.error("Failed to rename a partition. Rollback transaction");
        msdb.rollbackTransaction();
        if (dataWasMoved) {
          LOG.error("Revert the data move in renaming a partition.");
          try {
            if (destFs.exists(destPath)) {
              wh.renameDir(destPath, srcPath, false);
            }
          } catch (MetaException me) {
            LOG.error("Failed to restore partition data from " + destPath + " to " + srcPath
                +  " in alter partition failure. Manual restore is needed.");
          } catch (IOException ioe) {
            LOG.error("Failed to restore partition data from " + destPath + " to " + srcPath
                +  " in alter partition failure. Manual restore is needed.");
          }
        }
      }
    }
    return oldPart;
  }

  @Deprecated
  @Override
  public List<Partition> alterPartitions(final RawStore msdb, Warehouse wh, final String dbname,
    final String name, final List<Partition> new_parts,
    EnvironmentContext environmentContext)
      throws InvalidOperationException, InvalidObjectException, AlreadyExistsException, MetaException {
    return alterPartitions(msdb, wh, MetaStoreUtils.getDefaultCatalog(conf), dbname, name, new_parts,
        environmentContext, null, -1, null);
  }

  private Map<List<String>, Partition> getExistingPartitions(final RawStore msdb,
      final List<Partition> new_parts, final Table tbl, final String catName,
      final String dbname, final String name)
      throws MetaException, NoSuchObjectException, InvalidOperationException {

    // Get list of partition values
    List<String> partValues = new LinkedList<>();
    for (Partition tmpPart : new_parts) {
      partValues.add(Warehouse.makePartName(tbl.getPartitionKeys(), tmpPart.getValues()));
    }

    // Get existing partitions from store
    List<Partition> oldParts = msdb.getPartitionsByNames(catName, dbname, name, partValues);
    if (new_parts.size() != oldParts.size()) {
      throw new InvalidOperationException("Alter partition operation failed: "
          + "new parts size " + new_parts.size()
          + " not matching with old parts size " + oldParts.size());
    }
    return oldParts.stream().collect(Collectors.toMap(Partition::getValues, Partition -> Partition));
  }

  @Override
  public List<Partition> alterPartitions(final RawStore msdb, Warehouse wh, final String catName,
                                         final String dbname, final String name,
                                         final List<Partition> new_parts,
                                         EnvironmentContext environmentContext,
                                         String writeIdList, long writeId,
                                         IHMSHandler handler)
      throws InvalidOperationException, InvalidObjectException, AlreadyExistsException, MetaException {
    List<Partition> oldParts = new ArrayList<>();
    List<List<String>> partValsList = new ArrayList<>();
    List<TransactionalMetaStoreEventListener> transactionalListeners = null;
    if (handler != null) {
      transactionalListeners = handler.getTransactionalListeners();
    }

    boolean success = false;
    try {
      msdb.openTransaction();

      // Note: should we pass in write ID here? We only update stats on parts so probably not.
      Table tbl = msdb.getTable(catName, dbname, name, null);
      if (tbl == null) {
        throw new InvalidObjectException(
            "Unable to alter partitions because table or database does not exist.");
      }

      blockPartitionLocationChangesOnReplSource(msdb.getDatabase(catName, dbname), tbl,
                                                environmentContext);
      Map<List<String>, Partition> oldPartMap = getExistingPartitions(msdb, new_parts, tbl, catName, dbname, name);

      for (Partition tmpPart: new_parts) {
        // Set DDL time to now if not specified
        if (tmpPart.getParameters() == null ||
            tmpPart.getParameters().get(hive_metastoreConstants.DDL_TIME) == null ||
            Integer.parseInt(tmpPart.getParameters().get(hive_metastoreConstants.DDL_TIME)) == 0) {
          tmpPart.putToParameters(hive_metastoreConstants.DDL_TIME, Long.toString(System
              .currentTimeMillis() / 1000));
        }

        Partition oldTmpPart = oldPartMap.get(tmpPart.getValues());
        oldParts.add(oldTmpPart);
        partValsList.add(tmpPart.getValues());

        if (MetaStoreServerUtils.requireCalStats(oldTmpPart, tmpPart, tbl, environmentContext)) {
          // Check if stats are same, no need to update
          if (MetaStoreServerUtils.isFastStatsSame(oldTmpPart, tmpPart)) {
            MetaStoreServerUtils.updateBasicState(environmentContext, tmpPart.getParameters());
          } else {
            MetaStoreServerUtils.updatePartitionStatsFast(
                tmpPart, tbl, wh, false, true, environmentContext, false);
          }
        }

        // PartitionView does not have SD and we do not need to update its column stats
        if (oldTmpPart.getSd() != null) {
          updateOrGetPartitionColumnStats(msdb, catName, dbname, name, oldTmpPart.getValues(),
              oldTmpPart.getSd().getCols(), tbl, tmpPart, null, null);
        }
      }

      msdb.alterPartitions(catName, dbname, name, partValsList, new_parts, writeId, writeIdList);
      Iterator<Partition> oldPartsIt = oldParts.iterator();
      for (Partition newPart : new_parts) {
        Partition oldPart;
        if (oldPartsIt.hasNext()) {
          oldPart = oldPartsIt.next();
        } else {
          throw new InvalidOperationException("Missing old partition corresponding to new partition " +
              "when invoking MetaStoreEventListener for alterPartitions event.");
        }

        if (transactionalListeners != null && !transactionalListeners.isEmpty()) {
          MetaStoreListenerNotifier.notifyEvent(transactionalListeners, EventMessage.EventType.ALTER_PARTITION,
              new AlterPartitionEvent(oldPart, newPart, tbl, false, true, newPart.getWriteId(), handler),
              environmentContext);
        }
      }

      success = msdb.commitTransaction();
    } catch (InvalidObjectException | NoSuchObjectException e) {
      throw new InvalidOperationException("Alter partition operation failed: " + e);
    } finally {
      if(!success) {
        msdb.rollbackTransaction();
      }
    }

    return oldParts;
  }

  // Validate changes to partition's location to protect against errors on migration during
  // replication
  private void blockPartitionLocationChangesOnReplSource(Database db, Table tbl,
                                                         EnvironmentContext ec)
          throws InvalidOperationException {
    // If the database is not replication source, nothing to do
    if (!ReplChangeManager.isSourceOfReplication(db)) {
      return;
    }

    // Do not allow changing location of a managed table as alter event doesn't capture the
    // new files list. So, it may cause data inconsistency.
    if ((ec != null) && ec.isSetProperties()) {
      String alterType = ec.getProperties().get(ALTER_TABLE_OPERATION_TYPE);
      if (alterType != null && alterType.equalsIgnoreCase(ALTERLOCATION) &&
          tbl.getTableType().equalsIgnoreCase(TableType.MANAGED_TABLE.name())) {
        String tableName = TableName.getQualified(tbl.getCatName(), tbl.getDbName(), tbl.getTableName());
        throw new InvalidOperationException(
            "Cannot change location of a managed table " + tableName + " as it is enabled for replication.");
      }
    }
  }

  // Validate changes to a table to protect against errors on migration during replication.
  private void validateTableChangesOnReplSource(Database db, Table oldTbl, Table newTbl,
                                                EnvironmentContext ec)
          throws InvalidOperationException {
    // If the database is not replication source, nothing to do
    if (!ReplChangeManager.isSourceOfReplication(db)) {
      return;
    }

    // Do not allow changing location of a managed table as alter event doesn't capture the
    // new files list. So, it may cause data inconsistency. We do this whether or not strict
    // managed is true on the source cluster.
    if ((ec != null) && ec.isSetProperties()) {
      String alterType = ec.getProperties().get(ALTER_TABLE_OPERATION_TYPE);
      if (alterType != null && alterType.equalsIgnoreCase(ALTERLOCATION) &&
          oldTbl.getTableType().equalsIgnoreCase(TableType.MANAGED_TABLE.name())) {
        String tableName = TableName.getQualified(oldTbl.getCatName(), oldTbl.getDbName(), oldTbl.getTableName());
        throw new InvalidOperationException(
            "Cannot change location of a managed table " + tableName + " as it is enabled for replication.");
        }
    }

    // Rest of the changes need validation only when strict managed tables is false. That's
    // when there's scope for migration during replication, at least for now.
    if (conf.getBoolean(MetastoreConf.ConfVars.STRICT_MANAGED_TABLES.getHiveName(), false)) {
      return;
    }

    // Do not allow changing the type of table. This is to avoid migration scenarios which causes
    // Managed ACID table to be converted to external at replica. As ACID tables cannot be
    // converted to external table and vice versa, we need to restrict this conversion at primary
    // as well. Currently, table type conversion is allowed only between managed and external
    // table types. But, to be future proof, any table type conversion is restricted on a
    // replication enabled DB.
    if (!oldTbl.getTableType().equalsIgnoreCase(newTbl.getTableType())) {
      throw new InvalidOperationException("Table type cannot be changed from " + oldTbl.getTableType()
              + " to " + newTbl.getTableType() + " for the table " +
              TableName.getQualified(oldTbl.getCatName(), oldTbl.getDbName(), oldTbl.getTableName())
              + " as it is enabled for replication.");
    }

    // Also we do not allow changing a non-Acid managed table to acid table on source with strict
    // managed false. After replicating a non-acid managed table to a target with strict managed
    // true the table will be converted to acid or external table. So changing the transactional
    // property of table on source can conflict with resultant change in the target.
    if (!TxnUtils.isTransactionalTable(oldTbl) && TxnUtils.isTransactionalTable(newTbl)) {
      throw new InvalidOperationException("A non-Acid table cannot be converted to an Acid " +
              "table for the table " + TableName.getQualified(oldTbl.getCatName(),
              oldTbl.getDbName(), oldTbl.getTableName()) + " as it is enabled for replication.");
    }
  }

  private boolean checkPartialPartKeysEqual(List<FieldSchema> oldPartKeys,
      List<FieldSchema> newPartKeys) {
    //return true if both are null, or false if one is null and the other isn't
    if (newPartKeys == null || oldPartKeys == null) {
      return oldPartKeys == newPartKeys;
    }
    if (oldPartKeys.size() != newPartKeys.size()) {
      return false;
    }
    Iterator<FieldSchema> oldPartKeysIter = oldPartKeys.iterator();
    Iterator<FieldSchema> newPartKeysIter = newPartKeys.iterator();
    FieldSchema oldFs;
    FieldSchema newFs;
    while (oldPartKeysIter.hasNext()) {
      oldFs = oldPartKeysIter.next();
      newFs = newPartKeysIter.next();
      // Alter table can change the type of partition key now.
      // So check the column name only.
      if (!oldFs.getName().equals(newFs.getName())) {
        return false;
      }
    }

    return true;
  }

  /**
   * Uses the scheme and authority of the object's current location and the path constructed
   * using the object's new name to construct a path for the object's new location.
   */
  private Path constructRenamedPath(Path defaultNewPath, Path currentPath) {
    URI currentUri = currentPath.toUri();

    return new Path(currentUri.getScheme(), currentUri.getAuthority(),
        defaultNewPath.toUri().getPath());
  }

  public static List<ColumnStatistics> getColumnStats(RawStore msdb, Table oldTable)
      throws NoSuchObjectException, MetaException {
    String catName = normalizeIdentifier(oldTable.isSetCatName()
        ? oldTable.getCatName()
        : getDefaultCatalog(msdb.getConf()));
    String dbName = oldTable.getDbName().toLowerCase();
    String tableName = normalizeIdentifier(oldTable.getTableName());
    List<String> columnNames = oldTable.getSd().getCols().stream().map(FieldSchema::getName).collect(Collectors.toList());
    return msdb.getTableColumnStatistics(catName, dbName, tableName, columnNames);
  }

  @VisibleForTesting
  public static List<ColumnStatistics> deleteTableColumnStats(RawStore msdb, Table oldTable, Table newTable, List<ColumnStatistics> multiColStats)
      throws InvalidObjectException, MetaException {
    List<ColumnStatistics> newMultiColStats = new ArrayList<>();
    try {
      String catName = normalizeIdentifier(oldTable.isSetCatName()
          ? oldTable.getCatName()
          : getDefaultCatalog(msdb.getConf()));
      String dbName = oldTable.getDbName().toLowerCase();
      String tableName = normalizeIdentifier(oldTable.getTableName());
      String newDbName = newTable.getDbName().toLowerCase();
      String newTableName = normalizeIdentifier(newTable.getTableName());
      List<FieldSchema> oldTableCols = oldTable.getSd().getCols();
      List<FieldSchema> newTableCols = newTable.getSd().getCols();

      boolean nameChanged = !newDbName.equals(dbName) || !newTableName.equals(tableName);

      if ((nameChanged || !MetaStoreServerUtils.columnsIncludedByNameType(oldTableCols, newTableCols)) &&
          // Don't bother in the case of ACID conversion.
          TxnUtils.isAcidTable(oldTable) == TxnUtils.isAcidTable(newTable)) {
        for (ColumnStatistics colStats : multiColStats) {
          List<ColumnStatisticsObj> statsObjs = colStats.getStatsObj();
          List<ColumnStatisticsObj> newStatsObjs = new ArrayList<>();

          if (statsObjs != null) {
            for (ColumnStatisticsObj statsObj : statsObjs) {
              boolean found = newTableCols.stream().anyMatch(c -> statsObj.getColName().equalsIgnoreCase(c.getName()) &&
                  statsObj.getColType().equalsIgnoreCase(c.getType()));
              if (nameChanged || !found) {
                msdb.deleteTableColumnStatistics(catName, oldTable.getDbName().toLowerCase(),
                    normalizeIdentifier(oldTable.getTableName()), statsObj.getColName(), colStats.getEngine());
              }
              if (found) {
                newStatsObjs.add(statsObj);
              }
            }
            StatsSetupConst.removeColumnStatsState(newTable.getParameters(),
                statsObjs.stream().map(ColumnStatisticsObj::getColName).collect(Collectors.toList()));
          }
          ColumnStatisticsDesc statsDesc = colStats.getStatsDesc();
          statsDesc.setDbName(newDbName);
          statsDesc.setTableName(newTableName);
          colStats.setStatsObj(newStatsObjs);
          newMultiColStats.add(colStats);
        }
      }
    } catch (NoSuchObjectException nsoe) {
      LOG.debug("Could not find db entry." + nsoe);
    } catch (InvalidInputException e) {
      //should not happen since the input were verified before passed in
      throw new InvalidObjectException("Invalid inputs to update table column stats: " + e);
    }
    return newMultiColStats;
  }

  @VisibleForTesting
  public void updateTableColumnStats(RawStore msdb, Table newTable, String validWriteIds, List<ColumnStatistics> columnStatistics)
      throws MetaException, InvalidObjectException {
    Deadline.checkTimeout();
    // Change to new table and append stats for the new table
    for (ColumnStatistics colStats : columnStatistics) {
      try {
        msdb.updateTableColumnStatistics(colStats, validWriteIds, newTable.getWriteId());
      } catch (NoSuchObjectException nsoe) {
        LOG.debug("Could not find db entry." + nsoe);
      } catch (InvalidInputException e) {
        //should not happen since the input were verified before passed in
        throw new InvalidObjectException("Invalid inputs to update table column stats: " + e);
      }
    }
  }

  public static List<ColumnStatisticsObj> filterColumnStatsForTableColumns(List<FieldSchema> columns, ColumnStatistics colStats) {
    return colStats.getStatsObj()
        .stream()
        .filter(o -> columns
            .stream()
            .anyMatch(column -> o.getColName().equalsIgnoreCase(column.getName()) && o.getColType().equalsIgnoreCase(column.getType())))
        .collect(Collectors.toList());
  }

  public static List<ColumnStatistics> updateOrGetPartitionColumnStats(
      RawStore msdb, String catName, String dbname, String tblname, List<String> partVals,
      List<FieldSchema> oldCols, Table table, Partition part, List<FieldSchema> newCols, List<String> deletedCols)
          throws MetaException, InvalidObjectException {
    List<ColumnStatistics> newPartsColStats = new ArrayList<>();
    boolean updateColumnStats = true;
    try {
      // if newCols are not specified, use default ones.
      if (newCols == null) {
        newCols = part.getSd() == null ? new ArrayList<>() : part.getSd().getCols();
      }
      String oldPartName = Warehouse.makePartName(table.getPartitionKeys(), partVals);
      String newPartName = Warehouse.makePartName(table.getPartitionKeys(), part.getValues());
      boolean rename = !part.getDbName().equals(dbname) || !part.getTableName().equals(tblname)
          || !oldPartName.equals(newPartName);

      // do not need to update column stats if alter partition is not for rename or changing existing columns
      if (!rename && MetaStoreServerUtils.columnsIncludedByNameType(oldCols, newCols)) {
        return newPartsColStats;
      }
      List<String> oldColNames = new ArrayList<>(oldCols.size());
      for (FieldSchema oldCol : oldCols) {
        oldColNames.add(oldCol.getName());
      }
      List<String> oldPartNames = Lists.newArrayList(oldPartName);
      // TODO: doesn't take txn stats into account. This method can only remove stats.
      List<List<ColumnStatistics>> multiPartsColStats = msdb.getPartitionColumnStatistics(catName, dbname, tblname,
          oldPartNames, oldColNames);
      for (List<ColumnStatistics> partsColStats : multiPartsColStats) {
        assert (partsColStats.size() <= 1);

        // for out para, this value is initialized by caller.
        if (deletedCols == null) {
          deletedCols = new ArrayList<>();
        } else {
          // in case deletedCols is provided by caller, stats will be updated  by caller.
          updateColumnStats = false;
        }
        for (ColumnStatistics partColStats : partsColStats) { //actually only at most one loop
          List<ColumnStatisticsObj> newStatsObjs = new ArrayList<>();
          List<ColumnStatisticsObj> statsObjs = partColStats.getStatsObj();
          for (ColumnStatisticsObj statsObj : statsObjs) {
            boolean found = false;
            for (FieldSchema newCol : newCols) {
              if (statsObj.getColName().equalsIgnoreCase(newCol.getName())
                  && statsObj.getColType().equalsIgnoreCase(newCol.getType())) {
                found = true;
                break;
              }
            }
            Deadline.checkTimeout();
            if (found) {
              if (rename) {
                if (updateColumnStats) {
                  msdb.deletePartitionColumnStatistics(catName, dbname, tblname,
                      partColStats.getStatsDesc().getPartName(), partVals, statsObj.getColName(),
                      partColStats.getEngine());
                } else {
                  deletedCols.add(statsObj.getColName());
                }
                newStatsObjs.add(statsObj);
              }
            } else {
              if (updateColumnStats) {
                msdb.deletePartitionColumnStatistics(catName, dbname, tblname, partColStats.getStatsDesc().getPartName(),
                    partVals, statsObj.getColName(), partColStats.getEngine());
              }
              deletedCols.add(statsObj.getColName());
            }
          }
          if (updateColumnStats) {
            StatsSetupConst.removeColumnStatsState(part.getParameters(), deletedCols);
          }
          if (!newStatsObjs.isEmpty()) {
            partColStats.setStatsObj(newStatsObjs);
            newPartsColStats.add(partColStats);
          }
        }
      }
    } catch (NoSuchObjectException nsoe) {
      // ignore this exception, actually this exception won't be thrown from getPartitionColumnStatistics
    } catch (InvalidInputException iie) {
      throw new InvalidObjectException("Invalid input to delete partition column stats." + iie);
    }

    return newPartsColStats;
  }
}