@Override
  public void createTable(ObjectPath tablePath, CatalogBaseTable catalogTable, boolean ignoreIfExists)
      throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
    if (!databaseExists(tablePath.getDatabaseName())) {
      throw new DatabaseNotExistException(getName(), tablePath.getDatabaseName());
    }
    if (tableExists(tablePath)) {
      if (ignoreIfExists) {
        return;
      } else {
        throw new TableAlreadyExistException(getName(), tablePath);
      }
    }

    if (catalogTable instanceof CatalogView) {
      throw new UnsupportedOperationException(
          "Hudi catalog doesn't support to CREATE VIEW.");
    }

    ResolvedCatalogTable resolvedTable = (ResolvedCatalogTable) catalogTable;
    final String tablePathStr = inferTablePath(catalogPathStr, tablePath);
    Map<String, String> options = applyOptionsHook(tablePathStr, catalogTable.getOptions());
    Configuration conf = Configuration.fromMap(options);
    conf.setString(FlinkOptions.PATH, tablePathStr);
    ResolvedSchema resolvedSchema = resolvedTable.getResolvedSchema();
    if (!resolvedSchema.getPrimaryKey().isPresent() && !conf.containsKey(RECORD_KEY_FIELD.key())) {
      throw new CatalogException("Primary key definition is missing");
    }
    final String avroSchema = AvroSchemaConverter.convertToSchema(
        resolvedSchema.toPhysicalRowDataType().getLogicalType(),
        AvroSchemaUtils.getAvroRecordQualifiedName(tablePath.getObjectName())).toString();
    conf.setString(FlinkOptions.SOURCE_AVRO_SCHEMA, avroSchema);

    // stores two copies of options:
    // - partition keys
    // - primary keys
    // because the HoodieTableMetaClient is a heavy impl, we try to avoid initializing it
    // when calling #getTable.

    //set pk
    if (resolvedSchema.getPrimaryKey().isPresent()
            && !conf.containsKey(FlinkOptions.RECORD_KEY_FIELD.key())) {
      final String pkColumns = String.join(",", resolvedSchema.getPrimaryKey().get().getColumns());
      conf.setString(RECORD_KEY_FIELD, pkColumns);
    }

    if (resolvedSchema.getPrimaryKey().isPresent()) {
      options.put(TableOptionProperties.PK_CONSTRAINT_NAME, resolvedSchema.getPrimaryKey().get().getName());
    }
    if (conf.containsKey(RECORD_KEY_FIELD.key())) {
      options.put(TableOptionProperties.PK_COLUMNS, conf.getString(RECORD_KEY_FIELD));
    }

    // check preCombine
    StreamerUtil.checkPreCombineKey(conf, resolvedSchema.getColumnNames());

    if (resolvedTable.isPartitioned()) {
      final String partitions = String.join(",", resolvedTable.getPartitionKeys());
      conf.setString(FlinkOptions.PARTITION_PATH_FIELD, partitions);
      options.put(TableOptionProperties.PARTITION_COLUMNS, partitions);

      final String[] pks = conf.getString(FlinkOptions.RECORD_KEY_FIELD).split(",");
      boolean complexHoodieKey = pks.length > 1 || resolvedTable.getPartitionKeys().size() > 1;
      StreamerUtil.checkKeygenGenerator(complexHoodieKey, conf);
    } else {
      conf.setString(FlinkOptions.KEYGEN_CLASS_NAME.key(), NonpartitionedAvroKeyGenerator.class.getName());
    }
    conf.setString(FlinkOptions.TABLE_NAME, tablePath.getObjectName());
    try {
      StreamerUtil.initTableIfNotExists(conf);
      // prepare the non-table-options properties
      if (!StringUtils.isNullOrEmpty(resolvedTable.getComment())) {
        options.put(TableOptionProperties.COMMENT, resolvedTable.getComment());
      }
      TableOptionProperties.createProperties(tablePathStr, hadoopConf, options);
    } catch (IOException e) {
      throw new CatalogException(String.format("Initialize table path %s exception.", tablePathStr), e);
    }
  }
