// 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.impala.util;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Range;
import com.google.errorprone.annotations.Immutable;

import java.io.IOException;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.hadoop.hive.common.ValidWriteIdList.RangeResponse;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FileMetadataLoader.LoadStats;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.ScalarType;
import org.apache.impala.catalog.StructField;
import org.apache.impala.catalog.StructType;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.Pair;
import org.apache.impala.common.Reference;
import org.apache.impala.thrift.THdfsFileDesc;
import org.apache.impala.thrift.TPartialPartitionInfo;
import org.apache.impala.thrift.TTransactionalType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Predicate;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.annotation.Nullable;

/**
 * Contains utility functions for working with Acid tables.
 * <p>
 * The code is mostly copy pasted from Hive. Ideally we should use the
 * code directly from Hive.
 * </p>
 */
public class AcidUtils {
  private final static Logger LOG = LoggerFactory.getLogger(AcidUtils.class);
  // Constant also defined in TransactionalValidationListener
  public static final String INSERTONLY_TRANSACTIONAL_PROPERTY = "insert_only";
  // Constant also defined in hive_metastoreConstants
  public static final String TABLE_IS_TRANSACTIONAL = "transactional";
  public static final String TABLE_TRANSACTIONAL_PROPERTIES = "transactional_properties";

  /**
   * Transaction parameters needed for single table operations.
   */
  public static class TblTransaction {
    public long txnId;
    public boolean ownsTxn;
    public long writeId;
    public String validWriteIds;
  }

  // Regex pattern for files in base directories. The pattern matches strings like
  // "base_0000005/abc.txt",
  // "base_0000005/0000/abc.txt",
  // "base_0000003_v0003217/000000_0"
  private static final Pattern BASE_PATTERN = Pattern.compile(
      "base_" +
      "(?<writeId>\\d+)" +
      "(?:_v(?<visibilityTxnId>\\d+))?" +
      "(?:/.*)?");

  // Regex pattern for files in delta directories. The pattern matches strings like
  // "delta_0000006_0000006/000000_0",
  // "delta_0000009_0000009_0000/0000/def.txt"
  private static final String DELTA_STR =
      "delta_" +
      "(?<minWriteId>\\d+)_" +
      "(?<maxWriteId>\\d+)" +
      // Statement id, or visiblityTxnId
      "(?:_(?<statementId>\\d+)|_v(?<visibilityTxnId>\\d+))?" +
      // Optional path suffix.
      "(?:/.*)?";

  private static final Pattern DELTA_PATTERN = Pattern.compile(DELTA_STR);

  // Regex pattern for files in delete delta directories. The pattern is similar to
  // the 'DELTA_PATTERN', but starts with "delete_".
  private static final Pattern DELETE_DELTA_PATTERN = Pattern.compile(
    "delete_" + DELTA_STR);

  @VisibleForTesting
  static final long SENTINEL_BASE_WRITE_ID = Long.MIN_VALUE;

  // The code is same as what exists in AcidUtils.java in hive-exec.
  // Ideally we should move the AcidUtils code from hive-exec into
  // hive-standalone-metastore or some other jar and use it here.
  private static boolean isInsertOnlyTable(Map<String, String> props) {
    Preconditions.checkNotNull(props);
    if (!isTransactionalTable(props)) {
      return false;
    }
    String transactionalProp = props.get(TABLE_TRANSACTIONAL_PROPERTIES);
    return transactionalProp != null && INSERTONLY_TRANSACTIONAL_PROPERTY.
        equalsIgnoreCase(transactionalProp);
  }

  public static boolean isTransactionalTable(Map<String, String> props) {
    Preconditions.checkNotNull(props);
    String tableIsTransactional = props.get(TABLE_IS_TRANSACTIONAL);
    if (tableIsTransactional == null) {
      tableIsTransactional = props.get(TABLE_IS_TRANSACTIONAL.toUpperCase());
    }
    return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true");
  }

  public static boolean isFullAcidTable(Map<String, String> props) {
    return isTransactionalTable(props) && !isInsertOnlyTable(props);
  }

  public static Column getRowIdColumnType(int position) {
    StructType row__id = new StructType();
    row__id.addField(new StructField("operation", ScalarType.INT, ""));
    row__id.addField(new StructField("originaltransaction", ScalarType.BIGINT, ""));
    row__id.addField(new StructField("bucket", ScalarType.INT, ""));
    row__id.addField(new StructField("rowid", ScalarType.BIGINT, ""));
    row__id.addField(new StructField("currenttransaction", ScalarType.BIGINT, ""));
    return new Column("row__id", row__id, "", position);
  }

