impala-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tarmstr...@apache.org
Subject [06/14] incubator-impala git commit: IMPALA-3719: Simplify CREATE TABLE statements with Kudu tables
Date Sat, 22 Oct 2016 05:33:33 GMT
http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
index d55f8da..d0185b7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/KuduTable.java
@@ -19,21 +19,19 @@ package org.apache.impala.catalog;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 
 import javax.xml.bind.DatatypeConverter;
 
-import org.apache.hadoop.hive.metastore.IMetaStoreClient;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
-import org.apache.log4j.Logger;
-import org.apache.kudu.client.KuduClient;
-import org.apache.kudu.client.LocatedTablet;
-
+import org.apache.impala.analysis.ColumnDef;
+import org.apache.impala.analysis.DistributeParam;
+import org.apache.impala.analysis.ToSqlUtils;
 import org.apache.impala.common.ImpalaRuntimeException;
 import org.apache.impala.thrift.TCatalogObjectType;
 import org.apache.impala.thrift.TColumn;
+import org.apache.impala.thrift.TDistributeByHashParam;
+import org.apache.impala.thrift.TDistributeByRangeParam;
+import org.apache.impala.thrift.TDistributeParam;
 import org.apache.impala.thrift.TKuduTable;
 import org.apache.impala.thrift.TResultSet;
 import org.apache.impala.thrift.TResultSetMetadata;
@@ -42,76 +40,86 @@ import org.apache.impala.thrift.TTableDescriptor;
 import org.apache.impala.thrift.TTableType;
 import org.apache.impala.util.KuduUtil;
 import org.apache.impala.util.TResultRowBuilder;
