@Test
  public void testTransformerWithNonHiveCatalogs() throws Exception {
    try {
      resetHMSClient();
      Table table, tbl2;
      String tblName = "non_hive_exttable";
      String sparkDbName = "sparkdb";
      String catalog = "sparkcat";
      Map<String, Object> tProps = new HashMap<>();
      TableType type = TableType.EXTERNAL_TABLE;
      tProps.put("TBLNAME", tblName);
      tProps.put("TBLTYPE", type);
      tProps.put("CATALOG", catalog);
      tProps.put("DBNAME", sparkDbName);
      StringBuilder table_params = new StringBuilder();
      table_params.append("key1=val1");
      table_params.append(";");
      table_params.append("EXTERNAL").append("=").append("TRUE");
      tProps.put("PROPERTIES", table_params.toString());

      List<String> capabilities = new ArrayList<>();
      setHMSClient("TestCreateTableNonHive#1", (String[])(capabilities.toArray(new String[0])));

      try {
        table = createTableWithCapabilities(tProps);
        LOG.info("Create non-hive table is expected to succeed and has succeeded"); // no transformation for views
      } catch (Exception e) {
        fail("Create non-hive table expected to succeed but has failed. <" + e.getMessage() +">");
      }

      tbl2 = client.getTable(catalog, sparkDbName, tblName);
      assertEquals("TableName expected to be " + tblName, tblName, tbl2.getTableName());
      assertEquals("TableType expected to be EXTERNAL", TableType.EXTERNAL_TABLE.name(), tbl2.getTableType());
      assertNull("Table's ReadCapabilities is expected to be null", tbl2.getRequiredReadCapabilities());
      assertNull("Table's WriteCapabilities is expected to be null", tbl2.getRequiredWriteCapabilities());

      String newLocation = wh.getAbsolutePath().concat(File.separator).concat(sparkDbName).concat(File.separator)
          .concat(tblName);
      tbl2.getSd().setLocation(newLocation);

      setHMSClient("TestAlterTableNonHive#1", (String[])(capabilities.toArray(new String[0])));
      try {
        client.alter_table(catalog, sparkDbName, tblName, tbl2);
        LOG.info("alter_table succeeded with new location in managed warehouse as expected");
      } catch (Exception e) {
        fail("alter_table expected to succeed but failed with new location:" + newLocation);
      }

      tbl2 = client.getTable(catalog, sparkDbName, tblName);
      assertEquals("TableType expected to be EXTERNAL", TableType.EXTERNAL_TABLE.name(), tbl2.getTableType());
      int idx = (tbl2.getSd().getLocation().indexOf(":") > 0) ? tbl2.getSd().getLocation().indexOf(":") : 0;
      assertEquals("Table location expected to be in external warehouse", newLocation, tbl2.getSd().getLocation().substring(idx+1));

      tblName = "non_hive_mgdtable";
      tProps = new HashMap<>();
      type = TableType.MANAGED_TABLE;
      tProps.put("TBLNAME", tblName);
      tProps.put("TBLTYPE", type);
      tProps.put("CATALOG", catalog);
      tProps.put("DBNAME", sparkDbName);
      tProps.put("DROPDB", Boolean.FALSE);
      table_params = new StringBuilder();
      table_params.append("key1=val1");
      table_params.append(";");
      tProps.put("PROPERTIES", table_params.toString());

      capabilities.add("CONNECTORWRITE");
      setHMSClient("TestCreateTableNonHive#2", (String[])(capabilities.toArray(new String[0])));

      try {
        table = createTableWithCapabilities(tProps);
        LOG.info("Create non-hive MGD table is expected to succeed and has succeeded"); // no transformation for views
      } catch (Exception e) {
        fail("Create non-hive MGD table expected to succeed but has failed. <" + e.getMessage() +">");
      }

      tbl2 = client.getTable(catalog, sparkDbName, tblName);
      assertEquals("TableName expected to be " + tblName, tblName, tbl2.getTableName());
      assertEquals("TableType expected to be MANAGED", TableType.MANAGED_TABLE.name(), tbl2.getTableType());
      assertNull("Table's ReadCapabilities is expected to be null", tbl2.getRequiredReadCapabilities());
      assertNull("Table's WriteCapabilities is expected to be null", tbl2.getRequiredWriteCapabilities());


      // TESTS to ensure AlterTable does not go thru translation for non-hive catalog objects
      setHMSClient("TestAlterTableNonHive#2", (String[])(capabilities.toArray(new String[0])));
      tbl2 = client.getTable(catalog, sparkDbName, tblName);
      newLocation = ext_wh.getAbsolutePath().concat(File.separator).concat(sparkDbName).concat(File.separator)
          .concat(tblName);
      tbl2.getSd().setLocation(newLocation);

      try {
        client.alter_table(catalog, sparkDbName, tblName, tbl2);
        LOG.info("alter_table succeeded with new location in external warehouse as expected");
      } catch (Exception e) {
        fail("alter_table expected to succeed but failed with new location:" + newLocation);
      }

      tbl2 = client.getTable(catalog, sparkDbName, tblName);
      assertEquals("TableType expected to be MANAGED", TableType.MANAGED_TABLE.name(), tbl2.getTableType());
      idx = (tbl2.getSd().getLocation().indexOf(":") > 0) ? tbl2.getSd().getLocation().indexOf(":") : 0;
      assertEquals("Table location expected to be in managed warehouse", newLocation, tbl2.getSd().getLocation().substring(idx+1));

      // Test getTablesExt with many tables.
      sparkDbName = "sparkdbext";
      tblName = "test_get_tables_ext";
      int count = 10;

      tProps = new HashMap<>();
      capabilities = new ArrayList<>();
      capabilities.add("EXTREAD");
      tProps.put("CATALOG", catalog);
      tProps.put("DBNAME", sparkDbName);
      tProps.put("TBLNAME", tblName);
      type = TableType.MANAGED_TABLE;
      tProps.put("TABLECOUNT", count);
      tProps.put("TBLTYPE", type);
      table_params = new StringBuilder();
      table_params.append("key1=val1");
      table_params.append(";");
      tProps.put("PROPERTIES", table_params.toString());

      setHMSClient("test_get_tables_ext", (String[])(capabilities.toArray(new String[0])));

      List<String> tables = createTables(tProps);
      int requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.PROCESSOR_CAPABILITIES)).bitValue();
      List<ExtendedTableInfo> extTables = client.getTablesExt(catalog, sparkDbName, "*", requestedFields, 2000);
      LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
      assertEquals("Return list size does not match expected size:extTables", count, extTables.size());
      for (ExtendedTableInfo tableInfo : extTables) {
        assertNull("Return object should not have read capabilities", tableInfo.getRequiredReadCapabilities());
        assertNull("Return object should not have write capabilities", tableInfo.getRequiredWriteCapabilities());
        assertEquals("AccessType not expected to be set", 0, tableInfo.getAccessType());
      }

      requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.ACCESS_TYPE)).bitValue();
      extTables = client.getTablesExt(catalog, sparkDbName, "*", requestedFields, 2000);
      LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
      assertEquals("Return list size does not match expected size", count, extTables.size());
      for (ExtendedTableInfo tableInfo : extTables) {
        assertNull("Return object should not have read capabilities", tableInfo.getRequiredReadCapabilities());
        assertNull("Return object should not have write capabilities", tableInfo.getRequiredWriteCapabilities());
        assertTrue("AccessType not expected to be set", tableInfo.getAccessType() <= 0);
      }

      requestedFields = (new GetTablesRequestBuilder().with(GetTablesExtRequestFields.ALL)).bitValue();
      extTables = client.getTablesExt(catalog, sparkDbName, "*", requestedFields, 2000);
      LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
      assertEquals("Return list size does not match expected size", count, extTables.size());
      for (ExtendedTableInfo tableInfo : extTables) {
        assertTrue("AccessType not expected to be set", tableInfo.getAccessType() <= 0);
      }

      extTables = client.getTablesExt(catalog, sparkDbName, "*", requestedFields, (count - 3));
      LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
      assertEquals("Return list size does not match expected size", (count - 3), extTables.size());
      for (ExtendedTableInfo tableInfo : extTables) {
        assertTrue("AccessType not expected to be set", tableInfo.getAccessType() <= 0);
      }

      extTables = client.getTablesExt(catalog, sparkDbName, "*", requestedFields, -1);
      LOG.debug("Return list size=" + extTables.size() + ",bitValue=" + requestedFields);
      assertEquals("Return list size does not match expected size", count, extTables.size());

      count = 300;
      tProps.put("TBLNAME", "test_limit");
      tProps.put("TABLECOUNT", count);
      tables = createTables(tProps);
      assertEquals("Unexpected number of tables created", count, tables.size());

      extTables = client.getTablesExt(catalog, sparkDbName, "test_limit*", requestedFields, count);
      assertEquals("Unexpected number of tables returned", count, extTables.size());

      extTables = client.getTablesExt(catalog, sparkDbName, "test_limit*", requestedFields, (count/2));
      assertEquals("Unexpected number of tables returned", (count/2), extTables.size());

      extTables = client.getTablesExt(catalog, sparkDbName, "test_limit*", requestedFields, 1);
      assertEquals("Unexpected number of tables returned", 1, extTables.size());

    } catch (Exception e) {
      System.err.println(org.apache.hadoop.util.StringUtils.stringifyException(e));
      System.err.println("testCreateTable() failed.");
      fail("testCreateTable failed:" + e.getMessage());
    } finally {
      resetHMSClient();
    }
  }
EXTERNAL_TABLE
  1) Old table (name in upper case) with no capabilities
  2) Old table with no capabilities
  3a) New table with capabilities with no client requirements
  3b) New table with capabilities with no matching client requirements
  3c) New table with capabilities with partial match requirements
  3d) New table with capabilities with full match requirements
itests/hive-unit/src/test/java/org/apache/hadoop/hive/metastore/TestHiveMetastoreTransformer.java