public boolean createTable(CreateTableStmt stmt, List<Column> partitionColumns) throws DdlException {
        String dbName = stmt.getDbName();
        String tableName = stmt.getTableName();
        Map<String, String> properties = stmt.getProperties() != null ? stmt.getProperties() : new HashMap<>();
        Path tablePath = null;
        boolean tableLocationExists = false;
        if (!stmt.isExternal()) {
            checkLocationProperties(properties);
            if (!Strings.isNullOrEmpty(properties.get(LOCATION_PROPERTY))) {
                String tableLocationWithUserAssign = properties.get(LOCATION_PROPERTY);
                tablePath = new Path(tableLocationWithUserAssign);
                if (pathExists(tablePath, hadoopConf)) {
                    tableLocationExists = true;
                    if (!isEmpty(tablePath, hadoopConf)) {
                        throw new StarRocksConnectorException("not support creating table under non-empty directory: %s",
                                tableLocationWithUserAssign);
                    }
                }
            } else {
                tablePath = getDefaultLocation(dbName, tableName);
            }
        } else {
            // checkExternalLocationProperties(properties);
            if (properties.containsKey(EXTERNAL_LOCATION_PROPERTY)) {
                tablePath = new Path(properties.get(EXTERNAL_LOCATION_PROPERTY));
            } else if (properties.containsKey(LOCATION_PROPERTY)) {
                tablePath = new Path(properties.get(LOCATION_PROPERTY));
            }
            tableLocationExists = true;
        }

        HiveStorageFormat.check(properties);

        List<String> partitionColNames;
        if (partitionColumns.isEmpty()) {
            partitionColNames = stmt.getPartitionDesc() != null ?
                    ((ListPartitionDesc) stmt.getPartitionDesc()).getPartitionColNames() : new ArrayList<>();
        } else {
            partitionColNames = partitionColumns.stream().map(Column::getName).collect(Collectors.toList());
        }

        // default is managed table
        HiveTable.HiveTableType tableType = HiveTable.HiveTableType.MANAGED_TABLE;
        if (stmt.isExternal()) {
            tableType = HiveTable.HiveTableType.EXTERNAL_TABLE;
        }
        HiveTable.Builder builder = HiveTable.builder()
                .setId(ConnectorTableId.CONNECTOR_ID_GENERATOR.getNextId().asInt())
                .setTableName(tableName)
                .setCatalogName(catalogName)
                .setResourceName(toResourceName(catalogName, "hive"))
                .setHiveDbName(dbName)
                .setHiveTableName(tableName)
                .setPartitionColumnNames(partitionColNames)
                .setDataColumnNames(stmt.getColumns().stream()
                        .map(Column::getName)
                        .collect(Collectors.toList()).subList(0, stmt.getColumns().size() - partitionColNames.size()))
                .setFullSchema(stmt.getColumns())
                .setTableLocation(tablePath == null ? null : tablePath.toString())
                .setProperties(stmt.getProperties())
                .setStorageFormat(HiveStorageFormat.get(properties.getOrDefault(FILE_FORMAT, "parquet")))
                .setCreateTime(System.currentTimeMillis())
                .setHiveTableType(tableType);
        Table table = builder.build();
        try {
            if (!tableLocationExists) {
                createDirectory(tablePath, hadoopConf);
            }
            metastore.createTable(dbName, table);
        } catch (Exception e) {
            LOG.error("Failed to create table {}.{}", dbName, tableName);
            boolean shouldDelete;
            try {
                if (tableExists(dbName, tableName)) {
                    LOG.warn("Table {}.{} already exists. But some error occur such as accessing meta service timeout",
                            dbName, table, e);
                    return true;
                }
                FileSystem fileSystem = FileSystem.get(URI.create(tablePath.toString()), hadoopConf);
                shouldDelete = !fileSystem.listLocatedStatus(tablePath).hasNext() && !tableLocationExists;
                if (shouldDelete) {
                    fileSystem.delete(tablePath);
                }
            } catch (Exception e1) {
                LOG.error("Failed to delete table location {}", tablePath, e);
            }
            throw new DdlException(String.format("Failed to create table %s.%s. msg: %s", dbName, tableName, e.getMessage()));
        }

        return true;
    }
@Test
    public void testCreateTableWithLocation() throws DdlException {
        new MockUp<HiveWriteUtils>() {
            @Mock
            public void createDirectory(Path path, Configuration conf) {
            }

            @Mock
            public boolean pathExists(Path path, Configuration conf) {
                return true;
            }

            @Mock
            public boolean isEmpty(Path path, Configuration conf) {
                return true;
            }
        };

        HiveMetastoreOperations mockedHmsOps = new HiveMetastoreOperations(cachingHiveMetastore, true,
                new Configuration(), MetastoreType.HMS, "hive_catalog") {
            @Override
            public Path getDefaultLocation(String dbName, String tableName) {
                return new Path("mytable_locatino");
            }
        };

        Map<String, String> properties = Maps.newHashMap();
        properties.put("location", "hdfs://path_to_file/file_name");
        CreateTableStmt stmt = new CreateTableStmt(
                false,
                false,
                new TableName("hive_catalog", "hive_db", "hive_table"),
                Lists.newArrayList(
                        new ColumnDef("c1", TypeDef.create(PrimitiveType.INT)),
                        new ColumnDef("p1", TypeDef.create(PrimitiveType.INT))),
                "hive",
                null,
                new ListPartitionDesc(Lists.newArrayList("p1"), new ArrayList<>()),
                null,
                properties,
                new HashMap<>(),
                "my table comment");
        List<Column> columns = stmt.getColumnDefs().stream().map(def -> def.toColumn(null)).collect(Collectors.toList());
        stmt.setColumns(columns);

        Assert.assertTrue(mockedHmsOps.createTable(stmt));
    }