  // Sets transaction related table properties for new tables based on manually
  // set table properties and default transactional type.
  public static void setTransactionalProperties(Map<String, String> props,
      TTransactionalType defaultTransactionalType) {
    Preconditions.checkNotNull(props);
    if (props.get(TABLE_IS_TRANSACTIONAL) != null
        || props.get(TABLE_TRANSACTIONAL_PROPERTIES) != null) {
      // Table properties are set manually, ignore default.
      return;
    }

    switch (defaultTransactionalType) {
      case NONE: break;
      case INSERT_ONLY:
        props.put(TABLE_IS_TRANSACTIONAL, "true");
        props.put(TABLE_TRANSACTIONAL_PROPERTIES, INSERTONLY_TRANSACTIONAL_PROPERTY);
        break;
    }
  }

  /**
   * This method is copied from Hive's org.apache.hadoop.hive.ql.io.AcidUtils.java
   * (commit hash 17ac1d9f230b8d663c09c22016753012a9b91edf). It is used to generate
   * the ACID directory names to be added to the insert events on the transactional
   * tables.
   */
  //Get the first level acid directory (if any) from a given path
  public static String getFirstLevelAcidDirPath(Path dataPath, FileSystem fileSystem)
      throws IOException {
    if (dataPath == null) {
      return null;
    }
    String firstLevelAcidDir = getAcidSubDir(dataPath);
    if (firstLevelAcidDir != null) {
      return firstLevelAcidDir;
    }

    String acidDirPath = getFirstLevelAcidDirPath(dataPath.getParent(), fileSystem);
    if (acidDirPath == null) {
      return null;
    }

    // We need the path for directory so no need to append file name
    if (fileSystem.isDirectory(dataPath)) {
      return acidDirPath + Path.SEPARATOR + dataPath.getName();
    }
    return acidDirPath;
  }

  private static String getAcidSubDir(Path dataPath) {
    String dataDir = dataPath.getName();
    if (dataDir.startsWith("base_")
        || dataDir.startsWith("delta_")
        || dataDir.startsWith("delete_delta_")) {
      return dataDir;
    }
    return null;
  }

  /**
   * Predicate that checks if the file or directory is relevant for a given WriteId list.
   * The class does not implement a Predicate interface since we want to support a strict
   * mode which throw exception in certain cases.
   * <p>
   *  <b>Must be called only for ACID table.</b>
   *  Checks that the path conforms to ACID table dir structure, and includes only
   *  directories that correspond to valid committed transactions.
   * </p>
   */
  private static class WriteListBasedPredicate {

    @Nullable
    private final ValidTxnList validTxnList;
    private final ValidWriteIdList writeIdList;
    // when strict mode is turned on, it throws exceptions when a given base file
    // is invalid or a compacted delta file has some open writeIds.
    private final boolean doStrictCheck;

    /**
     * Creates a Predicate just based on WriteIdList. This is used to filter out
     * already cached filedescriptors where it is guaranteed that files related to
     * invalid transactions are not loaded.
     * @param writeIdList
     */
    WriteListBasedPredicate(ValidWriteIdList writeIdList, boolean strictMode) {
      this.validTxnList = null;
      this.writeIdList = Preconditions.checkNotNull(writeIdList);
      this.doStrictCheck = strictMode;
    }

    /**
     * Creates a Predicate based on a ValidTxnList and ValidWriteIdList. Useful when we
     * are filtering out the file listing directly from fileSystem which may included
     * compacted directories.
     *
     * @param validTxnList
     * @param writeIdList
     */
    WriteListBasedPredicate(ValidTxnList validTxnList, ValidWriteIdList writeIdList) {
      this.validTxnList = Preconditions.checkNotNull(validTxnList);
      this.writeIdList = Preconditions.checkNotNull(writeIdList);
      this.doStrictCheck = false;
    }

