// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package org.apache.impala.catalog;

import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.impala.analysis.TableName;
import org.apache.impala.authorization.AuthorizationChecker;
import org.apache.impala.authorization.AuthorizationPolicy;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.Pair;
import org.apache.impala.service.FeSupport;
import org.apache.impala.thrift.TAuthzCacheInvalidation;
import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TCatalogObjectType;
import org.apache.impala.thrift.TDataSource;
import org.apache.impala.thrift.TDatabase;
import org.apache.impala.thrift.TFunction;
import org.apache.impala.thrift.TGetPartitionStatsResponse;
import org.apache.impala.thrift.TTable;
import org.apache.impala.thrift.TUniqueId;
import org.apache.impala.thrift.TUpdateCatalogCacheRequest;
import org.apache.impala.thrift.TUpdateCatalogCacheResponse;
import org.apache.impala.util.PatternMatcher;
import org.apache.impala.util.TByteBuffer;
import org.apache.log4j.Logger;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;

import com.google.common.base.Preconditions;

/**
 * Thread safe Catalog for an Impalad.  The Impalad catalog can be updated either via
 * a StateStore heartbeat or by directly applying the result of a catalog operation to
 * the CatalogCache. All updates are applied using the updateCatalog() function.
 * Table metadata is loaded lazily. The CatalogServer initially broadcasts (via the
 * statestore) the known table names (as IncompleteTables). These table names are added
 * to the Impalad catalog cache and when one of the tables is accessed, the impalad will
 * make an RPC to the CatalogServer to request loading the complete table metadata.
 * In both cases, we need to ensure that work from one update is not "undone" by another
 * update. To handle this the ImpaladCatalog does the following:
 * - Tracks the overall catalog version last received in a state store heartbeat, this
 *   version is maintained by the catalog server and it is always guaranteed that
 *   this impalad's catalog will never contain any objects < than this version
 *   (any updates with a lower version number are ignored).
 * - For updated/new objects, check if the object already exists in the
 *   catalog cache. If it does, only apply the update if the catalog version is > the
 *   existing object's catalog version. Also keep a log of all dropped catalog objects
 *   (and the version they were dropped in). Before updating any object, check if it was
 *   dropped in a later version. If so, ignore the update.
 * - Before dropping any catalog object, see if the object already exists in the catalog
 *   cache. If it does, only drop the object if the version of the drop is > that
 *   object's catalog version.
 * The CatalogServiceId is also tracked to detect if a different instance of the catalog
 * service has been started, in which case a full topic update is required.
 */
public class ImpaladCatalog extends Catalog implements FeCatalog {
  private static final Logger LOG = Logger.getLogger(ImpaladCatalog.class);
  // The last known Catalog Service ID. If the ID changes, it indicates the CatalogServer
  // has restarted.
  private TUniqueId catalogServiceId_ = Catalog.INITIAL_CATALOG_SERVICE_ID;

  // The catalog version received in the last StateStore heartbeat. It is guaranteed
  // all objects in the catalog have at a minimum, this version. Because updates may
  // be applied out of band of a StateStore heartbeat, it is possible the catalog
  // contains some objects > than this version.
  private AtomicLong lastSyncedCatalogVersion_ =
      new AtomicLong(Catalog.INITIAL_CATALOG_VERSION);

  // Tracks modifications to this Impalad's catalog from direct updates to the cache.
  private final CatalogDeltaLog catalogDeltaLog_ = new CatalogDeltaLog();

  // Object that is used to synchronize on and signal when a catalog update is received.
  private final Object catalogUpdateEventNotifier_ = new Object();

  // The addresses of the Kudu masters to use if no Kudu masters were explicitly provided.
  // Used during table creation.
  private final String defaultKuduMasterHosts_;
  private final AtomicReference<? extends AuthorizationChecker> authzChecker_;

  public ImpaladCatalog(String defaultKuduMasterHosts,
      AtomicReference<? extends AuthorizationChecker> authzChecker) {
    super();
    authzChecker_ = authzChecker;
    addDb(BuiltinsDb.getInstance());
    defaultKuduMasterHosts_ = defaultKuduMasterHosts;
    // Ensure the contents of the CatalogObjectVersionSet instance are cleared when a
    // new instance of ImpaladCatalog is created (see IMPALA-6486).
    CatalogObjectVersionSet.INSTANCE.clear();
  }

