/*
 * 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.gobblin.iceberg.writer;

import java.io.IOException;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.avro.generic.GenericRecord;
import org.apache.avro.specific.SpecificData;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils;
import org.apache.iceberg.AppendFiles;
import org.apache.iceberg.CatalogUtil;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFiles;
import org.apache.iceberg.ExpireSnapshots;
import org.apache.iceberg.FindFiles;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.avro.AvroSchemaUtil;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.hive.HiveCatalog;
import org.apache.iceberg.types.Types;
import org.joda.time.DateTime;
import org.joda.time.format.PeriodFormatter;
import org.joda.time.format.PeriodFormatterBuilder;

import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Range;
import com.google.common.collect.Sets;
import com.google.common.io.Closer;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import org.apache.gobblin.completeness.verifier.KafkaAuditCountVerifier;
import org.apache.gobblin.configuration.ConfigurationKeys;
import org.apache.gobblin.configuration.State;
import org.apache.gobblin.data.management.copy.hive.WhitelistBlacklist;
import org.apache.gobblin.hive.AutoCloseableHiveLock;
import org.apache.gobblin.hive.HiveLock;
import org.apache.gobblin.hive.HivePartition;
import org.apache.gobblin.hive.metastore.HiveMetaStoreUtils;
import org.apache.gobblin.hive.spec.HiveSpec;
import org.apache.gobblin.hive.writer.HiveMetadataWriter;
import org.apache.gobblin.hive.writer.MetadataWriter;
import org.apache.gobblin.hive.writer.MetadataWriterKeys;
import org.apache.gobblin.iceberg.Utils.IcebergUtils;
import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
import org.apache.gobblin.metadata.OperationType;
import org.apache.gobblin.metrics.GobblinMetricsRegistry;
import org.apache.gobblin.metrics.MetricContext;
import org.apache.gobblin.metrics.Tag;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.metrics.event.GobblinEventBuilder;
import org.apache.gobblin.metrics.kafka.KafkaSchemaRegistry;
import org.apache.gobblin.metrics.kafka.SchemaRegistryException;
import org.apache.gobblin.source.extractor.extract.LongWatermark;
import org.apache.gobblin.stream.RecordEnvelope;
import org.apache.gobblin.util.AvroUtils;
import org.apache.gobblin.util.ClustersNames;
import org.apache.gobblin.util.HadoopUtils;
import org.apache.gobblin.util.ParallelRunner;
import org.apache.gobblin.util.WriterUtils;

import static org.apache.gobblin.iceberg.writer.IcebergMetadataWriterConfigKeys.*;

/**
 * This writer is used to calculate iceberg metadata from GMCE and register to iceberg
 * iceberg metadata here includes:
 * 1. Data files that contained by the table and the metrics of the data files
 * 2. Properties of the table (origin table properties, data offset range, high watermark of the GMCE and schema created time)
 * 3. Latest schema of the table
 */
@Slf4j
public class IcebergMetadataWriter implements MetadataWriter {

