[Enhancement] implement createRepositoryCommand in nereids

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

@@ -230,6 +230,7 @@ supportedCreateStatement
             passwordOption commentSpec?                                             #createUser
     | 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 DICTIONARY (IF NOT EXISTS)? name = multipartIdentifier
@@ -646,8 +647,8 @@ supportedGrantRevokeStatement
     | GRANT roles+=identifierOrText (COMMA roles+=identifierOrText)* TO userIdentify        #grantRole
     | REVOKE roles+=identifierOrText (COMMA roles+=identifierOrText)* FROM userIdentify     #revokeRole
     | REVOKE privilegeList ON
-            (RESOURCE | CLUSTER | COMPUTE GROUP | STAGE | STORAGE VAULT | WORKLOAD GROUP)
-            identifierOrTextOrAsterisk FROM (userIdentify | ROLE identifierOrText)          #revokeResourcePrivilege
+        (RESOURCE | CLUSTER | COMPUTE GROUP | STAGE | STORAGE VAULT | WORKLOAD GROUP)
+        identifierOrTextOrAsterisk FROM (userIdentify | ROLE identifierOrText)              #revokeResourcePrivilege
     ;
 
 unsupportedGrantRevokeStatement
@@ -810,8 +811,7 @@ analyzeProperties
     ;
 
 unsupportedCreateStatement
-    : CREATE (READ ONLY)? REPOSITORY name=identifier WITH storageBackend        #createRepository
-    | CREATE STORAGE VAULT (IF NOT EXISTS)?
+    : CREATE STORAGE VAULT (IF NOT EXISTS)?
         name=identifierOrText properties=propertyClause?                        #createStorageVault
     ;
 

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

@@ -92,6 +92,23 @@ public void analyze(Analyzer analyzer) throws UserException {
         checkPath(location, storageType, null);
     }
 
+    public void validate() throws UserException {
+        StorageBackend.StorageType storageType = storageDesc.getStorageType();
+        if (storageType != StorageType.BROKER && StringUtils.isEmpty(storageDesc.getName())) {
+            storageDesc.setName(storageType.name());
+        }
+        if (storageType != StorageType.BROKER && storageType != StorageType.S3
+                && storageType != StorageType.HDFS) {
+            throw new NotImplementedException(storageType.toString() + " is not support now.");
+        }
+        FeNameFormat.checkCommonName("repository", storageDesc.getName());
+
+        if (Strings.isNullOrEmpty(location)) {
+            throw new AnalysisException("You must specify a location on the repository");
+        }
+        checkPath(location, storageType, null);
+    }
+
     @Override
     public String toSql() {
         StringBuilder sb = new StringBuilder();

fe/fe-core/src/main/java/org/apache/doris/backup/BackupHandler.java

@@ -56,6 +56,7 @@
 import org.apache.doris.fsv2.remote.RemoteFileSystem;
 import org.apache.doris.fsv2.remote.S3FileSystem;
 import org.apache.doris.nereids.trees.plans.commands.CancelBackupCommand;
+import org.apache.doris.nereids.trees.plans.commands.CreateRepositoryCommand;
 import org.apache.doris.persist.BarrierLog;
 import org.apache.doris.task.DirMoveTask;
 import org.apache.doris.task.DownloadTask;
@@ -208,6 +209,38 @@ protected void runAfterCatalogReady() {
         }
     }
 