  /**
   * Utility class for sequencing the order in which a set of updated catalog objects
   * need to be applied to the catalog in order to satisfy referential constraints.
   *
   * If one type of object refers to another type of object, it needs to be added
   * after it and deleted before it.
   */
  public static class ObjectUpdateSequencer {
    private final ArrayDeque<TCatalogObject> updatedObjects = new ArrayDeque<>();
    private final ArrayDeque<TCatalogObject> deletedObjects = new ArrayDeque<>();

    public void add(TCatalogObject obj, boolean isDeleted) {
      if (!isDeleted) {
        // Update top-level objects first.
        if (isTopLevelCatalogObject(obj)) {
          updatedObjects.addFirst(obj);
        } else {
          updatedObjects.addLast(obj);
        }
      } else {
        // Remove low-level objects first.
        if (isTopLevelCatalogObject(obj)) {
          deletedObjects.addLast(obj);
        } else {
          deletedObjects.addFirst(obj);
        }
      }
    }

    public Iterable<TCatalogObject> getUpdatedObjects() { return updatedObjects; }
    public Iterable<TCatalogObject> getDeletedObjects() { return deletedObjects; }

    /**
     * Returns true if the given object does not depend on any other object already
     * existing in the catalog in order to be added.
     */
    private static boolean isTopLevelCatalogObject(TCatalogObject catalogObject) {
      return catalogObject.getType() == TCatalogObjectType.DATABASE ||
          catalogObject.getType() == TCatalogObjectType.DATA_SOURCE ||
          catalogObject.getType() == TCatalogObjectType.HDFS_CACHE_POOL ||
          catalogObject.getType() == TCatalogObjectType.PRINCIPAL ||
          catalogObject.getType() == TCatalogObjectType.AUTHZ_CACHE_INVALIDATION;
    }
  }

  /**
   * Update the catalog service Id. Trigger a full update if the service ID changes.
   */
  private void setCatalogServiceId(TUniqueId catalog_service_id) throws CatalogException {
    // Check for changes in the catalog service ID.
    if (!catalogServiceId_.equals(catalog_service_id)) {
      boolean firstRun = catalogServiceId_.equals(INITIAL_CATALOG_SERVICE_ID);
      catalogServiceId_ = catalog_service_id;
      if (!firstRun) {
        // Throw an exception which will trigger a full topic update request.
        throw new CatalogException("Detected catalog service ID change. Aborting " +
            "updateCatalog()");
      }
    }
  }

  /**
   * Updates the internal Catalog based on the given TCatalogUpdateReq.
   * This method:
   * 1) Calls NativeGetNextCatalogObjectUpdate() to get all the updates from the backend.
   * 2) Updates all top level objects (such as databases and roles).
   * 3) Updates all objects that depend on top level objects (such as functions, tables,
   *    privileges).
   * 4) Removes all dropped catalog objects.
   *
   * This method is called once per statestore heartbeat and is guaranteed the same
   * object will not be in both the "updated" list and the "removed" list (it is
   * a detail handled by the statestore).
   * Catalog objects are ordered by version, which is not necessarily the same as ordering
   * by dependency. This is handled by doing two passes and first updating the top level
   * objects, followed by updating the dependent objects. This method is synchronized
   * because updateCatalog() can be called by during a statestore update or during a
   * direct-DDL operation and catalogServiceId_ and lastSyncedCatalogVersion_ must be
   * protected.
   */
  public synchronized TUpdateCatalogCacheResponse updateCatalog(
    TUpdateCatalogCacheRequest req) throws CatalogException, TException {
    // For updates from catalog op results, the service ID is set in the request.
    if (req.isSetCatalog_service_id()) setCatalogServiceId(req.catalog_service_id);
    ObjectUpdateSequencer sequencer = new ObjectUpdateSequencer();
    long newCatalogVersion = lastSyncedCatalogVersion_.get();
    Pair<Boolean, ByteBuffer> update;
    while ((update = FeSupport.NativeGetNextCatalogObjectUpdate(req.native_iterator_ptr))
        != null) {
      TCatalogObject obj = new TCatalogObject();
      obj.read(new TBinaryProtocol(new TByteBuffer(update.second)));
      String key = Catalog.toCatalogObjectKey(obj);
      int len = update.second.capacity();
      if (len > 100 * 1024 * 1024 /* 100MB */) {
        LOG.info("Received large catalog object(>100mb): " + key + " is " + len +
            "bytes");
      }
      LOG.info((update.first ? "Deleting: " : "Adding: ") + key + " version: "
          + obj.catalog_version + " size: " + len);
      // For statestore updates, the service ID and updated version is wrapped in a
      // CATALOG catalog object.
      if (obj.type == TCatalogObjectType.CATALOG) {
        setCatalogServiceId(obj.catalog.catalog_service_id);
        newCatalogVersion = obj.catalog_version;
      } else {
        sequencer.add(obj, update.first);
      }
    }

    for (TCatalogObject catalogObject: sequencer.getUpdatedObjects()) {
      try {
        addCatalogObject(catalogObject);
      } catch (Exception e) {
        LOG.error("Error adding catalog object: " + e.getMessage(), e);
      }
    }

    for (TCatalogObject catalogObject: sequencer.getDeletedObjects()) {
      removeCatalogObject(catalogObject);
    }

    lastSyncedCatalogVersion_.set(newCatalogVersion);
    // Cleanup old entries in the log.
    catalogDeltaLog_.garbageCollect(newCatalogVersion);
    // Notify all the threads waiting on a catalog update.
    synchronized (catalogUpdateEventNotifier_) {
      catalogUpdateEventNotifier_.notifyAll();
    }
    return new TUpdateCatalogCacheResponse(catalogServiceId_,
        CatalogObjectVersionSet.INSTANCE.getMinimumVersion(), newCatalogVersion);
  }