+import org.apache.impala.service.CatalogOpExecutor;
+
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
+
+import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+import org.apache.kudu.client.LocatedTablet;
+import org.apache.kudu.client.PartitionSchema.HashBucketSchema;
+import org.apache.kudu.client.PartitionSchema.RangeSchema;
+import org.apache.kudu.client.PartitionSchema;
+import org.apache.log4j.Logger;
+import org.apache.thrift.TException;
 
 /**
- * Impala representation of a Kudu table.
- *
- * The Kudu-related metadata is stored in the Metastore table's table properties.
+ * Representation of a Kudu table in the catalog cache.
  */
 public class KuduTable extends Table {
-  private static final Logger LOG = Logger.getLogger(Table.class);
+
+  private static final Logger LOG = Logger.getLogger(KuduTable.class);
 
   // Alias to the string key that identifies the storage handler for Kudu tables.
   public static final String KEY_STORAGE_HANDLER =
       hive_metastoreConstants.META_TABLE_STORAGE;
 
-  // Key to access the table name from the table properties
+  // Key to access the table name from the table properties.
   public static final String KEY_TABLE_NAME = "kudu.table_name";
 
   // Key to access the columns used to build the (composite) key of the table.
-  // The order of the keys is important.
+  // Deprecated - Used only for error checking.
   public static final String KEY_KEY_COLUMNS = "kudu.key_columns";
 
-  // Key to access the master address from the table properties. Error handling for
+  // Key to access the master host from the table properties. Error handling for
   // this string is done in the KuduClient library.
-  // TODO we should have something like KuduConfig.getDefaultConfig()
-  public static final String KEY_MASTER_ADDRESSES = "kudu.master_addresses";
+  // TODO: Rename kudu.master_addresses to kudu.master_host will break compatibility
+  // with older versions.
+  public static final String KEY_MASTER_HOSTS = "kudu.master_addresses";
 
   // Kudu specific value for the storage handler table property keyed by
   // KEY_STORAGE_HANDLER.
+  // TODO: Fix the storage handler name (see IMPALA-4271).
   public static final String KUDU_STORAGE_HANDLER =
       "com.cloudera.kudu.hive.KuduStorageHandler";
 
   // Key to specify the number of tablet replicas.
-  // TODO(KUDU): Allow modification in alter table.
   public static final String KEY_TABLET_REPLICAS = "kudu.num_tablet_replicas";
 
   public static final long KUDU_RPC_TIMEOUT_MS = 50000;
 
-  // The name of the table in Kudu.
+  // Table name in the Kudu storage engine. It may not neccessarily be the same as the
+  // table name specified in the CREATE TABLE statement; the latter
+  // is stored in Table.name_. Reasons why KuduTable.kuduTableName_ and Table.name_ may
+  // differ:
+  // 1. For managed tables, 'kuduTableName_' is prefixed with 'impala::<db_name>' to
+  // avoid conficts. TODO: Remove this when Kudu supports databases.
+  // 2. The user may specify a table name using the 'kudu.table_name' table property.
   private String kuduTableName_;
 
   // Comma separated list of Kudu master hosts with optional ports.
   private String kuduMasters_;
 
-  // The set of columns that are key columns in Kudu.
-  private ImmutableList<String> kuduKeyColumnNames_;
+  // Primary key column names.
+  private final List<String> primaryKeyColumnNames_ = Lists.newArrayList();
+
+  // Distribution schemes of this Kudu table. Both range and hash-based distributions are
+  // supported.
+  private final List<DistributeParam> distributeBy_ = Lists.newArrayList();
 
   protected KuduTable(TableId id, org.apache.hadoop.hive.metastore.api.Table msTable,
       Db db, String name, String owner) {
     super(id, msTable, db, name, owner);
-  }
-
-  public TKuduTable getKuduTable() {
-    TKuduTable tbl = new TKuduTable();
-    tbl.setKey_columns(Preconditions.checkNotNull(kuduKeyColumnNames_));
-    tbl.setMaster_addresses(Lists.newArrayList(kuduMasters_.split(",")));
-    tbl.setTable_name(kuduTableName_);
-    return tbl;
-  }
-
-  @Override
-  public TTableDescriptor toThriftDescriptor(Set<Long> referencedPartitions) {
-    TTableDescriptor desc = new TTableDescriptor(id_.asInt(), TTableType.KUDU_TABLE,
-        getTColumnDescriptors(), numClusteringCols_, kuduTableName_, db_.getName());
-    desc.setKuduTable(getKuduTable());
-    return desc;
+    kuduTableName_ = msTable.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    kuduMasters_ = msTable.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
   }
 
   @Override
@@ -126,78 +134,149 @@ public class KuduTable extends Table {
   @Override
   public ArrayList<Column> getColumnsInHiveOrder() { return getColumns(); }
 
-  public static boolean isKuduTable(org.apache.hadoop.hive.metastore.api.Table mstbl) {
-    return KUDU_STORAGE_HANDLER.equals(mstbl.getParameters().get(KEY_STORAGE_HANDLER));
+  public static boolean isKuduTable(org.apache.hadoop.hive.metastore.api.Table msTbl) {
+    return KUDU_STORAGE_HANDLER.equals(msTbl.getParameters().get(KEY_STORAGE_HANDLER));
+  }
+
+  public String getKuduTableName() { return kuduTableName_; }
+  public String getKuduMasterHosts() { return kuduMasters_; }
+
+  public List<String> getPrimaryKeyColumnNames() {
+    return ImmutableList.copyOf(primaryKeyColumnNames_);
+  }
+
+  public List<DistributeParam> getDistributeBy() {
+    return ImmutableList.copyOf(distributeBy_);
   }
 
   /**
-   * Load the columns from the schema list
+   * Loads the metadata of a Kudu table.
+   *
+   * Schema and distribution schemes are loaded directly from Kudu whereas column stats
+   * are loaded from HMS. The function also updates the table schema in HMS in order to
+   * propagate alterations made to the Kudu table to HMS.
    */
-  private void loadColumns(List<FieldSchema> schema, IMetaStoreClient client,
-      Set<String> keyColumns) throws TableLoadingException {
+  @Override
+  public void load(boolean dummy /* not used */, IMetaStoreClient msClient,
+      org.apache.hadoop.hive.metastore.api.Table msTbl) throws TableLoadingException {
+    msTable_ = msTbl;
+    // This is set to 0 for Kudu tables.
+    // TODO: Change this to reflect the number of pk columns and modify all the
+    // places (e.g. insert stmt) that currently make use of this parameter.
+    numClusteringCols_ = 0;
+    kuduTableName_ = msTable_.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    Preconditions.checkNotNull(kuduTableName_);
+    kuduMasters_ = msTable_.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+    Preconditions.checkNotNull(kuduMasters_);
+    org.apache.kudu.client.KuduTable kuduTable = null;
+    numRows_ = getRowCount(msTable_.getParameters());
+
+    // Connect to Kudu to retrieve table metadata
+    try (KuduClient kuduClient = new KuduClient.KuduClientBuilder(
+        getKuduMasterHosts()).build()) {
+      kuduTable = kuduClient.openTable(kuduTableName_);
+    } catch (KuduException e) {
+      LOG.error("Error accessing Kudu table " + kuduTableName_);
+      throw new TableLoadingException(e.getMessage());
+    }
+    Preconditions.checkNotNull(kuduTable);
+
+    // Load metadata from Kudu and HMS
+    try {
+      loadSchema(kuduTable);
+      loadDistributeByParams(kuduTable);
+      loadAllColumnStats(msClient);
+    } catch (ImpalaRuntimeException e) {
+      LOG.error("Error loading metadata for Kudu table: " + kuduTableName_);
+      throw new TableLoadingException("Error loading metadata for Kudu table " +
+          kuduTableName_, e);
+    }
 
-    if (keyColumns.size() == 0 || keyColumns.size() > schema.size()) {
-      throw new TableLoadingException(String.format("Kudu tables must have at least one"
-          + "key column (had %d), and no more key columns than there are table columns "
-          + "(had %d).", keyColumns.size(), schema.size()));
+    // Update the table schema in HMS.
+    try {
+      long lastDdlTime = CatalogOpExecutor.calculateDdlTime(msTable_);
+      msTable_.putToParameters("transient_lastDdlTime", Long.toString(lastDdlTime));
+      msTable_.putToParameters(StatsSetupConst.DO_NOT_UPDATE_STATS,
+          StatsSetupConst.TRUE);
+      msClient.alter_table(msTable_.getDbName(), msTable_.getTableName(), msTable_);
+    } catch (TException e) {
+      throw new TableLoadingException(e.getMessage());
     }
+  }
 
+  /**
+   * Loads the schema from the Kudu table including column definitions and primary key
+   * columns. Replaces the columns in the HMS table with the columns from the Kudu table.
+   * Throws an ImpalaRuntimeException if Kudu column data types cannot be mapped to
+   * Impala data types.
+   */
+  private void loadSchema(org.apache.kudu.client.KuduTable kuduTable)
+      throws ImpalaRuntimeException {
+    Preconditions.checkNotNull(kuduTable);
     clearColumns();
-    Set<String> columnNames = Sets.newHashSet();
+    primaryKeyColumnNames_.clear();
+    List<FieldSchema> cols = msTable_.getSd().getCols();
+    cols.clear();
     int pos = 0;
-    for (FieldSchema field: schema) {
-      org.apache.impala.catalog.Type type = parseColumnType(field);
-      // TODO(kudu-merge): Check for decimal types?
-      boolean isKey = keyColumns.contains(field.getName());
-      KuduColumn col = new KuduColumn(field.getName(), isKey, !isKey, type,
-          field.getComment(), pos);
-      columnNames.add(col.getName());
-      addColumn(col);
+    for (ColumnSchema colSchema: kuduTable.getSchema().getColumns()) {
+      Type type = KuduUtil.toImpalaType(colSchema.getType());
+      String colName = colSchema.getName();
+      cols.add(new FieldSchema(colName, type.toSql().toLowerCase(), null));
+      boolean isKey = colSchema.isKey();
+      if (isKey) primaryKeyColumnNames_.add(colName);
+      addColumn(new KuduColumn(colName, isKey, !isKey, type, null, pos));
       ++pos;
     }
+  }
 
-    if (!columnNames.containsAll(keyColumns)) {
-      throw new TableLoadingException(String.format("Some key columns were not found in"
-              + " the set of columns. List of column names: %s, List of key column names:"
-              + " %s", Iterables.toString(columnNames), Iterables.toString(keyColumns)));
+  private void loadDistributeByParams(org.apache.kudu.client.KuduTable kuduTable) {
+    Preconditions.checkNotNull(kuduTable);
+    PartitionSchema partitionSchema = kuduTable.getPartitionSchema();
+    Preconditions.checkState(!colsByPos_.isEmpty());
+    distributeBy_.clear();
+    for (HashBucketSchema hashBucketSchema: partitionSchema.getHashBucketSchemas()) {
+      List<String> columnNames = Lists.newArrayList();
+      for (int colPos: hashBucketSchema.getColumnIds()) {
+        columnNames.add(colsByPos_.get(colPos).getName());
+      }
+      distributeBy_.add(
+          DistributeParam.createHashParam(columnNames, hashBucketSchema.getNumBuckets()));
     }
-
-    kuduKeyColumnNames_ = ImmutableList.copyOf(keyColumns);
-
-    loadAllColumnStats(client);
+    RangeSchema rangeSchema = partitionSchema.getRangeSchema();
+    List<Integer> columnIds = rangeSchema.getColumns();
+    if (columnIds.isEmpty()) return;
+    List<String> columnNames = Lists.newArrayList();
+    for (int colPos: columnIds) columnNames.add(colsByPos_.get(colPos).getName());
+    // We don't populate the split values because Kudu's API doesn't currently support
+    // retrieving the split values for range partitions.
+    // TODO: File a Kudu JIRA.
+    distributeBy_.add(DistributeParam.createRangeParam(columnNames, null));
   }
 
-  @Override
-  public void load(boolean reuseMetadata, IMetaStoreClient client,
-      org.apache.hadoop.hive.metastore.api.Table msTbl) throws TableLoadingException {
-    // TODO handle 'reuseMetadata'
-    if (getMetaStoreTable() == null || !tableParamsAreValid(msTbl.getParameters())) {
-      throw new TableLoadingException(String.format(
-          "Cannot load Kudu table %s, table is corrupt.", name_));
+  /**
+   * Creates a temporary KuduTable object populated with the specified properties but has
+   * an invalid TableId and is not added to the Kudu storage engine or the
+   * HMS. This is used for CTAS statements.
+   */
+  public static KuduTable createCtasTarget(Db db,
+      org.apache.hadoop.hive.metastore.api.Table msTbl, List<ColumnDef> columnDefs,
+      List<String> primaryKeyColumnNames, List<DistributeParam> distributeParams) {
+    KuduTable tmpTable = new KuduTable(TableId.createInvalidId(), msTbl, db,
+        msTbl.getTableName(), msTbl.getOwner());
+    int pos = 0;
+    for (ColumnDef colDef: columnDefs) {
+      tmpTable.addColumn(new Column(colDef.getColName(), colDef.getType(), pos++));
     }
-
-    msTable_ = msTbl;
-    kuduTableName_ = msTbl.getParameters().get(KEY_TABLE_NAME);
-    kuduMasters_ = msTbl.getParameters().get(KEY_MASTER_ADDRESSES);
-
-    String keyColumnsProp = Preconditions.checkNotNull(msTbl.getParameters()
-        .get(KEY_KEY_COLUMNS).toLowerCase(), "'kudu.key_columns' cannot be null.");
-    Set<String> keyColumns = KuduUtil.parseKeyColumns(keyColumnsProp);
-
-    // Load the rest of the data from the table parameters directly
-    loadColumns(msTbl.getSd().getCols(), client, keyColumns);
-
-    numClusteringCols_ = 0;
-
-    // Get row count from stats
-    numRows_ = getRowCount(getMetaStoreTable().getParameters());
+    tmpTable.primaryKeyColumnNames_.addAll(primaryKeyColumnNames);
+    tmpTable.distributeBy_.addAll(distributeParams);
+    return tmpTable;
   }
 
   @Override
   public TTable toThrift() {
     TTable table = super.toThrift();
     table.setTable_type(TTableType.KUDU_TABLE);
-    table.setKudu_table(getKuduTable());
+    table.setKudu_table(getTKuduTable());
     return table;
   }
 
@@ -207,33 +286,46 @@ public class KuduTable extends Table {
     TKuduTable tkudu = thriftTable.getKudu_table();
     kuduTableName_ = tkudu.getTable_name();
     kuduMasters_ = Joiner.on(',').join(tkudu.getMaster_addresses());
-    kuduKeyColumnNames_ = ImmutableList.copyOf(tkudu.getKey_columns());
+    primaryKeyColumnNames_.clear();
+    primaryKeyColumnNames_.addAll(tkudu.getKey_columns());
+    loadDistributeByParamsFromThrift(tkudu.getDistribute_by());
   }
 
-  public String getKuduTableName() { return kuduTableName_; }
-  public String getKuduMasterAddresses() { return kuduMasters_; }
-  public int getNumKeyColumns() { return kuduKeyColumnNames_.size(); }
-
-  /**
-   * Returns true if all required parameters are present in the given table properties
-   * map.
-   * TODO(kudu-merge) Return a more specific error string.
-   */
-  public static boolean tableParamsAreValid(Map<String, String> params) {
-    return params.get(KEY_TABLE_NAME) != null && params.get(KEY_TABLE_NAME).length() > 0
-        && params.get(KEY_MASTER_ADDRESSES) != null
-        && params.get(KEY_MASTER_ADDRESSES).length() > 0
-        && params.get(KEY_KEY_COLUMNS) != null
-        && params.get(KEY_KEY_COLUMNS).length() > 0;
-   }
+  private void loadDistributeByParamsFromThrift(List<TDistributeParam> params) {
+    distributeBy_.clear();
+    for (TDistributeParam param: params) {
+      if (param.isSetBy_hash_param()) {
+        TDistributeByHashParam hashParam = param.getBy_hash_param();
+        distributeBy_.add(DistributeParam.createHashParam(
+            hashParam.getColumns(), hashParam.getNum_buckets()));
+      } else {
+        Preconditions.checkState(param.isSetBy_range_param());
+        TDistributeByRangeParam rangeParam = param.getBy_range_param();
+        distributeBy_.add(DistributeParam.createRangeParam(rangeParam.getColumns(),
+            null));
+      }
+    }
+  }
 
-  /**
-   * The number of nodes is not know ahead of time and will be updated during computeStats
-   * in the scan node.
-   */
-  public int getNumNodes() { return -1; }
+  @Override
+  public TTableDescriptor toThriftDescriptor(Set<Long> referencedPartitions) {
+    TTableDescriptor desc = new TTableDescriptor(id_.asInt(), TTableType.KUDU_TABLE,
+        getTColumnDescriptors(), numClusteringCols_, kuduTableName_, db_.getName());
+    desc.setKuduTable(getTKuduTable());
+    return desc;
+  }
 
-  public List<String> getKuduKeyColumnNames() { return kuduKeyColumnNames_; }
+  private TKuduTable getTKuduTable() {
+    TKuduTable tbl = new TKuduTable();
+    tbl.setKey_columns(Preconditions.checkNotNull(primaryKeyColumnNames_));
+    tbl.setMaster_addresses(Lists.newArrayList(kuduMasters_.split(",")));
+    tbl.setTable_name(kuduTableName_);
+    Preconditions.checkNotNull(distributeBy_);
+    for (DistributeParam distributeParam: distributeBy_) {
+      tbl.addToDistribute_by(distributeParam.toThrift());
+    }
+    return tbl;
+  }
 
   public TResultSet getTableStats() throws ImpalaRuntimeException {
     TResultSet result = new TResultSet();
@@ -247,7 +339,7 @@ public class KuduTable extends Table {
     resultSchema.addToColumns(new TColumn("# Replicas", Type.INT.toThrift()));
 
     try (KuduClient client = new KuduClient.KuduClientBuilder(
-        getKuduMasterAddresses()).build()) {
+        getKuduMasterHosts()).build()) {
       org.apache.kudu.client.KuduTable kuduTable = client.openTable(kuduTableName_);
       List<LocatedTablet> tablets =
           kuduTable.getTabletsLocations(KUDU_RPC_TIMEOUT_MS);

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/Table.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/Table.java b/fe/src/main/java/org/apache/impala/catalog/Table.java
index 6145cc5..4b40b44 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Table.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Table.java
@@ -455,6 +455,11 @@ public abstract class Table implements CatalogObject {
   @Override
   public boolean isLoaded() { return true; }
 
+  public static boolean isExternalTable(
+      org.apache.hadoop.hive.metastore.api.Table msTbl) {
+    return msTbl.getTableType().equalsIgnoreCase(TableType.EXTERNAL_TABLE.toString());
+  }
+
   /**
    * If the table is cached, it returns a <cache pool name, replication factor> pair
    * and adds the table cached directive ID to 'cacheDirIds'. Otherwise, it

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/TableLoader.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/TableLoader.java b/fe/src/main/java/org/apache/impala/catalog/TableLoader.java
index 764abe0..8541a3a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/TableLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/TableLoader.java
@@ -18,7 +18,6 @@
 package org.apache.impala.catalog;
 
 import java.util.EnumSet;
-import java.util.Set;
 
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/Type.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/Type.java b/fe/src/main/java/org/apache/impala/catalog/Type.java
index 91fc2e3..05c71c7 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Type.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Type.java
@@ -317,6 +317,15 @@ public abstract class Type {
   }
 
   /**
+   * Checks if types t1 and t2 are assignment compatible, i.e. if both t1 and t2 can be
+   * assigned to a type t without an explicit cast and without any conversions that would
+   * result in loss of precision.
+   */
+  public static boolean areAssignmentCompatibleTypes(Type t1, Type t2) {
+    return getAssignmentCompatibleType(t1, t2, true) != ScalarType.INVALID;
+  }
+
+  /**
    * Returns true if this type exceeds the MAX_NESTING_DEPTH, false otherwise.
    */
   public boolean exceedsMaxNestingDepth() { return exceedsMaxNestingDepth(0); }

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/delegates/DdlDelegate.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/delegates/DdlDelegate.java b/fe/src/main/java/org/apache/impala/catalog/delegates/DdlDelegate.java
deleted file mode 100644
index 6c3ba8e..0000000
--- a/fe/src/main/java/org/apache/impala/catalog/delegates/DdlDelegate.java
+++ /dev/null
@@ -1,75 +0,0 @@
-// 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.delegates;
-
-import java.util.List;
-
-import org.apache.impala.thrift.TDistributeParam;
-import org.apache.hadoop.hive.metastore.api.Table;
-
-import org.apache.impala.common.ImpalaRuntimeException;
-import org.apache.impala.thrift.TAlterTableParams;
-
-/**
- * Abstract class to implement the storage specific portion of DDL requests.
- *
- * During catalog DDL operations the CatalogOpExecutor will instantiate the correct
- * subclass of this class to handle the DDL operation to the storage backend. See,
- * CatalogOpExecutor::createDDLDelegate() for details.
- *
- */
-public abstract class DdlDelegate {
-
-  protected Table msTbl_;
-  protected TAlterTableParams tAlterTableParams_;
-  protected List<TDistributeParam> distributeParams_;
-
-  /**
-   * Creates a new delegate to modify Table 'msTbl'.
-   */
-  public DdlDelegate setMsTbl(Table msTbl) {
-    msTbl_ = msTbl;
-    return this;
-  }
-
-  public DdlDelegate setAlterTableParams(TAlterTableParams p) {
-    tAlterTableParams_ = p;
-    return this;
-  }
-
-  public DdlDelegate setDistributeParams(List<TDistributeParam> p) {
-    distributeParams_ = p;
-    return this;
-  }
-
-  /**
-   * Creates the table.
-   */
-  public abstract void createTable() throws ImpalaRuntimeException;
-
-  /**
-   * Drops the table.
-   */
-  public abstract void dropTable() throws ImpalaRuntimeException;
-
-  /**
-   * Performs an alter table with the parameters set with setAlterTableParams().
-   */
-  public abstract boolean alterTable() throws ImpalaRuntimeException;
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/delegates/KuduDdlDelegate.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/delegates/KuduDdlDelegate.java b/fe/src/main/java/org/apache/impala/catalog/delegates/KuduDdlDelegate.java
deleted file mode 100644
index 8410868..0000000
--- a/fe/src/main/java/org/apache/impala/catalog/delegates/KuduDdlDelegate.java
+++ /dev/null
@@ -1,190 +0,0 @@
-// 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.delegates;
-
-import static org.apache.impala.util.KuduUtil.compareSchema;
-import static org.apache.impala.util.KuduUtil.fromImpalaType;
-import static org.apache.impala.util.KuduUtil.parseKeyColumns;
-
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-
-import org.apache.hadoop.hive.metastore.TableType;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.kudu.ColumnSchema;
-import org.apache.kudu.ColumnSchema.ColumnSchemaBuilder;
-import org.apache.kudu.Schema;
-import org.apache.kudu.Type;
-import org.apache.kudu.client.CreateTableOptions;
-import org.apache.kudu.client.KuduClient;
-import org.apache.kudu.client.PartialRow;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.impala.catalog.KuduTable;
-import org.apache.impala.common.ImpalaRuntimeException;
-import org.apache.impala.thrift.TDistributeParam;
-import org.apache.impala.util.KuduUtil;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-
-
-/**
- * Implementation of the Kudu DDL Delegate. Propagates create and drop table statements to
- * Kudu.
- */
-public class KuduDdlDelegate extends DdlDelegate {
-
-  private static final Logger LOG = LoggerFactory.getLogger(KuduDdlDelegate.class);
-
-  public KuduDdlDelegate(Table msTbl) {
-    setMsTbl(msTbl);
-  }
-
-  /**
-   * Creates the Kudu table if it does not exist and returns true. If the table exists and
-   * the table is not a managed table ignore and return false, otherwise throw an
-   * exception.
-   */
-  @Override
-  public void createTable()
-      throws ImpalaRuntimeException {
-
-    String kuduTableName = msTbl_.getParameters().get(KuduTable.KEY_TABLE_NAME);
-    String kuduMasters = msTbl_.getParameters().get(KuduTable.KEY_MASTER_ADDRESSES);
-
-    // Can be optional for un-managed tables
-    String kuduKeyCols = msTbl_.getParameters().get(KuduTable.KEY_KEY_COLUMNS);
-
-    String replication = msTbl_.getParameters().get(KuduTable.KEY_TABLET_REPLICAS);
-
-    try (KuduClient client = new KuduClient.KuduClientBuilder(kuduMasters).build()) {
-      // TODO should we throw if the table does not exist when its an external table?
-      if (client.tableExists(kuduTableName)) {
-        if (msTbl_.getTableType().equals(TableType.MANAGED_TABLE.toString())) {
-          throw new ImpalaRuntimeException(String.format(
-              "Table %s already exists in Kudu master %s.", kuduTableName, kuduMasters));
-        }
-
-        // Check if the external table matches the schema
-        org.apache.kudu.client.KuduTable kuduTable = client.openTable(kuduTableName);
-        if (!compareSchema(msTbl_, kuduTable)) {
-          throw new ImpalaRuntimeException(String.format(
-              "Table %s (%s) has a different schema in Kudu than in Hive.",
-              msTbl_.getTableName(), kuduTableName));
-        }
-        return;
-      }
-
-      HashSet<String> keyColNames = parseKeyColumns(kuduKeyCols);
-      List<ColumnSchema> keyColSchemas = new ArrayList<>();
-
-      // Create a new Schema and map the types accordingly
-      ArrayList<ColumnSchema> columns = Lists.newArrayList();
-      for (FieldSchema fieldSchema: msTbl_.getSd().getCols()) {
-        org.apache.impala.catalog.Type catalogType = org.apache.impala.catalog.Type
-            .parseColumnType(fieldSchema.getType());
-        if (catalogType == null) {
-          throw new ImpalaRuntimeException(String.format(
-              "Could not parse column type %s.", fieldSchema.getType()));
-        }
-        Type t = fromImpalaType(catalogType);
-        // Create the actual column and check if the column is a key column
-        ColumnSchemaBuilder csb = new ColumnSchemaBuilder(
-            fieldSchema.getName(), t);
-        boolean isKeyColumn = keyColNames.contains(fieldSchema.getName());
-        csb.key(isKeyColumn);
-        csb.nullable(!isKeyColumn);
-        ColumnSchema cs = csb.build();
-        columns.add(cs);
-        if (isKeyColumn) keyColSchemas.add(cs);
-      }
-
-      Schema schema = new Schema(columns);
-      CreateTableOptions cto = new CreateTableOptions();
-
-      // Handle auto-partitioning of the Kudu table
-      if (distributeParams_ != null) {
-        for (TDistributeParam param : distributeParams_) {
-          if (param.isSetBy_hash_param()) {
-            Preconditions.checkState(!param.isSetBy_range_param());
-            cto.addHashPartitions(param.getBy_hash_param().getColumns(),
-                param.getBy_hash_param().getNum_buckets());
-          } else {
-            Preconditions.checkState(param.isSetBy_range_param());
-            cto.setRangePartitionColumns(param.getBy_range_param().getColumns());
-            for (PartialRow p : KuduUtil.parseSplits(schema, param.getBy_range_param())) {
-              cto.addSplitRow(p);
-            }
-          }
-        }
-      }
-
-      if (!Strings.isNullOrEmpty(replication)) {
-        int r = Integer.parseInt(replication);
-        if (r <= 0) {
-          throw new ImpalaRuntimeException(
-              "Number of tablet replicas must be greater than zero. " +
-              "Given number of replicas is: " + Integer.toString(r));
-        }
-        cto.setNumReplicas(r);
-      }
-
-      client.createTable(kuduTableName, schema, cto);
-    } catch (ImpalaRuntimeException e) {
-      throw e;
-    } catch (Exception e) {
-      throw new ImpalaRuntimeException("Error creating Kudu table", e);
-    }
-  }
-
-  @Override
-  public void dropTable() throws ImpalaRuntimeException {
-    // If table is an external table, do not delete the data
-    if (msTbl_.getTableType().equals(TableType.EXTERNAL_TABLE.toString())) return;
-
-    String kuduTableName = msTbl_.getParameters().get(KuduTable.KEY_TABLE_NAME);
-    String kuduMasters = msTbl_.getParameters().get(KuduTable.KEY_MASTER_ADDRESSES);
-
-    try (KuduClient client = new KuduClient.KuduClientBuilder(kuduMasters).build()) {
-      if (!client.tableExists(kuduTableName)) {
-        LOG.warn("Table: %s is in inconsistent state. It does not exist in Kudu master(s)"
-            + " %s, but it exists in Hive metastore. Deleting from metastore only.",
-            kuduTableName, kuduMasters);
-        return;
-      }
-      client.deleteTable(kuduTableName);
-      return;
-    } catch (Exception e) {
-      throw new ImpalaRuntimeException("Error dropping Kudu table", e);
-    }
-  }
-
-  public static boolean canHandle(org.apache.hadoop.hive.metastore.api.Table msTbl) {
-    return KuduTable.isKuduTable(msTbl);
-  }
-
-  @Override
-  public boolean alterTable() throws ImpalaRuntimeException {
-    throw new ImpalaRuntimeException(
-        "Alter table operations are not supported for Kudu tables.");
-  }
-}

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/catalog/delegates/UnsupportedOpDelegate.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/delegates/UnsupportedOpDelegate.java b/fe/src/main/java/org/apache/impala/catalog/delegates/UnsupportedOpDelegate.java
deleted file mode 100644
index 8aabaa4..0000000
--- a/fe/src/main/java/org/apache/impala/catalog/delegates/UnsupportedOpDelegate.java
+++ /dev/null
@@ -1,35 +0,0 @@
-// 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.delegates;
-
-import org.apache.impala.common.ImpalaRuntimeException;
-
-/**
- * Empty implementation for the DdlDelegate interface that does nothing.
- */
-public class UnsupportedOpDelegate extends DdlDelegate {
-
-  @Override
-  public void createTable() throws ImpalaRuntimeException { }
-
-  @Override
-  public void dropTable() throws ImpalaRuntimeException { }
-
-  @Override
-  public boolean alterTable() throws ImpalaRuntimeException { return true; }
-}

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
index 3345c1b..8d15425 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsPartitionFilter.java
@@ -52,7 +52,7 @@ public class HdfsPartitionFilter {
 
   // lhs exprs of smap used in isMatch()
   private final ArrayList<SlotRef> lhsSlotRefs_ = Lists.newArrayList();
-  // indices into Table.getColumns()
+  // indices into Table.getColumnNames()
   private final ArrayList<Integer> refdKeys_ = Lists.newArrayList();
 
   public HdfsPartitionFilter(Expr predicate, HdfsTable tbl, Analyzer analyzer) {

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
index 64ef822..9434801 100644
--- a/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/KuduScanNode.java
@@ -107,7 +107,7 @@ public class KuduScanNode extends ScanNode {
     conjuncts_ = orderConjunctsByCost(conjuncts_);
 
     try (KuduClient client =
-         new KuduClientBuilder(kuduTable_.getKuduMasterAddresses()).build()) {
+         new KuduClientBuilder(kuduTable_.getKuduMasterHosts()).build()) {
       org.apache.kudu.client.KuduTable rpcTable =
           client.openTable(kuduTable_.getKuduTableName());
       validateSchema(rpcTable);

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 5743a59..54493d1 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -51,6 +51,11 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.StringColumnStatsData;
 import org.apache.log4j.Logger;
 import org.apache.thrift.TException;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 
 import org.apache.impala.analysis.FunctionName;
 import org.apache.impala.analysis.TableName;
@@ -61,7 +66,6 @@ import org.apache.impala.catalog.CatalogServiceCatalog;
 import org.apache.impala.catalog.Column;
 import org.apache.impala.catalog.ColumnNotFoundException;
 import org.apache.impala.catalog.DataSource;
-import org.apache.impala.catalog.DatabaseNotFoundException;
 import org.apache.impala.catalog.Db;
 import org.apache.impala.catalog.Function;
 import org.apache.impala.catalog.HBaseTable;
@@ -70,6 +74,7 @@ import org.apache.impala.catalog.HdfsPartition;
 import org.apache.impala.catalog.HdfsTable;
 import org.apache.impala.catalog.HiveStorageDescriptorFactory;
 import org.apache.impala.catalog.IncompleteTable;
+import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.catalog.MetaStoreClientPool.MetaStoreClient;
 import org.apache.impala.catalog.PartitionNotFoundException;
 import org.apache.impala.catalog.PartitionStatsUtil;
@@ -82,9 +87,6 @@ import org.apache.impala.catalog.TableLoadingException;
 import org.apache.impala.catalog.TableNotFoundException;
 import org.apache.impala.catalog.Type;
 import org.apache.impala.catalog.View;
-import org.apache.impala.catalog.delegates.DdlDelegate;
-import org.apache.impala.catalog.delegates.KuduDdlDelegate;
-import org.apache.impala.catalog.delegates.UnsupportedOpDelegate;
 import org.apache.impala.common.FileSystemUtil;
 import org.apache.impala.common.ImpalaException;
 import org.apache.impala.common.ImpalaRuntimeException;
@@ -121,7 +123,6 @@ import org.apache.impala.thrift.TCreateTableParams;
 import org.apache.impala.thrift.TDatabase;
 import org.apache.impala.thrift.TDdlExecRequest;
 import org.apache.impala.thrift.TDdlExecResponse;
-import org.apache.impala.thrift.TDistributeParam;
 import org.apache.impala.thrift.TDropDataSourceParams;
 import org.apache.impala.thrift.TDropDbParams;
 import org.apache.impala.thrift.TDropFunctionParams;
@@ -149,11 +150,6 @@ import org.apache.impala.thrift.TTruncateParams;
 import org.apache.impala.thrift.TUpdateCatalogRequest;
 import org.apache.impala.thrift.TUpdateCatalogResponse;
 import org.apache.impala.util.HdfsCachingUtil;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 
 /**
  * Class used to execute Catalog Operations, including DDL and refresh/invalidate
@@ -1103,8 +1099,7 @@ public class CatalogOpExecutor {
 
   /**
    * Drops a database from the metastore and removes the database's metadata from the
-   * internal cache. Re-throws any Hive Meta Store exceptions encountered during
-   * the drop.
+   * internal cache. Re-throws any HMS exceptions encountered during the drop.
    */
   private void dropDatabase(TDropDbParams params, TDdlExecResponse resp)
       throws ImpalaException {
@@ -1120,6 +1115,9 @@ public class CatalogOpExecutor {
 
     TCatalogObject removedObject = new TCatalogObject();
     synchronized (metastoreDdlLock_) {
+      // Remove all the Kudu tables of 'db' from the Kudu storage engine.
+      if (db != null && params.cascade) dropTablesFromKudu(db);
+
       try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
         msClient.getHiveClient().dropDatabase(
             params.getDb(), true, params.if_exists, params.cascade);
@@ -1144,6 +1142,44 @@ public class CatalogOpExecutor {
   }
 
   /**
+   * Drops all the Kudu tables of database 'db' from the Kudu storage engine. Retrieves
+   * the Kudu table name of each table in 'db' from HMS. Throws an ImpalaException if
+   * metadata for Kudu tables cannot be loaded from HMS or if an error occurs while
+   * trying to drop a table from Kudu.
+   */
+  private void dropTablesFromKudu(Db db) throws ImpalaException {
+    // If the table format isn't available, because the table hasn't been loaded yet,
+    // the metadata must be fetched from the Hive Metastore.
+    List<String> incompleteTableNames = Lists.newArrayList();
+    List<org.apache.hadoop.hive.metastore.api.Table> msTables = Lists.newArrayList();
+    for (Table table: db.getTables()) {
+      org.apache.hadoop.hive.metastore.api.Table msTable = table.getMetaStoreTable();
+      if (msTable == null) {
+        incompleteTableNames.add(table.getName());
+      } else {
+        msTables.add(msTable);
+      }
+    }
+    if (!incompleteTableNames.isEmpty()) {
+      try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+        msTables.addAll(msClient.getHiveClient().getTableObjectsByName(
+            db.getName(), incompleteTableNames));
+      } catch (TException e) {
+        LOG.error(String.format(HMS_RPC_ERROR_FORMAT_STR, "getTableObjectsByName") +
+            e.getMessage());
+      }
+    }
+    for (org.apache.hadoop.hive.metastore.api.Table msTable: msTables) {
+      if (!KuduTable.isKuduTable(msTable) || Table.isExternalTable(msTable)) continue;
+      // The operation will be aborted if the Kudu table cannot be dropped. If for
+      // some reason Kudu is permanently stuck in a non-functional state, the user is
+      // expected to ALTER TABLE to either set the table to UNMANAGED or set the format
+      // to something else.
+      KuduCatalogOpExecutor.dropTable(msTable, /*if exists*/ true);
+    }
+  }
+
+  /**
    * Drops a table or view from the metastore and removes it from the catalog.
    * Also drops all associated caching requests on the table and/or table's partitions,
    * uncaching all table data. If params.purge is true, table data is permanently
@@ -1157,17 +1193,6 @@ public class CatalogOpExecutor {
 
     TCatalogObject removedObject = new TCatalogObject();
     synchronized (metastoreDdlLock_) {
-
-      // Forward the DDL operation to the specified storage backend.
-      try {
-        org.apache.hadoop.hive.metastore.api.Table msTbl = getExistingTable(
-            tableName.getDb(), tableName.getTbl()).getMetaStoreTable();
-        DdlDelegate handler = createDdlDelegate(msTbl);
-        handler.dropTable();
-      } catch (TableNotFoundException | DatabaseNotFoundException e) {
-        // Do nothing
-      }
-
       Db db = catalog_.getDb(params.getTable_name().db_name);
       if (db == null) {
         if (params.if_exists) return;
@@ -1179,6 +1204,23 @@ public class CatalogOpExecutor {
         if (params.if_exists) return;
         throw new CatalogException("Table/View does not exist: " + tableName);
       }
+
+      // Retrieve the HMS table to determine if this is a Kudu table.
+      org.apache.hadoop.hive.metastore.api.Table msTbl = existingTbl.getMetaStoreTable();
+      if (msTbl == null) {
+        Preconditions.checkState(existingTbl instanceof IncompleteTable);
+        try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+          msTbl = msClient.getHiveClient().getTable(tableName.getDb(),
+              tableName.getTbl());
+        } catch (TException e) {
+          LOG.error(String.format(HMS_RPC_ERROR_FORMAT_STR, "getTable") + e.getMessage());
+        }
+      }
+      if (msTbl != null && KuduTable.isKuduTable(msTbl)
+          && !Table.isExternalTable(msTbl)) {
+        KuduCatalogOpExecutor.dropTable(msTbl, /* if exists */ true);
+      }
+
       // Check to make sure we don't drop a view with "drop table" statement and
       // vice versa. is_table field is marked optional in TDropTableOrViewParams to
       // maintain catalog api compatibility.
@@ -1343,7 +1385,8 @@ public class CatalogOpExecutor {
 
   /**
    * Creates a new table in the metastore and adds an entry to the metadata cache to
-   * lazily load the new metadata on the next access. Re-throws any Hive Meta Store
+   * lazily load the new metadata on the next access. If this is a managed Kudu table,
+   * the table is also created in the Kudu storage engine. Re-throws any HMS or Kudu
    * exceptions encountered during the create.
    */
   private boolean createTable(TCreateTableParams params, TDdlExecResponse response)
@@ -1351,9 +1394,8 @@ public class CatalogOpExecutor {
     Preconditions.checkNotNull(params);
     TableName tableName = TableName.fromThrift(params.getTable_name());
     Preconditions.checkState(tableName != null && tableName.isFullyQualified());
-    Preconditions.checkState(params.getColumns() != null &&
-        params.getColumns().size() > 0,
-        "Null or empty column list given as argument to Catalog.createTable");
+    Preconditions.checkState(params.getColumns() != null,
+        "Null column list given as argument to Catalog.createTable");
 
     if (params.if_not_exists &&
         catalog_.containsTable(tableName.getDb(), tableName.getTbl())) {
@@ -1362,11 +1404,161 @@ public class CatalogOpExecutor {
       response.getResult().setVersion(catalog_.getCatalogVersion());
       return false;
     }
-    org.apache.hadoop.hive.metastore.api.Table tbl =
-        createMetaStoreTable(params);
+    org.apache.hadoop.hive.metastore.api.Table tbl = createMetaStoreTable(params);
     LOG.debug(String.format("Creating table %s", tableName));
-    return createTable(tbl, params.if_not_exists, params.getCache_op(),
-        params.getDistribute_by(), response);
+    if (KuduTable.isKuduTable(tbl)) return createKuduTable(tbl, params, response);
+    Preconditions.checkState(params.getColumns().size() > 0,
+        "Empty column list given as argument to Catalog.createTable");
+    return createTable(tbl, params.if_not_exists, params.getCache_op(), response);
+  }
+
+  /**
+   * Utility function that creates a hive.metastore.api.Table object based on the given
+   * TCreateTableParams.
+   * TODO: Extract metastore object creation utility functions into a separate
+   * helper/factory class.
+   */
+  public static org.apache.hadoop.hive.metastore.api.Table createMetaStoreTable(
+      TCreateTableParams params) {
+    Preconditions.checkNotNull(params);
+    TableName tableName = TableName.fromThrift(params.getTable_name());
+    org.apache.hadoop.hive.metastore.api.Table tbl =
+        new org.apache.hadoop.hive.metastore.api.Table();
+    tbl.setDbName(tableName.getDb());
+    tbl.setTableName(tableName.getTbl());
+    tbl.setOwner(params.getOwner());
+    if (params.isSetTable_properties()) {
+      tbl.setParameters(params.getTable_properties());
+    } else {
+      tbl.setParameters(new HashMap<String, String>());
+    }
+
+    if (params.getComment() != null) {
+      tbl.getParameters().put("comment", params.getComment());
+    }
+    if (params.is_external) {
+      tbl.setTableType(TableType.EXTERNAL_TABLE.toString());
+      tbl.putToParameters("EXTERNAL", "TRUE");
+    } else {
+      tbl.setTableType(TableType.MANAGED_TABLE.toString());
+    }
+
+    tbl.setSd(createSd(params));
+    if (params.getPartition_columns() != null) {
+      // Add in any partition keys that were specified
+      tbl.setPartitionKeys(buildFieldSchemaList(params.getPartition_columns()));
+    } else {
+      tbl.setPartitionKeys(new ArrayList<FieldSchema>());
+    }
+    return tbl;
+  }
+
+  private static StorageDescriptor createSd(TCreateTableParams params) {
+    StorageDescriptor sd = HiveStorageDescriptorFactory.createSd(
+        params.getFile_format(), RowFormat.fromThrift(params.getRow_format()));
+    if (params.isSetSerde_properties()) {
+      if (sd.getSerdeInfo().getParameters() == null) {
+        sd.getSerdeInfo().setParameters(params.getSerde_properties());
+      } else {
+        sd.getSerdeInfo().getParameters().putAll(params.getSerde_properties());
+      }
+    }
+
+    if (params.getLocation() != null) sd.setLocation(params.getLocation());
+
+    // Add in all the columns
+    sd.setCols(buildFieldSchemaList(params.getColumns()));
+    return sd;
+  }
+
+  /**
+   * Creates a new Kudu table. The Kudu table is first created in the Kudu storage engine
+   * (only applicable to managed tables), then in HMS and finally in the catalog cache.
+   * Failure to add the table in HMS results in the table being dropped from Kudu.
+   * 'response' is populated with the results of this operation. Returns true if a new
+   * table was created as part of this call, false otherwise.
+   */
+  private boolean createKuduTable(org.apache.hadoop.hive.metastore.api.Table newTable,
+      TCreateTableParams params, TDdlExecResponse response) throws ImpalaException {
+    Preconditions.checkState(KuduTable.isKuduTable(newTable));
+    if (Table.isExternalTable(newTable)) {
+      KuduCatalogOpExecutor.populateColumnsFromKudu(newTable);
+    } else {
+      KuduCatalogOpExecutor.createManagedTable(newTable, params);
+    }
+    try {
+      // Add the table to the HMS and the catalog cache. Aquire metastoreDdlLock_ to
+      // ensure the atomicity of these operations.
+      synchronized (metastoreDdlLock_) {
+        try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+          msClient.getHiveClient().createTable(newTable);
+        }
+        // Add the table to the catalog cache
+        Table newTbl = catalog_.addTable(newTable.getDbName(), newTable.getTableName());
+        addTableToCatalogUpdate(newTbl, response.result);
+      }
+    } catch (Exception e) {
+      try {
+        // Error creating the table in HMS, drop the managed table from Kudu.
+        if (!Table.isExternalTable(newTable)) {
+          KuduCatalogOpExecutor.dropTable(newTable, false);
+        }
+      } catch (Exception logged) {
+        String kuduTableName = newTable.getParameters().get(KuduTable.KEY_TABLE_NAME);
+        LOG.error(String.format("Failed to drop Kudu table '%s'", kuduTableName),
+            logged);
+        throw new RuntimeException(String.format("Failed to create the table '%s' in " +
+            " the Metastore and the newly created Kudu table '%s' could not be " +
+            " dropped. The log contains more information.", newTable.getTableName(),
+            kuduTableName), e);
+      }
+      if (e instanceof AlreadyExistsException && params.if_not_exists) return false;
+      throw new ImpalaRuntimeException(
+          String.format(HMS_RPC_ERROR_FORMAT_STR, "createTable"), e);
+    }
+    return true;
+  }
+
+  /**
+   * Creates a new table. The table is initially created in HMS and, if that operation
+   * succeeds, it is then added in the catalog cache. It also sets HDFS caching if
+   * 'cacheOp' is not null. 'response' is populated with the results of this operation.
+   * Returns true if a new table was created as part of this call, false otherwise.
+   */
+  private boolean createTable(org.apache.hadoop.hive.metastore.api.Table newTable,
+      boolean if_not_exists, THdfsCachingOp cacheOp, TDdlExecResponse response)
+      throws ImpalaException {
+    Preconditions.checkState(!KuduTable.isKuduTable(newTable));
+    synchronized (metastoreDdlLock_) {
+      try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+        msClient.getHiveClient().createTable(newTable);
+        // If this table should be cached, and the table location was not specified by
+        // the user, an extra step is needed to read the table to find the location.
+        if (cacheOp != null && cacheOp.isSet_cached() &&
+            newTable.getSd().getLocation() == null) {
+          newTable = msClient.getHiveClient().getTable(
+              newTable.getDbName(), newTable.getTableName());
+        }
+      } catch (Exception e) {
+        if (e instanceof AlreadyExistsException && if_not_exists) return false;
+        throw new ImpalaRuntimeException(
+            String.format(HMS_RPC_ERROR_FORMAT_STR, "createTable"), e);
+      }
+
+      // Submit the cache request and update the table metadata.
+      if (cacheOp != null && cacheOp.isSet_cached()) {
+        short replication = cacheOp.isSetReplication() ? cacheOp.getReplication() :
+            JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR;
+        long id = HdfsCachingUtil.submitCacheTblDirective(newTable,
+            cacheOp.getCache_pool_name(), replication);
+        catalog_.watchCacheDirs(Lists.<Long>newArrayList(id),
+            new TTableName(newTable.getDbName(), newTable.getTableName()));
+        applyAlterTable(newTable);
+      }
+      Table newTbl = catalog_.addTable(newTable.getDbName(), newTable.getTableName());
+      addTableToCatalogUpdate(newTbl, response.result);
+    }
+    return true;
   }
 
   /**
@@ -1392,7 +1584,7 @@ public class CatalogOpExecutor {
         new org.apache.hadoop.hive.metastore.api.Table();
     setViewAttributes(params, view);
     LOG.debug(String.format("Creating view %s", tableName));
-    createTable(view, params.if_not_exists, null, null, response);
+    createTable(view, params.if_not_exists, null, response);
   }
 
   /**
@@ -1423,6 +1615,8 @@ public class CatalogOpExecutor {
     Table srcTable = getExistingTable(srcTblName.getDb(), srcTblName.getTbl());
     org.apache.hadoop.hive.metastore.api.Table tbl =
         srcTable.getMetaStoreTable().deepCopy();
+    Preconditions.checkState(!KuduTable.isKuduTable(tbl),
+        "CREATE TABLE LIKE is not supported for Kudu tables.");
     tbl.setDbName(tblName.getDb());
     tbl.setTableName(tblName.getTbl());
     tbl.setOwner(params.getOwner());
@@ -1460,7 +1654,7 @@ public class CatalogOpExecutor {
     tbl.getSd().setLocation(params.getLocation());
     if (fileFormat != null) {
       setStorageDescriptorFileFormat(tbl.getSd(), fileFormat);
-    } else if (fileFormat == null && srcTable instanceof View) {
+    } else if (srcTable instanceof View) {
       // Here, source table is a view which has no input format. So to be
       // consistent with CREATE TABLE, default input format is assumed to be
       // TEXT unless otherwise specified.
@@ -1469,85 +1663,7 @@ public class CatalogOpExecutor {
     // Set the row count of this table to unknown.
     tbl.putToParameters(StatsSetupConst.ROW_COUNT, "-1");
     LOG.debug(String.format("Creating table %s LIKE %s", tblName, srcTblName));
-    createTable(tbl, params.if_not_exists, null, null, response);
-  }
-
-  /**
-   * Creates a new table in the HMS. If ifNotExists=true, no error will be thrown if
-   * the table already exists, otherwise an exception will be thrown.
-   * Accepts an optional 'cacheOp' param, which if specified will cache the table's
-   * HDFS location according to the 'cacheOp' spec after creation.
-   * Stores details of the operations (such as the resulting catalog version) in
-   * 'response' output parameter.
-   * Returns true if a new table was created as part of this call, false otherwise.
-   */
-  private boolean createTable(org.apache.hadoop.hive.metastore.api.Table newTable,
-      boolean ifNotExists, THdfsCachingOp cacheOp, List<TDistributeParam> distribute_by,
-      TDdlExecResponse response)
-      throws ImpalaException {
-    synchronized (metastoreDdlLock_) {
-
-      try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
-        msClient.getHiveClient().createTable(newTable);
-        // If this table should be cached, and the table location was not specified by
-        // the user, an extra step is needed to read the table to find the location.
-        if (cacheOp != null && cacheOp.isSet_cached() &&
-            newTable.getSd().getLocation() == null) {
-          newTable = msClient.getHiveClient().getTable(newTable.getDbName(),
-              newTable.getTableName());
-        }
-      } catch (AlreadyExistsException e) {
-        if (!ifNotExists) {
-          throw new ImpalaRuntimeException(
-              String.format(HMS_RPC_ERROR_FORMAT_STR, "createTable"), e);
-        }
-        LOG.debug(String.format("Ignoring '%s' when creating table %s.%s because " +
-            "IF NOT EXISTS was specified.", e,
-            newTable.getDbName(), newTable.getTableName()));
-        return false;
-      } catch (TException e) {
-        throw new ImpalaRuntimeException(
-            String.format(HMS_RPC_ERROR_FORMAT_STR, "createTable"), e);
-      }
-
-      // Forward the operation to a specific storage backend. If the operation fails,
-      // delete the just created hive table to avoid inconsistencies.
-      try {
-        createDdlDelegate(newTable).setDistributeParams(distribute_by).createTable();
-      } catch (ImpalaRuntimeException e) {
-        try (MetaStoreClient c = catalog_.getMetaStoreClient()) {
-          c.getHiveClient().dropTable(newTable.getDbName(), newTable.getTableName(),
-              false, ifNotExists);
-        } catch (Exception hE) {
-          throw new ImpalaRuntimeException(String.format(HMS_RPC_ERROR_FORMAT_STR,
-              "dropTable"), hE);
-        }
-        throw e;
-      }
-
-      // Submit the cache request and update the table metadata.
-      if (cacheOp != null && cacheOp.isSet_cached()) {
-        short replication = cacheOp.isSetReplication() ? cacheOp.getReplication() :
-            JniCatalogConstants.HDFS_DEFAULT_CACHE_REPLICATION_FACTOR;
-        long id = HdfsCachingUtil.submitCacheTblDirective(newTable,
-            cacheOp.getCache_pool_name(), replication);
-        catalog_.watchCacheDirs(Lists.<Long>newArrayList(id),
-            new TTableName(newTable.getDbName(), newTable.getTableName()));
-        applyAlterTable(newTable);
-      }
-      Table newTbl = catalog_.addTable(newTable.getDbName(), newTable.getTableName());
-      addTableToCatalogUpdate(newTbl, response.result);
-    }
-    return true;
-  }
-
-  /**
-   * Instantiate the appropriate DDL delegate for the table. If no known delegate is
-   * available for the table, returns a UnsupportedOpDelegate instance.
-   */
-  private DdlDelegate createDdlDelegate(org.apache.hadoop.hive.metastore.api.Table tab) {
-    if (KuduDdlDelegate.canHandle(tab)) return new KuduDdlDelegate(tab);
-    return new UnsupportedOpDelegate();
+    createTable(tbl, params.if_not_exists, null, response);
   }
 
   /**
@@ -1967,6 +2083,9 @@ public class CatalogOpExecutor {
       switch (params.getTarget()) {
         case TBL_PROPERTY:
           msTbl.getParameters().putAll(properties);
+          if (KuduTable.isKuduTable(msTbl)) {
+            KuduCatalogOpExecutor.validateKuduTblExists(msTbl);
+          }
           break;
         case SERDE_PROPERTY:
           msTbl.getSd().getSerdeInfo().getParameters().putAll(properties);
@@ -2120,7 +2239,6 @@ public class CatalogOpExecutor {
     Preconditions.checkNotNull(cacheOp);
     Preconditions.checkNotNull(params.getPartition_spec());
     // Alter partition params.
-    final String RUNTIME_FILTER_FORMAT = "apply %s on %s";
     TableName tableName = tbl.getTableName();
     HdfsPartition partition = catalog_.getHdfsPartition(
         tableName.getDb(), tableName.getTbl(), params.getPartition_spec());
@@ -2535,16 +2653,6 @@ public class CatalogOpExecutor {
   }
 
   /**
-   * Returns a deep copy of the metastore.api.Table object for the given TableName.
-   */
-  private org.apache.hadoop.hive.metastore.api.Table getMetaStoreTable(
-      TableName tableName) throws CatalogException {
-    Preconditions.checkState(tableName != null && tableName.isFullyQualified());
-    return getExistingTable(tableName.getDb(), tableName.getTbl())
-        .getMetaStoreTable().deepCopy();
-  }
-
-  /**
    * Returns the metastore.api.Table object from the Hive Metastore for an existing
    * fully loaded table.
    */
@@ -2608,7 +2716,7 @@ public class CatalogOpExecutor {
   /**
    * Calculates the next transient_lastDdlTime value.
    */
-  private static long calculateDdlTime(
+  public static long calculateDdlTime(
       org.apache.hadoop.hive.metastore.api.Table msTbl) {
     long existingLastDdlTime = CatalogServiceCatalog.getLastDdlTime(msTbl);
     long currentTime = System.currentTimeMillis() / 1000;
@@ -2617,63 +2725,6 @@ public class CatalogOpExecutor {
   }
 
   /**
-   * Utility function that creates a hive.metastore.api.Table object based on the given
-   * TCreateTableParams.
-   * TODO: Extract metastore object creation utility functions into a separate
-   * helper/factory class.
-   */
-  public static org.apache.hadoop.hive.metastore.api.Table
-      createMetaStoreTable(TCreateTableParams params) {
-    Preconditions.checkNotNull(params);
-    TableName tableName = TableName.fromThrift(params.getTable_name());
-    org.apache.hadoop.hive.metastore.api.Table tbl =
-        new org.apache.hadoop.hive.metastore.api.Table();
-    tbl.setDbName(tableName.getDb());
-    tbl.setTableName(tableName.getTbl());
-    tbl.setOwner(params.getOwner());
-    if (params.isSetTable_properties()) {
-      tbl.setParameters(params.getTable_properties());
-    } else {
-      tbl.setParameters(new HashMap<String, String>());
-    }
-
-    if (params.getComment() != null) {
-      tbl.getParameters().put("comment", params.getComment());
-    }
-    if (params.is_external) {
-      tbl.setTableType(TableType.EXTERNAL_TABLE.toString());
-      tbl.putToParameters("EXTERNAL", "TRUE");
-    } else {
-      tbl.setTableType(TableType.MANAGED_TABLE.toString());
-    }
-
-    StorageDescriptor sd = HiveStorageDescriptorFactory.createSd(
-        params.getFile_format(), RowFormat.fromThrift(params.getRow_format()));
-
-    if (params.isSetSerde_properties()) {
-      if (sd.getSerdeInfo().getParameters() == null) {
-        sd.getSerdeInfo().setParameters(params.getSerde_properties());
-      } else {
-        sd.getSerdeInfo().getParameters().putAll(params.getSerde_properties());
-      }
-    }
-
-    if (params.getLocation() != null) {
-      sd.setLocation(params.getLocation());
-    }
-    // Add in all the columns
-    sd.setCols(buildFieldSchemaList(params.getColumns()));
-    tbl.setSd(sd);
-    if (params.getPartition_columns() != null) {
-      // Add in any partition keys that were specified
-      tbl.setPartitionKeys(buildFieldSchemaList(params.getPartition_columns()));
-    } else {
-      tbl.setPartitionKeys(new ArrayList<FieldSchema>());
-    }
-    return tbl;
-  }
-
-  /**
    * Executes a TResetMetadataRequest and returns the result as a
    * TResetMetadataResponse. Based on the request parameters, this operation
    * may do one of three things:

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/service/Frontend.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index 00a3d93..6d535fd 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -170,9 +170,11 @@ public class Frontend {
   private final AtomicReference<AuthorizationChecker> authzChecker_;
   private final ScheduledExecutorService policyReader_ =
       Executors.newScheduledThreadPool(1);
+  private final String defaultKuduMasterHosts_;
 
-  public Frontend(AuthorizationConfig authorizationConfig) {
-    this(authorizationConfig, new ImpaladCatalog());
+  public Frontend(AuthorizationConfig authorizationConfig,
+      String defaultKuduMasterHosts) {
+    this(authorizationConfig, new ImpaladCatalog(defaultKuduMasterHosts));
   }
 
   /**
@@ -181,6 +183,7 @@ public class Frontend {
   public Frontend(AuthorizationConfig authorizationConfig, ImpaladCatalog catalog) {
     authzConfig_ = authorizationConfig;
     impaladCatalog_ = catalog;
+    defaultKuduMasterHosts_ = catalog.getDefaultKuduMasterHosts();
     authzChecker_ = new AtomicReference<AuthorizationChecker>(
         new AuthorizationChecker(authzConfig_, impaladCatalog_.getAuthPolicy()));
     // If authorization is enabled, reload the policy on a regular basis.
@@ -226,7 +229,7 @@ public class Frontend {
 
     // If this is not a delta, this update should replace the current
     // Catalog contents so create a new catalog and populate it.
-    if (!req.is_delta) catalog = new ImpaladCatalog();
+    if (!req.is_delta) catalog = new ImpaladCatalog(defaultKuduMasterHosts_);
 
     TUpdateCatalogCacheResponse response = catalog.updateCatalog(req);
 

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/service/JniCatalog.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index fc8deaf..7d0af54 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -102,7 +102,7 @@ public class JniCatalog {
     try {
       catalog_.reset();
     } catch (CatalogException e) {
-      LOG.error("Error initialializing Catalog. Please run 'invalidate metadata'", e);
+      LOG.error("Error initializing Catalog. Please run 'invalidate metadata'", e);
     }
     catalogOpExecutor_ = new CatalogOpExecutor(catalog_);
   }

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/service/JniFrontend.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/JniFrontend.java b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
index 07d6ec6..0d502e5 100644
--- a/fe/src/main/java/org/apache/impala/service/JniFrontend.java
+++ b/fe/src/main/java/org/apache/impala/service/JniFrontend.java
@@ -117,7 +117,8 @@ public class JniFrontend {
    */
   public JniFrontend(boolean lazy, String serverName, String authorizationPolicyFile,
       String sentryConfigFile, String authPolicyProviderClass, int impalaLogLevel,
-      int otherLogLevel, boolean allowAuthToLocal) throws InternalException {
+      int otherLogLevel, boolean allowAuthToLocal, String defaultKuduMasterHosts)
+      throws InternalException {
     BackendConfig.setAuthToLocal(allowAuthToLocal);
     GlogAppender.Install(TLogLevel.values()[impalaLogLevel],
         TLogLevel.values()[otherLogLevel]);
@@ -136,7 +137,7 @@ public class JniFrontend {
     }
     LOG.info(JniUtil.getJavaVersion());
 
-    frontend_ = new Frontend(authConfig);
+    frontend_ = new Frontend(authConfig, defaultKuduMasterHosts);
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
new file mode 100644
index 0000000..bd6d0fe
--- /dev/null
+++ b/fe/src/main/java/org/apache/impala/service/KuduCatalogOpExecutor.java
@@ -0,0 +1,240 @@
+// 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.service;
+
+import java.lang.NumberFormatException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.impala.analysis.ToSqlUtils;
+import org.apache.impala.catalog.KuduTable;
+import org.apache.impala.catalog.Table;
+import org.apache.impala.catalog.TableNotFoundException;
+import org.apache.impala.catalog.Type;
+import org.apache.impala.common.ImpalaRuntimeException;
+import org.apache.impala.thrift.TCreateTableParams;
+import org.apache.impala.thrift.TDistributeParam;
+import org.apache.impala.util.KuduUtil;
+import org.apache.kudu.ColumnSchema.ColumnSchemaBuilder;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.client.CreateTableOptions;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.PartialRow;
+import org.apache.log4j.Logger;
+
+/**
+ * This is a helper for the CatalogOpExecutor to provide Kudu related DDL functionality
+ * such as creating and dropping tables from Kudu.
+ */
+public class KuduCatalogOpExecutor {
+  public static final Logger LOG = Logger.getLogger(KuduCatalogOpExecutor.class);
+
+  /**
+   * Create a table in Kudu with a schema equivalent to the schema stored in 'msTbl'.
+   * Throws an exception if 'msTbl' represents an external table or if the table couldn't
+   * be created in Kudu.
+   */
+  static void createManagedTable(org.apache.hadoop.hive.metastore.api.Table msTbl,
+      TCreateTableParams params) throws ImpalaRuntimeException {
+    Preconditions.checkState(!Table.isExternalTable(msTbl));
+    String kuduTableName = msTbl.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    String masterHosts = msTbl.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+    LOG.debug(String.format("Creating table '%s' in master '%s'", kuduTableName,
+        masterHosts));
+    try (KuduClient kudu = new KuduClient.KuduClientBuilder(masterHosts).build()) {
+      // TODO: The IF NOT EXISTS case should be handled by Kudu to ensure atomicity.
+      // (see KUDU-1710).
+      if (kudu.tableExists(kuduTableName)) {
+        if (params.if_not_exists) return;
+        throw new ImpalaRuntimeException(String.format(
+            "Table '%s' already exists in Kudu.", kuduTableName));
+      }
+      Schema schema = createTableSchema(msTbl, params);
+      CreateTableOptions tableOpts = buildTableOptions(msTbl, params, schema);
+      kudu.createTable(kuduTableName, schema, tableOpts);
+    } catch (Exception e) {
+      throw new ImpalaRuntimeException(String.format("Error creating table '%s'",
+          kuduTableName), e);
+    }
+  }
+
+  /**
+   * Creates the schema of a new Kudu table.
+   */
+  private static Schema createTableSchema(
+      org.apache.hadoop.hive.metastore.api.Table msTbl, TCreateTableParams params)
+      throws ImpalaRuntimeException {
+    Set<String> keyColNames = new HashSet<>(params.getPrimary_key_column_names());
+    List<FieldSchema> fieldSchemas = msTbl.getSd().getCols();
+    List<ColumnSchema> colSchemas = new ArrayList<>(fieldSchemas.size());
+    for (FieldSchema fieldSchema : fieldSchemas) {
+      Type type = Type.parseColumnType(fieldSchema.getType());
+      Preconditions.checkState(type != null);
+      org.apache.kudu.Type kuduType = KuduUtil.fromImpalaType(type);
+      // Create the actual column and check if the column is a key column
+      ColumnSchemaBuilder csb =
+          new ColumnSchemaBuilder(fieldSchema.getName(), kuduType);
+      boolean isKeyCol = keyColNames.contains(fieldSchema.getName());
+      csb.key(isKeyCol);
+      csb.nullable(!isKeyCol);
+      colSchemas.add(csb.build());
+    }
+    return new Schema(colSchemas);
+  }
+
+  /**
+   * Builds the table options of a new Kudu table.
+   */
+  private static CreateTableOptions buildTableOptions(
+      org.apache.hadoop.hive.metastore.api.Table msTbl,
+      TCreateTableParams params, Schema schema) throws ImpalaRuntimeException {
+    CreateTableOptions tableOpts = new CreateTableOptions();
+    // Set the distribution schemes
+    List<TDistributeParam> distributeParams = params.getDistribute_by();
+    if (distributeParams != null) {
+      boolean hasRangePartitioning = false;
+      for (TDistributeParam distParam : distributeParams) {
+        if (distParam.isSetBy_hash_param()) {
+          Preconditions.checkState(!distParam.isSetBy_range_param());
+          tableOpts.addHashPartitions(distParam.getBy_hash_param().getColumns(),
+              distParam.getBy_hash_param().getNum_buckets());
+        } else {
+          Preconditions.checkState(distParam.isSetBy_range_param());
+          hasRangePartitioning = true;
+          tableOpts.setRangePartitionColumns(
+              distParam.getBy_range_param().getColumns());
+          for (PartialRow partialRow :
+              KuduUtil.parseSplits(schema, distParam.getBy_range_param())) {
+            tableOpts.addSplitRow(partialRow);
+          }
+        }
+      }
+      // If no range-based distribution is specified in a CREATE TABLE statement, Kudu
+      // generates one by default that includes all the primary key columns. To prevent
+      // this from happening, explicitly set the range partition columns to be
+      // an empty list.
+      if (!hasRangePartitioning) {
+        tableOpts.setRangePartitionColumns(Collections.<String>emptyList());
+      }
+    }
+
+    // Set the number of table replicas, if specified.
+    String replication = msTbl.getParameters().get(KuduTable.KEY_TABLET_REPLICAS);
+    if (!Strings.isNullOrEmpty(replication)) {
+      try {
+        int r = Integer.parseInt(replication);
+        Preconditions.checkState(r > 0);
+        tableOpts.setNumReplicas(r);
+      } catch (NumberFormatException e) {
+        throw new ImpalaRuntimeException(String.format("Invalid number of table " +
+            "replicas specified: '%s'", replication), e);
+      }
+    }
+    return tableOpts;
+  }
+
+  /**
+   * Drops the table in Kudu. If the table does not exist and 'ifExists' is false, a
+   * TableNotFoundException is thrown. If the table exists and could not be dropped,
+   * an ImpalaRuntimeException is thrown.
+   */
+  static void dropTable(org.apache.hadoop.hive.metastore.api.Table msTbl,
+      boolean ifExists) throws ImpalaRuntimeException, TableNotFoundException {
+    Preconditions.checkState(!Table.isExternalTable(msTbl));
+    String tableName = msTbl.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    String masterHosts = msTbl.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+    LOG.debug(String.format("Dropping table '%s' from master '%s'", tableName,
+        masterHosts));
+    try (KuduClient kudu = new KuduClient.KuduClientBuilder(masterHosts).build()) {
+      Preconditions.checkState(!Strings.isNullOrEmpty(tableName));
+      // TODO: The IF EXISTS case should be handled by Kudu to ensure atomicity.
+      // (see KUDU-1710).
+      if (kudu.tableExists(tableName)) {
+        kudu.deleteTable(tableName);
+      } else if (!ifExists) {
+        throw new TableNotFoundException(String.format(
+            "Table '%s' does not exist in Kudu master(s) '%s'.", tableName, masterHosts));
+      }
+    } catch (Exception e) {
+      throw new ImpalaRuntimeException(String.format("Error dropping table '%s'",
+          tableName), e);
+    }
+  }
+
+  /**
+   * Reads the column definitions from a Kudu table and populates 'msTbl' with
+   * an equivalent schema. Throws an exception if any errors are encountered.
+   */
+  public static void populateColumnsFromKudu(
+      org.apache.hadoop.hive.metastore.api.Table msTbl) throws ImpalaRuntimeException {
+    org.apache.hadoop.hive.metastore.api.Table msTblCopy = msTbl.deepCopy();
+    List<FieldSchema> cols = msTblCopy.getSd().getCols();
+    String kuduTableName = msTblCopy.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    Preconditions.checkState(!Strings.isNullOrEmpty(kuduTableName));
+    String masterHosts = msTblCopy.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+    LOG.debug(String.format("Loading schema of table '%s' from master '%s'",
+        kuduTableName, masterHosts));
+    try (KuduClient kudu = new KuduClient.KuduClientBuilder(masterHosts).build()) {
+      if (!kudu.tableExists(kuduTableName)) {
+        throw new ImpalaRuntimeException(String.format("Table does not exist in Kudu: " +
+            "'%s'", kuduTableName));
+      }
+      org.apache.kudu.client.KuduTable kuduTable = kudu.openTable(kuduTableName);
+      // Replace the columns in the Metastore table with the columns from the recently
+      // accessed Kudu schema.
+      cols.clear();
+      for (ColumnSchema colSchema : kuduTable.getSchema().getColumns()) {
+        Type type = KuduUtil.toImpalaType(colSchema.getType());
+        cols.add(new FieldSchema(colSchema.getName(), type.toSql().toLowerCase(), null));
+      }
+    } catch (Exception e) {
+      throw new ImpalaRuntimeException(String.format("Error loading schema of table " +
+          "'%s'", kuduTableName), e);
+    }
+    List<FieldSchema> newCols = msTbl.getSd().getCols();
+    newCols.clear();
+    newCols.addAll(cols);
+  }
+
+  /**
+   * Validates the table properties of a Kudu table. It checks that the specified master
+   * addresses point to valid Kudu masters and that the table exists.
+   * Throws an ImpalaRuntimeException if this is not the case.
+   */
+  public static void validateKuduTblExists(
+      org.apache.hadoop.hive.metastore.api.Table msTbl) throws ImpalaRuntimeException {
+    String masterHosts = msTbl.getParameters().get(KuduTable.KEY_MASTER_HOSTS);
+    Preconditions.checkState(!Strings.isNullOrEmpty(masterHosts));
+    String kuduTableName = msTbl.getParameters().get(KuduTable.KEY_TABLE_NAME);
+    Preconditions.checkState(!Strings.isNullOrEmpty(kuduTableName));
+    try (KuduClient kudu = new KuduClient.KuduClientBuilder(masterHosts).build()) {
+      kudu.tableExists(kuduTableName);
+    } catch (Exception e) {
+      throw new ImpalaRuntimeException(String.format("Kudu table '%s' does not exist " +
+          "on master '%s'", kuduTableName, masterHosts), e);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/041fa6d9/fe/src/main/java/org/apache/impala/util/KuduUtil.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/util/KuduUtil.java b/fe/src/main/java/org/apache/impala/util/KuduUtil.java
index b9f8653..a679032 100644
--- a/fe/src/main/java/org/apache/impala/util/KuduUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/KuduUtil.java
@@ -18,14 +18,15 @@
 package org.apache.impala.util;
 
 import java.io.StringReader;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import javax.json.Json;
 import javax.json.JsonArray;
 import javax.json.JsonReader;
 
+import org.apache.impala.catalog.Catalog;
 import org.apache.impala.catalog.ScalarType;
+import org.apache.impala.catalog.Type;
 import org.apache.impala.common.ImpalaRuntimeException;
 import org.apache.impala.thrift.TDistributeByRangeParam;
 import org.apache.impala.thrift.TRangeLiteral;
@@ -33,48 +34,17 @@ import org.apache.impala.thrift.TRangeLiteralList;
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
-import org.apache.hadoop.hive.metastore.api.FieldSchema;
-import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.kudu.ColumnSchema;
 import org.apache.kudu.Schema;
-import org.apache.kudu.Type;
-import org.apache.kudu.client.KuduTable;
 import org.apache.kudu.client.PartialRow;
 
-import static org.apache.impala.catalog.Type.parseColumnType;
 import static java.lang.String.format;
 
 public class KuduUtil {
 
   private static final String SPLIT_KEYS_ERROR_MESSAGE = "Error parsing splits keys.";
-
-  /**
-   * Compare the schema of a HMS table and a Kudu table. Returns true if both tables have
-   * a matching schema.
-   */
-  public static boolean compareSchema(Table msTable, KuduTable kuduTable)
-      throws ImpalaRuntimeException {
-    List<FieldSchema> msFields = msTable.getSd().getCols();
-    List<ColumnSchema> kuduFields = kuduTable.getSchema().getColumns();
-    if (msFields.size() != kuduFields.size()) return false;
-
-    HashMap<String, ColumnSchema> kuduFieldMap = Maps.newHashMap();
-    for (ColumnSchema kuduField : kuduFields) {
-      kuduFieldMap.put(kuduField.getName().toUpperCase(), kuduField);
-    }
-
-    for (FieldSchema msField : msFields) {
-      ColumnSchema kuduField = kuduFieldMap.get(msField.getName().toUpperCase());
-      if (kuduField == null
-          || fromImpalaType(parseColumnType(msField.getType())) != kuduField.getType()) {
-        return false;
-      }
-    }
-
-    return true;
-  }
+  private static final String KUDU_TABLE_NAME_PREFIX = "impala::";
 
   /**
    * Parses split keys from statements.
@@ -145,10 +115,9 @@ public class KuduUtil {
   /**
    * Sets the value in 'key' at 'pos', given the json representation.
    */
-  private static void setKey(Type type, JsonArray array, int pos, PartialRow key)
-      throws ImpalaRuntimeException {
+  private static void setKey(org.apache.kudu.Type type, JsonArray array, int pos,
+      PartialRow key) throws ImpalaRuntimeException {
     switch (type) {
-      case BOOL: key.addBoolean(pos, array.getBoolean(pos)); break;
       case INT8: key.addByte(pos, (byte) array.getInt(pos)); break;
       case INT16: key.addShort(pos, (short) array.getInt(pos)); break;
       case INT32: key.addInt(pos, array.getInt(pos)); break;
@@ -163,13 +132,9 @@ public class KuduUtil {
   /**
    * Sets the value in 'key' at 'pos', given the range literal.
    */
-  private static void setKey(Type type, TRangeLiteral literal, int pos, String colName,
-      PartialRow key) throws ImpalaRuntimeException {
+  private static void setKey(org.apache.kudu.Type type, TRangeLiteral literal, int pos,
+      String colName, PartialRow key) throws ImpalaRuntimeException {
     switch (type) {
-      case BOOL:
-        checkCorrectType(literal.isSetBool_literal(), type, colName, literal);
-        key.addBoolean(pos, literal.isBool_literal());
-        break;
       case INT8:
         checkCorrectType(literal.isSetInt_literal(), type, colName, literal);
         key.addByte(pos, (byte) literal.getInt_literal());
@@ -200,8 +165,8 @@ public class KuduUtil {
    * If correctType is true, returns. Otherwise throws a formatted error message
    * indicating problems with the type of the literal of the range literal.
    */
-  private static void checkCorrectType(boolean correctType, Type t, String colName,
-      TRangeLiteral literal) throws ImpalaRuntimeException {
+  private static void checkCorrectType(boolean correctType, org.apache.kudu.Type t,
+      String colName, TRangeLiteral literal) throws ImpalaRuntimeException {
     if (correctType) return;
     throw new ImpalaRuntimeException(
         format("Expected %s literal for column '%s' got '%s'", t.getName(), colName,
@@ -220,11 +185,24 @@ public class KuduUtil {
     return Lists.newArrayList(Splitter.on(",").trimResults().split(cols.toLowerCase()));
   }
 
+  public static boolean isSupportedKeyType(org.apache.impala.catalog.Type type) {
+    return type.isIntegerType() || type.isStringType();
+  }
+
+  /**
+   * Return the name that should be used in Kudu when creating a table, assuming a custom
+   * name was not provided.
+   */
+  public static String getDefaultCreateKuduTableName(String metastoreDbName,
+      String metastoreTableName) {
+    return KUDU_TABLE_NAME_PREFIX + metastoreDbName + "." + metastoreTableName;
+  }
+
   /**
    * Converts a given Impala catalog type to the Kudu type. Throws an exception if the
    * type cannot be converted.
    */
-  public static Type fromImpalaType(org.apache.impala.catalog.Type t)
+  public static org.apache.kudu.Type fromImpalaType(Type t)
       throws ImpalaRuntimeException {
     if (!t.isScalarType()) {
       throw new ImpalaRuntimeException(format(
@@ -232,16 +210,16 @@ public class KuduUtil {
     }
     ScalarType s = (ScalarType) t;
     switch (s.getPrimitiveType()) {
-      case TINYINT: return Type.INT8;
-      case SMALLINT: return Type.INT16;
-      case INT: return Type.INT32;
-      case BIGINT: return Type.INT64;
-      case BOOLEAN: return Type.BOOL;
-      case CHAR: return Type.STRING;
-      case STRING: return Type.STRING;
-      case VARCHAR: return Type.STRING;
-      case DOUBLE: return Type.DOUBLE;
-      case FLOAT: return Type.FLOAT;
+      case TINYINT: return org.apache.kudu.Type.INT8;
+      case SMALLINT: return org.apache.kudu.Type.INT16;
+      case INT: return org.apache.kudu.Type.INT32;
+      case BIGINT: return org.apache.kudu.Type.INT64;
+      case BOOLEAN: return org.apache.kudu.Type.BOOL;
+      case CHAR: return org.apache.kudu.Type.STRING;
+      case STRING: return org.apache.kudu.Type.STRING;
+      case VARCHAR: return org.apache.kudu.Type.STRING;
+      case DOUBLE: return org.apache.kudu.Type.DOUBLE;
+      case FLOAT: return org.apache.kudu.Type.FLOAT;
         /* Fall through below */
       case INVALID_TYPE:
       case NULL_TYPE:
@@ -256,11 +234,27 @@ public class KuduUtil {
     }
   }
 
+  public static Type toImpalaType(org.apache.kudu.Type t)
+      throws ImpalaRuntimeException {
+    switch (t) {
+      case BOOL: return Type.BOOLEAN;
+      case DOUBLE: return Type.DOUBLE;
+      case FLOAT: return Type.FLOAT;
+      case INT8: return Type.TINYINT;
+      case INT16: return Type.SMALLINT;
+      case INT32: return Type.INT;
+      case INT64: return Type.BIGINT;
+      case STRING: return Type.STRING;
+      default:
+        throw new ImpalaRuntimeException(String.format(
+            "Kudu type %s is not supported in Impala", t));
+    }
+  }
+
   /**
    * Returns the string value of the RANGE literal.
    */
   static String toString(TRangeLiteral l) throws ImpalaRuntimeException {
-    if (l.isSetBool_literal()) return String.valueOf(l.bool_literal);
     if (l.isSetString_literal()) return String.valueOf(l.string_literal);
     if (l.isSetInt_literal()) return String.valueOf(l.int_literal);
     throw new ImpalaRuntimeException("Unsupported type for RANGE literal.");



Mime
View raw message