    public boolean check(String dirPath) throws CatalogException {
      ParsedBase parsedBase = parseBase(dirPath);
      if (parsedBase.writeId != SENTINEL_BASE_WRITE_ID) {
        boolean isValid = writeIdList.isValidBase(parsedBase.writeId) &&
               isTxnValid(parsedBase.visibilityTxnId);
        if (doStrictCheck && !isValid) {
          throw new CatalogException("Invalid base file found " + dirPath);
        }
        return isValid;
      } else {
        ParsedDelta pd = parseDelta(dirPath);
        if (pd == null) pd = parseDeleteDelta(dirPath);
        if (pd != null) {
          if (!isTxnValid(pd.visibilityTxnId)) return false;
          ValidWriteIdList.RangeResponse rr =
              writeIdList.isWriteIdRangeValid(pd.minWriteId, pd.maxWriteId);
          if (rr == RangeResponse.ALL) return true;
          if (rr == RangeResponse.NONE) return false;
          // either this is compacted delta file whose visibility transaction id is
          // valid or a delta file generated by Hive Streaming engine.
          // We allow the delta files for streaming engine which have open writeIds since
          // backend code handles such writeIds appropriately.
          if (!pd.isCompactedDeltaFile()) return true;
          // This is a compacted delta file and has some writeIds which are not valid.
          // We allow only aborted and committed writeIds in compacted files (no open
          // writeIds)
          for (long writeId = pd.minWriteId; writeId<=pd.maxWriteId; writeId++) {
            if (!writeIdList.isWriteIdValid(writeId) && !writeIdList
                .isWriteIdAborted(writeId)) {
              if (doStrictCheck) {
                throw new CatalogException(
                    "Open writeId " + writeId + " found in compacted delta file "
                        + dirPath);
              }
              return false;
            }
          }
          return true;
        }
      }
      // If it wasn't in a base or a delta directory, we should include it.
      // This allows post-upgrade tables to be read.
      // TODO(todd) add an e2e test for this.
      return true;
    }

    /**
     * The ACID compactor process does not change the writeIds of the compacted files (eg.
     * delta_0001 and delta_0002 will be compacted to delta_0001_0002_v0123 where v0123 is
     * the visibility txn id for the compaction itself). While this compaction is in
     * progress, we need to make sure that we ignore all such files. This is where
     * the TxnList is useful. We use the validTxnList to determine if the compaction
     * process is committed or not. If its not, we ignore the files which
     * are being compacted.
     *
     * @param visibilityTxnId TransactionID derived from the directory name on the
     *                        filesystem.
     * @return true if the given visibilityTxnId is valid. In case either the
     * visibilityTxnId is -1 or validTxnList is null, we return true. False otherwise.
     */
    private boolean isTxnValid(long visibilityTxnId) {
      return validTxnList == null ||
        visibilityTxnId == -1 || validTxnList.isTxnValid(visibilityTxnId);
    }
  }

  @Immutable
  private static final class ParsedBase {
    final long writeId;
    final long visibilityTxnId;

    ParsedBase(long writeId, long visibilityTxnId) {
      this.writeId = writeId;
      this.visibilityTxnId = visibilityTxnId;
    }
  }

  @VisibleForTesting
  static ParsedBase parseBase(String relPath) {
    Matcher baseMatcher = BASE_PATTERN.matcher(relPath);
    if (baseMatcher.matches()) {
      long writeId = Long.valueOf(baseMatcher.group("writeId"));
      long visibilityTxnId = -1;
      String visibilityTxnIdStr = baseMatcher.group("visibilityTxnId");
      if (visibilityTxnIdStr != null) {
        visibilityTxnId = Long.valueOf(visibilityTxnIdStr);
      }
      return new ParsedBase(writeId, visibilityTxnId);
    }
    return new ParsedBase(SENTINEL_BASE_WRITE_ID, -1);
  }

  @VisibleForTesting
  static long getBaseWriteId(String relPath) {
    return parseBase(relPath).writeId;
  }

  @Immutable
  private static final class ParsedDelta {
    final long minWriteId;
    final long maxWriteId;
    /// Value -1 means there is no statement id.
    final long statementId;
    /// Value -1 means there is no visibility txn id.
    final long visibilityTxnId;

    ParsedDelta(long minWriteId, long maxWriteId, long statementId,
        long visibilityTxnId) {
      this.minWriteId = minWriteId;
      this.maxWriteId = maxWriteId;
      this.statementId = statementId;
      this.visibilityTxnId = visibilityTxnId;
    }

    private boolean isCompactedDeltaFile() {
      return visibilityTxnId != -1;
    }
  }