+    // handle create repository command
+    public void createRepository(CreateRepositoryCommand command) throws DdlException {
+        if (!env.getBrokerMgr().containsBroker(command.getBrokerName())
+                && command.getStorageType() == StorageBackend.StorageType.BROKER) {
+            ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR,
+                    "broker does not exist: " + command.getBrokerName());
+        }
+
+        RemoteFileSystem fileSystem;
+        try {
+            fileSystem = FileSystemFactory.get(command.getStorageType(), command.getProperties());
+        } catch (UserException e) {
+            throw new DdlException("Failed to initialize remote file system: " + e.getMessage());
+        }
+        org.apache.doris.fs.remote.RemoteFileSystem oldfs = org.apache.doris.fs.FileSystemFactory
+                .get(command.getBrokerName(), command.getStorageType(),
+                command.getProperties());
+        long repoId = env.getNextId();
+        Repository repo = new Repository(repoId, command.getName(), command.isReadOnly(), command.getLocation(),
+                fileSystem, oldfs);
+
+        Status st = repoMgr.addAndInitRepoIfNotExist(repo, false);
+        if (!st.ok()) {
+            ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR,
+                    "Failed to create repository: " + st.getErrMsg());
+        }
+        if (!repo.ping()) {
+            ErrorReport.reportDdlException(ErrorCode.ERR_COMMON_ERROR,
+                    "Failed to create repository: failed to connect to the repo");
+        }
+    }
+
     // handle create repository stmt
     public void createRepository(CreateRepositoryStmt stmt) throws DdlException {
         if (!env.getBrokerMgr().containsBroker(stmt.getBrokerName())

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

@@ -607,6 +607,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.CreateRepositoryCommand;
 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;
@@ -6834,6 +6835,35 @@ public LogicalPlan visitCreateDatabase(DorisParser.CreateDatabaseContext ctx) {
             Maps.newHashMap(properties));
     }
 
+    @Override
+    public LogicalPlan visitCreateRepository(DorisParser.CreateRepositoryContext ctx) {
+        String name = stripQuotes(ctx.name.getText());
+        String location = stripQuotes(ctx.storageBackend().STRING_LITERAL().getText());
+
+        String storageName = "";
+        if (ctx.storageBackend().brokerName != null) {
+            storageName = stripQuotes(ctx.storageBackend().brokerName.getText());
+        }
+
+        Map<String, String> properties = visitPropertyClause(ctx.storageBackend().properties);
+
+        StorageBackend.StorageType storageType = null;
+        if (ctx.storageBackend().BROKER() != null) {
+            storageType = StorageBackend.StorageType.BROKER;
+        } else if (ctx.storageBackend().S3() != null) {
+            storageType = StorageBackend.StorageType.S3;
+        } else if (ctx.storageBackend().HDFS() != null) {
+            storageType = StorageBackend.StorageType.HDFS;
+        } else if (ctx.storageBackend().LOCAL() != null) {
+            storageType = StorageBackend.StorageType.LOCAL;
+        }
+
+        return new CreateRepositoryCommand(
+            ctx.READ() != null,
+            name,
+                new StorageBackend(storageName, location, storageType, Maps.newHashMap(properties)));
+    }
+
     @Override
     public LogicalPlan visitShowColumnHistogramStats(ShowColumnHistogramStatsContext ctx) {
         TableNameInfo tableNameInfo = new TableNameInfo(visitMultipartIdentifier(ctx.tableName));

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

@@ -195,6 +195,7 @@ public enum PlanType {
     CLEAN_LABEL_COMMAND,
     CREATE_ROLE_COMMAND,
     CREATE_DATABASE_COMMAND,
+    CREATE_REPOSITORY_COMMAND,
     ALTER_CATALOG_RENAME_COMMAND,
     ALTER_ROLE_COMMAND,
     ALTER_VIEW_COMMAND,

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

@@ -0,0 +1,106 @@
+// 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.analysis.StorageBackend;
+import org.apache.doris.catalog.Env;
+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.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.StmtExecutor;
+
+import java.util.Map;
+
+/**
+ * CreateRepositoryCommand
+ */
+public class CreateRepositoryCommand extends Command implements ForwardWithSync, NeedAuditEncryption {
+    private boolean isReadOnly;
+    private String name;
+    private StorageBackend storage;
+
+    public CreateRepositoryCommand(boolean isReadOnly, String name, StorageBackend storage) {
+        super(PlanType.CREATE_REPOSITORY_COMMAND);
+        this.isReadOnly = isReadOnly;
+        this.name = name;
+        this.storage = storage;
+    }
+
+    @Override
+    public void run(ConnectContext ctx, StmtExecutor executor) throws Exception {
+        validate();
+        Env.getCurrentEnv().getBackupHandler().createRepository(this);
+    }
+
+    /**
+     * validate
+     */
+    public void validate() throws UserException {
+        storage.validate();
+        // check auth
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ConnectContext.get(), PrivPredicate.ADMIN)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
+        }
+        FeNameFormat.checkRepositoryName(name);
+    }
+
+    @Override
+    public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+        return visitor.visitCreateRepositoryCommand(this, context);
+    }
+
+    @Override
+    public boolean needAuditEncryption() {
+        return true;
+    }
+
+    @Override
+    public StmtType stmtType() {
+        return StmtType.CREATE;
+    }
+
+    public boolean isReadOnly() {
+        return isReadOnly;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public String getBrokerName() {
+        return storage.getStorageDesc().getName();
+    }
+
+    public StorageBackend.StorageType getStorageType() {
+        return storage.getStorageDesc().getStorageType();
+    }
+
+    public String getLocation() {
+        return storage.getLocation();
+    }
+
+    public Map<String, String> getProperties() {
+        return storage.getStorageDesc().getProperties();
+    }
+}

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

@@ -114,10 +114,8 @@ public void validate() throws AnalysisException {
 
         if (resourcePattern.isPresent()) {
             PrivBitSet.convertResourcePrivToCloudPriv(resourcePattern.get(), privileges);
-            // modify when grantCommand merged
             GrantResourcePrivilegeCommand.checkResourcePrivileges(privileges, resourcePattern.get());
         } else if (workloadGroupPattern.isPresent()) {
-            // modify when grantCommand merged
             GrantResourcePrivilegeCommand.checkWorkloadGroupPrivileges(privileges, workloadGroupPattern.get());
         }
     }

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

@@ -69,6 +69,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.CreateRepositoryCommand;
 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;
@@ -526,6 +527,10 @@ default R visitCreateDatabaseCommand(CreateDatabaseCommand command, C context) {
         return visitCommand(command, context);
     }
 
+    default R visitCreateRepositoryCommand(CreateRepositoryCommand command, C context) {
+        return visitCommand(command, context);
+    }
+
     default R visitCreateTableLikeCommand(CreateTableLikeCommand createTableLikeCommand, C context) {
         return visitCommand(createTableLikeCommand, context);
     }
@@ -1195,9 +1200,8 @@ default R visitGrantTablePrivilegeCommand(GrantTablePrivilegeCommand grantTableP
         return visitCommand(grantTablePrivilegeCommand, context);
     }
 
-    default R visitGrantResourcePrivilegeCommand(GrantResourcePrivilegeCommand grantResourcePrivilegeCommand,
-            C context) {
-        return visitCommand(grantResourcePrivilegeCommand, context);
+    default R visitGrantResourcePrivilegeCommand(GrantResourcePrivilegeCommand command, C context) {
+        return visitCommand(command, context);
     }
 
     default R visitRevokeRoleCommand(RevokeRoleCommand revokeRoleCommand, C context) {

fe/fe-core/src/test/java/org/apache/doris/nereids/parser/NereidsParserTest.java

@@ -653,7 +653,12 @@ public void testCreateUser() {
     @Test
     public void testCreateRepository() {
         NereidsParser nereidsParser = new NereidsParser();
-        String sql = "create repository a with S3 on location 's3://xxx' properties('k'='v')";
+        String sql = "create repository a with S3 on location 's3://s3-repo' "
+                + "properties("
+                + "'AWS_ENDPOINT' = 'http://s3.us-east-1.amazonaws.com',"
+                + "'AWS_ACCESS_KEY' = 'akk',"
+                + "'AWS_SECRET_KEY'='skk',"
+                + "'AWS_REGION' = 'us-east-1')";
         nereidsParser.parseSingle(sql);
     }