[bugfix](external)add check of engine and catalog types (#39343)

## Proposed changes

Currently, we can only create hive table under hms type catalog and
iceberg table under iceberg catalog.

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java

@@ -225,6 +225,21 @@ public List<String> getTableNameParts() {
         return ImmutableList.of(tableName);
     }
 
+    private void checkEngineWithCatalog() {
+        if (engineName.equals(ENGINE_OLAP)) {
+            if (!ctlName.equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
+                throw new AnalysisException("Cannot create olap table out of internal catalog."
+                    + " Make sure 'engine' type is specified when use the catalog: " + ctlName);
+            }
+        }
+        CatalogIf catalog = Env.getCurrentEnv().getCatalogMgr().getCatalog(ctlName);
+        if (catalog instanceof HMSExternalCatalog && !engineName.equals(ENGINE_HIVE)) {
+            throw new AnalysisException("Hms type catalog can only use `hive` engine.");
+        } else if (catalog instanceof IcebergExternalCatalog && !engineName.equals(ENGINE_ICEBERG)) {
+            throw new AnalysisException("Iceberg type catalog can only use `iceberg` engine.");
+        }
+    }
+
     /**
      * analyze create table info
      */
@@ -262,12 +277,7 @@ public void validate(ConnectContext ctx) {
             throw new AnalysisException(e.getMessage(), e);
         }
 
-        if (engineName.equals(ENGINE_OLAP)) {
-            if (!ctlName.equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
-                throw new AnalysisException("Cannot create olap table out of internal catalog."
-                    + " Make sure 'engine' type is specified when use the catalog: " + ctlName);
-            }
-        }
+        checkEngineWithCatalog();
 
         // analyze table name
         if (Strings.isNullOrEmpty(dbName)) {

regression-test/suites/external_table_p0/hive/ddl/test_hive_ddl.groovy

@@ -682,6 +682,24 @@ suite("test_hive_ddl", "p0,external,hive,external_docker,external_docker_hive")
             sql """ drop database if exists `test_hive_db_tbl` """;
         }
 
+        def test_error_create = { String catalog_name ->
+            sql """switch `${catalog_name}`"""
+            sql """ drop database if exists test_hive_db_error_tbl """
+            sql """ create database `test_hive_db_error_tbl` """;
+
+            test {
+                sql """ create table err_tb (id int) engine = iceberg """
+                exception "java.sql.SQLException: errCode = 2, detailMessage = Hms type catalog can only use `hive` engine."
+            }
+
+            test {
+                sql """ create table err_tb (id int) engine = jdbc """
+                exception "java.sql.SQLException: errCode = 2, detailMessage = Hms type catalog can only use `hive` engine."
+            }
+
+            sql """ drop database test_hive_db_error_tbl """
+        }
+
 
         try {
             String hms_port = context.config.otherConfigs.get("hive2HmsPort")
@@ -717,6 +735,7 @@ suite("test_hive_ddl", "p0,external,hive,external_docker,external_docker_hive")
                 }
                 test_create_tbl_cross_catalog(file_format, catalog_name)
             }
+            test_error_create(catalog_name)
             sql """drop catalog if exists ${catalog_name}"""
         } finally {
         }

regression-test/suites/external_table_p0/iceberg/write/test_iceberg_create_table.groovy

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_create_table", "p0,external,doris,external_docker,external_docker_doris") {
+    String enabled = context.config.otherConfigs.get("enableIcebergTest")
+    if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+        logger.info("disable iceberg test.")
+        return
+    }
+
+    String rest_port = context.config.otherConfigs.get("iceberg_rest_uri_port")
+    String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
+    String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+    String catalog_name = "test_iceberg_create_table"
+
+    sql """drop catalog if exists ${catalog_name}"""
+    sql """
+    CREATE CATALOG ${catalog_name} PROPERTIES (
+        'type'='iceberg',
+        'iceberg.catalog.type'='rest',
+        'uri' = 'http://${externalEnvIp}:${rest_port}',
+        "s3.access_key" = "admin",
+        "s3.secret_key" = "password",
+        "s3.endpoint" = "http://${externalEnvIp}:${minio_port}",
+        "s3.region" = "us-east-1"
+    );"""
+
+    sql """ switch ${catalog_name} """
+
+    String db1 = catalog_name + "_db1"
+    String tb1 = db1 + "_tb1"
+    String tb2 = db1 + "_tb2"
+
+    sql """ drop table if exists ${db1}.${tb1} """
+    sql """ drop table if exists ${db1}.${tb2} """
+    sql """ drop database if exists ${db1} """
+
+    sql """ create database ${db1} """
+
+    test {
+        sql """ create table ${db1}.${tb1} (id int) engine = olap """
+        exception "Cannot create olap table out of internal catalog. Make sure 'engine' type is specified when use the catalog: ${catalog_name}"
+    }
+
+    test {
+        sql """ create table ${db1}.${tb1} (id int) engine = hive """
+        exception "java.sql.SQLException: errCode = 2, detailMessage = Iceberg type catalog can only use `iceberg` engine."
+    }
+
+    test {
+        sql """ create table ${db1}.${tb1} (id int) engine = jdbc """
+        exception "java.sql.SQLException: errCode = 2, detailMessage = Iceberg type catalog can only use `iceberg` engine."
+    }
+
+    sql """ create table ${db1}.${tb1} (id int) engine = iceberg """
+    sql """ create table ${db1}.${tb2} (id int) """
+
+    sql """ drop table ${db1}.${tb1} """
+    sql """ drop table ${db1}.${tb2} """
+    sql """ drop database ${db1} """
+
+}