[Enhancement] (nereids)implement CreateResourceCommand in nereids

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -220,6 +220,8 @@ supportedCreateStatement
     | CREATE USER (IF NOT EXISTS)? grantUserIdentify
             (SUPERUSER | DEFAULT ROLE role=STRING_LITERAL)?
             passwordOption commentSpec?                            #createUser
+    | CREATE EXTERNAL? RESOURCE (IF NOT EXISTS)?
+            name=identifierOrText properties=propertyClause?                        #createResource
     ;
 
 supportedAlterStatement
@@ -782,8 +784,6 @@ unsupportedCreateStatement
     : CREATE (DATABASE | SCHEMA) (IF NOT EXISTS)? name=multipartIdentifier
         properties=propertyClause?                                              #createDatabase
     | CREATE (READ ONLY)? REPOSITORY name=identifier WITH storageBackend        #createRepository
-    | CREATE EXTERNAL? RESOURCE (IF NOT EXISTS)?
-        name=identifierOrText properties=propertyClause?                        #createResource
     | CREATE STORAGE VAULT (IF NOT EXISTS)?
         name=identifierOrText properties=propertyClause?                        #createStorageVault
     | CREATE WORKLOAD POLICY (IF NOT EXISTS)? name=identifierOrText

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

@@ -26,6 +26,7 @@
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.common.proc.BaseProcResult;
 import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateResourceInfo;
 import org.apache.doris.persist.gson.GsonPostProcessable;
 import org.apache.doris.persist.gson.GsonUtils;
 
@@ -131,6 +132,14 @@ public static Resource fromStmt(CreateResourceStmt stmt) throws DdlException {
         return resource;
     }
 
+    public static Resource fromInfo(CreateResourceInfo info) throws DdlException {
+        Resource resource = getResourceInstance(info.getResourceType(), info.getResourceName());
+        resource.id = Env.getCurrentEnv().getNextId();
+        resource.version = 0;
+        resource.setProperties(info.getProperties());
+        return resource;
+    }
+
     public long getId() {
         return this.id;
     }

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

@@ -30,7 +30,9 @@
 import org.apache.doris.common.proc.ProcNodeInterface;
 import org.apache.doris.common.proc.ProcResult;
 import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.plans.commands.CreateResourceCommand;
 import org.apache.doris.nereids.trees.plans.commands.DropResourceCommand;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateResourceInfo;
 import org.apache.doris.persist.DropResourceOperationLog;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.policy.Policy;
@@ -83,6 +85,18 @@ public void createResource(CreateResourceStmt stmt) throws DdlException {
         }
     }
 
+    public void createResource(CreateResourceCommand command) throws DdlException {
+        CreateResourceInfo info = command.getInfo();
+        if (info.getResourceType() == ResourceType.UNKNOWN) {
+            throw new DdlException("Only support SPARK, ODBC_CATALOG ,JDBC, S3_COOLDOWN, S3, HDFS and HMS resource.");
+        }
+        Resource resource = Resource.fromInfo(info);
+        if (createResource(resource, info.isIfNotExists())) {
+            Env.getCurrentEnv().getEditLog().logCreateResource(resource);
+            LOG.info("Create resource success. Resource: {}", resource.getName());
+        }
+    }
+
     // Return true if the resource is truly added,
     // otherwise, return false or throw exception.
     public boolean createResource(Resource resource, boolean ifNotExists) throws DdlException {

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -567,6 +567,7 @@
 import org.apache.doris.nereids.trees.plans.commands.CreateMaterializedViewCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreatePolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateProcedureCommand;
+import org.apache.doris.nereids.trees.plans.commands.CreateResourceCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
@@ -737,6 +738,7 @@
 import org.apache.doris.nereids.trees.plans.commands.info.CreateIndexOp;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateJobInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateMTMVInfo;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateResourceInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateTableLikeInfo;
@@ -6621,5 +6623,20 @@ public UserDesc visitGrantUserIdentify(DorisParser.GrantUserIdentifyContext ctx)
 
         return new UserDesc(userIdentity, new PassVar(password, isPlain));
     }
+
+    @Override
+    public LogicalPlan visitCreateResource(DorisParser.CreateResourceContext ctx) {
+        String resourceName = visitIdentifierOrText(ctx.name);
+        Map<String, String> properties = new HashMap<>(visitPropertyClause(ctx.properties));
+
+        CreateResourceInfo createResourceInfo = new CreateResourceInfo(
+                ctx.EXTERNAL() != null,
+                ctx.IF() != null,
+                resourceName,
+                properties
+        );
+
+        return new CreateResourceCommand(createResourceInfo);
+    }
 }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java

@@ -340,5 +340,6 @@ public enum PlanType {
     TRANSACTION_ROLLBACK_COMMAND,
     KILL_ANALYZE_JOB_COMMAND,
     DROP_ANALYZE_JOB_COMMAND,
-    CREATE_USER_COMMAND
+    CREATE_USER_COMMAND,
+    CREATE_RESOURCE_COMMAND
 }

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