  @Override // FeCatalog
  public void prioritizeLoad(Set<TableName> tableNames) throws InternalException {
    FeSupport.PrioritizeLoad(tableNames);
  }

  @Override // FeCatalog
  public TGetPartitionStatsResponse getPartitionStats(
      TableName table) throws InternalException {
    return FeSupport.GetPartitionStats(table);
  }

  @Override // FeCatalog
  public void waitForCatalogUpdate(long timeoutMs) {
    synchronized (catalogUpdateEventNotifier_) {
      try {
        catalogUpdateEventNotifier_.wait(timeoutMs);
      } catch (InterruptedException e) {
        // Ignore
      }
    }
  }

  /**
   *  Adds the given TCatalogObject to the catalog cache. The update may be ignored
   *  (considered out of date) if:
   *  1) An item exists in the catalog cache with a version > than the given
   *     TCatalogObject's version.
   *  2) The catalogDeltaLog_ contains an entry for this object with a version
   *     > than the given TCatalogObject's version.
   */
  private void addCatalogObject(TCatalogObject catalogObject)
      throws TableLoadingException {
    // This item is out of date and should not be applied to the catalog.
    if (catalogDeltaLog_.wasObjectRemovedAfter(catalogObject)) {
      if (LOG.isTraceEnabled()) {
        LOG.trace(String.format("Skipping update because a matching object was removed " +
            "in a later catalog version: %s", catalogObject));
      }
      return;
    }

    switch(catalogObject.getType()) {
      case DATABASE:
        addDb(catalogObject.getDb(), catalogObject.getCatalog_version());
        break;
      case TABLE:
      case VIEW:
        addTable(catalogObject.getTable(), catalogObject.getCatalog_version());
        break;
      case FUNCTION:
        // Remove the function first, in case there is an existing function with the same
        // name and signature.
        removeFunction(catalogObject.getFn(), catalogObject.getCatalog_version());
        addFunction(catalogObject.getFn(), catalogObject.getCatalog_version());
        break;
      case DATA_SOURCE:
        addDataSource(catalogObject.getData_source(), catalogObject.getCatalog_version());
        break;
      case PRINCIPAL:
        Principal principal = Principal.fromThrift(catalogObject.getPrincipal());
        principal.setCatalogVersion(catalogObject.getCatalog_version());
        authPolicy_.addPrincipal(principal);
        break;
      case PRIVILEGE:
        PrincipalPrivilege privilege =
            PrincipalPrivilege.fromThrift(catalogObject.getPrivilege());
        privilege.setCatalogVersion(catalogObject.getCatalog_version());
        try {
          authPolicy_.addPrivilege(privilege);
        } catch (CatalogException e) {
          LOG.error("Error adding privilege: ", e);
        }
        break;
      case HDFS_CACHE_POOL:
        HdfsCachePool cachePool = new HdfsCachePool(catalogObject.getCache_pool());
        cachePool.setCatalogVersion(catalogObject.getCatalog_version());
        hdfsCachePools_.add(cachePool);
        break;
      case AUTHZ_CACHE_INVALIDATION:
        AuthzCacheInvalidation authzCacheInvalidation = new AuthzCacheInvalidation(
            catalogObject.getAuthz_cache_invalidation());
        authzCacheInvalidation.setCatalogVersion(catalogObject.getCatalog_version());
        authzCacheInvalidation_.add(authzCacheInvalidation);
        authzChecker_.get().invalidateAuthorizationCache();
        break;
      default:
        throw new IllegalStateException(
            "Unexpected TCatalogObjectType: " + catalogObject.getType());
    }
  }