  // Critical when there's dataset-level ACL enforced for both data and Iceberg metadata
  public static final String USE_DATA_PATH_AS_TABLE_LOCATION = "use.data.path.as.table.location";
  public static final String TABLE_LOCATION_SUFFIX = "/_iceberg_metadata/%s";
  public static final String GMCE_HIGH_WATERMARK_KEY = "gmce.high.watermark.%s";
  public static final String GMCE_LOW_WATERMARK_KEY = "gmce.low.watermark.%s";
  private final static String EXPIRE_SNAPSHOTS_LOOKBACK_TIME = "gobblin.iceberg.dataset.expire.snapshots.lookBackTime";
  private final static String DEFAULT_EXPIRE_SNAPSHOTS_LOOKBACK_TIME = "3d";
  private static final String ICEBERG_REGISTRATION_BLACKLIST = "iceberg.registration.blacklist";
  private static final String ICEBERG_REGISTRATION_WHITELIST = "iceberg.registration.whitelist";
  private static final String ICEBERG_REGISTRATION_AUDIT_COUNT_BLACKLIST = "iceberg.registration.audit.count.blacklist";
  private static final String ICEBERG_REGISTRATION_AUDIT_COUNT_WHITELIST = "iceberg.registration.audit.count.whitelist";
  private static final String ICEBERG_METADATA_FILE_PERMISSION = "iceberg.metadata.file.permission";
  private static final String CREATE_TABLE_TIME = "iceberg.create.table.time";
  private static final String SCHEMA_CREATION_TIME_KEY = "schema.creation.time";
  private static final String ADDED_FILES_CACHE_EXPIRING_TIME = "added.files.cache.expiring.time";
  private static final int DEFAULT_ADDED_FILES_CACHE_EXPIRING_TIME = 1;
  private static final String OFFSET_RANGE_KEY_PREFIX = "offset.range.";
  private static final String OFFSET_RANGE_KEY_FORMAT = OFFSET_RANGE_KEY_PREFIX + "%s";

  private static final String DEFAULT_CREATION_TIME = "0";
  private static final String SNAPSHOT_EXPIRE_THREADS = "snapshot.expire.threads";
  private static final long DEFAULT_WATERMARK = -1L;

  /* one of the fields in DataFile entry to describe the location URI of a data file with FS Scheme */
  private static final String ICEBERG_FILE_PATH_COLUMN = DataFile.FILE_PATH.name();

  private final boolean completenessEnabled;
  private final boolean totalCountCompletenessEnabled;
  private final WhitelistBlacklist completenessWhitelistBlacklist;
  private final WhitelistBlacklist totalCountBasedCompletenessWhitelistBlacklist;
  private final String timeZone;
  private final DateTimeFormatter HOURLY_DATEPARTITION_FORMAT;
  private final String newPartitionColumn;
  private final String newPartitionColumnType;
  private final boolean newPartitionEnabled;
  private final WhitelistBlacklist newPartitionTableWhitelistBlacklist;
  private Optional<KafkaAuditCountVerifier> auditCountVerifier;
  private final String auditCheckGranularity;

  protected final MetricContext metricContext;
  protected EventSubmitter eventSubmitter;
  private final WhitelistBlacklist whitelistBlacklist;
  private final WhitelistBlacklist auditWhitelistBlacklist;
  private final Closer closer = Closer.create();

  // Mapping between table-id and currently processed watermark
  private final Map<TableIdentifier, Long> tableCurrentWatermarkMap;
  // Used to store the relationship between table and the gmce topicPartition
  private final Map<TableIdentifier, String> tableTopicPartitionMap;
  @Getter
  private final KafkaSchemaRegistry schemaRegistry;
  protected final Map<TableIdentifier, TableMetadata> tableMetadataMap;
  @Setter
  protected Catalog catalog;
  protected final Configuration conf;
  protected final ReadWriteLock readWriteLock;
  private final HiveLock locks;
  private final boolean useDataLocationAsTableLocation;
  private final ParallelRunner parallelRunner;
  private FsPermission permission;
  protected State state;