@@ -0,0 +1,63 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands;
+
+import org.apache.doris.analysis.StmtType;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.commands.info.CreateResourceInfo;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+/**
+ * create resource command
+ */
+public class CreateResourceCommand extends Command implements ForwardWithSync, NeedAuditEncryption {
+    private final CreateResourceInfo info;
+
+    public CreateResourceCommand(CreateResourceInfo info) {
+        super(PlanType.CREATE_RESOURCE_COMMAND);
+        this.info = info;
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitCreateResourceCommand(this, context);
+    }
+
+    @Override
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        info.validate();
+        Env.getCurrentEnv().getResourceMgr().createResource(this);
+    }
+
+    @Override
+    public boolean needAuditEncryption() {
+        return true;
+    }
+
+    @Override
+    public StmtType stmtType() {
+        return StmtType.CREATE;
+    }
+
+    public CreateResourceInfo getInfo() {
+        return info;
+    }
+}

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

@@ -0,0 +1,132 @@
+// 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.
+
+package org.apache.doris.nereids.trees.plans.commands.info;
+
+import org.apache.doris.analysis.ResourceTypeEnum;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Resource.ResourceType;
+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;
+import org.apache.doris.common.UserException;
+import org.apache.doris.datasource.property.constants.AzureProperties;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.base.Strings;
+
+import java.util.Map;
+
+/**
+ * create resource info
+ */
+public class CreateResourceInfo {
+    private static final String TYPE = "type";
+    private final boolean isExternal;
+    private final boolean ifNotExists;
+    private final String resourceName;
+    private final Map<String, String> properties;
+    private ResourceType resourceType;
+
+    /**
+     * CreateResourceInfo
+     */
+    public CreateResourceInfo(boolean isExternal, boolean ifNotExists, String resourceName,
+            Map<String, String> properties) {
+        this.isExternal = isExternal;
+        this.ifNotExists = ifNotExists;
+        this.resourceName = resourceName;
+        this.properties = properties;
+        this.resourceType = ResourceType.UNKNOWN;
+    }
+
+    /**
+     * analyze createResourceInfo
+     */
+    public void validate() throws UserException {
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+
+        // check name
+        FeNameFormat.checkResourceName(resourceName, ResourceTypeEnum.GENERAL);
+
+        // check type in properties
+        if (properties == null || properties.isEmpty()) {
+            throw new AnalysisException("Resource properties can't be null");
+        }
+
+        analyzeResourceType();
+    }
+
+    /**
+     * analyze Resource Type
+     */
+    public void analyzeResourceType() throws UserException {
+        String type = null;
+        for (Map.Entry<String, String> property : properties.entrySet()) {
+            if (property.getKey().equalsIgnoreCase(TYPE)) {
+                type = property.getValue();
+            }
+        }
+        if (Strings.isNullOrEmpty(type)) {
+            throw new AnalysisException("Resource type can't be null");
+        }
+
+        if (AzureProperties.checkAzureProviderPropertyExist(properties)) {
+            resourceType = ResourceType.AZURE;
+            return;
+        }
+
+        resourceType = ResourceType.fromString(type);
+        if (resourceType == ResourceType.UNKNOWN) {
+            throw new AnalysisException("Unsupported resource type: " + type);
+        }
+        if (resourceType == ResourceType.SPARK && !isExternal) {
+            throw new AnalysisException("Spark is external resource");
+        }
+        if (resourceType == ResourceType.ODBC_CATALOG && !Config.enable_odbc_mysql_broker_table) {
+            throw new AnalysisException("ODBC table is deprecated, use JDBC instead. Or you can set "
+                + "`enable_odbc_mysql_broker_table=true` in fe.conf to enable ODBC again.");
+        }
+    }
+
+    public boolean isExternal() {
+        return isExternal;
+    }
+
+    public boolean isIfNotExists() {
+        return ifNotExists;
+    }
+
+    public String getResourceName() {
+        return resourceName;
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+
+    public ResourceType getResourceType() {
+        return resourceType;
+    }
+
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/CommandVisitor.java

@@ -56,6 +56,7 @@
 import org.apache.doris.nereids.trees.plans.commands.CreateMaterializedViewCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreatePolicyCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateProcedureCommand;
+import org.apache.doris.nereids.trees.plans.commands.CreateResourceCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateRoleCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateSqlBlockRuleCommand;
 import org.apache.doris.nereids.trees.plans.commands.CreateTableCommand;
@@ -979,4 +980,8 @@ default R visitDropAnalyzeJobCommand(DropAnalyzeJobCommand dropAnalyzeJobCommand
     default R visitCreateUserCommand(CreateUserCommand createUserCommand, C context) {
         return visitCommand(createUserCommand, context);
     }
+
+    default R visitCreateResourceCommand(CreateResourceCommand createResourceCommand, C context) {
+        return visitCommand(createResourceCommand, context);
+    }
 }