  /**
   *  Removes the matching TCatalogObject from the catalog, if one exists and its
   *  catalog version is < the catalog version of this drop operation.
   */
  private void removeCatalogObject(TCatalogObject catalogObject) {
    Preconditions.checkState(catalogObject.getCatalog_version() != 0);
    long dropCatalogVersion = catalogObject.getCatalog_version();
    switch(catalogObject.getType()) {
      case DATABASE:
        removeDb(catalogObject.getDb(), dropCatalogVersion);
        break;
      case TABLE:
      case VIEW:
        removeTable(catalogObject.getTable(), dropCatalogVersion);
        break;
      case FUNCTION:
        removeFunction(catalogObject.getFn(), dropCatalogVersion);
        break;
      case DATA_SOURCE:
        removeDataSource(catalogObject.getData_source(), dropCatalogVersion);
        break;
      case PRINCIPAL:
        authPolicy_.removePrincipalIfLowerVersion(catalogObject.getPrincipal(),
            dropCatalogVersion);
        break;
      case PRIVILEGE:
        authPolicy_.removePrivilegeIfLowerVersion(catalogObject.getPrivilege(),
            dropCatalogVersion);
        break;
      case HDFS_CACHE_POOL:
        HdfsCachePool existingItem =
            hdfsCachePools_.get(catalogObject.getCache_pool().getPool_name());
        if (existingItem.getCatalogVersion() <= catalogObject.getCatalog_version()) {
          hdfsCachePools_.remove(catalogObject.getCache_pool().getPool_name());
        }
        break;
      case AUTHZ_CACHE_INVALIDATION:
        removeAuthzCacheInvalidation(catalogObject.getAuthz_cache_invalidation(),
            dropCatalogVersion);
        break;
      default:
        throw new IllegalStateException(
            "Unexpected TCatalogObjectType: " + catalogObject.getType());
    }

    if (catalogObject.getCatalog_version() > lastSyncedCatalogVersion_.get()) {
      catalogDeltaLog_.addRemovedObject(catalogObject);
    }
  }

  private void addDb(TDatabase thriftDb, long catalogVersion) {
    Db existingDb = getDb(thriftDb.getDb_name());
    if (existingDb == null ||
        existingDb.getCatalogVersion() < catalogVersion) {
      Db newDb = Db.fromTDatabase(thriftDb);
      newDb.setCatalogVersion(catalogVersion);
      if (existingDb != null) {
        CatalogObjectVersionSet.INSTANCE.updateVersions(
            existingDb.getCatalogVersion(), catalogVersion);
        CatalogObjectVersionSet.INSTANCE.removeAll(existingDb.getTables());
        CatalogObjectVersionSet.INSTANCE.removeAll(
            existingDb.getFunctions(null, new PatternMatcher()));
        // IMPALA-8434: add back the existing tables/functions. Note that their version
        // counters in CatalogObjectVersionSet have been decreased by the above removeAll
        // statements, meaning their references from the old db are deleted since the old
        // db object has been replaced by newDb. addTable and addFunction will add their
        // versions back.
        for (Table tbl: existingDb.getTables()) {
          newDb.addTable(tbl);
        }
        for (List<Function> functionList: existingDb.getAllFunctions().values()) {
          for (Function func: functionList) {
            newDb.addFunction(func);
          }
        }
      } else {
        CatalogObjectVersionSet.INSTANCE.addVersion(catalogVersion);
      }
      addDb(newDb);
    }
  }

  private void addTable(TTable thriftTable, long catalogVersion)
      throws TableLoadingException {
    Db db = getDb(thriftTable.db_name);
    if (db == null) {
      if (LOG.isTraceEnabled()) {
        LOG.trace("Parent database of table does not exist: " +
            thriftTable.db_name + "." + thriftTable.tbl_name);
      }
      return;
    }

    Table newTable = Table.fromThrift(db, thriftTable);
    newTable.setCatalogVersion(catalogVersion);
    db.addTable(newTable);
  }

  private void addFunction(TFunction fn, long catalogVersion) {
    LibCacheSetNeedsRefresh(fn.hdfs_location);
    Function function = Function.fromThrift(fn);
    function.setCatalogVersion(catalogVersion);
    Db db = getDb(function.getFunctionName().getDb());
    if (db == null) {
      if (LOG.isTraceEnabled()) {
        LOG.trace("Parent database of function does not exist: " + function.getName());
      }
      return;
    }
    Function existingFn = db.getFunction(fn.getSignature());
    if (existingFn == null ||
        existingFn.getCatalogVersion() < catalogVersion) {
      db.addFunction(function);
      if (existingFn != null) {
        CatalogObjectVersionSet.INSTANCE.updateVersions(
            existingFn.getCatalogVersion(), catalogVersion);
      } else {
        CatalogObjectVersionSet.INSTANCE.addVersion(catalogVersion);
      }
    }
  }