  public IcebergMetadataWriter(State state) throws IOException {
    this.state = state;
    this.schemaRegistry = KafkaSchemaRegistry.get(state.getProperties());
    conf = HadoopUtils.getConfFromState(state);
    initializeCatalog();
    tableTopicPartitionMap = new HashMap<>();
    tableMetadataMap = new HashMap<>();
    tableCurrentWatermarkMap = new HashMap<>();
    List<Tag<?>> tags = Lists.newArrayList();
    String clusterIdentifier = ClustersNames.getInstance().getClusterName();
    tags.add(new Tag<>(MetadataWriterKeys.CLUSTER_IDENTIFIER_KEY_NAME, clusterIdentifier));
    metricContext = closer.register(
        GobblinMetricsRegistry.getInstance().getMetricContext(state, IcebergMetadataWriter.class, tags));
    this.eventSubmitter =
        new EventSubmitter.Builder(this.metricContext, MetadataWriterKeys.METRICS_NAMESPACE_ICEBERG_WRITER).build();
    this.whitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_REGISTRATION_WHITELIST, ""),
        state.getProp(ICEBERG_REGISTRATION_BLACKLIST, ""));
    this.auditWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_REGISTRATION_AUDIT_COUNT_WHITELIST, ""),
        state.getProp(ICEBERG_REGISTRATION_AUDIT_COUNT_BLACKLIST, ""));

    // Use rw-lock to make it thread-safe when flush and write(which is essentially aggregate & reading metadata),
    // are called in separate threads.
    readWriteLock = new ReentrantReadWriteLock();
    this.locks = new HiveLock(state.getProperties());
    parallelRunner = closer.register(new ParallelRunner(state.getPropAsInt(SNAPSHOT_EXPIRE_THREADS, 20),
        FileSystem.get(HadoopUtils.getConfFromState(state))));
    useDataLocationAsTableLocation = state.getPropAsBoolean(USE_DATA_PATH_AS_TABLE_LOCATION, false);
    if (useDataLocationAsTableLocation) {
      permission =
          HadoopUtils.deserializeFsPermission(state, ICEBERG_METADATA_FILE_PERMISSION,
              FsPermission.getDefault());
    }
    this.completenessEnabled = state.getPropAsBoolean(ICEBERG_COMPLETENESS_ENABLED, DEFAULT_ICEBERG_COMPLETENESS);
    this.totalCountCompletenessEnabled = state.getPropAsBoolean(ICEBERG_TOTAL_COUNT_COMPLETENESS_ENABLED,
        DEFAULT_ICEBERG_TOTAL_COUNT_COMPLETENESS);
    this.completenessWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_COMPLETENESS_WHITELIST, ""),
        state.getProp(ICEBERG_COMPLETENESS_BLACKLIST, ""));
    this.totalCountBasedCompletenessWhitelistBlacklist = new WhitelistBlacklist(
        state.getProp(ICEBERG_TOTAL_COUNT_COMPLETENESS_WHITELIST, ""),
        state.getProp(ICEBERG_TOTAL_COUNT_COMPLETENESS_BLACKLIST, ""));
    this.timeZone = state.getProp(TIME_ZONE_KEY, DEFAULT_TIME_ZONE);
    this.HOURLY_DATEPARTITION_FORMAT = DateTimeFormatter.ofPattern(DATEPARTITION_FORMAT)
        .withZone(ZoneId.of(this.timeZone));
    this.auditCountVerifier = Optional.fromNullable(this.completenessEnabled ? new KafkaAuditCountVerifier(state) : null);
    this.newPartitionColumn = state.getProp(NEW_PARTITION_KEY, DEFAULT_NEW_PARTITION);
    this.newPartitionColumnType = state.getProp(NEW_PARTITION_TYPE_KEY, DEFAULT_PARTITION_COLUMN_TYPE);
    this.newPartitionEnabled = state.getPropAsBoolean(ICEBERG_NEW_PARTITION_ENABLED, DEFAULT_ICEBERG_NEW_PARTITION_ENABLED);
    this.newPartitionTableWhitelistBlacklist = new WhitelistBlacklist(state.getProp(ICEBERG_NEW_PARTITION_WHITELIST, ""),
        state.getProp(ICEBERG_NEW_PARTITION_BLACKLIST, ""));
    this.auditCheckGranularity = state.getProp(AUDIT_CHECK_GRANULARITY, DEFAULT_AUDIT_CHECK_GRANULARITY);
  }

  @VisibleForTesting
  protected void setAuditCountVerifier(KafkaAuditCountVerifier verifier) {
    this.auditCountVerifier = Optional.of(verifier);
  }

  protected void initializeCatalog() {
    catalog = CatalogUtil.loadCatalog(HiveCatalog.class.getName(), "HiveCatalog", new HashMap<>(), conf);
  }

  private org.apache.iceberg.Table getIcebergTable(TableIdentifier tid) throws NoSuchTableException {
    TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata(this.conf));
    if (!tableMetadata.table.isPresent()) {
      tableMetadata.table = Optional.of(catalog.loadTable(tid));
    }
    return tableMetadata.table.get();
  }

  /**
   *  The method is used to get current watermark of the gmce topic partition for a table, and persist the value
   *  in the {@link #tableMetadataMap} as a side effect.
   *
   *  Make the watermark config name contains topicPartition in case we change the gmce topic name for some reason
   */
  private Long getAndPersistCurrentWatermark(TableIdentifier tid, String topicPartition) {
    if (tableCurrentWatermarkMap.containsKey(tid)) {
      return tableCurrentWatermarkMap.get(tid);
    }
    org.apache.iceberg.Table icebergTable;
    Long currentWatermark = DEFAULT_WATERMARK;
    try {
      icebergTable = getIcebergTable(tid);
    } catch (NoSuchTableException e) {
      return currentWatermark;
    }
    currentWatermark =
        icebergTable.properties().containsKey(String.format(GMCE_HIGH_WATERMARK_KEY, topicPartition)) ? Long.parseLong(
            icebergTable.properties().get(String.format(GMCE_HIGH_WATERMARK_KEY, topicPartition))) : DEFAULT_WATERMARK;
    return currentWatermark;
  }

  /**
   * The write method will be responsible for processing gmce and aggregating the metadata.
   * The logic of this function will be:
   * 1. Check whether a table exists, if not then create a iceberg table
   *    - If completeness is enabled, Add new parititon column to
   *      table {#NEW_PARTITION_KEY}
   * 2. Compute schema from the gmce and update the cache for candidate schemas
   * 3. Do the required operation of the gmce, i.e. addFile, rewriteFile, dropFile or change_property.
   *
   * Note: this method only aggregates the metadata in cache without committing,
   * while the actual commit will be done in flush method (except rewrite and drop methods where preserving older file
   * information increases the memory footprints, therefore we would like to flush them eagerly).
   */
  public void write(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> newSpecsMap,
      Map<String, Collection<HiveSpec>> oldSpecsMap, HiveSpec tableSpec) throws IOException {
    TableIdentifier tid = TableIdentifier.of(tableSpec.getTable().getDbName(), tableSpec.getTable().getTableName());
    TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata(this.conf));
    Table table;
    try {
      table = getIcebergTable(tid);
    } catch (NoSuchTableException e) {
      try {
        if (gmce.getOperationType() == OperationType.drop_files ||
            gmce.getOperationType() == OperationType.change_property) {
          log.warn("Table {} does not exist, skip processing this {} event", tid.toString(), gmce.getOperationType());
          return;
        }
        table = createTable(gmce, tableSpec);
        tableMetadata.table = Optional.of(table);
      } catch (Exception e1) {
        log.error("skip processing {} for table {}.{} due to error when creating table", gmce.toString(),
            tableSpec.getTable().getDbName(), tableSpec.getTable().getTableName());
        log.debug(e1.toString());
        return;
      }
    }
    if(tableMetadata.completenessEnabled) {
      tableMetadata.completionWatermark = Long.parseLong(table.properties().getOrDefault(COMPLETION_WATERMARK_KEY,
          String.valueOf(DEFAULT_COMPLETION_WATERMARK)));

      if (tableMetadata.totalCountCompletenessEnabled) {
        tableMetadata.totalCountCompletionWatermark = Long.parseLong(
            table.properties().getOrDefault(TOTAL_COUNT_COMPLETION_WATERMARK_KEY,
                String.valueOf(DEFAULT_COMPLETION_WATERMARK)));
      }
    }

    computeCandidateSchema(gmce, tid, tableSpec);
    tableMetadata.ensureTxnInit();
    tableMetadata.lowestGMCEEmittedTime = Long.min(tableMetadata.lowestGMCEEmittedTime, gmce.getGMCEmittedTime());
    switch (gmce.getOperationType()) {
      case add_files: {
        updateTableProperty(tableSpec, tid, gmce);
        addFiles(gmce, newSpecsMap, table, tableMetadata);
        if (gmce.getAuditCountMap() != null && auditWhitelistBlacklist.acceptTable(tableSpec.getTable().getDbName(),
            tableSpec.getTable().getTableName())) {
          tableMetadata.serializedAuditCountMaps.add(gmce.getAuditCountMap());
        }
        if (gmce.getTopicPartitionOffsetsRange() != null) {
          mergeOffsets(gmce, tid);
        }
        break;
      }
      case rewrite_files: {
        updateTableProperty(tableSpec, tid, gmce);
        rewriteFiles(gmce, newSpecsMap, oldSpecsMap, table, tableMetadata);
        break;
      }
      case drop_files: {
        dropFiles(gmce, oldSpecsMap, table, tableMetadata, tid);
        break;
      }
      case change_property: {
        updateTableProperty(tableSpec, tid, gmce);
        if (gmce.getTopicPartitionOffsetsRange() != null) {
          mergeOffsets(gmce, tid);
        }
        log.info("No file operation need to be performed by Iceberg Metadata Writer at this point.");
        break;
      }
      default: {
        log.error("unsupported operation {}", gmce.getOperationType().toString());
        return;
      }
    }
  }

  private HashMap<String, List<Range>> getLastOffset(TableMetadata tableMetadata) {
    HashMap<String, List<Range>> offsets = new HashMap<>();
    if (tableMetadata.lastProperties.isPresent()) {
      for (Map.Entry<String, String> entry : tableMetadata.lastProperties.get().entrySet()) {
        if (entry.getKey().startsWith(OFFSET_RANGE_KEY_PREFIX)) {
          List<Range> ranges = Arrays.asList(entry.getValue().split(ConfigurationKeys.LIST_DELIMITER_KEY))
              .stream()
              .map(s -> {
                List<String> rangePair = Splitter.on(ConfigurationKeys.RANGE_DELIMITER_KEY).splitToList(s);
                return Range.openClosed(Long.parseLong(rangePair.get(0)), Long.parseLong(rangePair.get(1)));})
              .collect(Collectors.toList());
          offsets.put(entry.getKey().substring(OFFSET_RANGE_KEY_PREFIX.length()), ranges);
        }
      }
    }
    return offsets;
  }

  /**
   * The side effect of this method is to update the offset-range of the table identified by
   * the given {@link TableIdentifier} with the input {@link GobblinMetadataChangeEvent}
   */
  private void mergeOffsets(GobblinMetadataChangeEvent gmce, TableIdentifier tid) {
    TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata(this.conf));
    tableMetadata.dataOffsetRange = Optional.of(tableMetadata.dataOffsetRange.or(() -> getLastOffset(tableMetadata)));
    Map<String, List<Range>> offsets = tableMetadata.dataOffsetRange.get();
    for (Map.Entry<String, String> entry : gmce.getTopicPartitionOffsetsRange().entrySet()) {
      List<String> rangePair = Splitter.on(ConfigurationKeys.RANGE_DELIMITER_KEY).splitToList(entry.getValue());
      Range range = Range.openClosed(Long.parseLong(rangePair.get(0)), Long.parseLong(rangePair.get(1)));
      if (range.lowerEndpoint().equals(range.upperEndpoint())) {
        //Ignore this case
        continue;
      }
      List<Range> existRanges = offsets.getOrDefault(entry.getKey(), new ArrayList<>());
      List<Range> newRanges = new ArrayList<>();
      for (Range r : existRanges) {
        if (range.isConnected(r)) {
          range = range.span(r);
        } else {
          newRanges.add(r);
        }
      }
      newRanges.add(range);
      Collections.sort(newRanges, new Comparator<Range>() {
        @Override
        public int compare(Range o1, Range o2) {
          return o1.lowerEndpoint().compareTo(o2.lowerEndpoint());
        }
      });
      offsets.put(entry.getKey(), newRanges);
    }
  }

  protected void updateTableProperty(HiveSpec tableSpec, TableIdentifier tid, GobblinMetadataChangeEvent gmce) {
    org.apache.hadoop.hive.metastore.api.Table table = HiveMetaStoreUtils.getTable(tableSpec.getTable());
    TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata(this.conf));
    tableMetadata.newProperties = Optional.of(IcebergUtils.getTableProperties(table));
    String nativeName = tableMetadata.datasetName;
    String topic = nativeName.substring(nativeName.lastIndexOf("/") + 1);
    tableMetadata.newProperties.get().put(TOPIC_NAME_KEY, topic);
  }

  /**
   * Compute the candidate schema from the gmce.
   * If the schema source is schemaRegistry, we will use the schema creation time as the schema version to compute candidate schema and determine latest schema
   * If the schema does not contain creation time, we will treat it the same as when schema source is event
   * If the schema source is event, we will put it as default creation time, during flush, if we only have one candidate with default creation time,
   * we'll use that to update schema.
   * @param gmce
   * @param tid
   */
  private void computeCandidateSchema(GobblinMetadataChangeEvent gmce, TableIdentifier tid, HiveSpec spec) {
    Table table = getIcebergTable(tid);
    TableMetadata tableMetadata = tableMetadataMap.computeIfAbsent(tid, t -> new TableMetadata(this.conf));
    org.apache.hadoop.hive.metastore.api.Table hiveTable = HiveMetaStoreUtils.getTable(spec.getTable());
    tableMetadata.lastProperties = Optional.of(tableMetadata.lastProperties.or(() -> table.properties()));
    Map<String, String> props = tableMetadata.lastProperties.get();
    tableMetadata.lastSchemaVersion = Optional.of(
        tableMetadata.lastSchemaVersion.or(() -> props.getOrDefault(SCHEMA_CREATION_TIME_KEY, DEFAULT_CREATION_TIME)));
    String lastSchemaVersion = tableMetadata.lastSchemaVersion.get();
    tableMetadata.candidateSchemas = Optional.of(tableMetadata.candidateSchemas.or(() -> CacheBuilder.newBuilder()
        .expireAfterAccess(conf.getInt(MetadataWriter.CACHE_EXPIRING_TIME,
            MetadataWriter.DEFAULT_CACHE_EXPIRING_TIME), TimeUnit.HOURS)
        .build()));
    Cache<String, Pair<Schema, String>> candidate = tableMetadata.candidateSchemas.get();
    try {
      switch (gmce.getSchemaSource()) {
        case SCHEMAREGISTRY: {
          org.apache.avro.Schema schema = new org.apache.avro.Schema.Parser().parse(gmce.getTableSchema());
          String createdOn = AvroUtils.getSchemaCreationTime(schema);
          if (createdOn == null) {
            candidate.put(DEFAULT_CREATION_TIME,
                Pair.of(IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema, gmce.getTableSchema()));
          } else if (!createdOn.equals(lastSchemaVersion)) {
            candidate.put(createdOn, Pair.of(IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema, gmce.getTableSchema()));
          }
          break;
        }
        case EVENT: {
          candidate.put(DEFAULT_CREATION_TIME,
              Pair.of(IcebergUtils.getIcebergSchema(gmce.getTableSchema(), hiveTable).tableSchema, gmce.getTableSchema()));
          break;
        }
        case NONE: {
          log.debug("Schema source set to be none, will ignore the schema");
          break;
        }
        default: {
          throw new IOException(String.format("unsupported schema source %s", gmce.getSchemaSource()));
        }
      }
    } catch (Exception e) {
      log.error("Cannot get candidate schema from event due to", e);
    }
  }

  /**
   * Add a partition column to the schema and partition spec
   * @param table incoming iceberg table
   * @param fieldName name of partition column
   * @param type datatype of partition column
   * @return table with updated schema and partition spec
   */
  private Table addPartitionToIcebergTable(Table table, String fieldName, String type) {
    boolean isTableUpdated = false;
    if(!table.schema().columns().stream().anyMatch(x -> x.name().equalsIgnoreCase(fieldName))) {
      table.updateSchema().addColumn(fieldName, Types.fromPrimitiveString(type)).commit();
      isTableUpdated = true;
    }
    if(!table.spec().fields().stream().anyMatch(x -> x.name().equalsIgnoreCase(fieldName))) {
      table.updateSpec().addField(fieldName).commit();
      isTableUpdated = true;
    }

    if (isTableUpdated) {
      table.refresh();
    }

    return table;
  }

  protected Table createTable(GobblinMetadataChangeEvent gmce, HiveSpec spec) throws IOException {
    String schema = gmce.getTableSchema();
    org.apache.hadoop.hive.metastore.api.Table table = HiveMetaStoreUtils.getTable(spec.getTable());
    IcebergUtils.IcebergDataAndPartitionSchema schemas = IcebergUtils.getIcebergSchema(schema, table);
    TableIdentifier tid = TableIdentifier.of(table.getDbName(), table.getTableName());
    Schema tableSchema = schemas.tableSchema;
    Preconditions.checkState(tableSchema != null, "Table schema cannot be null when creating a table");
    PartitionSpec partitionSpec = IcebergUtils.getPartitionSpec(tableSchema, schemas.partitionSchema);
    Table icebergTable = null;
    String tableLocation = null;
    if (useDataLocationAsTableLocation) {
      tableLocation = gmce.getDatasetIdentifier().getNativeName() + String.format(TABLE_LOCATION_SUFFIX, table.getDbName());
      //Set the path permission
      Path tablePath = new Path(tableLocation);
      WriterUtils.mkdirsWithRecursivePermission(tablePath.getFileSystem(conf), tablePath, permission);
    }
    try (Timer.Context context = metricContext.timer(CREATE_TABLE_TIME).time()) {
      icebergTable =
          catalog.createTable(tid, tableSchema, partitionSpec, tableLocation, IcebergUtils.getTableProperties(table));
      // We should set the avro schema literal when creating the table.
      icebergTable.updateProperties().set(AvroSerdeUtils.AvroTableProperties.SCHEMA_LITERAL.getPropName(), schema).commit();
      log.info("Created table {}, schema: {} partition spec: {}", tid, tableSchema, partitionSpec);
    } catch (AlreadyExistsException e) {
      log.warn("table {} already exist, there may be some other process try to create table concurrently", tid);
    }
    return icebergTable;
  }

  protected void rewriteFiles(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> newSpecsMap,
      Map<String, Collection<HiveSpec>> oldSpecsMap, Table table, TableMetadata tableMetadata) throws IOException {
    PartitionSpec partitionSpec = table.spec();
    tableMetadata.ensureTxnInit();
    Set<DataFile> newDataFiles = new HashSet<>();
    getIcebergDataFilesToBeAddedHelper(gmce, table, newSpecsMap, tableMetadata)
        .forEach(dataFile -> {
          newDataFiles.add(dataFile);
          tableMetadata.addedFiles.put(dataFile.path(), "");
        });
    Set<DataFile> oldDataFiles = getIcebergDataFilesToBeDeleted(gmce, table, newSpecsMap, oldSpecsMap, partitionSpec);

    // Dealing with the case when old file doesn't exist, in which it could either be converted into noop or AppendFile.
    if (oldDataFiles.isEmpty() && !newDataFiles.isEmpty()) {
      //We randomly check whether one of the new data files already exists in the db to avoid reprocessing re-write events
      DataFile dataFile = newDataFiles.iterator().next();
      Expression exp = Expressions.startsWith(ICEBERG_FILE_PATH_COLUMN, dataFile.path().toString());
      if (FindFiles.in(table).withMetadataMatching(exp).collect().iterator().hasNext()) {
        //This means this re-write event is duplicated with the one we already handled, so noop.
        return;
      }
      // This is the case when the files to be deleted do not exist in table
      // So we directly call addFiles interface to add new files into the table.
      // Note that this AppendFiles won't be committed here, in contrast to a real rewrite operation
      // where the commit will be called at once to save memory footprints.
      AppendFiles appendFiles = tableMetadata.getOrInitAppendFiles();
      newDataFiles.forEach(appendFiles::appendFile);
      return;
    }

    tableMetadata.transaction.get().newRewrite().rewriteFiles(oldDataFiles, newDataFiles).commit();
  }

  /**
   * Given the GMCE, get the iceberg schema with the origin ID specified by data pipeline which
   * is corresponding to the file metrics index.
   * @param gmce GMCE emitted by data pipeline
   * @return iceberg schema with the origin ID
   */
  private Schema getSchemaWithOriginId(GobblinMetadataChangeEvent gmce) {
    Schema schemaWithOriginId = null;
    if (gmce.getAvroSchemaWithIcebergSchemaID() != null) {
      org.apache.iceberg.shaded.org.apache.avro.Schema avroSchema =
          new org.apache.iceberg.shaded.org.apache.avro.Schema.Parser().parse(gmce.getAvroSchemaWithIcebergSchemaID());
      schemaWithOriginId = AvroSchemaUtil.toIceberg(avroSchema);
    }
    return schemaWithOriginId;
  }

  /**
   * Deal with both regular file deletions manifested by GMCE(aggregation but no commit),
   * and expiring older snapshots(commit).
   */
  protected void dropFiles(GobblinMetadataChangeEvent gmce, Map<String, Collection<HiveSpec>> oldSpecsMap, Table table,
      TableMetadata tableMetadata, TableIdentifier tid) throws IOException {
    PartitionSpec partitionSpec = table.spec();

    // Update DeleteFiles in tableMetadata: This is regular file deletion
    DeleteFiles deleteFiles = tableMetadata.getOrInitDeleteFiles();
    Set<DataFile> oldDataFiles =
        getIcebergDataFilesToBeDeleted(gmce, table, new HashMap<>(), oldSpecsMap, partitionSpec);
    oldDataFiles.forEach(deleteFiles::deleteFile);

    // Update ExpireSnapshots and commit the updates at once: This is for expiring snapshots that are
    // beyond look-back allowance for time-travel.
    parallelRunner.submitCallable(new Callable<Void>() {
      @Override
      public Void call() throws Exception {
        try {
          long olderThan = getExpireSnapshotTime();
          long start = System.currentTimeMillis();
          ExpireSnapshots expireSnapshots = table.expireSnapshots();
          final Table tmpTable = table;

          expireSnapshots.deleteWith(new Consumer<String>() {
            @Override
            public void accept(String file) {
              if (file.startsWith(tmpTable.location())) {
                tmpTable.io().deleteFile(file);
              }
            }
          }).expireOlderThan(olderThan).commit();
          //TODO: emit these metrics to Ingraphs, in addition to metrics for publishing new snapshots and other Iceberg metadata operations.
          log.info("Spent {} ms to expire snapshots older than {} ({}) in table {}", System.currentTimeMillis() - start,
              new DateTime(olderThan).toString(), olderThan, tid.toString());
        } catch (Exception e) {
          log.error(String.format("Fail to expire snapshots for table %s due to exception ", tid.toString()), e);
        }
        return null;
      }
    }, tid.toString());
  }

  private long getExpireSnapshotTime() {
    PeriodFormatter periodFormatter = new PeriodFormatterBuilder().appendYears()
        .appendSuffix("y")
        .appendMonths()
        .appendSuffix("M")
        .appendDays()
        .appendSuffix("d")
        .appendHours()
        .appendSuffix("h")
        .appendMinutes()
