public static void createTable(String databaseName, String tableName, RowType schema,
            PartitionPolicy partitionPolicy, String hiveVersion, String inputFormat, String outputFormat,
            String serializationLib) {
        HiveConf hiveConf = HiveTableInlongFactory.getHiveConf();
        try (HiveMetastoreClientWrapper client = HiveMetastoreClientFactory.create(hiveConf, hiveVersion)) {

            List<String> dbs = client.getAllDatabases();
            boolean dbExist = dbs.stream().anyMatch(databaseName::equalsIgnoreCase);
            // create database if not exists
            if (!dbExist) {
                Database defaultDb = client.getDatabase("default");
                Database targetDb = new Database();
                targetDb.setName(databaseName);
                targetDb.setLocationUri(defaultDb.getLocationUri() + "/" + databaseName + ".db");
                client.createDatabase(targetDb);
            }

            String sinkPartitionName = hiveConf.get(SINK_PARTITION_NAME.key(), SINK_PARTITION_NAME.defaultValue());
            FieldSchema defaultPartition = new FieldSchema(sinkPartitionName, "string", "");

            List<FieldSchema> fieldSchemaList = new ArrayList<>();
            for (RowField field : schema.getFields()) {
                if (!field.getName().equals(sinkPartitionName)) {
                    FieldSchema hiveFieldSchema = new FieldSchema(field.getName(), flinkType2HiveType(field.getType()),
                            "");
                    fieldSchemaList.add(hiveFieldSchema);
                }
            }

            Table table = new Table();
            table.setDbName(databaseName);
            table.setTableName(tableName);

            if (PartitionPolicy.NONE != partitionPolicy) {
                table.setPartitionKeys(Collections.singletonList(defaultPartition));
            }

            StorageDescriptor sd = new StorageDescriptor();
            table.setSd(sd);
            sd.setCols(fieldSchemaList);
            sd.setInputFormat(inputFormat);
            sd.setOutputFormat(outputFormat);
            sd.setSerdeInfo(new SerDeInfo());
            sd.getSerdeInfo().setSerializationLib(serializationLib);
            client.createTable(table);
            LOG.info("create table {}.{}", databaseName, tableName);
        } catch (TException e) {
            throw new CatalogException("Failed to create database", e);
        }
    }
create hive table with default `pt` partition field

@param databaseName database name
@param tableName table name
@param schema flink field type
@param partitionPolicy policy of partitioning table
@param hiveVersion hive version
@param inputFormat the input format of storage descriptor
@param outputFormat the output format of storage descriptor
@param serializationLib the serialization library of storage descriptor
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/util/HiveTableUtil.java