  private static ParsedDelta matcherToParsedDelta(Matcher deltaMatcher) {
    if (!deltaMatcher.matches()) {
      return null;
    }
    long minWriteId = Long.valueOf(deltaMatcher.group("minWriteId"));
    long maxWriteId = Long.valueOf(deltaMatcher.group("maxWriteId"));
    String statementIdStr = deltaMatcher.group("statementId");
    long statementId = statementIdStr != null ? Long.valueOf(statementIdStr) : -1;
    String visibilityTxnIdStr = deltaMatcher.group("visibilityTxnId");
    long visibilityTxnId = visibilityTxnIdStr != null ?
        Long.valueOf(visibilityTxnIdStr) : -1;
    return new ParsedDelta(minWriteId, maxWriteId, statementId, visibilityTxnId);
  }

  private static ParsedDelta parseDelta(String dirPath) {
    return matcherToParsedDelta(DELTA_PATTERN.matcher(dirPath));
  }

  private static ParsedDelta parseDeleteDelta(String dirPath) {
    return matcherToParsedDelta(DELETE_DELTA_PATTERN.matcher(dirPath));
  }

  private static String getFirstDirName(String relPath) {
    int slashIdx = relPath.indexOf("/");
    if (slashIdx != -1) {
      return relPath.substring(0, slashIdx);
    } else {
      return null;
    }
  }

  /**
   * Returns true if 'fd' refers to a delete delta file.
   */
  public static boolean isDeleteDeltaFd(FileDescriptor fd) {
    return fd.getRelativePath().startsWith("delete_delta_");
  }

  /**
   * This method is similar to {@link AcidUtils#filterFilesForAcidState} with the
   * difference that it expects input to be valid file descriptors from a loaded table.
   * This means that file descriptors are already pre-vetted and are consistent with
   * respect to some ValidWriteIdList and ValidTxnList. All this method does is to try to
   * filter such file descriptors for a different ValidWriteIdList.
   *
   * @param fds Input list of File descriptors to be filtered in-place.
   * @param validWriteIdList The ValidWriteIdList for which we filter the fds.
   * @return The number of file descriptors which were filtered out.
   * @throws CatalogException if any of the provided FileDescriptor could be included or
   * excluded since it contains some writeIds which are invalid for the given
   * ValidWriteIdList.
   */
  public static int filterFdsForAcidState(List<FileDescriptor> fds,
      ValidWriteIdList validWriteIdList) throws CatalogException {
    Preconditions.checkNotNull(fds);

    if (validWriteIdList == null) return 0;

    WriteListBasedPredicate writeListBasedPredicate = new WriteListBasedPredicate(
        validWriteIdList, true);
    Iterator<FileDescriptor> it = fds.iterator();
    int numRemoved = 0;
    while (it.hasNext()) {
      if (!writeListBasedPredicate.check(it.next().getRelativePath())) {
        it.remove();
        numRemoved++;
      }
    }
    return numRemoved;
  }

  /**
   * Filters the files based on Acid state.
   * @param stats the FileStatuses obtained from recursively listing the directory
   * @param baseDir the base directory for the partition (or table, in the case of
   *   unpartitioned tables)
   * @param writeIds the valid write IDs for the table
   * @param loadStats stats to add counts of skipped files to. May be null.
   * @return the FileStatuses that is a subset of passed in descriptors that
   *    must be used.
   * @throws CatalogException on ACID error. TODO: Remove throws clause once IMPALA-9042
   * is resolved.
   */
  public static List<FileStatus> filterFilesForAcidState(List<FileStatus> stats,
      Path baseDir, ValidTxnList validTxnList, ValidWriteIdList writeIds,
      @Nullable LoadStats loadStats) throws CatalogException {
    // First filter out any paths that are not considered valid write IDs.
    // At the same time, calculate the max valid base write ID and collect the names of
    // the delta directories.
    WriteListBasedPredicate pred = new WriteListBasedPredicate(validTxnList, writeIds);
    long maxBaseWriteId = Long.MIN_VALUE;
    Set<String> deltaDirNames = new HashSet<>();
    for (Iterator<FileStatus> it = stats.iterator(); it.hasNext();) {
      FileStatus stat = it.next();
      String relPath = FileSystemUtil.relativizePath(stat.getPath(), baseDir);
      if (!pred.check(relPath)) {
        it.remove();
        if (loadStats != null) loadStats.uncommittedAcidFilesSkipped++;
        continue;
      }
      maxBaseWriteId = Math.max(getBaseWriteId(relPath), maxBaseWriteId);
      String dirName = getFirstDirName(relPath);
      if (dirName != null && (dirName.startsWith("delta_") ||
          dirName.startsWith("delete_delta_"))) {
        deltaDirNames.add(dirName);
      }
    }
    // Get a list of all valid delta directories.
    List<Pair<String, ParsedDelta>> deltas =
        getValidDeltaDirsOrdered(deltaDirNames, maxBaseWriteId);
    // Filter out delta directories superceded by major/minor compactions.
    Set<String> filteredDeltaDirs =
        getFilteredDeltaDirs(deltas, maxBaseWriteId, writeIds);
    // Filter out any files that are superceded by the latest valid base or not located
    // in 'filteredDeltaDirs'.
    return filterFilesForAcidState(stats, baseDir, maxBaseWriteId, filteredDeltaDirs,
        loadStats);
  }

