Catalog.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.catalog;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.log4j.Logger;

import com.cloudera.impala.analysis.FunctionName;
import com.cloudera.impala.catalog.MetaStoreClientPool.MetaStoreClient;
import com.cloudera.impala.thrift.TCatalogObject;
import com.cloudera.impala.thrift.TFunction;
import com.cloudera.impala.thrift.TPartitionKeyValue;
import com.cloudera.impala.thrift.TTableName;
import com.cloudera.impala.util.PatternMatcher;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;

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

  // Initial catalog version.
  public final static long INITIAL_CATALOG_VERSION = 0L;
  public static final String DEFAULT_DB = "default";
  private static final int META_STORE_CLIENT_POOL_SIZE = 5;

  public static final String BUILTINS_DB = "_impala_builtins";

  protected final MetaStoreClientPool metaStoreClientPool_ = new MetaStoreClientPool(0);

  // Cache of authorization policy metadata. Populated from data retried from the
  // Sentry Service, if configured.
  protected AuthorizationPolicy authPolicy_ = new AuthorizationPolicy();

  // Thread safe cache of database metadata. Uses an AtomicReference so reset()
  // operations can atomically swap dbCache_ references.
  // TODO: Update this to use a CatalogObjectCache?
  protected AtomicReference<ConcurrentHashMap<String, Db>> dbCache_ =
      new AtomicReference<ConcurrentHashMap<String, Db>>(
          new ConcurrentHashMap<String, Db>());

  // DB that contains all builtins
  private static Db builtinsDb_;

  // Cache of data sources.
  protected final CatalogObjectCache<DataSource> dataSources_;

  // Cache of known HDFS cache pools. Allows for checking the existence
  // of pools without hitting HDFS.
  protected final CatalogObjectCache<HdfsCachePool> hdfsCachePools_ =
      new CatalogObjectCache<HdfsCachePool>(false);

  public Catalog(boolean initMetastoreClientPool) {
    if (initMetastoreClientPool) {
      metaStoreClientPool_.addClients(META_STORE_CLIENT_POOL_SIZE);
    }
    dataSources_ = new CatalogObjectCache<DataSource>();
    builtinsDb_ = new BuiltinsDb(BUILTINS_DB, this);
    addDb(builtinsDb_);
  }

  public Db getBuiltinsDb() { return builtinsDb_; }

  public Db addDb(Db db) {
    return dbCache_.get().put(db.getName().toLowerCase(), db);
  }

  public Db getDb(String dbName) {
    Preconditions.checkState(dbName != null && !dbName.isEmpty(),
        "Null or empty database name given as argument to Catalog.getDb");
    return dbCache_.get().get(dbName.toLowerCase());
  }

  public Db removeDb(String dbName) {
    return dbCache_.get().remove(dbName.toLowerCase());
  }

  public List<String> getDbNames(String dbPattern) {
    return filterStringsByPattern(dbCache_.get().keySet(), dbPattern);
  }

  public Table getTable(String dbName, String tableName) throws
      CatalogException {
    Db db = getDb(dbName);
    if (db == null) {
      throw new DatabaseNotFoundException("Database '" + dbName + "' not found");
    }
    return db.getTable(tableName);
  }

  public Table removeTable(TTableName tableName) {
    // Remove the old table name from the cache and add the new table.
    Db db = getDb(tableName.getDb_name());
    if (db == null) return null;
    return db.removeTable(tableName.getTable_name());
  }

  public List<String> getTableNames(String dbName, String tablePattern)
      throws DatabaseNotFoundException {
    Preconditions.checkNotNull(dbName);
    Db db = getDb(dbName);
    if (db == null) {
      throw new DatabaseNotFoundException("Database '" + dbName + "' not found");
    }
    return filterStringsByPattern(db.getAllTableNames(), tablePattern);
  }

  public boolean containsTable(String dbName, String tableName) {
    Db db = getDb(dbName);
    return (db == null) ? false : db.containsTable(tableName);
  }

  public boolean addDataSource(DataSource dataSource) {
    return dataSources_.add(dataSource);
  }

  public DataSource removeDataSource(String dataSourceName) {
    Preconditions.checkNotNull(dataSourceName);
    return dataSources_.remove(dataSourceName.toLowerCase());
  }

  public DataSource getDataSource(String dataSourceName) {
    Preconditions.checkNotNull(dataSourceName);
    return dataSources_.get(dataSourceName.toLowerCase());
  }

  public List<DataSource> getDataSources() {
    return dataSources_.getValues();
  }

  public List<String> getDataSourceNames(String pattern) {
    return filterStringsByPattern(dataSources_.keySet(), pattern);
  }

  public List<DataSource> getDataSources(String pattern) {
    List<String> names = filterStringsByPattern(dataSources_.keySet(), pattern);
    List<DataSource> dataSources = Lists.newArrayListWithCapacity(names.size());
    for (String name: names) {
      dataSources.add(dataSources_.get(name));
    }
    return dataSources;
  }

  public boolean addFunction(Function fn) {
    Db db = getDb(fn.dbName());
    if (db == null) return false;
    return db.addFunction(fn);
  }

  public Function getFunction(Function desc, Function.CompareMode mode) {
    Db db = getDb(desc.dbName());
    if (db == null) return null;
    return db.getFunction(desc, mode);
  }

  public static Function getBuiltin(Function desc, Function.CompareMode mode) {
    return builtinsDb_.getFunction(desc, mode);
  }

  public Function removeFunction(Function desc) {
    Db db = getDb(desc.dbName());
    if (db == null) return null;
    return db.removeFunction(desc);
  }

  public boolean containsFunction(FunctionName name) {
    Db db = getDb(name.getDb());
    if (db == null) return false;
    return db.containsFunction(name.getFunction());
  }

  public boolean addHdfsCachePool(HdfsCachePool cachePool) {
    return hdfsCachePools_.add(cachePool);
  }

  public HdfsCachePool getHdfsCachePool(String poolName) {
    return hdfsCachePools_.get(poolName);
  }

  public void close() { metaStoreClientPool_.close(); }


  public MetaStoreClient getMetaStoreClient() { return metaStoreClientPool_.getClient(); }

  private List<String> filterStringsByPattern(Iterable<String> candidates,
      String matchPattern) {
    List<String> filtered = Lists.newArrayList();
    if (matchPattern == null) {
      filtered = Lists.newArrayList(candidates);
    } else {
      PatternMatcher matcher = PatternMatcher.createHivePatternMatcher(matchPattern);
      for (String candidate: candidates) {
        if (matcher.matches(candidate)) filtered.add(candidate);
      }
    }
    Collections.sort(filtered, String.CASE_INSENSITIVE_ORDER);
    return filtered;
  }

  public HdfsPartition getHdfsPartition(String dbName, String tableName,
      List<TPartitionKeyValue> partitionSpec) throws CatalogException {
    String partitionNotFoundMsg =
        "Partition not found: " + Joiner.on(", ").join(partitionSpec);
    Table table = getTable(dbName, tableName);
    // This is not an Hdfs table, throw an error.
    if (!(table instanceof HdfsTable)) {
      throw new PartitionNotFoundException(partitionNotFoundMsg);
    }
    // Get the HdfsPartition object for the given partition spec.
    HdfsPartition partition =
        ((HdfsTable) table).getPartitionFromThriftPartitionSpec(partitionSpec);
    if (partition == null) throw new PartitionNotFoundException(partitionNotFoundMsg);
    return partition;
  }

  public boolean containsHdfsPartition(String dbName, String tableName,
      List<TPartitionKeyValue> partitionSpec) throws CatalogException {
    try {
      return getHdfsPartition(dbName, tableName, partitionSpec) != null;
    } catch (PartitionNotFoundException e) {
      return false;
    }
  }

  public TCatalogObject getTCatalogObject(TCatalogObject objectDesc)
      throws CatalogException {
    TCatalogObject result = new TCatalogObject();
    switch (objectDesc.getType()) {
      case DATABASE: {
        Db db = getDb(objectDesc.getDb().getDb_name());
        if (db == null) {
          throw new CatalogException(
              "Database not found: " + objectDesc.getDb().getDb_name());
        }
        result.setType(db.getCatalogObjectType());
        result.setCatalog_version(db.getCatalogVersion());
        result.setDb(db.toThrift());
        break;
      }
      case TABLE:
      case VIEW: {
        Table table = getTable(objectDesc.getTable().getDb_name(),
            objectDesc.getTable().getTbl_name());
        if (table == null) {
          throw new CatalogException("Table not found: " +
              objectDesc.getTable().getTbl_name());
        }
        result.setType(table.getCatalogObjectType());
        result.setCatalog_version(table.getCatalogVersion());
        result.setTable(table.toThrift());
        break;
      }
      case FUNCTION: {
        TFunction tfn = objectDesc.getFn();
        Function desc = Function.fromThrift(tfn);
        Function fn = getFunction(desc, Function.CompareMode.IS_INDISTINGUISHABLE);
        if (fn == null) {
          throw new CatalogException("Function not found: " + tfn);
        }
        result.setType(fn.getCatalogObjectType());
        result.setCatalog_version(fn.getCatalogVersion());
        result.setFn(fn.toThrift());
        break;
      }
      case DATA_SOURCE: {
        String dataSrcName = objectDesc.getData_source().getName();
        DataSource dataSrc = getDataSource(dataSrcName);
        if (dataSrc == null) {
          throw new CatalogException("Data source not found: " + dataSrcName);
        }
        result.setType(dataSrc.getCatalogObjectType());
        result.setCatalog_version(dataSrc.getCatalogVersion());
        result.setData_source(dataSrc.toThrift());
        break;
      }
      case HDFS_CACHE_POOL: {
        HdfsCachePool pool = getHdfsCachePool(objectDesc.getCache_pool().getPool_name());
        if (pool == null) {
          throw new CatalogException(
              "Hdfs cache pool not found: " + objectDesc.getCache_pool().getPool_name());
        }
        result.setType(pool.getCatalogObjectType());
        result.setCatalog_version(pool.getCatalogVersion());
        result.setCache_pool(pool.toThrift());
        break;
      }
      case ROLE:
        Role role = authPolicy_.getRole(objectDesc.getRole().getRole_name());
        if (role == null) {
          throw new CatalogException("Role not found: " +
              objectDesc.getRole().getRole_name());
        }
        result.setType(role.getCatalogObjectType());
        result.setCatalog_version(role.getCatalogVersion());
        result.setRole(role.toThrift());
        break;
      case PRIVILEGE:
        Role tmpRole = authPolicy_.getRole(objectDesc.getPrivilege().getRole_id());
        if (tmpRole == null) {
          throw new CatalogException("No role associated with ID: " +
              objectDesc.getPrivilege().getRole_id());
        }
        for (RolePrivilege p: tmpRole.getPrivileges()) {
          if (p.getName().equalsIgnoreCase(
              objectDesc.getPrivilege().getPrivilege_name())) {
            result.setType(p.getCatalogObjectType());
            result.setCatalog_version(p.getCatalogVersion());
            result.setPrivilege(p.toThrift());
            return result;
          }
        }
        throw new CatalogException(String.format("Role '%s' does not contain " +
            "privilege: '%s'", tmpRole.getName(),
            objectDesc.getPrivilege().getPrivilege_name()));
      default: throw new IllegalStateException(
          "Unexpected TCatalogObject type: " + objectDesc.getType());
    }
    return result;
  }
}