[refactor](iceberg-hudi) disable iceberg and hudi table by default

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java

@@ -274,11 +274,6 @@ public void analyze(Analyzer analyzer) throws UserException {
 
         analyzeEngineName();
 
-        // TODO(wyb): spark-load
-        if (engineName.equals("hive") && !Config.enable_spark_load) {
-            throw new AnalysisException("Spark Load from hive table is coming soon");
-        }
-
         // `analyzeUniqueKeyMergeOnWrite` would modify `properties`, which will be used later,
         // so we just clone a properties map here.
         boolean enableUniqueKeyMergeOnWrite = false;
@@ -531,6 +526,13 @@ private void analyzeEngineName() throws AnalysisException {
                 throw new AnalysisException("Do not support external table with engine name = olap");
             }
         }
+
+        if (Config.disable_iceberg_hudi_table && (engineName.equals("iceberg") || engineName.equals("hudi"))) {
+            throw new AnalysisException(
+                    "iceberg and hudi table is no longer supported. Use multi catalog feature instead."
+                            + ". Or you can temporarily set 'disable_iceberg_hudi_table=false'"
+                            + " in fe.conf to reopen this feature.");
+        }
     }
 
     public static CreateTableStmt read(DataInput in) throws IOException {

fe/fe-core/src/main/java/org/apache/doris/analysis/GrantStmt.java

@@ -21,7 +21,6 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeNameFormat;
@@ -109,10 +108,6 @@ public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
         if (tblPattern != null) {
             tblPattern.analyze(analyzer);
         } else {
-            // TODO(wyb): spark-load
-            if (!Config.enable_spark_load) {
-                throw new AnalysisException("GRANT ON RESOURCE is coming soon");
-            }
             resourcePattern.analyze();
         }
 

fe/fe-core/src/main/java/org/apache/doris/analysis/LoadStmt.java

@@ -22,7 +22,6 @@
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.PrintableMap;
@@ -365,10 +364,6 @@ public void analyze(Analyzer analyzer) throws UserException {
         if (resourceDesc != null) {
             resourceDesc.analyze();
             etlJobType = resourceDesc.getEtlJobType();
-            // TODO(wyb): spark-load
-            if (!Config.enable_spark_load) {
-                throw new AnalysisException("Spark Load is coming soon");
-            }
             // check resource usage privilege
             if (!Env.getCurrentEnv().getAuth().checkResourcePriv(ConnectContext.get(),
                                                                          resourceDesc.getName(),

fe/fe-core/src/main/java/org/apache/doris/analysis/RevokeStmt.java

@@ -20,7 +20,6 @@
 import org.apache.doris.catalog.AccessPrivilege;
 import org.apache.doris.cluster.ClusterNamespace;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.FeNameFormat;
 import org.apache.doris.mysql.privilege.PaloPrivilege;
 import org.apache.doris.mysql.privilege.PrivBitSet;
@@ -99,10 +98,6 @@ public void analyze(Analyzer analyzer) throws AnalysisException {
         if (tblPattern != null) {
             tblPattern.analyze(analyzer);
         } else {
-            // TODO(wyb): spark-load
-            if (!Config.enable_spark_load) {
-                throw new AnalysisException("REVOKE ON RESOURCE is coming soon");
-            }
             resourcePattern.analyze();
         }
 

fe/fe-core/src/main/java/org/apache/doris/catalog/DatabaseProperty.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.catalog;
 
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.io.Text;
 import org.apache.doris.common.io.Writable;
@@ -85,7 +86,14 @@ public DatabaseProperty checkAndBuildProperties() throws DdlException {
         Map<String, String> icebergProperties = new HashMap<>();
         for (Map.Entry<String, String> entry : this.properties.entrySet()) {
             if (entry.getKey().startsWith(ICEBERG_PROPERTY_PREFIX)) {
-                icebergProperties.put(entry.getKey(), entry.getValue());
+                if (Config.disable_iceberg_hudi_table) {
+                    throw new DdlException(
+                            "database for iceberg is no longer supported. Use multi catalog feature instead."
+                                    + ". Or you can temporarily set 'disable_iceberg_hudi_table=false'"
+                                    + " in fe.conf to reopen this feature.");
+                } else {
+                    icebergProperties.put(entry.getKey(), entry.getValue());
+                }
             }
         }
         if (icebergProperties.size() > 0) {

fe/fe-core/src/main/java/org/apache/doris/common/Config.java

@@ -1314,18 +1314,6 @@ public class Config extends ConfigBase {
     @ConfField(mutable = true, masterOnly = true)
     public static boolean drop_backend_after_decommission = true;
 
-    /**
-     * enable spark load for temporary use
-     */
-    @ConfField(mutable = true, masterOnly = true)
-    public static boolean enable_spark_load = true;
-
-    /**
-     * enable use odbc table
-     */
-    @ConfField(mutable = true, masterOnly = true)
-    public static boolean enable_odbc_table = true;
-
     /**
      * Define thrift server's server model, default is TThreadPoolServer model
      */
@@ -1829,4 +1817,11 @@ public class Config extends ConfigBase {
      */
     @ConfField(mutable = true, masterOnly = true)
     public static long max_backend_heartbeat_failure_tolerance_count = 1;
+
+    /**
+     * The iceberg and hudi table will be removed in v1.3
+     * Use multi catalog instead.
+     */
+    @ConfField(mutable = true, masterOnly = false)
+    public static boolean disable_iceberg_hudi_table = true;
 }

fe/fe-core/src/main/java/org/apache/doris/planner/SingleNodePlanner.java

@@ -1925,8 +1925,7 @@ private PlanNode createScanNode(Analyzer analyzer, TableRef tblRef, SelectStmt s
                         null, -1);
                 break;
             case ICEBERG:
-                scanNode = new IcebergScanNode(ctx.getNextNodeId(), tblRef.getDesc(), "IcebergScanNode",
-                        null, -1);
+                scanNode = new ExternalFileScanNode(ctx.getNextNodeId(), tblRef.getDesc());
                 break;
             case HUDI:
                 scanNode = new HudiScanNode(ctx.getNextNodeId(), tblRef.getDesc(), "HudiScanNode",
@@ -1948,7 +1947,8 @@ private PlanNode createScanNode(Analyzer analyzer, TableRef tblRef, SelectStmt s
             default:
                 break;
         }
-        if (scanNode instanceof OlapScanNode || scanNode instanceof EsScanNode || scanNode instanceof HiveScanNode) {
+        if (scanNode instanceof OlapScanNode || scanNode instanceof EsScanNode || scanNode instanceof HiveScanNode
+                || scanNode instanceof ExternalFileScanNode) {
             if (analyzer.enableInferPredicate()) {
                 PredicatePushDown.visitScanNode(scanNode, tblRef.getJoinOp(), analyzer);
             }

fe/fe-core/src/test/java/org/apache/doris/analysis/CreateTableStmtTest.java

@@ -23,6 +23,7 @@
 import org.apache.doris.catalog.PrimitiveType;
 import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.ExceptionChecker;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.PropertyAnalyzer;
@@ -309,6 +310,7 @@ public void testBmpHllIncAgg() throws Exception {
 
     @Test
     public void testCreateIcebergTable() throws UserException {
+        Config.disable_iceberg_hudi_table = false;
         Map<String, String> properties = new HashMap<>();
         properties.put("iceberg.database", "doris");
         properties.put("iceberg.table", "test");
@@ -324,6 +326,7 @@ public void testCreateIcebergTable() throws UserException {
 
     @Test
     public void testCreateHudiTable() throws UserException {
+        Config.disable_iceberg_hudi_table = false;
         Map<String, String> properties = new HashMap<>();
         properties.put("hudi.database", "doris");
         properties.put("hudi.table", "test");
@@ -340,6 +343,7 @@ public void testCreateHudiTable() throws UserException {
 
     @Test
     public void testCreateHudiTableWithSchema() throws UserException {
+        Config.disable_iceberg_hudi_table = false;
         Map<String, String> properties = new HashMap<>();
         properties.put("hudi.database", "doris");
         properties.put("hudi.table", "test");

fe/fe-core/src/test/java/org/apache/doris/analysis/GrantStmtTest.java

@@ -20,7 +20,6 @@
 import org.apache.doris.catalog.AccessPrivilege;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.UserException;
 import org.apache.doris.mysql.privilege.PaloAuth;
 import org.apache.doris.qe.ConnectContext;
@@ -95,9 +94,6 @@ public void testNormal() throws AnalysisException, UserException {
 
     @Test
     public void testResourceNormal() throws UserException {
-        // TODO(wyb): spark-load
-        Config.enable_spark_load = true;
-
         String resourceName = "spark0";
         List<AccessPrivilege> privileges = Lists.newArrayList(AccessPrivilege.USAGE_PRIV);
         GrantStmt stmt = new GrantStmt(new UserIdentity("testUser", "%"), null, new ResourcePattern(resourceName), privileges);

fe/fe-core/src/test/java/org/apache/doris/analysis/LoadStmtTest.java

@@ -21,7 +21,6 @@
 import org.apache.doris.catalog.ResourceMgr;
 import org.apache.doris.catalog.SparkResource;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.SqlParserUtils;
 import org.apache.doris.load.EtlJobType;
@@ -115,8 +114,6 @@ public void testNormal(@Injectable DataDescription desc, @Mocked Env env,
                 + "(XXX)", stmt.toString());
 
         // test ResourceDesc
-        // TODO(wyb): spark-load
-        Config.enable_spark_load = true;
         stmt = new LoadStmt(new LabelName("testDb", "testLabel"), dataDescriptionList,
                             new ResourceDesc(resourceName, null), null);
         stmt.analyze(analyzer);

fe/fe-core/src/test/java/org/apache/doris/mysql/privilege/AuthTest.java

@@ -32,7 +32,6 @@
 import org.apache.doris.catalog.DomainResolver;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
-import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.ExceptionChecker;
 import org.apache.doris.common.UserException;
@@ -1400,8 +1399,6 @@ public void testResource() {
         ResourcePattern anyResourcePattern = new ResourcePattern(anyResource);
         List<AccessPrivilege> usagePrivileges = Lists.newArrayList(AccessPrivilege.USAGE_PRIV);
         UserDesc userDesc = new UserDesc(userIdentity, "12345", true);
-        // TODO(wyb): spark-load
-        Config.enable_spark_load = true;
 
         // ------ grant|revoke resource to|from user ------
         // 1. create user with no role