  private static List<FileStatus> filterFilesForAcidState(List<FileStatus> stats,
      Path baseDir, long maxBaseWriteId, Set<String> deltaDirs,
      @Nullable LoadStats loadStats) throws CatalogException {
    List<FileStatus> validStats = new ArrayList<>(stats);
    for (Iterator<FileStatus> it = validStats.iterator(); it.hasNext();) {
      FileStatus stat = it.next();

      if (stat.isDirectory()) {
        it.remove();
        continue;
      }

      String relPath = FileSystemUtil.relativizePath(stat.getPath(), baseDir);
      if (relPath.startsWith("delta_") ||
          relPath.startsWith("delete_delta_")) {
        String dirName = getFirstDirName(relPath);
        if (dirName != null && !deltaDirs.contains(dirName)) {
          it.remove();
          if (loadStats != null) loadStats.filesSupersededByAcidState++;
        }
        if (relPath.endsWith("_flush_length")) {
          throw new CatalogException("Found Hive Streaming side-file: " +
              stat.getPath() + " It means that the contents of the directory are " +
              "currently being written, therefore Impala is not able to read it. " +
              "Please try to load the table again once Hive Streaming commits " +
              "the transaction.");
        }
        continue;
      }
      long baseWriteId = getBaseWriteId(relPath);
      if (baseWriteId != SENTINEL_BASE_WRITE_ID) {
        if (baseWriteId < maxBaseWriteId) {
          it.remove();
          if (loadStats != null) loadStats.filesSupersededByAcidState++;
        }
        continue;
      }

      // Not in a base or a delta directory. In that case, it's probably a
      // post-upgrade file.
      // If there is no valid base: we should read the file (assuming that
      // hive.mm.allow.originals == true)
      // If there is a valid base: the file should be merged to the base by the
      // compaction, so we can assume that the file is no longer valid and just
      // waits to be deleted.
      if (maxBaseWriteId != SENTINEL_BASE_WRITE_ID) it.remove();
    }
    return validStats;
  }

