diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index 96c9d92c8eb6..d94da91ef4c5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -125,26 +125,30 @@ public List<Long> listAllIds() {
         }
     }
 
-    /** Check TableScheme is be modified. */
-    public void checkTableSchema(TableSchema oldSchema, TableSchema newSchema) {
-        boolean isCommon =
-                oldSchema.version() == newSchema.version()
-                        && Objects.equals(oldSchema.fields(), newSchema.fields())
-                        && oldSchema.highestFieldId() == newSchema.highestFieldId()
-                        && Objects.equals(oldSchema.partitionKeys(), newSchema.partitionKeys())
-                        && Objects.equals(oldSchema.primaryKeys(), newSchema.primaryKeys());
-
-        if (!isCommon) {
-            throw new IllegalStateException(
-                    "Schema in filesystem exists, please use updating,"
-                            + " latest schema is: "
-                            + oldSchema);
-        }
-    }
-
     /** Create a new schema from {@link Schema}. */
     public TableSchema createTable(Schema schema) throws Exception {
+        return createTable(schema, false);
+    }
+
+    public TableSchema createTable(Schema schema, boolean ignoreIfExistsSame) throws Exception {
         while (true) {
+            Optional<TableSchema> latest = latest();
+            if (latest.isPresent()) {
+                TableSchema oldSchema = latest.get();
+                boolean isSame =
+                        Objects.equals(oldSchema.fields(), schema.fields())
+                                && Objects.equals(oldSchema.partitionKeys(), schema.partitionKeys())
+                                && Objects.equals(oldSchema.primaryKeys(), schema.primaryKeys())
+                                && Objects.equals(oldSchema.options(), schema.options());
+                if (ignoreIfExistsSame && isSame) {
+                    return oldSchema;
+                }
+
+                throw new IllegalStateException(
+                        "Schema in filesystem exists, please use updating,"
+                                + " latest schema is: "
+                                + oldSchema);
+            }
 
             List<DataField> fields = schema.fields();
             List<String> partitionKeys = schema.partitionKeys();
@@ -162,11 +166,6 @@ public TableSchema createTable(Schema schema) throws Exception {
                             options,
                             schema.comment());
 
-            if (latest().isPresent()) {
-                checkTableSchema(latest().get(), newSchema);
-                return newSchema;
-            }
-
             boolean success = commit(newSchema);
             if (success) {
                 return newSchema;
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 372bfedefb88..32d25e7db199 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -339,6 +339,14 @@ public TableSchema getDataTableSchema(Identifier identifier) throws TableNotExis
                         () -> new RuntimeException("There is no paimon table in " + tableLocation));
     }
 
+    private boolean usingExternalTable() {
+        TableType tableType =
+                OptionsUtils.convertToEnum(
+                        hiveConf.get(TABLE_TYPE.key(), TableType.MANAGED.toString()),
+                        TableType.class);
+        return TableType.EXTERNAL.equals(tableType);
+    }
+
     @Override
     protected void dropTableImpl(Identifier identifier) {
         try {
@@ -347,11 +355,7 @@ protected void dropTableImpl(Identifier identifier) {
 
             // When drop a Hive external table, only the hive metadata is deleted and the data files
             // are not deleted.
-            TableType tableType =
-                    OptionsUtils.convertToEnum(
-                            hiveConf.get(TABLE_TYPE.key(), TableType.MANAGED.toString()),
-                            TableType.class);
-            if (TableType.EXTERNAL.equals(tableType)) {
+            if (usingExternalTable()) {
                 return;
             }
 
@@ -377,7 +381,7 @@ protected void createTableImpl(Identifier identifier, Schema schema) {
         // if changes on Hive fails there is no harm to perform the same changes to files again
         TableSchema tableSchema;
         try {
-            tableSchema = schemaManager(identifier).createTable(schema);
+            tableSchema = schemaManager(identifier).createTable(schema, usingExternalTable());
         } catch (Exception e) {
             throw new RuntimeException(
                     "Failed to commit changes of table "
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/CreateTableITCase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/CreateTableITCase.java
index 88c4762ac536..15856c3c06cd 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/CreateTableITCase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/CreateTableITCase.java
@@ -148,11 +148,9 @@ public void testCallCreateTableToCreatHiveExternalTable() throws Exception {
         hiveCatalog.createTable(identifier, schema, false);
 
         // Drop hive external table
-        String hiveSql = String.join("\n", Arrays.asList("DROP TABLE " + tableName));
-        assertThatCode(() -> hiveShell.execute(hiveSql)).doesNotThrowAnyException();
+        hiveShell.execute("DROP TABLE " + tableName);
 
-        assertThatCode(() -> hiveCatalog.createTable(identifier, schema, false))
-                .doesNotThrowAnyException();
+        hiveCatalog.createTable(identifier, schema, false);
     }
 
     @Test