HdfsCachingUtil.java

Go to the documentation of this file.

// Copyright 2012 Cloudera Inc.
//
// Licensed 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 com.cloudera.impala.util;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.CacheDirectiveEntry;
import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo;
import org.apache.hadoop.hdfs.protocol.CacheDirectiveInfo.Expiration;
import org.apache.hadoop.ipc.RemoteException;
import org.apache.log4j.Logger;

import com.cloudera.impala.analysis.TableName;
import com.cloudera.impala.catalog.HdfsPartition;
import com.cloudera.impala.common.FileSystemUtil;
import com.cloudera.impala.common.ImpalaException;
import com.cloudera.impala.common.ImpalaRuntimeException;
import com.cloudera.impala.thrift.JniCatalogConstants;
import com.cloudera.impala.thrift.THdfsCachingOp;
import com.google.common.base.Preconditions;

public class HdfsCachingUtil {
  private static final Logger LOG = Logger.getLogger(HdfsCachingUtil.class);

  // The key name used to save cache directive IDs in table/partition properties.
  private final static String CACHE_DIR_ID_PROP_NAME = "cache_directive_id";

  // The key name used to store the replication factor for cached files
  private final static String CACHE_DIR_REPLICATION_PROP_NAME = "cache_replication";

  // The number of caching refresh intervals that can go by when waiting for data to
  // become cached before assuming no more progress is being made.
  private final static int MAX_UNCHANGED_CACHING_REFRESH_INTERVALS = 5;

  private final static DistributedFileSystem dfs;
  static {
    try {
      dfs = FileSystemUtil.getDistributedFileSystem();
    } catch (IOException e) {
      throw new RuntimeException("HdfsCachingUtil failed to initialize the " +
          "DistributedFileSystem: ", e);
    }
  }

  public static long submitCacheTblDirective(
      org.apache.hadoop.hive.metastore.api.Table table,
      String poolName, short replication) throws ImpalaRuntimeException {
    long id = HdfsCachingUtil.submitDirective(new Path(table.getSd().getLocation()),
        poolName, replication);
    table.putToParameters(CACHE_DIR_ID_PROP_NAME, Long.toString(id));
    table.putToParameters(CACHE_DIR_REPLICATION_PROP_NAME, Long.toString(replication));
    return id;
  }

  public static long submitCachePartitionDirective(HdfsPartition part,
      String poolName, short replication) throws ImpalaRuntimeException {
    long id = HdfsCachingUtil.submitDirective(new Path(part.getLocation()),
        poolName, replication);
    part.putToParameters(CACHE_DIR_ID_PROP_NAME, Long.toString(id));
    part.putToParameters(CACHE_DIR_REPLICATION_PROP_NAME, Long.toString(replication));
    return id;
  }

  public static long submitCachePartitionDirective(
      org.apache.hadoop.hive.metastore.api.Partition part,
      String poolName, short replication) throws ImpalaRuntimeException {
    long id = HdfsCachingUtil.submitDirective(new Path(part.getSd().getLocation()),
        poolName, replication);
    part.putToParameters(CACHE_DIR_ID_PROP_NAME, Long.toString(id));
    part.putToParameters(CACHE_DIR_REPLICATION_PROP_NAME, Long.toString(replication));
    return id;
  }

  public static void uncacheTbl(org.apache.hadoop.hive.metastore.api.Table table)
      throws ImpalaRuntimeException {
    Preconditions.checkNotNull(table);
    LOG.debug("Uncaching table: " + table.getDbName() + "." + table.getTableName());
    Long id = getCacheDirectiveId(table.getParameters());
    if (id == null) return;
    HdfsCachingUtil.removeDirective(id);
    table.getParameters().remove(CACHE_DIR_ID_PROP_NAME);
    table.getParameters().remove(CACHE_DIR_REPLICATION_PROP_NAME);
  }