  private static List<Pair<String, ParsedDelta>> getValidDeltaDirsOrdered(
      Set<String> deltaDirNames, long baseWriteId)
      throws CatalogException {
    List <Pair<String, ParsedDelta>> deltas = new ArrayList<>();
    for (Iterator<String> it = deltaDirNames.iterator(); it.hasNext();) {
      String dirname = it.next();
      ParsedDelta parsedDelta = parseDelta(dirname);
      if (parsedDelta == null) parsedDelta = parseDeleteDelta(dirname);
      if (parsedDelta != null) {
        if (parsedDelta.minWriteId <= baseWriteId) {
          Preconditions.checkState(parsedDelta.maxWriteId <= baseWriteId);
          it.remove();
          continue;
        }
        deltas.add(new Pair<String, ParsedDelta>(dirname, parsedDelta));
        continue;
      }
    }

    deltas.sort(new Comparator<Pair<String, ParsedDelta>>() {
      // This compare method is based on Hive (d6ad73c3615)
      // AcidUtils.ParsedDeltaLight.compareTo()
      // One additon to it is to take the visbilityTxnId into consideration. Hence if
      // there's delta_N_M and delta_N_M_v001234 then delta_N_M_v001234 must be ordered
      // before.
      @Override
      public int compare(Pair<String, ParsedDelta> o1, Pair<String, ParsedDelta> o2) {
        ParsedDelta pd1 = o1.second;
        ParsedDelta pd2 = o2.second;
        if (pd1.minWriteId != pd2.minWriteId) {
          if (pd1.minWriteId < pd2.minWriteId) {
            return -1;
          } else {
            return 1;
          }
        } else if (pd1.maxWriteId != pd2.maxWriteId) {
          if (pd1.maxWriteId < pd2.maxWriteId) {
            return 1;
          } else {
            return -1;
          }
        } else if (pd1.statementId != pd2.statementId) {
          /**
           * We want deltas after minor compaction (w/o statementId) to sort earlier so
           * that getAcidState() considers compacted files (into larger ones) obsolete
           * Before compaction, include deltas with all statementIds for a given writeId.
           */
          if (pd1.statementId < pd2.statementId) {
            return -1;
          } else {
            return 1;
          }
        } else if (pd1.visibilityTxnId != pd2.visibilityTxnId) {
          // This is an alteration from Hive's algorithm. If everything is the same then
          // the higher visibilityTxnId wins (since no visibiltyTxnId is -1).
          // Currently this cannot happen since Hive doesn't minor compact standalone
          // delta directories of streaming ingestion, i.e. the following cannot happen:
          // delta_1_5 => delta_1_5_v01234
          // However, it'd make sense because streaming ingested ORC files doesn't use
          // advanced features like dictionary encoding or statistics. Hence Hive might
          // do that in the future and that'd make Impala seeing duplicate rows.
          // So I'd be cautious here in case they forget to tell us.
          if (pd1.visibilityTxnId < pd2.visibilityTxnId) {
            return 1;
          } else {
            return -1;
          }
        } else {
          return o1.first.compareTo(o2.first);
        }
      }
    });
    return deltas;
  }

  /**
   * The algorithm is copied from Hive's (d6ad73c3615)
   * org.apache.hadoop.hive.ql.io.AcidUtils.getAcidState()
   * One additon to it is to take the visbilityTxnId into consideration. Hence if
   * there's delta_N_M_v001234 and delta_N_M then it ignores delta_N_M.
   */
  private static Set<String> getFilteredDeltaDirs(List<Pair<String, ParsedDelta>> deltas,
      long baseWriteId, ValidWriteIdList writeIds) {
    long current = baseWriteId;
    long lastStmtId = -1;
    ParsedDelta prev = null;
    Set<String> filteredDeltaDirs = new HashSet<>();
    for (Pair<String, ParsedDelta> pathDelta : deltas) {
      ParsedDelta next = pathDelta.second;
      if (next.maxWriteId > current) {
        // are any of the new transactions ones that we care about?
        if (writeIds.isWriteIdRangeValid(current + 1, next.maxWriteId) !=
            ValidWriteIdList.RangeResponse.NONE) {
          filteredDeltaDirs.add(pathDelta.first);
          current = next.maxWriteId;
          lastStmtId = next.statementId;
          prev = next;
        }
      } else if (next.maxWriteId == current && lastStmtId >= 0) {
        // make sure to get all deltas within a single transaction; multi-statement txn
        // generate multiple delta files with the same txnId range
        // of course, if maxWriteId has already been minor compacted, all per statement
        // deltas are obsolete
        filteredDeltaDirs.add(pathDelta.first);
        prev = next;
      } else if (prev != null && next.maxWriteId == prev.maxWriteId &&
          next.minWriteId == prev.minWriteId &&
          next.statementId == prev.statementId &&
          // If visibilityTxnId differs, then 'pathDelta' is probably a streaming ingested
          // delta directory and 'prev' is the compacted version of it.
          next.visibilityTxnId == prev.visibilityTxnId) {
        // The 'next' parsedDelta may have everything equal to the 'prev' parsedDelta,
        // except
        // the path. This may happen when we have split update and we have two types of
        // delta
        // directories- 'delta_x_y' and 'delete_delta_x_y' for the SAME txn range.

        // Also note that any delete_deltas in between a given delta_x_y range would be
        // made
        // obsolete. For example, a delta_30_50 would make delete_delta_40_40 obsolete.
        // This is valid because minor compaction always compacts the normal deltas and
        // the delete deltas for the same range. That is, if we had 3 directories,
        // delta_30_30, delete_delta_40_40 and delta_50_50, then running minor compaction
        // would produce delta_30_50 and delete_delta_30_50.
        filteredDeltaDirs.add(pathDelta.first);
        prev = next;
      }
    }
    return filteredDeltaDirs;
  }