  private void addDataSource(TDataSource thrift, long catalogVersion) {
    LibCacheSetNeedsRefresh(thrift.hdfs_location);
    DataSource dataSource = DataSource.fromThrift(thrift);
    dataSource.setCatalogVersion(catalogVersion);
    addDataSource(dataSource);
  }

  private void removeDataSource(TDataSource thrift, long dropCatalogVersion) {
    DataSource src = dataSources_.get(thrift.name);
    if (src != null && src.getCatalogVersion() < dropCatalogVersion) {
      LibCacheRemoveEntry(src.getLocation());
    }
    removeDataSource(thrift.getName());
  }

  private void removeDb(TDatabase thriftDb, long dropCatalogVersion) {
    Db db = getDb(thriftDb.getDb_name());
    if (db != null && db.getCatalogVersion() < dropCatalogVersion) {
      removeDb(db.getName());
      CatalogObjectVersionSet.INSTANCE.removeVersion(
          db.getCatalogVersion());
      CatalogObjectVersionSet.INSTANCE.removeAll(db.getTables());
      CatalogObjectVersionSet.INSTANCE.removeAll(
          db.getFunctions(null, new PatternMatcher()));
    }
  }

  private void removeTable(TTable thriftTable, long dropCatalogVersion) {
    Db db = getDb(thriftTable.db_name);
    // The parent database doesn't exist, nothing to do.
    if (db == null) return;

    Table table = db.getTable(thriftTable.getTbl_name());
    if (table != null && table.getCatalogVersion() < dropCatalogVersion) {
      db.removeTable(thriftTable.tbl_name);
    }
  }

  private void removeFunction(TFunction thriftFn, long dropCatalogVersion) {
    Db db = getDb(thriftFn.name.getDb_name());
    // The parent database doesn't exist, nothing to do.
    if (db == null) return;

    // If the function exists and it has a catalog version less than the
    // version of the drop, remove the function.
    Function fn = db.getFunction(thriftFn.getSignature());
    if (fn != null && fn.getCatalogVersion() < dropCatalogVersion) {
      LibCacheRemoveEntry(fn.getLocation().getLocation());
      db.removeFunction(thriftFn.getSignature());
      CatalogObjectVersionSet.INSTANCE.removeVersion(
          fn.getCatalogVersion());
    }
  }

  private void removeAuthzCacheInvalidation(
      TAuthzCacheInvalidation authzCacheInvalidation, long dropCatalogVersion) {
    AuthzCacheInvalidation existingItem = authzCacheInvalidation_.get(
        authzCacheInvalidation.getMarker_name());
    if (existingItem != null && existingItem.getCatalogVersion() < dropCatalogVersion) {
      authzCacheInvalidation_.remove(authzCacheInvalidation.getMarker_name());
    }
  }

  @Override // FeCatalog
  public boolean isReady() {
    return lastSyncedCatalogVersion_.get() > INITIAL_CATALOG_VERSION;
  }

  // Only used for testing.
  @Override // FeCatalog
  public void setIsReady(boolean isReady) {
    lastSyncedCatalogVersion_.incrementAndGet();
    synchronized (catalogUpdateEventNotifier_) {
      catalogUpdateEventNotifier_.notifyAll();
    }
  }
  @Override // FeCatalog
  public AuthorizationPolicy getAuthPolicy() { return authPolicy_; }
  @Override // FeCatalog
  public String getDefaultKuduMasterHosts() { return defaultKuduMasterHosts_; }

  private void LibCacheSetNeedsRefresh(String hdfsLocation) {
    if (!FeSupport.NativeLibCacheSetNeedsRefresh(hdfsLocation)) {
      LOG.error("NativeLibCacheSetNeedsRefresh(" + hdfsLocation + ") failed.");
    }
  }
  private void LibCacheRemoveEntry(String hdfsLibFile) {
    if (!FeSupport.NativeLibCacheRemoveEntry(hdfsLibFile)) {
      LOG.error("LibCacheRemoveEntry(" + hdfsLibFile + ") failed.");
    }
  }

  @Override
  public TUniqueId getCatalogServiceId() { return catalogServiceId_; }
}