@Override
  public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)
      throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
    checkNotNull(tablePath, "Table path cannot be null");
    checkNotNull(table, "Table cannot be null");

    if (!databaseExists(tablePath.getDatabaseName())) {
      throw new DatabaseNotExistException(getName(), tablePath.getDatabaseName());
    }

    if (!table.getOptions().getOrDefault(CONNECTOR.key(), "").equalsIgnoreCase(HoodieTableFactory.FACTORY_ID)) {
      throw new HoodieCatalogException(String.format("Unsupported connector identity %s, supported identity is %s",
          table.getOptions().getOrDefault(CONNECTOR.key(), ""), HoodieTableFactory.FACTORY_ID));
    }

    if (table instanceof CatalogView) {
      throw new HoodieCatalogException("CREATE VIEW is not supported.");
    }

    try {
      boolean isMorTable = OptionsResolver.isMorTable(table.getOptions());
      Table hiveTable = instantiateHiveTable(tablePath, table, inferTablePath(tablePath, table), isMorTable);
      //create hive table
      client.createTable(hiveTable);
      //init hoodie metaClient
      initTableIfNotExists(tablePath, (CatalogTable) table);
    } catch (AlreadyExistsException e) {
      if (!ignoreIfExists) {
        throw new TableAlreadyExistException(getName(), tablePath, e);
      }
    } catch (Exception e) {
      throw new HoodieCatalogException(
          String.format("Failed to create table %s", tablePath.getFullName()), e);
    }
  }
@Test
  public void testCreateHoodieTableWithWrongTableType() {
    HashMap<String,String> properties = new HashMap<>();
    properties.put(FactoryUtil.CONNECTOR.key(), "hudi");
    properties.put("table.type","wrong type");
    CatalogTable table =
            new CatalogTableImpl(schema,  properties, "hudi table");
    assertThrows(HoodieCatalogException.class, () -> hoodieCatalog.createTable(tablePath, table, false));
  }