/*
 * 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.iceberg.mr.hive;

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.commons.collections4.ListUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.StringUtils;
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.common.type.Date;
import org.apache.hadoop.hive.common.type.SnapshotContext;
import org.apache.hadoop.hive.common.type.Timestamp;
import org.apache.hadoop.hive.conf.Constants;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.InvalidObjectException;
import org.apache.hadoop.hive.metastore.api.LockType;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.metastore.utils.MetaStoreServerUtils;
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.Context.Operation;
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.QueryState;
import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
import org.apache.hadoop.hive.ql.ddl.table.AbstractAlterTableDesc;
import org.apache.hadoop.hive.ql.ddl.table.AlterTableType;
import org.apache.hadoop.hive.ql.ddl.table.create.like.CreateTableLikeDesc;
import org.apache.hadoop.hive.ql.ddl.table.misc.properties.AlterTableSetPropertiesDesc;
import org.apache.hadoop.hive.ql.exec.FetchOperator;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.io.IOConstants;
import org.apache.hadoop.hive.ql.io.StorageFormatDescriptor;
import org.apache.hadoop.hive.ql.io.parquet.vector.VectorizedParquetRecordReader;
import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.HiveStoragePredicateHandler;
import org.apache.hadoop.hive.ql.metadata.HiveUtils;
import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.parse.AlterTableExecuteSpec;
import org.apache.hadoop.hive.ql.parse.AlterTableSnapshotRefSpec;
import org.apache.hadoop.hive.ql.parse.PartitionTransform;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.parse.StorageFormat;
import org.apache.hadoop.hive.ql.parse.StorageFormat.StorageHandlerTypes;
import org.apache.hadoop.hive.ql.parse.TransformSpec;
import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDynamicListDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.HiveOperation;
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionStateUtil;
import org.apache.hadoop.hive.ql.stats.Partish;
import org.apache.hadoop.hive.serde2.AbstractSerDe;
import org.apache.hadoop.hive.serde2.DefaultFetchFormatter;
import org.apache.hadoop.hive.serde2.Deserializer;
import org.apache.hadoop.hive.serde2.FetchFormatter;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobContext;
import org.apache.hadoop.mapred.JobContextImpl;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.iceberg.BaseMetastoreTableOperations;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.ExpireSnapshots;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.NullOrder;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.PartitionSpecParser;
import org.apache.iceberg.RowLevelOperationMode;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SortDirection;
import org.apache.iceberg.SortField;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.hadoop.HadoopConfigurable;
import org.apache.iceberg.mr.Catalogs;
import org.apache.iceberg.mr.InputFormatConfig;
import org.apache.iceberg.puffin.Blob;
import org.apache.iceberg.puffin.BlobMetadata;
import org.apache.iceberg.puffin.Puffin;
import org.apache.iceberg.puffin.PuffinCompressionCodec;
import org.apache.iceberg.puffin.PuffinReader;
import org.apache.iceberg.puffin.PuffinWriter;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.base.Splitter;
import org.apache.iceberg.relocated.com.google.common.base.Throwables;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.relocated.com.google.common.collect.Streams;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.ByteBuffers;
import org.apache.iceberg.util.Pair;
import org.apache.iceberg.util.SerializationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, HiveStorageHandler {
  private static final Logger LOG = LoggerFactory.getLogger(HiveIcebergStorageHandler.class);

  private static final String ICEBERG_URI_PREFIX = "iceberg://";
  private static final Splitter TABLE_NAME_SPLITTER = Splitter.on("..");
  private static final String TABLE_NAME_SEPARATOR = "..";
  // Column index for partition metadata table
  private static final int SPEC_IDX = 1;
  private static final int PART_IDX = 0;
  public static final String COPY_ON_WRITE = "copy-on-write";
  public static final String MERGE_ON_READ = "merge-on-read";
  public static final String STATS = "/stats/";

  /**
   * Function template for producing a custom sort expression function:
   * Takes the source column index and the bucket count to creat a function where Iceberg bucket UDF is used to build
   * the sort expression, e.g. iceberg_bucket(_col2, 5)
   */
  private static final transient BiFunction<Integer, Integer, Function<List<ExprNodeDesc>, ExprNodeDesc>>
      BUCKET_SORT_EXPR =
          (idx, bucket) -> cols -> {
            try {
              ExprNodeDesc icebergBucketSourceCol = cols.get(idx);
              return ExprNodeGenericFuncDesc.newInstance(new GenericUDFIcebergBucket(), "iceberg_bucket",
                  Lists.newArrayList(
                      icebergBucketSourceCol,
                      new ExprNodeConstantDesc(TypeInfoFactory.intTypeInfo, bucket)
                  ));
            } catch (UDFArgumentException e) {
              throw new RuntimeException(e);
            }
          };

  private static final List<VirtualColumn> ACID_VIRTUAL_COLS = ImmutableList.of(VirtualColumn.PARTITION_SPEC_ID,
      VirtualColumn.PARTITION_HASH, VirtualColumn.FILE_PATH, VirtualColumn.ROW_POSITION);
  private static final List<FieldSchema> ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA = ACID_VIRTUAL_COLS.stream()
      .map(v -> new FieldSchema(v.getName(), v.getTypeInfo().getTypeName(), ""))
      .collect(Collectors.toList());


  private Configuration conf;

  @Override
  public Class<? extends InputFormat> getInputFormatClass() {
    return HiveIcebergInputFormat.class;
  }

  @Override
  public Class<? extends OutputFormat> getOutputFormatClass() {
    return HiveIcebergOutputFormat.class;
  }

  @Override
  public Class<? extends AbstractSerDe> getSerDeClass() {
    return HiveIcebergSerDe.class;
  }

  @Override
  public HiveMetaHook getMetaHook() {
    // Make sure to always return a new instance here, as HiveIcebergMetaHook might hold state relevant for the
    // operation.
    return new HiveIcebergMetaHook(conf);
  }

  @Override
  public HiveAuthorizationProvider getAuthorizationProvider() {
    return null;
  }

  @Override
  public void configureInputJobProperties(TableDesc tableDesc, Map<String, String> map) {
    overlayTableProperties(conf, tableDesc, map);
    // Until the vectorized reader can handle delete files, let's fall back to non-vector mode for V2 tables
    fallbackToNonVectorizedModeBasedOnProperties(tableDesc.getProperties());
  }

  @Override
  public void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> map) {
    overlayTableProperties(conf, tableDesc, map);
    // Until the vectorized reader can handle delete files, let's fall back to non-vector mode for V2 tables
    fallbackToNonVectorizedModeBasedOnProperties(tableDesc.getProperties());
    // For Tez, setting the committer here is enough to make sure it'll be part of the jobConf
    map.put("mapred.output.committer.class", HiveIcebergNoJobCommitter.class.getName());
    // For MR, the jobConf is set only in configureJobConf, so we're setting the write key here to detect it over there
    String opType = getOperationType();
    map.put(InputFormatConfig.OPERATION_TYPE_PREFIX + tableDesc.getTableName(), opType);
    // Putting the key into the table props as well, so that projection pushdown can be determined on a
    // table-level and skipped only for output tables in HiveIcebergSerde. Properties from the map will be present in
    // the serde config for all tables in the query, not just the output tables, so we can't rely on that in the serde.
    tableDesc.getProperties().put(InputFormatConfig.OPERATION_TYPE_PREFIX + tableDesc.getTableName(), opType);
  }

  /**
   * Committer with no-op job commit. We can pass this into the Tez AM to take care of task commits/aborts, as well
   * as aborting jobs reliably if an execution error occurred. However, we want to execute job commits on the
   * HS2-side during the MoveTask, so we will use the full-featured HiveIcebergOutputCommitter there.
   */
  static class HiveIcebergNoJobCommitter extends HiveIcebergOutputCommitter {
    @Override
    public void commitJob(JobContext originalContext) throws IOException {
      // do nothing
    }
  }

  @Override
  public void configureTableJobProperties(TableDesc tableDesc, Map<String, String> map) {

  }

  // Override annotation commented out, since this interface method has been introduced only in Hive 3
  // @Override
  public void configureInputJobCredentials(TableDesc tableDesc, Map<String, String> secrets) {

  }

  @Override
  public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
    setCommonJobConf(jobConf);
    if (tableDesc != null && tableDesc.getProperties() != null &&
        tableDesc.getProperties().get(InputFormatConfig.OPERATION_TYPE_PREFIX + tableDesc.getTableName()) != null) {
      String tableName = tableDesc.getTableName();
      String opKey = InputFormatConfig.OPERATION_TYPE_PREFIX + tableName;
      // set operation type into job conf too
      jobConf.set(opKey, tableDesc.getProperties().getProperty(opKey));
      Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
          "Can not handle table " + tableName + ". Its name contains '" + TABLE_NAME_SEPARATOR + "'");
      String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
      tables = tables == null ? tableName : tables + TABLE_NAME_SEPARATOR + tableName;
      jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);

      String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
      if (catalogName != null) {
        jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName, catalogName);
      }
    }
    try {
      if (!jobConf.getBoolean(HiveConf.ConfVars.HIVE_IN_TEST_IDE.varname, false)) {
        // For running unit test this won't work as maven surefire CP is different than what we have on a cluster:
        // it places the current projects' classes and test-classes to top instead of jars made from these...
        Utilities.addDependencyJars(jobConf, HiveIcebergStorageHandler.class);
      }
    } catch (IOException e) {
      Throwables.propagate(e);
    }
  }

  @Override
  public boolean directInsert() {
    return true;
  }

  @Override
  public boolean alwaysUnpartitioned() {
    return true;
  }

  @Override
  public Configuration getConf() {
    return conf;
  }

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

  @Override
  public String toString() {
    return this.getClass().getName();
  }

  /**
   * @param jobConf Job configuration for InputFormat to access
   * @param deserializer Deserializer
   * @param exprNodeDesc Filter expression extracted by Hive
   * @return Entire filter to take advantage of Hive's pruning as well as Iceberg's pruning.
   */
  @Override
  public DecomposedPredicate decomposePredicate(JobConf jobConf, Deserializer deserializer, ExprNodeDesc exprNodeDesc) {
    DecomposedPredicate predicate = new DecomposedPredicate();
    predicate.residualPredicate = (ExprNodeGenericFuncDesc) exprNodeDesc;
    predicate.pushedPredicate = (ExprNodeGenericFuncDesc) exprNodeDesc;
    return predicate;
  }

  @Override
  public boolean canProvideBasicStatistics() {
    return true;
  }

  @Override
  public StorageFormatDescriptor getStorageFormatDescriptor(org.apache.hadoop.hive.metastore.api.Table table)
      throws SemanticException {
    if (table.getParameters() != null) {
      String format = table.getParameters().getOrDefault(TableProperties.DEFAULT_FILE_FORMAT, IOConstants.PARQUET);
      return StorageFormat.getDescriptor(format, TableProperties.DEFAULT_FILE_FORMAT);
    }
    return null;
  }

  public boolean supportsAppendData(org.apache.hadoop.hive.metastore.api.Table table, boolean withPartClause)
      throws SemanticException {
    Table icebergTbl = IcebergTableUtil.getTable(conf, table);
    if (icebergTbl.spec().isUnpartitioned()) {
      return true;
    }
    // If it is a table which has undergone partition evolution, return false;
    if (icebergTbl.currentSnapshot() != null) {
      if (icebergTbl.currentSnapshot().allManifests(icebergTbl.io()).parallelStream()
          .map(ManifestFile::partitionSpecId)
          .anyMatch(id -> id < icebergTbl.spec().specId())) {
        if (withPartClause) {
          throw new SemanticException("Can not Load into an iceberg table, which has undergone partition evolution " +
              "using the PARTITION clause");
        }
        return false;
      }
    }
    return withPartClause;
  }

  public void appendFiles(org.apache.hadoop.hive.metastore.api.Table table, URI fromURI, boolean isOverwrite,
      Map<String, String> partitionSpec)
      throws SemanticException {
    Table icebergTbl = IcebergTableUtil.getTable(conf, table);
    String format = table.getParameters().get(TableProperties.DEFAULT_FILE_FORMAT);
    HiveTableUtil.appendFiles(fromURI, format, icebergTbl, isOverwrite, partitionSpec, conf);
  }

  @Override
  public Map<String, String> getBasicStatistics(Partish partish) {
    org.apache.hadoop.hive.ql.metadata.Table hmsTable = partish.getTable();
    // For write queries where rows got modified, don't fetch from cache as values could have changed.
    Table table = getTable(hmsTable);
    Map<String, String> stats = Maps.newHashMap();
    if (getStatsSource().equals(Constants.ICEBERG)) {
      if (table.currentSnapshot() != null) {
        Map<String, String> summary = table.currentSnapshot().summary();
        if (summary != null) {

          if (summary.containsKey(SnapshotSummary.TOTAL_DATA_FILES_PROP)) {
            stats.put(StatsSetupConst.NUM_FILES, summary.get(SnapshotSummary.TOTAL_DATA_FILES_PROP));
          }

          if (summary.containsKey(SnapshotSummary.TOTAL_RECORDS_PROP)) {
            long totalRecords = Long.parseLong(summary.get(SnapshotSummary.TOTAL_RECORDS_PROP));
            if (summary.containsKey(SnapshotSummary.TOTAL_EQ_DELETES_PROP) &&
                summary.containsKey(SnapshotSummary.TOTAL_POS_DELETES_PROP)) {

              long totalEqDeletes = Long.parseLong(summary.get(SnapshotSummary.TOTAL_EQ_DELETES_PROP));
              long totalPosDeletes = Long.parseLong(summary.get(SnapshotSummary.TOTAL_POS_DELETES_PROP));

              long actualRecords = totalRecords - (totalEqDeletes > 0 ? 0 : totalPosDeletes);
              totalRecords = actualRecords > 0 ? actualRecords : totalRecords;
              // actualRecords maybe -ve in edge cases
            }
            stats.put(StatsSetupConst.ROW_COUNT, String.valueOf(totalRecords));
          }

          if (summary.containsKey(SnapshotSummary.TOTAL_FILE_SIZE_PROP)) {
            stats.put(StatsSetupConst.TOTAL_SIZE, summary.get(SnapshotSummary.TOTAL_FILE_SIZE_PROP));
          }
        }
      } else {
        stats.put(StatsSetupConst.NUM_FILES, "0");
        stats.put(StatsSetupConst.ROW_COUNT, "0");
        stats.put(StatsSetupConst.TOTAL_SIZE, "0");
      }
    }
    return stats;
  }

  private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    Table table;
    final Optional<QueryState> queryState = SessionStateUtil.getQueryState(conf);
    if (!queryState.isPresent() || queryState.get().getNumModifiedRows() > 0) {
      table = IcebergTableUtil.getTable(conf, hmsTable.getTTable(), true);
    } else {
      table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
    }
    return table;
  }

  @Override
  public boolean canSetColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
    return table.currentSnapshot() != null && getStatsSource().equals(Constants.ICEBERG);
  }

  @Override
  public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable, List<ColumnStatistics> colStats) {
    Table tbl = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
    String snapshotId = String.format("%s-STATS-%d", tbl.name(), tbl.currentSnapshot().snapshotId());
    return writeColStats(colStats.get(0), tbl, snapshotId);
  }

  private boolean writeColStats(ColumnStatistics tableColStats, Table tbl, String snapshotId) {
    try {
      boolean rewriteStats = removeColStatsIfExists(tbl);
      if (!rewriteStats) {
        checkAndMergeColStats(tableColStats, tbl);
      }
      // Currently, we are only serializing table level stats.
      byte[] serializeColStats = SerializationUtils.serialize(tableColStats);
      try (PuffinWriter writer = Puffin.write(tbl.io().newOutputFile(getColStatsPath(tbl).toString()))
          .createdBy(Constants.HIVE_ENGINE).build()) {
        writer.add(new Blob(tbl.name() + "-" + snapshotId, ImmutableList.of(1), tbl.currentSnapshot().snapshotId(),
            tbl.currentSnapshot().sequenceNumber(), ByteBuffer.wrap(serializeColStats), PuffinCompressionCodec.NONE,
            ImmutableMap.of()));
        writer.finish();
        return true;
      } catch (IOException e) {
        LOG.warn("Unable to write stats to puffin file", e.getMessage());
        return false;
      }
    } catch (InvalidObjectException | IOException e) {
      LOG.warn("Unable to invalidate or merge stats: ", e.getMessage());
      return false;
    }
  }

  @Override
  public boolean canProvideColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
    return canSetColStatistics(hmsTable) && canProvideColStats(table, table.currentSnapshot().snapshotId());
  }

  private boolean canProvideColStats(Table table, long snapshotId) {
    Path statsPath = getColStatsPath(table, snapshotId);
    try {
      FileSystem fs = statsPath.getFileSystem(conf);
      return  fs.exists(statsPath);
    } catch (IOException e) {
      LOG.warn("Exception when trying to find Iceberg column stats for table:{} , snapshot:{} , " +
          "statsPath: {} , stack trace: {}", table.name(), table.currentSnapshot(), statsPath, e);
    }
    return false;
  }

  @Override
  public List<ColumnStatisticsObj> getColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    Table table = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
    Path statsPath = getColStatsPath(table);
    LOG.info("Using stats from puffin file at: {}", statsPath);
    return readColStats(table, statsPath).getStatsObj();
  }

  private ColumnStatistics readColStats(Table table, Path statsPath) {
    try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath.toString())).build()) {
      List<BlobMetadata> blobMetadata = reader.fileMetadata().blobs();
      Map<BlobMetadata, ColumnStatistics> collect = Streams.stream(reader.readAll(blobMetadata)).collect(
          Collectors.toMap(Pair::first, blobMetadataByteBufferPair -> SerializationUtils.deserialize(
              ByteBuffers.toByteArray(blobMetadataByteBufferPair.second()))));
      return collect.get(blobMetadata.get(0));
    } catch (IOException | IndexOutOfBoundsException e) {
      LOG.warn(" Unable to read iceberg col stats from puffin files: ", e);
      return new ColumnStatistics();
    }
  }

  @Override
  public boolean canComputeQueryUsingStats(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    if (getStatsSource().equals(Constants.ICEBERG)) {
      Table table = getTable(hmsTable);
      if (table.currentSnapshot() != null) {
        Map<String, String> summary = table.currentSnapshot().summary();
        if (summary != null && summary.containsKey(SnapshotSummary.TOTAL_EQ_DELETES_PROP) &&
            summary.containsKey(SnapshotSummary.TOTAL_POS_DELETES_PROP)) {

          long totalEqDeletes = Long.parseLong(summary.get(SnapshotSummary.TOTAL_EQ_DELETES_PROP));
          long totalPosDeletes = Long.parseLong(summary.get(SnapshotSummary.TOTAL_POS_DELETES_PROP));
          return totalEqDeletes + totalPosDeletes == 0;
        }
      }
    }
    return false;
  }

  private String getStatsSource() {
    return HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_ICEBERG_STATS_SOURCE, Constants.ICEBERG).toLowerCase();
  }

  private Path getColStatsPath(Table table) {
    return getColStatsPath(table, table.currentSnapshot().snapshotId());
  }

  private Path getColStatsPath(Table table, long snapshotId) {
    return new Path(table.location() + STATS + table.name() + snapshotId);
  }

  private boolean removeColStatsIfExists(Table tbl) throws IOException {
    Path statsPath = getColStatsPath(tbl);
    FileSystem fs = statsPath.getFileSystem(conf);
    if (fs.exists(statsPath)) {
      // Analyze table and stats updater thread
      return fs.delete(statsPath, true);
    }
    return SessionStateUtil.getQueryState(conf).map(QueryState::getHiveOperation)
      .filter(opType -> HiveOperation.ANALYZE_TABLE == opType)
      .isPresent();
  }

  private void checkAndMergeColStats(ColumnStatistics statsObjNew, Table tbl) throws InvalidObjectException {
    Long previousSnapshotId = tbl.currentSnapshot().parentId();
    if (previousSnapshotId != null && canProvideColStats(tbl, previousSnapshotId)) {
      ColumnStatistics statsObjOld = readColStats(tbl, getColStatsPath(tbl, previousSnapshotId));
      if (statsObjOld != null && statsObjOld.getStatsObjSize() != 0 && !statsObjNew.getStatsObj().isEmpty()) {
        MetaStoreServerUtils.mergeColStats(statsObjNew, statsObjOld);
      }
    }
  }

  /**
   * No need for exclusive locks when writing, since Iceberg tables use optimistic concurrency when writing
   * and only lock the table during the commit operation.
   */
  @Override
  public LockType getLockType(WriteEntity writeEntity) {
    return LockType.SHARED_READ;
  }

  @Override
  public boolean supportsPartitionTransform() {
    return true;
  }

  @Override
  public List<TransformSpec> getPartitionTransformSpec(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    return table.spec().fields().stream().map(f ->
      getTransformSpec(table, f.transform().toString().toUpperCase(), f.sourceId())
    ).collect(Collectors.toList());
  }

  private List<TransformSpec> getSortTransformSpec(Table table) {
    return table.sortOrder().fields().stream().map(s ->
      getTransformSpec(table, s.transform().toString().toUpperCase(), s.sourceId())
    ).collect(Collectors.toList());
  }

  private TransformSpec getTransformSpec(Table table, String transformName, int sourceId) {
    TransformSpec spec = new TransformSpec();
    spec.setColumnName(table.schema().findColumnName(sourceId));
    // if the transform name contains '[' it means it has some config params
    if (transformName.contains("[")) {
      spec.setTransformType(TransformSpec.TransformType
          .valueOf(transformName.substring(0, transformName.indexOf("["))));
      spec.setTransformParam(Optional.of(Integer
          .valueOf(transformName.substring(transformName.indexOf("[") + 1, transformName.indexOf("]")))));
    } else {
      spec.setTransformType(TransformSpec.TransformType.valueOf(transformName));
      spec.setTransformParam(Optional.empty());
    }

    return spec;
  }

  @Override
  public DynamicPartitionCtx createDPContext(
          HiveConf hiveConf, org.apache.hadoop.hive.ql.metadata.Table hmsTable, Operation writeOperation)
      throws SemanticException {
    // delete records are already clustered by partition spec id and the hash of the partition struct
    // there is no need to do any additional sorting based on partition columns
    if (writeOperation == Operation.DELETE) {
      return null;
    }

    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());

    DynamicPartitionCtx dpCtx = new DynamicPartitionCtx(Maps.newLinkedHashMap(),
        hiveConf.getVar(HiveConf.ConfVars.DEFAULTPARTITIONNAME),
        hiveConf.getIntVar(HiveConf.ConfVars.DYNAMICPARTITIONMAXPARTSPERNODE));
    List<Function<List<ExprNodeDesc>, ExprNodeDesc>> customSortExprs = Lists.newLinkedList();
    dpCtx.setCustomSortExpressions(customSortExprs);

    if (table.spec().isPartitioned()) {
      addCustomSortExpr(table, hmsTable, writeOperation, customSortExprs, getPartitionTransformSpec(hmsTable));
    }

    SortOrder sortOrder = table.sortOrder();
    if (sortOrder.isSorted()) {
      List<Integer> customSortPositions = Lists.newLinkedList();
      List<Integer> customSortOrder = Lists.newLinkedList();
      dpCtx.setCustomSortOrder(customSortOrder);
      List<Integer> customSortNullOrder = Lists.newLinkedList();
      dpCtx.setCustomSortNullOrder(customSortNullOrder);
      for (SortField sortField : sortOrder.fields()) {
        int pos = 0;
        for (Types.NestedField field : table.schema().columns()) {
          if (sortField.sourceId() == field.fieldId()) {
            customSortPositions.add(pos);
            customSortOrder.add(sortField.direction() == SortDirection.ASC ? 1 : 0);
            customSortNullOrder.add(sortField.nullOrder() == NullOrder.NULLS_FIRST ? 0 : 1);
            break;
          }
          pos++;
        }
      }

      addCustomSortExpr(table, hmsTable, writeOperation, customSortExprs, getSortTransformSpec(table));
    }

    return dpCtx;
  }

  private void addCustomSortExpr(Table table,  org.apache.hadoop.hive.ql.metadata.Table hmsTable,
      Operation writeOperation, List<Function<List<ExprNodeDesc>, ExprNodeDesc>> customSortExprs,
      List<TransformSpec> transformSpecs) {
    Map<String, Integer> fieldOrderMap = Maps.newHashMap();
    List<Types.NestedField> fields = table.schema().columns();
    for (int i = 0; i < fields.size(); ++i) {
      fieldOrderMap.put(fields.get(i).name(), i);
    }

    int offset = acidSelectColumns(hmsTable, writeOperation).size();

    for (TransformSpec spec : transformSpecs) {
      int order = fieldOrderMap.get(spec.getColumnName());
      if (TransformSpec.TransformType.BUCKET.equals(spec.getTransformType())) {
        customSortExprs.add(BUCKET_SORT_EXPR.apply(order + offset, spec.getTransformParam().get()));
      } else {
        customSortExprs.add(cols -> cols.get(order + offset).clone());
      }
    }
  }

  @Override
  public String getFileFormatPropertyKey() {
    return TableProperties.DEFAULT_FILE_FORMAT;
  }

  @Override
  public boolean commitInMoveTask() {
    return true;
  }

  @Override
  public void storageHandlerCommit(Properties commitProperties, Operation operation)
        throws HiveException {
    String tableName = commitProperties.getProperty(Catalogs.NAME);
    String location = commitProperties.getProperty(Catalogs.LOCATION);
    String snapshotRef = commitProperties.getProperty(Catalogs.SNAPSHOT_REF);
    Configuration configuration = SessionState.getSessionConf();
    if (location != null) {
      HiveTableUtil.cleanupTableObjectFile(location, configuration);
    }
    List<JobContext> jobContextList = generateJobContext(configuration, tableName, snapshotRef);
    if (jobContextList.isEmpty()) {
      return;
    }

    HiveIcebergOutputCommitter committer = new HiveIcebergOutputCommitter();
    try {
      committer.commitJobs(jobContextList, operation);
    } catch (Throwable e) {
      String ids = jobContextList
          .stream().map(jobContext -> jobContext.getJobID().toString()).collect(Collectors.joining(", "));
      // Aborting the job if the commit has failed
      LOG.error("Error while trying to commit job: {}, starting rollback changes for table: {}",
          ids, tableName, e);
      try {
        committer.abortJobs(jobContextList);
      } catch (IOException ioe) {
        LOG.error("Error while trying to abort failed job. There might be uncleaned data files.", ioe);
        // no throwing here because the original exception should be propagated
      }
      throw new HiveException(
          "Error committing job: " + ids + " for table: " + tableName, e);
    }
  }

  @Override
  public boolean isAllowedAlterOperation(AlterTableType opType) {
    return HiveIcebergMetaHook.SUPPORTED_ALTER_OPS.contains(opType);
  }

  @Override
  public boolean supportsTruncateOnNonNativeTables() {
    return true;
  }

  @Override
  public boolean isTimeTravelAllowed() {
    return true;
  }

  @Override
  public boolean isTableMetaRefSupported() {
    return true;
  }

  @Override
  public void executeOperation(org.apache.hadoop.hive.ql.metadata.Table hmsTable, AlterTableExecuteSpec executeSpec) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table icebergTable = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    switch (executeSpec.getOperationType()) {
      case ROLLBACK:
        LOG.info("Executing rollback operation on iceberg table. If you would like to revert rollback you could " +
              "try altering the metadata location to the current metadata location by executing the following query:" +
              "ALTER TABLE {}.{} SET TBLPROPERTIES('metadata_location'='{}'). This operation is supported for Hive " +
              "Catalog tables.", hmsTable.getDbName(), hmsTable.getTableName(),
            ((BaseTable) icebergTable).operations().current().metadataFileLocation());
        AlterTableExecuteSpec.RollbackSpec rollbackSpec =
            (AlterTableExecuteSpec.RollbackSpec) executeSpec.getOperationParams();
        IcebergTableUtil.rollback(icebergTable, rollbackSpec.getRollbackType(), rollbackSpec.getParam());
        break;
      case EXPIRE_SNAPSHOT:
        LOG.info("Executing expire snapshots operation on iceberg table {}.{}", hmsTable.getDbName(),
            hmsTable.getTableName());
        AlterTableExecuteSpec.ExpireSnapshotsSpec expireSnapshotsSpec =
            (AlterTableExecuteSpec.ExpireSnapshotsSpec) executeSpec.getOperationParams();
        int numThreads = conf.getInt(HiveConf.ConfVars.HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS.varname,
            HiveConf.ConfVars.HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS.defaultIntVal);
        expireSnapshot(icebergTable, expireSnapshotsSpec, numThreads);
        break;
      case SET_CURRENT_SNAPSHOT:
        AlterTableExecuteSpec.SetCurrentSnapshotSpec setSnapshotVersionSpec =
            (AlterTableExecuteSpec.SetCurrentSnapshotSpec) executeSpec.getOperationParams();
        LOG.debug("Executing set current snapshot operation on iceberg table {}.{} to version {}", hmsTable.getDbName(),
            hmsTable.getTableName(), setSnapshotVersionSpec.getSnapshotId());
        IcebergTableUtil.setCurrentSnapshot(icebergTable, setSnapshotVersionSpec.getSnapshotId());
        break;
      case FAST_FORWARD:
        AlterTableExecuteSpec.FastForwardSpec fastForwardSpec =
            (AlterTableExecuteSpec.FastForwardSpec) executeSpec.getOperationParams();
        IcebergTableUtil.fastForwardBranch(icebergTable, fastForwardSpec.getSourceBranch(),
            fastForwardSpec.getTargetBranch());
        break;
      default:
        throw new UnsupportedOperationException(
            String.format("Operation type %s is not supported", executeSpec.getOperationType().name()));
    }
  }

  private void expireSnapshot(Table icebergTable, AlterTableExecuteSpec.ExpireSnapshotsSpec expireSnapshotsSpec,
      int numThreads) {
    ExecutorService deleteExecutorService = null;
    try {
      if (numThreads > 0) {
        LOG.info("Executing expire snapshots on iceberg table {} with {} threads", icebergTable.name(), numThreads);
        deleteExecutorService = getDeleteExecutorService(icebergTable.name(), numThreads);
      }
      if (expireSnapshotsSpec.isExpireByIds()) {
        expireSnapshotByIds(icebergTable, expireSnapshotsSpec.getIdsToExpire(), deleteExecutorService);
      } else {
        expireSnapshotOlderThanTimestamp(icebergTable, expireSnapshotsSpec.getTimestampMillis(), deleteExecutorService);
      }
    } finally {
      if (deleteExecutorService != null) {
        deleteExecutorService.shutdown();
      }
    }
  }

  private void expireSnapshotOlderThanTimestamp(Table icebergTable, Long timestamp,
      ExecutorService deleteExecutorService) {
    ExpireSnapshots expireSnapshots = icebergTable.expireSnapshots().expireOlderThan(timestamp);
    if (deleteExecutorService != null) {
      expireSnapshots.executeDeleteWith(deleteExecutorService);
    }
    expireSnapshots.commit();
  }

  private void expireSnapshotByIds(Table icebergTable, String[] idsToExpire,
      ExecutorService deleteExecutorService) {
    if (idsToExpire.length != 0) {
      ExpireSnapshots expireSnapshots = icebergTable.expireSnapshots();
      for (String id : idsToExpire) {
        expireSnapshots.expireSnapshotId(Long.parseLong(id));
      }
      LOG.info("Expiring snapshot on {} for snapshot Ids: {}", icebergTable.name(), Arrays.toString(idsToExpire));
      if (deleteExecutorService != null) {
        expireSnapshots = expireSnapshots.executeDeleteWith(deleteExecutorService);
      }
      expireSnapshots.commit();
    }
  }

  private ExecutorService getDeleteExecutorService(String completeName, int numThreads) {
    AtomicInteger deleteThreadsIndex = new AtomicInteger(0);
    return Executors.newFixedThreadPool(numThreads, runnable -> {
      Thread thread = new Thread(runnable);
      thread.setName("remove-snapshot-" + completeName + "-" + deleteThreadsIndex.getAndIncrement());
      return thread;
    });
  }

  @Override
  public void alterTableSnapshotRefOperation(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
      AlterTableSnapshotRefSpec alterTableSnapshotRefSpec) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table icebergTable = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    Optional.ofNullable(icebergTable.currentSnapshot()).orElseThrow(() ->
        new UnsupportedOperationException(String.format("Cannot alter %s on iceberg table %s.%s which has no snapshot",
            alterTableSnapshotRefSpec.getOperationType().getName(), hmsTable.getDbName(), hmsTable.getTableName())));

    switch (alterTableSnapshotRefSpec.getOperationType()) {
      case CREATE_BRANCH:
        AlterTableSnapshotRefSpec.CreateSnapshotRefSpec createBranchSpec =
            (AlterTableSnapshotRefSpec.CreateSnapshotRefSpec) alterTableSnapshotRefSpec.getOperationParams();
        IcebergBranchExec.createBranch(icebergTable, createBranchSpec);
        break;
      case CREATE_TAG:
        AlterTableSnapshotRefSpec.CreateSnapshotRefSpec createTagSpec =
            (AlterTableSnapshotRefSpec.CreateSnapshotRefSpec) alterTableSnapshotRefSpec.getOperationParams();
        IcebergTagExec.createTag(icebergTable, createTagSpec);
        break;
      case DROP_BRANCH:
        AlterTableSnapshotRefSpec.DropSnapshotRefSpec dropBranchSpec =
            (AlterTableSnapshotRefSpec.DropSnapshotRefSpec) alterTableSnapshotRefSpec.getOperationParams();
        IcebergBranchExec.dropBranch(icebergTable, dropBranchSpec);
        break;
      case DROP_TAG:
        AlterTableSnapshotRefSpec.DropSnapshotRefSpec dropTagSpec =
            (AlterTableSnapshotRefSpec.DropSnapshotRefSpec) alterTableSnapshotRefSpec.getOperationParams();
        IcebergTagExec.dropTag(icebergTable, dropTagSpec);
        break;
      default:
        throw new UnsupportedOperationException(String.format(
            "Operation type %s is not supported", alterTableSnapshotRefSpec.getOperationType().getName()));
    }
  }

  @Override
  public boolean isValidMetadataTable(String metaTableName) {
    return IcebergMetadataTables.isValidMetaTable(metaTableName);
  }

  @Override
  public org.apache.hadoop.hive.ql.metadata.Table checkAndSetTableMetaRef(
      org.apache.hadoop.hive.ql.metadata.Table hmsTable, String tableMetaRef) throws SemanticException {
    String refName = HiveUtils.getTableSnapshotRef(tableMetaRef);
    if (refName != null) {
      Table tbl = IcebergTableUtil.getTable(conf, hmsTable.getTTable());
      if (tbl.snapshot(refName) != null) {
        hmsTable.setSnapshotRef(tableMetaRef);
        return hmsTable;
      }
      throw new SemanticException(String.format("Cannot use snapshotRef (does not exist): %s", refName));
    }
    if (IcebergMetadataTables.isValidMetaTable(tableMetaRef)) {
      hmsTable.setMetaTable(tableMetaRef);
      return hmsTable;
    }
    throw new SemanticException(ErrorMsg.INVALID_METADATA_TABLE_NAME, tableMetaRef);
  }

  @Override
  public URI getURIForAuth(org.apache.hadoop.hive.metastore.api.Table hmsTable) throws URISyntaxException {
    String dbName = hmsTable.getDbName();
    String tableName = hmsTable.getTableName();
    StringBuilder authURI =
        new StringBuilder(ICEBERG_URI_PREFIX).append(encodeString(dbName)).append("/").append(encodeString(tableName))
            .append("?snapshot=");
    Optional<String> locationProperty = SessionStateUtil.getProperty(conf, hive_metastoreConstants.META_TABLE_LOCATION);
    if (locationProperty.isPresent()) {
      Preconditions.checkArgument(locationProperty.get() != null,
          "Table location is not set in SessionState. Authorization URI cannot be supplied.");
      // this property is set during the create operation before the hive table was created
      // we are returning a dummy iceberg metadata file
      authURI.append(encodeString(URI.create(locationProperty.get()).getPath()))
          .append(encodeString("/metadata/dummy.metadata.json"));
    } else {
      Table table = IcebergTableUtil.getTable(conf, hmsTable);
      authURI.append(
          encodeString(URI.create(((BaseTable) table).operations().current().metadataFileLocation()).getPath()));
    }
    LOG.debug("Iceberg storage handler authorization URI {}", authURI);
    return new URI(authURI.toString());
  }

  @VisibleForTesting
  static String encodeString(String rawString) {
    if (rawString == null) {
      return null;
    }
    return HiveConf.EncoderDecoderFactory.URL_ENCODER_DECODER.encode(rawString);
  }


  @Override
  public void validateSinkDesc(FileSinkDesc sinkDesc) throws SemanticException {
    HiveStorageHandler.super.validateSinkDesc(sinkDesc);
    if (sinkDesc.getInsertOverwrite()) {
      Table table = IcebergTableUtil.getTable(conf, sinkDesc.getTableInfo().getProperties());
      if (table.currentSnapshot() != null &&
          Long.parseLong(table.currentSnapshot().summary().get(SnapshotSummary.TOTAL_RECORDS_PROP)) == 0) {
        // If the table is empty we don't have any danger that some data can get lost.
        return;
      }
      if (IcebergTableUtil.isBucketed(table)) {
        throw new SemanticException("Cannot perform insert overwrite query on bucket partitioned Iceberg table.");
      }
      if (table.currentSnapshot() != null) {
        if (table.currentSnapshot().allManifests(table.io()).parallelStream().map(ManifestFile::partitionSpecId)
            .anyMatch(id -> id < table.spec().specId())) {
          throw new SemanticException(
              "Cannot perform insert overwrite query on Iceberg table where partition evolution happened. In order " +
              "to successfully carry out any insert overwrite operation on this table, the data has to be rewritten " +
              "conforming to the latest spec. ");
        }
      }
    }
  }

  @Override
  public AcidSupportType supportsAcidOperations(org.apache.hadoop.hive.ql.metadata.Table table,
      boolean isWriteOperation) {
    if (table.getParameters() != null && "2".equals(table.getParameters().get(TableProperties.FORMAT_VERSION))) {
      if (isWriteOperation) {
        checkDMLOperationMode(table);
      }
      return AcidSupportType.WITHOUT_TRANSACTIONS;
    }

    return AcidSupportType.NONE;
  }

  // TODO: remove the checks as copy-on-write mode implementation for these DML ops get added
  private static void checkDMLOperationMode(org.apache.hadoop.hive.ql.metadata.Table table) {
    Map<String, String> opTypes = ImmutableMap.of(
        TableProperties.MERGE_MODE, TableProperties.MERGE_MODE_DEFAULT,
        TableProperties.UPDATE_MODE, TableProperties.UPDATE_MODE_DEFAULT);

    for (Map.Entry<String, String> opType : opTypes.entrySet()) {
      String mode = table.getParameters().get(opType.getKey());
      RowLevelOperationMode rowLevelOperationMode = RowLevelOperationMode.fromName(
          mode != null ? mode : opType.getValue()
      );
      if (RowLevelOperationMode.COPY_ON_WRITE.equals(rowLevelOperationMode)) {
        throw new UnsupportedOperationException(
            String.format("Hive doesn't support copy-on-write mode as %s. Please set '%s'='merge-on-read' on %s " +
                "before running ACID operations on it.", opType.getKey(), opType.getKey(), table.getTableName())
        );
      }
    }
  }

  @Override
  public List<VirtualColumn> acidVirtualColumns() {
    return ACID_VIRTUAL_COLS;
  }

  @Override
  public List<FieldSchema> acidSelectColumns(org.apache.hadoop.hive.ql.metadata.Table table, Operation operation) {
    switch (operation) {
      case DELETE:
      case UPDATE:
        // TODO: make it configurable whether we want to include the table columns in the select query.
        // It might make delete writes faster if we don't have to write out the row object
        return ListUtils.union(ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA, table.getCols());
      case MERGE:
        return ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA;
      default:
        return ImmutableList.of();
    }
  }

  @Override
  public List<FieldSchema> acidSortColumns(org.apache.hadoop.hive.ql.metadata.Table table, Operation operation) {
    switch (operation) {
      case DELETE:
        return ACID_VIRTUAL_COLS_AS_FIELD_SCHEMA;
      default:
        // For update operations we use the same sort order defined by
        // {@link #createDPContext(HiveConf, org.apache.hadoop.hive.ql.metadata.Table)}
        return ImmutableList.of();
    }
  }

  @Override
  public boolean supportsSortColumns() {
    return true;
  }

  @Override
  public List<FieldSchema> sortColumns(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    if (table.sortOrder().isUnsorted()) {
      return Collections.emptyList();
    }

    Schema schema = table.schema();
    return table.sortOrder().fields().stream().map(s -> new FieldSchema(schema.findColumnName(s.sourceId()),
        schema.findType(s.sourceId()).toString(),
        String.format("Transform: %s, Sort direction: %s, Null sort order: %s",
        s.transform().toString(), s.direction().name(), s.nullOrder().name()))).collect(Collectors.toList());
  }

  private void setCommonJobConf(JobConf jobConf) {
    jobConf.set("tez.mrreader.config.update.properties", "hive.io.file.readcolumn.names,hive.io.file.readcolumn.ids");
  }

  public StorageHandlerTypes getType() {
    return StorageHandlerTypes.ICEBERG;
  }

  public boolean addDynamicSplitPruningEdge(org.apache.hadoop.hive.ql.metadata.Table table,
                                            ExprNodeDesc syntheticFilterPredicate) {
    try {
      Collection<String> partitionColumns = ((HiveIcebergSerDe) table.getDeserializer()).partitionColumns();
      if (partitionColumns.size() > 0) {
        // The filter predicate contains ExprNodeDynamicListDesc object(s) for places where we will substitute
        // dynamic values later during execution. For Example:
        // GenericUDFIn(Column[ss_sold_date_sk], RS[5] <-- This is an ExprNodeDynamicListDesc)
        //
        // We would like to check if we will be able to convert these expressions to Iceberg filters when the
        // actual values will be available, so in this check we replace the ExprNodeDynamicListDesc with dummy
        // values and check whether the conversion will be possible or not.
        ExprNodeDesc clone = syntheticFilterPredicate.clone();

        String filterColumn = collectColumnAndReplaceDummyValues(clone, null);

        // If the filter is for a partition column then it could be worthwhile to try dynamic partition pruning
        if (partitionColumns.contains(filterColumn)) {
          // Check if we can convert the expression to a valid Iceberg filter
          SearchArgument sarg = ConvertAstToSearchArg.create(conf, (ExprNodeGenericFuncDesc) clone);
          HiveIcebergFilterFactory.generateFilterExpression(sarg);
          LOG.debug("Found Iceberg partition column to prune with predicate {}", syntheticFilterPredicate);
          return true;
        }
      }
    } catch (UnsupportedOperationException uoe) {
      // If we can not convert the filter, we do not prune
      LOG.debug("Unsupported predicate {}", syntheticFilterPredicate, uoe);
    }

    // There is nothing to prune, or we could not use the filter
    LOG.debug("Not found Iceberg partition columns to prune with predicate {}", syntheticFilterPredicate);
    return false;
  }

  /**
   * Returns the Table serialized to the configuration based on the table name.
   * If configuration is missing from the FileIO of the table, it will be populated with the input config.
   *
   * @param config The configuration used to get the data from
   * @param name The name of the table we need as returned by TableDesc.getTableName()
   * @return The Table
   */
  public static Table table(Configuration config, String name) {
    Table table = SerializationUtil.deserializeFromBase64(config.get(InputFormatConfig.SERIALIZED_TABLE_PREFIX + name));
    if (table == null &&
            config.getBoolean(hive_metastoreConstants.TABLE_IS_CTAS, false) &&
            StringUtils.isNotBlank(config.get(InputFormatConfig.TABLE_LOCATION))) {
      table = HiveTableUtil.readTableObjectFromFile(config);
    }
    checkAndSetIoConfig(config, table);
    return table;
  }

  /**
   * If enabled, it populates the FileIO's hadoop configuration with the input config object.
   * This might be necessary when the table object was serialized without the FileIO config.
   *
   * @param config Configuration to set for FileIO, if enabled
   * @param table The Iceberg table object
   */
  public static void checkAndSetIoConfig(Configuration config, Table table) {
    if (table != null && config.getBoolean(InputFormatConfig.CONFIG_SERIALIZATION_DISABLED,
        InputFormatConfig.CONFIG_SERIALIZATION_DISABLED_DEFAULT) && table.io() instanceof HadoopConfigurable) {
      ((HadoopConfigurable) table.io()).setConf(config);
    }
  }

  /**
   * If enabled, it ensures that the FileIO's hadoop configuration will not be serialized.
   * This might be desirable for decreasing the overall size of serialized table objects.
   *
   * Note: Skipping FileIO config serialization in this fashion might in turn necessitate calling
   * {@link #checkAndSetIoConfig(Configuration, Table)} on the deserializer-side to enable subsequent use of the FileIO.
   *
   * @param config Configuration to set for FileIO in a transient manner, if enabled
   * @param table The Iceberg table object
   */
  public static void checkAndSkipIoConfigSerialization(Configuration config, Table table) {
    if (table != null && config.getBoolean(InputFormatConfig.CONFIG_SERIALIZATION_DISABLED,
            InputFormatConfig.CONFIG_SERIALIZATION_DISABLED_DEFAULT) && table.io() instanceof HadoopConfigurable) {
      ((HadoopConfigurable) table.io()).serializeConfWith(conf -> new NonSerializingConfig(config)::get);
    }
  }

  /**
   * Returns the names of the output tables stored in the configuration.
   * @param config The configuration used to get the data from
   * @return The collection of the table names as returned by TableDesc.getTableName()
   */
  public static Set<String> outputTables(Configuration config) {
    return Sets.newHashSet(TABLE_NAME_SPLITTER.split(config.get(InputFormatConfig.OUTPUT_TABLES)));
  }

  /**
   * Returns the catalog name serialized to the configuration.
   * @param config The configuration used to get the data from
   * @param name The name of the table we neeed as returned by TableDesc.getTableName()
   * @return catalog name
   */
  public static String catalogName(Configuration config, String name) {
    return config.get(InputFormatConfig.TABLE_CATALOG_PREFIX + name);
  }

  /**
   * Returns the Table Schema serialized to the configuration.
   * @param config The configuration used to get the data from
   * @return The Table Schema object
   */
  public static Schema schema(Configuration config) {
    return SchemaParser.fromJson(config.get(InputFormatConfig.TABLE_SCHEMA));
  }

  /**
   * Stores the serializable table data in the configuration.
   * Currently the following is handled:
   * <ul>
   *   <li>- Table - in case the table is serializable</li>
   *   <li>- Location</li>
   *   <li>- Schema</li>
   *   <li>- Partition specification</li>
   *   <li>- FileIO for handling table files</li>
   *   <li>- Location provider used for file generation</li>
   *   <li>- Encryption manager for encryption handling</li>
   * </ul>
   * @param configuration The configuration storing the catalog information
   * @param tableDesc The table which we want to store to the configuration
   * @param map The map of the configuration properties which we append with the serialized data
   */
  @VisibleForTesting
  static void overlayTableProperties(Configuration configuration, TableDesc tableDesc, Map<String, String> map) {
    Properties props = tableDesc.getProperties();

    Maps.fromProperties(props).entrySet().stream()
        .filter(entry -> !map.containsKey(entry.getKey())) // map overrides tableDesc properties
        .forEach(entry -> map.put(entry.getKey(), entry.getValue()));

    String location;
    Schema schema;
    PartitionSpec spec;
    try {
      Table table = IcebergTableUtil.getTable(configuration, props);
      location = table.location();
      schema = table.schema();
      spec = table.spec();

      // serialize table object into config
      Table serializableTable = SerializableTable.copyOf(table);
      checkAndSkipIoConfigSerialization(configuration, serializableTable);
      map.put(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableDesc.getTableName(),
          SerializationUtil.serializeToBase64(serializableTable));
    } catch (NoSuchTableException ex) {
      if (!HiveTableUtil.isCtas(props)) {
        throw ex;
      }

      if (!Catalogs.hiveCatalog(configuration, props)) {
        throw new UnsupportedOperationException(HiveIcebergSerDe.CTAS_EXCEPTION_MSG);
      }

      location = map.get(hive_metastoreConstants.META_TABLE_LOCATION);

      map.put(InputFormatConfig.SERIALIZED_TABLE_PREFIX + tableDesc.getTableName(),
              SerializationUtil.serializeToBase64(null));

      try {
        AbstractSerDe serDe = tableDesc.getDeserializer(configuration);
        HiveIcebergSerDe icebergSerDe = (HiveIcebergSerDe) serDe;
        schema = icebergSerDe.getTableSchema();
        spec = IcebergTableUtil.spec(configuration, icebergSerDe.getTableSchema());
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }

    map.put(InputFormatConfig.TABLE_IDENTIFIER, props.getProperty(Catalogs.NAME));
    if (StringUtils.isNotBlank(location)) {
      map.put(InputFormatConfig.TABLE_LOCATION, location);
    }
    String schemaJson = SchemaParser.toJson(schema);
    map.put(InputFormatConfig.TABLE_SCHEMA, schemaJson);
    // save schema into table props as well to avoid repeatedly hitting the HMS during serde initializations
    // this is an exception to the interface documentation, but it's a safe operation to add this property
    props.put(InputFormatConfig.TABLE_SCHEMA, schemaJson);
    if (spec == null) {
      spec = PartitionSpec.unpartitioned();
    }
    props.put(InputFormatConfig.PARTITION_SPEC, PartitionSpecParser.toJson(spec));

    // We need to remove this otherwise the job.xml will be invalid as column comments are separated with '\0' and
    // the serialization utils fail to serialize this character
    map.remove("columns.comments");
  }

  /**
   * Recursively replaces the ExprNodeDynamicListDesc nodes by a dummy ExprNodeConstantDesc so we can test if we can
   * convert the predicate to an Iceberg predicate when pruning the partitions later. Also collects the column names
   * in the filter.
   * <p>
   * Please make sure that it is ok to change the input node (clone if needed)
   * @param node The node we are traversing
   * @param foundColumn The column we already found
   */
  private String collectColumnAndReplaceDummyValues(ExprNodeDesc node, String foundColumn) {
    String column = foundColumn;
    List<ExprNodeDesc> children = node.getChildren();
    if (children != null && !children.isEmpty()) {
      ListIterator<ExprNodeDesc> iterator = children.listIterator();
      while (iterator.hasNext()) {
        ExprNodeDesc child = iterator.next();
        if (child instanceof ExprNodeDynamicListDesc) {
          Object dummy;
          switch (((PrimitiveTypeInfo) child.getTypeInfo()).getPrimitiveCategory()) {
            case INT:
            case SHORT:
              dummy = 1;
              break;
            case LONG:
              dummy = 1L;
              break;
            case TIMESTAMP:
            case TIMESTAMPLOCALTZ:
              dummy = new Timestamp();
              break;
            case CHAR:
            case VARCHAR:
            case STRING:
              dummy = "1";
              break;
            case DOUBLE:
            case FLOAT:
            case DECIMAL:
              dummy = 1.1;
              break;
            case DATE:
              dummy = new Date();
              break;
            case BOOLEAN:
              dummy = true;
              break;
            default:
              throw new UnsupportedOperationException("Not supported primitive type in partition pruning: " +
                  child.getTypeInfo());
          }

          iterator.set(new ExprNodeConstantDesc(child.getTypeInfo(), dummy));
        } else {
          String newColumn;
          if (child instanceof ExprNodeColumnDesc) {
            newColumn = ((ExprNodeColumnDesc) child).getColumn();
          } else {
            newColumn = collectColumnAndReplaceDummyValues(child, column);
          }

          if (column != null && newColumn != null && !newColumn.equals(column)) {
            throw new UnsupportedOperationException("Partition pruning does not support filtering for more columns");
          }

          if (column == null) {
            column = newColumn;
          }
        }
      }
    }

    return column;
  }

  /**
   * If any of the following checks is true we fall back to non vectorized mode:
   * <ul>
   *   <li>fileformat is set to avro</li>
   *   <li>querying metadata tables</li>
   *   <li>fileformat is set to ORC, and table schema has time type column</li>
   *   <li>fileformat is set to PARQUET, and table schema has a list type column, that has a complex type element</li>
   * </ul>
   * @param tableProps table properties, must be not null
   */
  private void fallbackToNonVectorizedModeBasedOnProperties(Properties tableProps) {
    Schema tableSchema = SchemaParser.fromJson(tableProps.getProperty(InputFormatConfig.TABLE_SCHEMA));
    if (FileFormat.AVRO.name().equalsIgnoreCase(tableProps.getProperty(TableProperties.DEFAULT_FILE_FORMAT)) ||
        (tableProps.containsKey("metaTable") && isValidMetadataTable(tableProps.getProperty("metaTable"))) ||
        hasOrcTimeInSchema(tableProps, tableSchema) ||
        !hasParquetNestedTypeWithinListOrMap(tableProps, tableSchema)) {
      conf.setBoolean(HiveConf.ConfVars.HIVE_VECTORIZATION_ENABLED.varname, false);
    }
  }

  /**
   * Iceberg Time type columns are written as longs into ORC files. There is no Time type in Hive, so it is represented
   * as String instead. For ORC there's no automatic conversion from long to string during vectorized reading such as
   * for example in Parquet (in Parquet files Time type is an int64 with 'time' logical annotation).
   * @param tableProps iceberg table properties
   * @param tableSchema iceberg table schema
   * @return
   */
  private static boolean hasOrcTimeInSchema(Properties tableProps, Schema tableSchema) {
    if (!FileFormat.ORC.name().equalsIgnoreCase(tableProps.getProperty(TableProperties.DEFAULT_FILE_FORMAT))) {
      return false;
    }
    return tableSchema.columns().stream().anyMatch(f -> Types.TimeType.get().typeId() == f.type().typeId());
  }

  /**
   * Vectorized reads of parquet files from columns with list or map type is only supported if the nested types are of
   * primitive type category
   * check {@link VectorizedParquetRecordReader#checkListColumnSupport} for details on nested types under lists
   * @param tableProps iceberg table properties
   * @param tableSchema iceberg table schema
   * @return
   */
  private static boolean hasParquetNestedTypeWithinListOrMap(Properties tableProps, Schema tableSchema) {
    if (!FileFormat.PARQUET.name().equalsIgnoreCase(tableProps.getProperty(TableProperties.DEFAULT_FILE_FORMAT))) {
      return true;
    }

    for (Types.NestedField field : tableSchema.columns()) {
      if (field.type().isListType() || field.type().isMapType()) {
        for (Types.NestedField nestedField : field.type().asNestedType().fields()) {
          if (!nestedField.type().isPrimitiveType()) {
            return false;
          }
        }
      }
    }

    return true;
  }

  /**
   * Generates {@link JobContext}s for the OutputCommitter for the specific table.
   * @param configuration The configuration used for as a base of the JobConf
   * @param tableName The name of the table we are planning to commit
   * @param overwrite If we have to overwrite the existing table or just add the new data
   * @return The generated Optional JobContext list or empty if not presents.
   */
  private List<JobContext> generateJobContext(Configuration configuration, String tableName,
      String branchName) {
    JobConf jobConf = new JobConf(configuration);
    Optional<Map<String, SessionStateUtil.CommitInfo>> commitInfoMap =
        SessionStateUtil.getCommitInfo(jobConf, tableName);
    if (commitInfoMap.isPresent()) {
      List<JobContext> jobContextList = Lists.newLinkedList();
      for (SessionStateUtil.CommitInfo commitInfo : commitInfoMap.get().values()) {
        JobID jobID = JobID.forName(commitInfo.getJobIdStr());
        commitInfo.getProps().forEach(jobConf::set);

        // we should only commit this current table because
        // for multi-table inserts, this hook method will be called sequentially for each target table
        jobConf.set(InputFormatConfig.OUTPUT_TABLES, tableName);
        if (branchName != null) {
          jobConf.set(InputFormatConfig.OUTPUT_TABLE_SNAPSHOT_REF, branchName);
        }

        jobContextList.add(new JobContextImpl(jobConf, jobID, null));
      }
      return jobContextList;
    } else {
      // most likely empty write scenario
      LOG.debug("Unable to find commit information in query state for table: {}", tableName);
      return Collections.emptyList();
    }
  }

  private String getOperationType() {
    return SessionStateUtil.getProperty(conf, Operation.class.getSimpleName())
        .orElse(Operation.OTHER.name());
  }

  private static class NonSerializingConfig implements Serializable {

    private final transient Configuration conf;

    NonSerializingConfig(Configuration conf) {
      this.conf = conf;
    }

    public Configuration get() {
      if (conf == null) {
        throw new IllegalStateException("Configuration was not serialized on purpose but was not set manually either");
      }

      return conf;
    }
  }

  @Override
  public boolean areSnapshotsSupported() {
    return true;
  }

  @Override
  public SnapshotContext getCurrentSnapshotContext(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    Snapshot current = table.currentSnapshot();
    if (current == null) {
      return null;
    }
    return new SnapshotContext(current.snapshotId());
  }

  @Override
  public void prepareAlterTableEnvironmentContext(AbstractAlterTableDesc alterTableDesc,
      EnvironmentContext environmentContext) {
    if (alterTableDesc instanceof AlterTableSetPropertiesDesc &&
        alterTableDesc.getProps().containsKey(BaseMetastoreTableOperations.METADATA_LOCATION_PROP)) {
      // signal manual iceberg metadata location updated by user
      environmentContext.putToProperties(HiveIcebergMetaHook.MANUAL_ICEBERG_METADATA_LOCATION_CHANGE, "true");
    }
  }

  @Override
  public void setTableParametersForCTLT(org.apache.hadoop.hive.ql.metadata.Table tbl, CreateTableLikeDesc desc,
      Map<String, String> origParams) {
    // Preserve the format-version of the iceberg table and filter out rest.
    String formatVersion = origParams.get(TableProperties.FORMAT_VERSION);
    if ("2".equals(formatVersion)) {
      tbl.getParameters().put(TableProperties.FORMAT_VERSION, formatVersion);
      tbl.getParameters().put(TableProperties.DELETE_MODE, MERGE_ON_READ);
      tbl.getParameters().put(TableProperties.UPDATE_MODE, MERGE_ON_READ);
      tbl.getParameters().put(TableProperties.MERGE_MODE, MERGE_ON_READ);
    }

    // check if the table is being created as managed table, in that case we translate it to external
    if (!desc.isExternal()) {
      tbl.getParameters().put("TRANSLATED_TO_EXTERNAL", "TRUE");
      desc.setIsExternal(true);
    }

    // If source is Iceberg table set the schema and the partition spec
    if ("ICEBERG".equalsIgnoreCase(origParams.get("table_type"))) {
      tbl.getParameters()
          .put(InputFormatConfig.TABLE_SCHEMA, origParams.get(InputFormatConfig.TABLE_SCHEMA));
      tbl.getParameters()
          .put(InputFormatConfig.PARTITION_SPEC, origParams.get(InputFormatConfig.PARTITION_SPEC));
    } else {
      // if the source is partitioned non-iceberg table set the partition transform spec and set the table as
      // unpartitioned
      List<TransformSpec> spec = PartitionTransform.getPartitionTransformSpec(tbl.getPartitionKeys());
      SessionStateUtil.addResourceOrThrow(conf, hive_metastoreConstants.PARTITION_TRANSFORM_SPEC, spec);
      tbl.getSd().getCols().addAll(tbl.getPartitionKeys());
      tbl.getTTable().setPartitionKeysIsSet(false);
    }
  }

  @Override
  public Map<String, String> getNativeProperties(org.apache.hadoop.hive.ql.metadata.Table table) {
    Table origTable = IcebergTableUtil.getTable(conf, table.getTTable());
    Map<String, String> props = Maps.newHashMap();
    props.put(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(origTable.schema()));
    props.put(InputFormatConfig.PARTITION_SPEC, PartitionSpecParser.toJson(origTable.spec()));
    return props;
  }

  @Override
  public boolean shouldOverwrite(org.apache.hadoop.hive.ql.metadata.Table mTable, String operationName) {
    String mode = null;
    String formatVersion = mTable.getTTable().getParameters().get(TableProperties.FORMAT_VERSION);
    // As of now only delete mode is supported, for all others return false
    if ("2".equals(formatVersion) && operationName.equalsIgnoreCase(Context.Operation.DELETE.toString())) {
      mode = mTable.getTTable().getParameters()
          .getOrDefault(TableProperties.DELETE_MODE, TableProperties.DELETE_MODE_DEFAULT);
    }
    return COPY_ON_WRITE.equalsIgnoreCase(mode);
  }

  @Override
  public Boolean hasAppendsOnly(org.apache.hadoop.hive.ql.metadata.Table hmsTable, SnapshotContext since) {
    TableDesc tableDesc = Utilities.getTableDesc(hmsTable);
    Table table = IcebergTableUtil.getTable(conf, tableDesc.getProperties());
    return hasAppendsOnly(table.snapshots(), since);
  }

  @VisibleForTesting
  Boolean hasAppendsOnly(Iterable<Snapshot> snapshots, SnapshotContext since) {
    boolean foundSince = since == null;
    for (Snapshot snapshot : snapshots) {
      if (!foundSince) {
        if (snapshot.snapshotId() == since.getSnapshotId()) {
          foundSince = true;
        }
      } else {
        if (!"append".equals(snapshot.operation())) {
          return false;
        }
      }
    }

    if (foundSince) {
      return true;
    }

    return null;
  }

  @Override
  public List<String> showPartitions(DDLOperationContext context, org.apache.hadoop.hive.ql.metadata.Table hmstbl)
      throws HiveException {
    Configuration confs = context.getConf();
    JobConf job = HiveTableUtil.getPartJobConf(confs, hmstbl);
    Class<? extends InputFormat> formatter = hmstbl.getInputFormatClass();

    try {
      InputFormat inputFormat = FetchOperator.getInputFormatFromCache(formatter, job);
      InputSplit[] splits = inputFormat.getSplits(job, 1);
      try (RecordReader<WritableComparable, Writable> reader = inputFormat.getRecordReader(splits[0], job,
          Reporter.NULL)) {
        return getPartitions(context, job, reader, hmstbl);
      }
    } catch (Exception e) {
      throw new HiveException(e, ErrorMsg.GENERIC_ERROR,
          "show partitions for table " + hmstbl.getTableName() + ". " + ErrorMsg.TABLE_NOT_PARTITIONED +
              " or the table is empty ");
    }
  }

  private List<String> getPartitions(DDLOperationContext context, Configuration job,
      RecordReader<WritableComparable, Writable> reader, org.apache.hadoop.hive.ql.metadata.Table hmstbl)
      throws Exception {

    List<String> parts = Lists.newArrayList();
    Writable value = reader.createValue();
    WritableComparable key = reader.createKey();

    try (FetchFormatter fetcher = new DefaultFetchFormatter()) {
      fetcher.initialize(job, HiveTableUtil.getSerializationProps());
      org.apache.hadoop.hive.ql.metadata.Table metaDataPartTable =
          context.getDb().getTable(hmstbl.getDbName(), hmstbl.getTableName(), "partitions", true);
      Deserializer currSerDe = metaDataPartTable.getDeserializer();
      ObjectMapper mapper = new ObjectMapper();
      Table tbl = getTable(hmstbl);
      while (reader.next(key, value)) {
        String[] row =
            fetcher.convert(currSerDe.deserialize(value), currSerDe.getObjectInspector())
                .toString().split("\t");
        parts.add(HiveTableUtil.getParseData(row[PART_IDX], row[SPEC_IDX], mapper, tbl.spec().specId()));
      }
    }
    Collections.sort(parts);
    return parts;
  }

}