  public static void uncachePartition(HdfsPartition part) throws ImpalaException {
    Preconditions.checkNotNull(part);
    Long id = getCacheDirectiveId(part.getParameters());
    if (id == null) return;
    HdfsCachingUtil.removeDirective(id);
    part.getParameters().remove(CACHE_DIR_ID_PROP_NAME);
    part.getParameters().remove(CACHE_DIR_REPLICATION_PROP_NAME);
  }

  public static void uncachePartition(
    org.apache.hadoop.hive.metastore.api.Partition part) throws ImpalaException {
    Preconditions.checkNotNull(part);
    Long id = getCacheDirectiveId(part.getParameters());
    if (id == null) return;
    HdfsCachingUtil.removeDirective(id);
    part.getParameters().remove(CACHE_DIR_ID_PROP_NAME);
    part.getParameters().remove(CACHE_DIR_REPLICATION_PROP_NAME);
  }

  public static Long getCacheDirectiveId(Map<String, String> params) {
    if (params == null) return null;
    String idStr = params.get(CACHE_DIR_ID_PROP_NAME);
    if (idStr == null) return null;
    try {
      return Long.parseLong(idStr);
    } catch (NumberFormatException e) {
      return null;
    }
  }

  public static String getCachePool(long directiveId)
      throws ImpalaRuntimeException {
    CacheDirectiveEntry entry = getDirective(directiveId);
    return entry == null ? null : entry.getInfo().getPool();
  }

  public static Short getCacheReplication(long directiveId)
      throws ImpalaRuntimeException {
    CacheDirectiveEntry entry = getDirective(directiveId);
    return entry != null ? entry.getInfo().getReplication() : null;
  }

  public static Short getCachedCacheReplication(Map<String, String> params) {
    Preconditions.checkNotNull(params);
    String replication = params.get(CACHE_DIR_REPLICATION_PROP_NAME);
    if (replication == null) {
      return JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR;
    }
    try {
      return Short.parseShort(replication);
    } catch (NumberFormatException e) {
      return JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR;
    }
  }

  public static void waitForDirective(long directiveId)
      throws ImpalaRuntimeException  {
    long bytesNeeded = 0L;
    long currentBytesCached = 0L;
    CacheDirectiveEntry cacheDir = getDirective(directiveId);
    if (cacheDir == null) return;

    bytesNeeded = cacheDir.getStats().getBytesNeeded();
    currentBytesCached = cacheDir.getStats().getBytesCached();
    LOG.debug(String.format("Waiting on cache directive id: %d. Bytes " +
        "cached (%d) / needed (%d)", directiveId, currentBytesCached, bytesNeeded));
    // All the bytes are cached, just return.
    if (bytesNeeded == currentBytesCached) return;

    // The refresh interval is how often HDFS will update cache directive stats. We use
    // this value to determine how frequently we should poll for changes.
    long hdfsRefreshIntervalMs = dfs.getConf().getLong(
        DFSConfigKeys.DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS,
        DFSConfigKeys.DFS_NAMENODE_PATH_BASED_CACHE_REFRESH_INTERVAL_MS_DEFAULT);
    Preconditions.checkState(hdfsRefreshIntervalMs > 0);

    // Loop until either MAX_UNCHANGED_CACHING_REFRESH_INTERVALS have passed with no
    // changes or all required data is cached.
    int unchangedCounter = 0;
    while (unchangedCounter < MAX_UNCHANGED_CACHING_REFRESH_INTERVALS) {
      long previousBytesCached = currentBytesCached;
      cacheDir = getDirective(directiveId);
      if (cacheDir == null) return;
      currentBytesCached = cacheDir.getStats().getBytesCached();
      bytesNeeded = cacheDir.getStats().getBytesNeeded();
      if (currentBytesCached == bytesNeeded) {
        LOG.debug(String.format("Cache directive id: %d has completed." +
            "Bytes cached (%d) / needed (%d)", directiveId, currentBytesCached,
            bytesNeeded));
        return;
      }

      if (currentBytesCached == previousBytesCached) {
        ++unchangedCounter;
      } else {
        unchangedCounter = 0;
      }
      try {
        // Sleep for the refresh interval + a little bit more to ensure a full interval
        // has completed. A value of 25% the refresh interval was arbitrarily chosen.
        Thread.sleep((long) (hdfsRefreshIntervalMs * 1.25));
      } catch (InterruptedException e) { /* ignore */ }
    }
    LOG.warn(String.format("No changes in cached bytes in: %d(ms). All data may not " +
        "be cached. Final stats for cache directive id: %d. Bytes cached (%d)/needed " +
        "(%d)", hdfsRefreshIntervalMs * MAX_UNCHANGED_CACHING_REFRESH_INTERVALS,
        directiveId, currentBytesCached, bytesNeeded));
  }