  /**
   * This method compares the writeIdList of the given table if it is loaded and is a
   * transactional table with the given ValidWriteIdList. If the tbl metadata is a
   * superset of the metadata view represented by the given validWriteIdList this
   * method returns a value greater than 0. If they are an exact match of each other,
   * it returns 0 and if the table ValidWriteIdList is behind the provided
   * validWriteIdList this return -1. This information useful to determine if the
   * cached table can be used to construct a consistent snapshot corresponding to the
   * given validWriteIdList. The ValidWriteIdList is compared only if the table id
   * matches with the given tableId.
   */
  public static int compare(HdfsTable tbl, ValidWriteIdList validWriteIdList,
      long tableId) {
    Preconditions.checkState(tbl != null && tbl.getMetaStoreTable() != null);
    // if tbl is not a transactional, there is nothing to compare against and we return 0
    if (!isTransactionalTable(tbl.getMetaStoreTable().getParameters())) return 0;
    Preconditions.checkNotNull(tbl.getValidWriteIds());
    // if the provided table id does not match with what CatalogService has we return
    // -1 indicating that cached table is stale.
    if (tableId != CatalogServiceCatalog.TABLE_ID_UNAVAILABLE
        && tbl.getMetaStoreTable().getId() != tableId) {
      return -1;
    }
    return compare(tbl.getValidWriteIds(), validWriteIdList);
  }

  /*** This method is mostly copied from {@link org.apache.hive.common.util.TxnIdUtils}
   * (e649562) with the exception that the table names for both the input writeIdList
   * must be the same to have a valid comparison.
   * //TODO source this directly from hive-exec so that future changes to this are
   * automatically imported.
   *
   * @param a
   * @param b
   * @return 0, if a and b are equivalent
   * 1, if a is more recent
   * -1, if b is more recent
   ***/
  @VisibleForTesting
  public static int compare(ValidWriteIdList a, ValidWriteIdList b) {
    Preconditions.checkState(a.getTableName().equalsIgnoreCase(b.getTableName()));
    // The algorithm assumes invalidWriteIds are sorted and values are less or equal than
    // hwm, here is how the algorithm works:
    // 1. Compare two invalidWriteIds until one the list ends, difference means the
    // mismatch writeid is committed in one ValidWriteIdList but not the other, the
    // comparison end
    // 2. Every writeid from the last writeid in the short invalidWriteIds till its
    // hwm should be committed in the other ValidWriteIdList, otherwise the comparison
    // end
    // 3. Every writeid from lower hwm to higher hwm should be invalid, otherwise, the
    // comparison end
    int minLen = Math.min(a.getInvalidWriteIds().length, b.getInvalidWriteIds().length);
    for (int i = 0; i < minLen; i++) {
      if (a.getInvalidWriteIds()[i] == b.getInvalidWriteIds()[i]) {
        continue;
      }
      return a.getInvalidWriteIds()[i] > b.getInvalidWriteIds()[i] ? 1 : -1;
    }
    if (a.getInvalidWriteIds().length == b.getInvalidWriteIds().length) {
      return Long.signum(a.getHighWatermark() - b.getHighWatermark());
    }
    if (a.getInvalidWriteIds().length == minLen) {
      if (a.getHighWatermark() != b.getInvalidWriteIds()[minLen] - 1) {
        return Long.signum(a.getHighWatermark() - (b.getInvalidWriteIds()[minLen] - 1));
      }
      if (allInvalidFrom(b.getInvalidWriteIds(), minLen, b.getHighWatermark())) {
        return 0;
      } else {
        return -1;
      }
    } else {
      if (b.getHighWatermark() != a.getInvalidWriteIds()[minLen] - 1) {
        return Long.signum(b.getHighWatermark() - (a.getInvalidWriteIds()[minLen] - 1));
      }
      if (allInvalidFrom(a.getInvalidWriteIds(), minLen, a.getHighWatermark())) {
        return 0;
      } else {
        return 1;
      }
    }
  }
  private static boolean allInvalidFrom(long[] invalidIds, int start, long hwm) {
    for (int i=start+1;i<invalidIds.length;i++) {
      if (invalidIds[i] != (invalidIds[i-1]+1)) {
        return false;
      }
    }
    return invalidIds[invalidIds.length-1] == hwm;
  }
}