permission check

Author: yiminpeng

server/odc-service/src/main/java/com/oceanbase/odc/service/flow/FlowInstanceService.java

@@ -974,8 +974,6 @@ private void checkCreateFlowInstancePermission(CreateFlowInstanceReq req) {
             MultipleDatabaseChangeParameters parameters = (MultipleDatabaseChangeParameters) req.getParameters();
             databaseIds =
                     parameters.getOrderedDatabaseIds().stream().flatMap(Collection::stream).collect(Collectors.toSet());
-        } else if (taskType == TaskType.LOGICAL_DATABASE_CHANGE) {
-            return;
         }
         permissionHelper.checkDBPermissions(databaseIds, DatabasePermissionType.from(req.getTaskType()));
     }

server/odc-service/src/main/java/com/oceanbase/odc/service/permission/DBResourcePermissionHelper.java

@@ -130,10 +130,12 @@ public void checkDBPermissions(Collection<Long> databaseIds, Collection<Database
             if (e.getProjectId() == null) {
                 throw new AccessDeniedException("Database is not belong to any project");
             }
-            DialectType dialectType = id2Entity.get(e.getConnectionId()).getDialectType();
-            if (permissionCheckWhitelist.containsDatabase(e.getName(), dialectType)
+            if (!id2Entity.isEmpty()) {
+                DialectType dialectType = id2Entity.get(e.getConnectionId()).getDialectType();
+                if (permissionCheckWhitelist.containsDatabase(e.getName(), dialectType)
                     || projectIds.contains(e.getProjectId())) {
-                continue;
+                    continue;
+                }
             }
             toCheckDatabaseIds.add(e.getId());
         }