  private static long submitDirective(Path path, String poolName, short replication)
      throws ImpalaRuntimeException {
    Preconditions.checkNotNull(path);
    Preconditions.checkState(poolName != null && !poolName.isEmpty());
    CacheDirectiveInfo info = new CacheDirectiveInfo.Builder()
        .setExpiration(Expiration.NEVER)
        .setPool(poolName)
        .setReplication(replication)
        .setPath(path).build();
    LOG.debug("Submitting cache directive: " + info.toString());
    try {
      return dfs.addCacheDirective(info);
    } catch (IOException e) {
      throw new ImpalaRuntimeException(e.getMessage(), e);
    }
  }

  public static long modifyCacheDirective(Long id,
      org.apache.hadoop.hive.metastore.api.Table table,
      String poolName, short replication) throws ImpalaRuntimeException {
    Preconditions.checkNotNull(id);
    HdfsCachingUtil.modifyCacheDirective(id, new Path(table.getSd().getLocation()),
        poolName, replication);
    table.putToParameters(CACHE_DIR_ID_PROP_NAME, Long.toString(id));
    table.putToParameters(CACHE_DIR_REPLICATION_PROP_NAME, Long.toString(replication));
    return id;
  }

  public static long modifyCacheDirective(Long id, HdfsPartition part, String poolName,
      short replication) throws ImpalaRuntimeException {
    Preconditions.checkNotNull(id);
    HdfsCachingUtil.modifyCacheDirective(id, new Path(part.getLocation()),
        poolName, replication);
    part.putToParameters(CACHE_DIR_ID_PROP_NAME, Long.toString(id));
    part.putToParameters(CACHE_DIR_REPLICATION_PROP_NAME, Long.toString(replication));
    return id;
  }

  private static void modifyCacheDirective(Long id, Path path, String poolName,
      short replication) throws ImpalaRuntimeException {
    Preconditions.checkNotNull(path);
    Preconditions.checkNotNull(id);
    Preconditions.checkState(poolName != null && !poolName.isEmpty());
    CacheDirectiveInfo info = new CacheDirectiveInfo.Builder()
        .setId(id)
        .setExpiration(Expiration.NEVER)
        .setPool(poolName)
        .setReplication(replication)
        .setPath(path).build();
    LOG.debug("Modifying cache directive: " + info.toString());
    try {
      dfs.modifyCacheDirective(info);
    } catch (IOException e) {
      throw new ImpalaRuntimeException(e.getMessage(), e);
    }
  }

  private static void removeDirective(long directiveId) throws ImpalaRuntimeException {
    LOG.debug("Removing cache directive id: " + directiveId);
    try {
      dfs.removeCacheDirective(directiveId);
    } catch (IOException e) {
      // There is no special exception type for the case where a directive ID does not
      // exist so we must inspect the error message.
      if (e.getMessage().contains("No directive with ID")) return;
      throw new ImpalaRuntimeException(e.getMessage(), e);
    }
  }

  private static CacheDirectiveEntry getDirective(long directiveId)
      throws ImpalaRuntimeException {
    LOG.trace("Getting cache directive id: " + directiveId);
    CacheDirectiveInfo filter = new CacheDirectiveInfo.Builder()
        .setId(directiveId)
        .build();
    try {
      RemoteIterator<CacheDirectiveEntry> itr = dfs.listCacheDirectives(filter);
      if (itr.hasNext()) return itr.next();
    } catch (IOException e) {
      // Handle connection issues with e.g. HDFS and possible not found errors
      throw new ImpalaRuntimeException(e.getMessage(), e);
    }
    throw new ImpalaRuntimeException(
        "HDFS cache directive filter returned empty result. This must not happen");
  }

