IMPALA-7257. Support Kudu tables in LocalCatalog

This adds support for querying Kudu tables when --use_local_catalog is
enabled.

With this change, most of the kudu e2e tests pass. Those that don't are
related to missing support for CREATE TABLE AS SELECT, which currently
has some downcasts to 'Db' causing issues. That will be addressed in a
separate patch along with fixing CTAS for FS-backed tables.

Change-Id: I5b6a317ee895e43e00ade953e814867b56b4e6dd
Reviewed-on: http://gerrit.cloudera.org:8080/10912
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Reviewed-by: Tianyi Wang <twang@cloudera.com>

Author: toddlipcon

fe/src/main/java/org/apache/impala/analysis/AlterTableAddDropRangePartitionStmt.java

@@ -20,8 +20,8 @@
 import java.util.List;
 
 import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableAddDropRangePartitionParams;
 import org.apache.impala.thrift.TAlterTableParams;
@@ -88,12 +88,12 @@ public TAlterTableParams toThrift() {
   public void analyze(Analyzer analyzer) throws AnalysisException {
     super.analyze(analyzer);
     FeTable table = getTargetTable();
-    if (!(table instanceof KuduTable)) {
+    if (!(table instanceof FeKuduTable)) {
       throw new AnalysisException(String.format("Table %s does not support range " +
           "partitions: RANGE %s", table.getFullName(), rangePartitionSpec_.toSql()));
     }
-    KuduTable kuduTable = (KuduTable) table;
-    List<String> colNames = kuduTable.getRangePartitioningColNames();
+    FeKuduTable kuduTable = (FeKuduTable) table;
+    List<String> colNames = FeKuduTable.Utils.getRangePartitioningColNames(kuduTable);
     if (colNames.isEmpty()) {
       throw new AnalysisException(String.format("Cannot add/drop partition %s: " +
           "Kudu table %s doesn't have a range-based partitioning.",

fe/src/main/java/org/apache/impala/analysis/AlterTableAddPartitionStmt.java

@@ -21,6 +21,7 @@
 import com.google.common.base.Joiner;
 import com.google.common.collect.Sets;
 
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
@@ -84,7 +85,7 @@ public TAlterTableParams toThrift() {
   public void analyze(Analyzer analyzer) throws AnalysisException {
     super.analyze(analyzer);
     FeTable table = getTargetTable();
-    if (table instanceof KuduTable) {
+    if (table instanceof FeKuduTable) {
       throw new AnalysisException("ALTER TABLE ADD PARTITION is not supported for " +
           "Kudu tables: " + table.getTableName());
     }

fe/src/main/java/org/apache/impala/analysis/AlterTableAddReplaceColsStmt.java

@@ -23,9 +23,9 @@
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 
 import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.HBaseTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableAddReplaceColsParams;
 import org.apache.impala.thrift.TAlterTableParams;
@@ -80,7 +80,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
           "supported on HBase tables.");
     }
 
-    boolean isKuduTable = t instanceof KuduTable;
+    boolean isKuduTable = t instanceof FeKuduTable;
     if (isKuduTable && replaceExistingCols_) {
       throw new AnalysisException("ALTER TABLE REPLACE COLUMNS is not " +
           "supported on Kudu tables.");

fe/src/main/java/org/apache/impala/analysis/AlterTableAlterColStmt.java

@@ -20,10 +20,10 @@
 import java.util.Map;
 
 import org.apache.impala.catalog.Column;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.HBaseTable;
 import org.apache.impala.catalog.KuduColumn;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableAlterColParams;
 import org.apache.impala.thrift.TAlterTableParams;
@@ -134,7 +134,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
     }
     if (newColDef_.hasKuduOptions()) {
       // Disallow Kudu options on non-Kudu tables.
-      if (!(t instanceof KuduTable)) {
+      if (!(t instanceof FeKuduTable)) {
         if (isDropDefault_) {
           throw new AnalysisException(String.format(
               "Unsupported column option for non-Kudu table: DROP DEFAULT"));
@@ -153,7 +153,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
                 newColDef_.toString()));
       }
     }
-    if (t instanceof KuduTable) {
+    if (t instanceof FeKuduTable) {
       KuduColumn col = (KuduColumn) t.getColumn(colName_);
       if (!col.getType().equals(newColDef_.getType())) {
         throw new AnalysisException(String.format("Cannot change the type of a Kudu " +

fe/src/main/java/org/apache/impala/analysis/AlterTableDropPartitionStmt.java

@@ -18,8 +18,8 @@
 package org.apache.impala.analysis;
 
 import org.apache.impala.authorization.Privilege;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableDropPartitionParams;
 import org.apache.impala.thrift.TAlterTableParams;
@@ -75,7 +75,7 @@ public TAlterTableParams toThrift() {
   public void analyze(Analyzer analyzer) throws AnalysisException {
     super.analyze(analyzer);
     FeTable table = getTargetTable();
-    if (table instanceof KuduTable) {
+    if (table instanceof FeKuduTable) {
       throw new AnalysisException("ALTER TABLE DROP PARTITION is not supported for " +
           "Kudu tables: " + partitionSet_.toSql());
     }

fe/src/main/java/org/apache/impala/analysis/AlterTableSetFileFormatStmt.java

@@ -17,8 +17,8 @@
 
 package org.apache.impala.analysis;
 
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableParams;
 import org.apache.impala.thrift.TAlterTableSetFileFormatParams;
@@ -56,7 +56,7 @@ public TAlterTableParams toThrift() {
   public void analyze(Analyzer analyzer) throws AnalysisException {
     super.analyze(analyzer);
     FeTable tbl = getTargetTable();
-    if (tbl instanceof KuduTable) {
+    if (tbl instanceof FeKuduTable) {
       throw new AnalysisException("ALTER TABLE SET FILEFORMAT is not supported " +
           "on Kudu tables: " + tbl.getFullName());
     }

fe/src/main/java/org/apache/impala/analysis/AlterTableSetLocationStmt.java

@@ -23,10 +23,10 @@
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.impala.authorization.Privilege;
 import org.apache.impala.catalog.FeFsPartition;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.HdfsPartition;
 import org.apache.impala.catalog.HdfsTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableParams;
 import org.apache.impala.thrift.TAlterTableSetLocationParams;
@@ -111,7 +111,7 @@ public String apply(FeFsPartition hdfsPartition) {
             "uncache before changing the location using: ALTER TABLE %s SET UNCACHED",
             table.getFullName()));
       }
-    } else if (table instanceof KuduTable) {
+    } else if (table instanceof FeKuduTable) {
       throw new AnalysisException("ALTER TABLE SET LOCATION is not supported on Kudu " +
           "tables: " + table.getFullName());
     }

fe/src/main/java/org/apache/impala/analysis/AlterTableSetTblProperties.java

@@ -27,6 +27,7 @@
 import org.apache.impala.authorization.PrivilegeRequestBuilder;
 import org.apache.impala.catalog.Column;
 import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.HBaseTable;
 import org.apache.impala.catalog.HdfsTable;
@@ -93,7 +94,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
           hive_metastoreConstants.META_TABLE_STORAGE));
     }
 
-    if (getTargetTable() instanceof KuduTable) analyzeKuduTable(analyzer);
+    if (getTargetTable() instanceof FeKuduTable) analyzeKuduTable(analyzer);
 
     // Check avro schema when it is set in avro.schema.url or avro.schema.literal to
     // avoid potential metadata corruption (see IMPALA-2042).
@@ -211,7 +212,7 @@ public static List<Integer> analyzeSortColumns(FeTable table,
     // AlterTableSetStmt::analyze().
     Preconditions.checkState(!(table instanceof HBaseTable));
 
-    if (table instanceof KuduTable) {
+    if (table instanceof FeKuduTable) {
       throw new AnalysisException(String.format("'%s' table property is not supported " +
           "for Kudu tables.", AlterTableSortByStmt.TBL_PROP_SORT_COLUMNS));
     }

fe/src/main/java/org/apache/impala/analysis/AlterTableSortByStmt.java

@@ -20,9 +20,9 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.HBaseTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.common.AnalysisException;
 import org.apache.impala.thrift.TAlterTableParams;
 import org.apache.impala.thrift.TAlterTableSetTblPropertiesParams;
@@ -71,7 +71,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
     if (targetTable instanceof HBaseTable) {
       throw new AnalysisException("ALTER TABLE SORT BY not supported on HBase tables.");
     }
-    if (targetTable instanceof KuduTable) {
+    if (targetTable instanceof FeKuduTable) {
       throw new AnalysisException("ALTER TABLE SORT BY not supported on Kudu tables.");
     }
 

fe/src/main/java/org/apache/impala/analysis/Analyzer.java

@@ -38,17 +38,16 @@
 import org.apache.impala.authorization.PrivilegeRequestBuilder;
 import org.apache.impala.authorization.User;
 import org.apache.impala.catalog.Column;
-import org.apache.impala.catalog.DataSourceTable;
 import org.apache.impala.catalog.DatabaseNotFoundException;
 import org.apache.impala.catalog.FeCatalog;
 import org.apache.impala.catalog.FeDataSourceTable;
 import org.apache.impala.catalog.FeDb;
 import org.apache.impala.catalog.FeFsTable;
+import org.apache.impala.catalog.FeKuduTable;
 import org.apache.impala.catalog.FeTable;
 import org.apache.impala.catalog.FeView;
 import org.apache.impala.catalog.HBaseTable;
 import org.apache.impala.catalog.IncompleteTable;
-import org.apache.impala.catalog.KuduTable;
 import org.apache.impala.catalog.TableLoadingException;
 import org.apache.impala.catalog.Type;
 import org.apache.impala.common.AnalysisException;
@@ -589,7 +588,7 @@ public TableRef resolveTableRef(TableRef tableRef) throws AnalysisException {
       if (table instanceof FeView) return new InlineViewRef((FeView) table, tableRef);
       // The table must be a base table.
       Preconditions.checkState(table instanceof FeFsTable ||
-          table instanceof KuduTable ||
+          table instanceof FeKuduTable ||
           table instanceof HBaseTable ||
           table instanceof FeDataSourceTable);
       return new BaseTableRef(tableRef, resolvedPath);