diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 4eb267e524..45bf426984 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2216,6 +2216,12 @@ public class Config extends ConfigBase {
         "Sample size for hive row count estimation."})
     public static int hive_stats_partition_sample_size = 3000;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "启用外表DDL",
+            "Enable external table DDL"})
+    public static boolean enable_external_ddl = false;
+
+
     @ConfField(mutable = true, masterOnly = true, description = {
             "Hive创建外部表默认指定的input format",
             "Default hive input format for creating table."})
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
index 6f43fab385..bf0dc906fd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java
@@ -254,6 +254,10 @@ public class CreateTableStmt extends DdlStmt {
         return engineName;
     }
 
+    public String getCatalogName() {
+        return tableName.getCtl();
+    }
+
     public String getDbName() {
         return tableName.getDb();
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DropTableStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DropTableStmt.java
index 70167d744d..ff679adc37 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DropTableStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DropTableStmt.java
@@ -61,6 +61,10 @@ public class DropTableStmt extends DdlStmt {
         return tableName.getTbl();
     }
 
+    public String getCatalogName() {
+        return tableName.getCtl();
+    }
+
     public boolean isView() {
         return isView;
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index bbcb14207d..effd7b6ca6 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -3039,7 +3039,9 @@ public class Env {
      * 11. add this table to ColocateGroup if necessary
      */
     public void createTable(CreateTableStmt stmt) throws UserException {
-        getCurrentCatalog().createTable(stmt);
+        CatalogIf<?> catalogIf = catalogMgr.getCatalogOrException(stmt.getCatalogName(),
+                catalog -> new DdlException(("Unknown catalog " + catalog)));
+        catalogIf.createTable(stmt);
     }
 
     public void createTableLike(CreateTableLikeStmt stmt) throws DdlException {
@@ -3659,7 +3661,9 @@ public class Env {
 
     // Drop table
     public void dropTable(DropTableStmt stmt) throws DdlException {
-        getCurrentCatalog().dropTable(stmt);
+        CatalogIf<?> catalogIf = catalogMgr.getCatalogOrException(stmt.getCatalogName(),
+                catalog -> new DdlException(("Unknown catalog " + catalog)));
+        catalogIf.dropTable(stmt);
     }
 
     public boolean unprotectDropTable(Database db, Table table, boolean isForceDrop, boolean isReplay,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
index 0eabffce44..783f8a0fdf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java
@@ -28,6 +28,7 @@ import org.apache.doris.catalog.InfoSchemaDb;
 import org.apache.doris.catalog.Resource;
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.cluster.ClusterNamespace;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.io.Text;
@@ -592,9 +593,12 @@ public abstract class ExternalCatalog
 
     @Override
     public void createDb(CreateDbStmt stmt) throws DdlException {
+        if (!Config.enable_external_ddl) {
+            throw new DdlException("Experimental. The config enable_external_ddl needs to be set to true.");
+        }
         makeSureInitialized();
         if (metadataOps == null) {
-            LOG.warn("dropDatabase not implemented");
+            LOG.warn("createDb not implemented");
             return;
         }
         try {
@@ -607,9 +611,12 @@ public abstract class ExternalCatalog
 
     @Override
     public void dropDb(DropDbStmt stmt) throws DdlException {
+        if (!Config.enable_external_ddl) {
+            throw new DdlException("Experimental. The config enable_external_ddl needs to be set to true.");
+        }
         makeSureInitialized();
         if (metadataOps == null) {
-            LOG.warn("dropDatabase not implemented");
+            LOG.warn("dropDb not implemented");
             return;
         }
         try {
@@ -622,6 +629,9 @@ public abstract class ExternalCatalog
 
     @Override
     public void createTable(CreateTableStmt stmt) throws UserException {
+        if (!Config.enable_external_ddl) {
+            throw new DdlException("Experimental. The config enable_external_ddl needs to be set to true.");
+        }
         makeSureInitialized();
         if (metadataOps == null) {
             LOG.warn("createTable not implemented");
@@ -637,6 +647,9 @@ public abstract class ExternalCatalog
 
     @Override
     public void dropTable(DropTableStmt stmt) throws DdlException {
+        if (!Config.enable_external_ddl) {
+            throw new DdlException("Experimental. The config enable_external_ddl needs to be set to true.");
+        }
         makeSureInitialized();
         if (metadataOps == null) {
             LOG.warn("dropTable not implemented");