@Test
  public void testCreateTable() throws Exception {
    ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
    // test create table
    catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);

    // test table exist
    assertTrue(catalog.tableExists(tablePath));

    // validate the full name of table create schema
    HoodieTableConfig tableConfig = StreamerUtil.getTableConfig(
        catalog.getTable(tablePath).getOptions().get(FlinkOptions.PATH.key()),
        HadoopConfigurations.getHadoopConf(new Configuration())).get();
    Option<org.apache.avro.Schema> tableCreateSchema = tableConfig.getTableCreateSchema();
    assertTrue(tableCreateSchema.isPresent(), "Table should have been created");
    assertThat(tableCreateSchema.get().getFullName(), is("hoodie.tb1.tb1_record"));

    // test create exist table
    assertThrows(TableAlreadyExistException.class,
        () -> catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, false));

    // validate key generator for partitioned table
    HoodieTableMetaClient metaClient = createMetaClient(
        new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())),
        catalog.inferTablePath(catalogPathStr, tablePath));
    String keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
    assertEquals(keyGeneratorClassName, SimpleAvroKeyGenerator.class.getName());

    // validate single key and multiple partition for partitioned table
    ObjectPath singleKeyMultiplePartitionPath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb_skmp" + System.currentTimeMillis());
    final ResolvedCatalogTable singleKeyMultiplePartitionTable = new ResolvedCatalogTable(
        CatalogTable.of(
            Schema.newBuilder().fromResolvedSchema(CREATE_TABLE_SCHEMA).build(),
            "test",
            Lists.newArrayList("par1", "par2"),
            EXPECTED_OPTIONS),
        CREATE_TABLE_SCHEMA
    );

    catalog.createTable(singleKeyMultiplePartitionPath, singleKeyMultiplePartitionTable, false);
    metaClient = createMetaClient(
        new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())),
        catalog.inferTablePath(catalogPathStr, singleKeyMultiplePartitionPath));
    keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
    assertThat(keyGeneratorClassName, is(ComplexAvroKeyGenerator.class.getName()));

    // validate multiple key and single partition for partitioned table
    ObjectPath multipleKeySinglePartitionPath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb_mksp" + System.currentTimeMillis());
    final ResolvedCatalogTable multipleKeySinglePartitionTable = new ResolvedCatalogTable(
        CatalogTable.of(
            Schema.newBuilder().fromResolvedSchema(CREATE_MULTI_KEY_TABLE_SCHEMA).build(),
            "test",
            Lists.newArrayList("par1"),
            EXPECTED_OPTIONS),
        CREATE_TABLE_SCHEMA
    );

    catalog.createTable(multipleKeySinglePartitionPath, multipleKeySinglePartitionTable, false);
    metaClient = createMetaClient(
        new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())),
        catalog.inferTablePath(catalogPathStr, singleKeyMultiplePartitionPath));
    keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
    assertThat(keyGeneratorClassName, is(ComplexAvroKeyGenerator.class.getName()));

    // validate key generator for non partitioned table
    ObjectPath nonPartitionPath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb");
    final ResolvedCatalogTable nonPartitionCatalogTable = new ResolvedCatalogTable(
        CatalogTable.of(
            Schema.newBuilder().fromResolvedSchema(CREATE_TABLE_SCHEMA).build(),
            "test",
            new ArrayList<>(),
            EXPECTED_OPTIONS),
        CREATE_TABLE_SCHEMA
    );

    catalog.createTable(nonPartitionPath, nonPartitionCatalogTable, false);

    metaClient = createMetaClient(
        new HadoopStorageConfiguration(HadoopConfigurations.getHadoopConf(new Configuration())),
        catalog.inferTablePath(catalogPathStr, nonPartitionPath));
    keyGeneratorClassName = metaClient.getTableConfig().getKeyGeneratorClassName();
    assertEquals(keyGeneratorClassName, NonpartitionedAvroKeyGenerator.class.getName());
  }