  public static boolean isSamePool(String poolName, Long directiveId)
      throws ImpalaRuntimeException {
    return poolName.equals(getCachePool(directiveId));
  }

  public static short getReplicationOrDefault(THdfsCachingOp op) {
    return op.isSetReplication() ? op.getReplication() :
      JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR;
  }

  public static boolean isUpdateOp(THdfsCachingOp op, Map<String, String> params)
      throws ImpalaRuntimeException {

    Long directiveId = Long.parseLong(params.get(CACHE_DIR_ID_PROP_NAME));
    CacheDirectiveEntry entry = getDirective(directiveId);
    Preconditions.checkNotNull(entry);

    // Verify cache pool
    if (!op.getCache_pool_name().equals(entry.getInfo().getPool())) {
      return false;
    }

    // Check cache replication factor
    if ((op.isSetReplication() && op.getReplication() !=
        entry.getInfo().getReplication()) || ( !op.isSetReplication() &&
        entry.getInfo().getReplication() !=
        JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR)) {
      return true;
    }
    return false;
  }

  public static void validateCachePool(THdfsCachingOp op, Long directiveId,
      TableName table, HdfsPartition partition)
      throws ImpalaRuntimeException {

    CacheDirectiveEntry entry = getDirective(directiveId);
    Preconditions.checkNotNull(entry);

    if (!op.getCache_pool_name().equals(entry.getInfo().getPool())) {
      throw new ImpalaRuntimeException(String.format("Cannot cache partition in " +
          "pool '%s' because it is already cached in '%s'. To change the cache " +
          "pool for this partition, first uncache using: ALTER TABLE %s.%s " +
          "%sSET UNCACHED", op.getCache_pool_name(),
          entry.getInfo().getPool(), table.getDb(), table,
          // Insert partition string if partition non null
          partition != null ? String.format(" PARTITION(%s) ",
          partition.getPartitionName().replaceAll("/", ", ")) : ""));
    }
  }

  public static void validateCachePool(THdfsCachingOp op, Long directiveId,
      TableName table) throws ImpalaRuntimeException {
    validateCachePool(op, directiveId, table, null);
  }

  public static boolean validateCacheParams(Map<String, String> params) {
    Long directiveId = getCacheDirectiveId(params);
    if (directiveId == null) return false;

    CacheDirectiveEntry entry = null;
    try {
      entry = getDirective(directiveId);
    } catch (ImpalaRuntimeException e) {
      if (e.getCause() != null && e.getCause() instanceof RemoteException) {
        // This exception signals that the cache directive no longer exists.
        LOG.error("Cache directive does not exist", e);
        params.remove(CACHE_DIR_ID_PROP_NAME);
        params.remove(CACHE_DIR_REPLICATION_PROP_NAME);
      } else {
        // This exception signals that there was a connection problem with HDFS.
        LOG.error("IO Exception, possible connectivity issues with HDFS", e);
      }
      return false;
    }
    Preconditions.checkNotNull(entry);

    // On the upgrade path the property might not exist, if it exists
    // and is different from the one from the meta store, issue a warning.
    String replicationFactor = params.get(CACHE_DIR_REPLICATION_PROP_NAME);
    if (replicationFactor != null &&
        Short.parseShort(replicationFactor) != entry.getInfo().getReplication()) {
      LOG.info("Replication factor for entry in HDFS differs from value in Hive MS: " +
          entry.getInfo().getPath().toString() + " " +
          entry.getInfo().getReplication().toString() + " != " +
          params.get(CACHE_DIR_REPLICATION_PROP_NAME));
    }
    params.put(CACHE_DIR_REPLICATION_PROP_NAME,
        String.valueOf(entry.getInfo().getReplication()));
    return true;
  }
}