It is necessary to check whether the resource is valid (#2399)

* init full_name in dml of version 1.2.2

* redefine request parameter resourceIds

* redefine request parameter resourceIds

* Actually grant resource file if choose the directory

* To cancel authorized resource need check whether it is used by the process definition which is online

* If release the process definition online,It is necessary to check whether resource is valid

* update ResourceServiceTest and ResourceMapperTest

* add batchUpdateResourceTest

* add getHdfsFileNameTest and getHdfsResourceFileNameTest

* update ResourceServiceTest and ResourceMapperTest

* extract getResourceProcessMap to ResourceProcessDefinitonUtils
Co-authored-by: Ndailidong <dailidong66@gmail.com>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java

@@ -190,6 +190,7 @@ public enum Status {
     UDF_RESOURCE_IS_BOUND(20013, "udf resource file is bound by UDF functions:{0}","udf函数绑定了资源文件[{0}]"),
     RESOURCE_IS_USED(20014, "resource file is used by process definition","资源文件被上线的流程定义使用了"),
     PARENT_RESOURCE_NOT_EXIST(20015, "parent resource not exist","父资源文件不存在"),
+    RESOURCE_NOT_EXIST_OR_NO_PERMISSION(20016, "resource not exist or no permission,please view the task node and remove error resource","请检查任务节点并移除无权限或者已删除的资源"),
 
 
     USER_NO_OPERATION_PERM(30001, "user has no operation privilege", "当前用户没有操作权限"),

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ProcessDefinitionService.java

@@ -44,6 +44,7 @@ import org.apache.dolphinscheduler.common.utils.*;
 import org.apache.dolphinscheduler.dao.entity.*;
 import org.apache.dolphinscheduler.dao.mapper.*;
 import org.apache.dolphinscheduler.dao.utils.DagHelper;
+import org.apache.dolphinscheduler.service.permission.PermissionCheck;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -143,6 +144,7 @@ public class ProcessDefinitionService extends BaseDAGService {
         processDefine.setTimeout(processData.getTimeout());
         processDefine.setTenantId(processData.getTenantId());
         processDefine.setModifyBy(loginUser.getUserName());
+        processDefine.setResourceIds(getResourceIds(processData));
 
         //custom global params
         List<Property> globalParamsList = processData.getGlobalParams();
@@ -333,6 +335,7 @@ public class ProcessDefinitionService extends BaseDAGService {
         processDefine.setTimeout(processData.getTimeout());
         processDefine.setTenantId(processData.getTenantId());
         processDefine.setModifyBy(loginUser.getUserName());
+        processDefine.setResourceIds(getResourceIds(processData));
 
         //custom global params
         List<Property> globalParamsList = new ArrayList<>();
@@ -476,6 +479,20 @@ public class ProcessDefinitionService extends BaseDAGService {
 
         switch (state) {
             case ONLINE:
+                // To check resources whether they are already cancel authorized or deleted
+                String resourceIds = processDefinition.getResourceIds();
+                if (StringUtils.isNotBlank(resourceIds)) {
+                    Integer[] resourceIdArray = Arrays.stream(resourceIds.split(",")).map(Integer::parseInt).toArray(Integer[]::new);
+                    PermissionCheck<Integer> permissionCheck = new PermissionCheck(AuthorizationType.RESOURCE_FILE_ID,processService,resourceIdArray,loginUser.getId(),logger);
+                    try {
+                        permissionCheck.checkPermission();
+                    } catch (Exception e) {
+                        logger.error(e.getMessage(),e);
+                        putMsg(result, Status.RESOURCE_NOT_EXIST_OR_NO_PERMISSION, "releaseState");
+                        return result;
+                    }
+                }
+
                 processDefinition.setReleaseState(state);
                 processDefineMapper.updateById(processDefinition);
                 break;

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java

@@ -36,6 +36,7 @@ import org.apache.dolphinscheduler.dao.entity.Tenant;
 import org.apache.dolphinscheduler.dao.entity.UdfFunc;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.mapper.*;
+import org.apache.dolphinscheduler.dao.utils.ResourceProcessDefinitionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -176,6 +177,21 @@ public class ResourcesService extends BaseService {
             putMsg(result, Status.HDFS_NOT_STARTUP);
             return result;
         }
+
+        if (pid != -1) {
+            Resource parentResource = resourcesMapper.selectById(pid);
+
+            if (parentResource == null) {
+                putMsg(result, Status.PARENT_RESOURCE_NOT_EXIST);
+                return result;
+            }
+
+            if (!hasPerm(loginUser, parentResource.getUserId())) {
+                putMsg(result, Status.USER_NO_OPERATION_PERM);
+                return result;
+            }
+        }
+
         // file is empty
         if (file.isEmpty()) {
             logger.error("file is empty: {}", file.getOriginalFilename());
@@ -416,6 +432,14 @@ public class ResourcesService extends BaseService {
         if (isAdmin(loginUser)) {
             userId= 0;
         }
+        if (direcotryId != -1) {
+            Resource directory = resourcesMapper.selectById(direcotryId);
+            if (directory == null) {
+                putMsg(result, Status.RESOURCE_NOT_EXIST);
+                return result;
+            }
+        }
+
         IPage<Resource> resourceIPage = resourcesMapper.queryResourcePaging(page,
                 userId,direcotryId, type.ordinal(), searchVal);
         PageInfo pageInfo = new PageInfo<Resource>(pageNo, pageSize);
@@ -505,8 +529,12 @@ public class ResourcesService extends BaseService {
 
         Map<String, Object> result = new HashMap<>(5);
 
-        Set<Resource> allResourceList = getAllResources(loginUser, type);
-        Visitor resourceTreeVisitor = new ResourceTreeVisitor(new ArrayList<>(allResourceList));
+        int userId = loginUser.getId();
+        if(isAdmin(loginUser)){
+            userId = 0;
+        }
+        List<Resource> allResourceList = resourcesMapper.queryResourceListAuthored(userId, type.ordinal(),0);
+        Visitor resourceTreeVisitor = new ResourceTreeVisitor(allResourceList);
         //JSONArray jsonArray = JSON.parseArray(JSON.toJSONString(resourceTreeVisitor.visit().getChildren(), SerializerFeature.SortField));
         result.put(Constants.DATA_LIST, resourceTreeVisitor.visit().getChildren());
         putMsg(result,Status.SUCCESS);
@@ -519,7 +547,7 @@ public class ResourcesService extends BaseService {
      * @param loginUser     login user
      * @return all resource set
      */
-    private Set<Resource> getAllResources(User loginUser, ResourceType type) {
+    /*private Set<Resource> getAllResources(User loginUser, ResourceType type) {
         int userId = loginUser.getId();
         boolean listChildren = true;
         if(isAdmin(loginUser)){
@@ -540,7 +568,7 @@ public class ResourcesService extends BaseService {
             }
         }
         return allResourceList;
-    }
+    }*/
 
     /**
      * query resource list
@@ -553,7 +581,7 @@ public class ResourcesService extends BaseService {
 
         Map<String, Object> result = new HashMap<>(5);
 
-        Set<Resource> allResourceList = getAllResources(loginUser, type);
+        List<Resource> allResourceList = resourcesMapper.queryResourceListAuthored(loginUser.getId(), type.ordinal(),0);
         List<Resource> resources = new ResourceFilter(".jar",new ArrayList<>(allResourceList)).filter();
         Visitor resourceTreeVisitor = new ResourceTreeVisitor(resources);
         result.put(Constants.DATA_LIST, resourceTreeVisitor.visit().getChildren());
@@ -592,15 +620,6 @@ public class ResourcesService extends BaseService {
             putMsg(result, Status.USER_NO_OPERATION_PERM);
             return result;
         }
-        //if resource type is UDF,need check whether it is bound by UDF functon
-        if (resource.getType() == (ResourceType.UDF)) {
-            List<UdfFunc> udfFuncs = udfFunctionMapper.listUdfByResourceId(new int[]{resourceId});
-            if (CollectionUtils.isNotEmpty(udfFuncs)) {
-                logger.error("can't be deleted,because it is bound by UDF functions:{}",udfFuncs.toString());
-                putMsg(result,Status.UDF_RESOURCE_IS_BOUND,udfFuncs.get(0).getFuncName());
-                return result;
-            }
-        }
 
         String tenantCode = getTenantCode(resource.getUserId(),result);
         if (StringUtils.isEmpty(tenantCode)){
@@ -608,10 +627,22 @@ public class ResourcesService extends BaseService {
         }
 
         // get all resource id of process definitions those is released
-        Map<Integer, Set<Integer>> resourceProcessMap = getResourceProcessMap();
+        List<Map<String, Object>> list = processDefinitionMapper.listResources();
+        Map<Integer, Set<Integer>> resourceProcessMap = ResourceProcessDefinitionUtils.getResourceProcessDefinitionMap(list);
         Set<Integer> resourceIdSet = resourceProcessMap.keySet();
         // get all children of the resource
         List<Integer> allChildren = listAllChildren(resource);
+        Integer[] needDeleteResourceIdArray = allChildren.toArray(new Integer[allChildren.size()]);
+
+        //if resource type is UDF,need check whether it is bound by UDF functon
+        if (resource.getType() == (ResourceType.UDF)) {
+            List<UdfFunc> udfFuncs = udfFunctionMapper.listUdfByResourceId(needDeleteResourceIdArray);
+            if (CollectionUtils.isNotEmpty(udfFuncs)) {
+                logger.error("can't be deleted,because it is bound by UDF functions:{}",udfFuncs.toString());
+                putMsg(result,Status.UDF_RESOURCE_IS_BOUND,udfFuncs.get(0).getFuncName());
+                return result;
+            }
+        }
 
         if (resourceIdSet.contains(resource.getPid())) {
             logger.error("can't be deleted,because it is used of process definition");
@@ -632,8 +663,8 @@ public class ResourcesService extends BaseService {
         String hdfsFilename = HadoopUtils.getHdfsFileName(resource.getType(), tenantCode, resource.getFullName());
 
         //delete data in database
-        resourcesMapper.deleteIds(allChildren.toArray(new Integer[allChildren.size()]));
-        resourceUserMapper.deleteResourceUser(0, resourceId);
+        resourcesMapper.deleteIds(needDeleteResourceIdArray);
+        resourceUserMapper.deleteResourceUserArray(0, needDeleteResourceIdArray);
 
         //delete file on hdfs
         HadoopUtils.getInstance().delete(hdfsFilename, true);
@@ -1191,38 +1222,4 @@ public class ResourcesService extends BaseService {
         }
     }
 
-    /**
-     * get resource process map key is resource id,value is the set of process definition
-     * @return resource process definition map
-     */
-    private Map<Integer,Set<Integer>> getResourceProcessMap(){
-        Map<Integer, String> map = new HashMap<>();
-        Map<Integer, Set<Integer>> result = new HashMap<>();
-        List<Map<String, Object>> list = processDefinitionMapper.listResources();
-        if (CollectionUtils.isNotEmpty(list)) {
-            for (Map<String, Object> tempMap : list) {
-
-                map.put((Integer) tempMap.get("id"), (String)tempMap.get("resource_ids"));
-            }
-        }
-
-        for (Map.Entry<Integer, String> entry : map.entrySet()) {
-            Integer mapKey = entry.getKey();
-            String[] arr = entry.getValue().split(",");
-            Set<Integer> mapValues = Arrays.stream(arr).map(Integer::parseInt).collect(Collectors.toSet());
-            for (Integer value : mapValues) {
-                if (result.containsKey(value)) {
-                    Set<Integer> set = result.get(value);
-                    set.add(mapKey);
-                    result.put(value, set);
-                } else {
-                    Set<Integer> set = new HashSet<>();
-                    set.add(mapKey);
-                    result.put(value, set);
-                }
-            }
-        }
-        return result;
-    }
-
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/UsersService.java

@@ -16,6 +16,8 @@
  */
 package org.apache.dolphinscheduler.api.service;
 
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.CheckUtils;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
@@ -23,15 +25,10 @@ import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ResourceType;
 import org.apache.dolphinscheduler.common.enums.UserType;
-import org.apache.dolphinscheduler.common.utils.CollectionUtils;
-import org.apache.dolphinscheduler.common.utils.EncryptionUtils;
-import org.apache.dolphinscheduler.common.utils.HadoopUtils;
-import org.apache.dolphinscheduler.common.utils.PropertyUtils;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.dolphinscheduler.common.utils.StringUtils;
+import org.apache.dolphinscheduler.common.utils.*;
 import org.apache.dolphinscheduler.dao.entity.*;
 import org.apache.dolphinscheduler.dao.mapper.*;
+import org.apache.dolphinscheduler.dao.utils.ResourceProcessDefinitionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -39,6 +36,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * user service
@@ -72,6 +70,9 @@ public class UsersService extends BaseService {
     @Autowired
     private AlertGroupMapper alertGroupMapper;
 
+    @Autowired
+    private ProcessDefinitionMapper processDefinitionMapper;
+
 
     /**
      * create user, only system admin have permission
@@ -483,23 +484,71 @@ public class UsersService extends BaseService {
             return result;
         }
 
+        String[] resourceFullIdArr = resourceIds.split(",");
+        // need authorize resource id set
+        Set<Integer> needAuthorizeResIds = new HashSet();
+        for (String resourceFullId : resourceFullIdArr) {
+            String[] resourceIdArr = resourceFullId.split("-");
+            for (int i=0;i<=resourceIdArr.length-1;i++) {
+                int resourceIdValue = Integer.parseInt(resourceIdArr[i]);
+                needAuthorizeResIds.add(resourceIdValue);
+            }
+        }
+
+        //get the authorized resource id list by user id
+        List<Resource> oldAuthorizedRes = resourceMapper.queryAuthorizedResourceList(userId);
+        //if resource type is UDF,need check whether it is bound by UDF functon
+        Set<Integer> oldAuthorizedResIds = oldAuthorizedRes.stream().map(t -> t.getId()).collect(Collectors.toSet());
+
+        //get the unauthorized resource id list
+        oldAuthorizedResIds.removeAll(needAuthorizeResIds);
+
+        if (CollectionUtils.isNotEmpty(oldAuthorizedResIds)) {
+
+            // get all resource id of process definitions those is released
+            List<Map<String, Object>> list = processDefinitionMapper.listResources();
+            Map<Integer, Set<Integer>> resourceProcessMap = ResourceProcessDefinitionUtils.getResourceProcessDefinitionMap(list);
+            Set<Integer> resourceIdSet = resourceProcessMap.keySet();
+
+            resourceIdSet.retainAll(oldAuthorizedResIds);
+            if (CollectionUtils.isNotEmpty(resourceIdSet)) {
+                logger.error("can't be deleted,because it is used of process definition");
+                for (Integer resId : resourceIdSet) {
+                    logger.error("resource id:{} is used of process definition {}",resId,resourceProcessMap.get(resId));
+                }
+                putMsg(result, Status.RESOURCE_IS_USED);
+                return result;
+            }
+
+        }
+
         resourcesUserMapper.deleteResourceUser(userId, 0);
 
         if (check(result, StringUtils.isEmpty(resourceIds), Status.SUCCESS)) {
             return result;
         }
 
-        String[] resourcesIdArr = resourceIds.split(",");
+        for (int resourceIdValue : needAuthorizeResIds) {
+            Resource resource = resourceMapper.selectById(resourceIdValue);
+            if (resource == null) {
+                putMsg(result, Status.RESOURCE_NOT_EXIST);
+                return result;
+            }
 
-        for (String resourceId : resourcesIdArr) {
             Date now = new Date();
             ResourcesUser resourcesUser = new ResourcesUser();
             resourcesUser.setUserId(userId);
-            resourcesUser.setResourcesId(Integer.parseInt(resourceId));
-            resourcesUser.setPerm(7);
+            resourcesUser.setResourcesId(resourceIdValue);
+            if (resource.isDirectory()) {
+                resourcesUser.setPerm(Constants.AUTHORIZE_READABLE_PERM);
+            }else{
+                resourcesUser.setPerm(Constants.AUTHORIZE_WRITABLE_PERM);
+            }
+
             resourcesUser.setCreateTime(now);
             resourcesUser.setUpdateTime(now);
             resourcesUserMapper.insert(resourcesUser);
+
         }
 
         putMsg(result, Status.SUCCESS);

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java

@@ -242,7 +242,7 @@ public class ResourcesServiceTest {
         User loginUser = new User();
         loginUser.setId(0);
         loginUser.setUserType(UserType.ADMIN_USER);
-        Mockito.when(resourcesMapper.queryResourceListAuthored(0, 0)).thenReturn(getResourceList());
+        Mockito.when(resourcesMapper.queryResourceListAuthored(0, 0,0)).thenReturn(getResourceList());
         Map<String, Object> result = resourcesService.queryResourceList(loginUser, ResourceType.FILE);
         logger.info(result.toString());
         Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UsersServiceTest.java

@@ -18,13 +18,16 @@ package org.apache.dolphinscheduler.api.service;
 
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import org.apache.avro.generic.GenericData;
 import org.apache.dolphinscheduler.api.enums.Status;
 import org.apache.dolphinscheduler.api.utils.PageInfo;
 import org.apache.dolphinscheduler.api.utils.Result;
 import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ResourceType;
 import org.apache.dolphinscheduler.common.enums.UserType;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
 import org.apache.dolphinscheduler.common.utils.EncryptionUtils;
+import org.apache.dolphinscheduler.dao.entity.Resource;
 import org.apache.dolphinscheduler.dao.entity.Tenant;
 import org.apache.dolphinscheduler.dao.entity.User;
 import org.apache.dolphinscheduler.dao.mapper.*;
@@ -68,6 +71,8 @@ public class UsersServiceTest {
     private DataSourceUserMapper datasourceUserMapper;
     @Mock
     private AlertGroupMapper alertGroupMapper;
+    @Mock
+    private ResourceMapper resourceMapper;
 
     private String queueName ="UsersServiceTestQueue";
 
@@ -301,9 +306,13 @@ public class UsersServiceTest {
         logger.info(result.toString());
         Assert.assertEquals(Status.USER_NOT_EXIST, result.get(Constants.STATUS));
         //success
+        when(resourceMapper.queryAuthorizedResourceList(1)).thenReturn(new ArrayList<Resource>());
+
+        when(resourceMapper.selectById(Mockito.anyInt())).thenReturn(getResource());
         result = usersService.grantResources(loginUser, 1, resourceIds);
         logger.info(result.toString());
         Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+
     }
 
 
@@ -476,11 +485,30 @@ public class UsersServiceTest {
         return user;
     }
 
-
+    /**
+     * get tenant
+     * @return tenant
+     */
     private Tenant getTenant(){
         Tenant tenant = new Tenant();
         tenant.setId(1);
         return tenant;
     }
 
+    /**
+     * get resource
+     * @return resource
+     */
+    private Resource getResource(){
+
+        Resource resource = new Resource();
+        resource.setPid(-1);
+        resource.setUserId(1);
+        resource.setDescription("ResourcesServiceTest.jar");
+        resource.setAlias("ResourcesServiceTest.jar");
+        resource.setFullName("/ResourcesServiceTest.jar");
+        resource.setType(ResourceType.FILE);
+        return resource;
+    }
+
 }
\ No newline at end of file

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java

@@ -946,4 +946,12 @@ public final class Constants {
      * schedule time
      */
     public static final String PARAMETER_SHECDULE_TIME = "schedule.time";
+    /**
+     * authorize writable perm
+     */
+    public static final int AUTHORIZE_WRITABLE_PERM=7;
+    /**
+     * authorize readable perm
+     */
+    public static final int AUTHORIZE_READABLE_PERM=4;
 }

dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/HadoopUtilsTest.java

@@ -16,6 +16,7 @@
  */
 package org.apache.dolphinscheduler.common.utils;
 
+import org.apache.dolphinscheduler.common.enums.ResourceType;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -85,4 +86,19 @@ public class HadoopUtilsTest {
         List<String> stringList = HadoopUtils.getInstance().catFile("/dolphinscheduler/hdfs/resources/WCSparkPython.py", 0, 1000);
         logger.info(String.join(",",stringList));
     }
+
+    @Test
+    public void getHdfsFileNameTest(){
+        logger.info(HadoopUtils.getHdfsFileName(ResourceType.FILE,"test","/test"));
+    }
+
+    @Test
+    public void getHdfsResourceFileNameTest(){
+        logger.info(HadoopUtils.getHdfsResourceFileName("test","/test"));
+    }
+
+    @Test
+    public void getHdfsUdfFileNameTest(){
+        logger.info(HadoopUtils.getHdfsUdfFileName("test","/test.jar"));
+    }
 }
\ No newline at end of file

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ResourceMapper.java

@@ -43,11 +43,13 @@ public interface ResourceMapper extends BaseMapper<Resource> {
      * query resource list
      * @param userId userId
      * @param type type
+     * @param perm perm
      * @return resource list
      */
     List<Resource> queryResourceListAuthored(
                                      @Param("userId") int userId,
-                                     @Param("type") int type);
+                                     @Param("type") int type,
+                                     @Param("perm") int perm);
 
 
     /**

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ResourceUserMapper.java

@@ -34,4 +34,13 @@ public interface ResourceUserMapper extends BaseMapper<ResourcesUser> {
     int deleteResourceUser(@Param("userId") int userId,
                            @Param("resourceId") int resourceId);
 
+    /**
+     * delete resource user relation
+     * @param userId userId
+     * @param resIds resource Ids
+     * @return delete result
+     */
+    int deleteResourceUserArray(@Param("userId") int userId,
+                           @Param("resIds") Integer[] resIds);
+
 }

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/UdfFuncMapper.java

@@ -81,9 +81,8 @@ public interface UdfFuncMapper extends BaseMapper<UdfFunc> {
     /**
      * list authorized UDF function
      * @param userId userId
-     * @param udfIds udfIds
-     * @param <T> T
-     * @return Udf function list
+     * @param udfIds UDF function id array
+     * @return UDF function list
      */
     <T> List<UdfFunc> listAuthorizedUdfFunc (@Param("userId") int userId,@Param("udfIds")T[] udfIds);
 
@@ -92,7 +91,7 @@ public interface UdfFuncMapper extends BaseMapper<UdfFunc> {
      * @param   resourceIds  resource id array
      * @return  UDF function list
      */
-    List<UdfFunc> listUdfByResourceId(@Param("resourceIds") int[] resourceIds);
+    List<UdfFunc> listUdfByResourceId(@Param("resourceIds") Integer[] resourceIds);
 
     /**
      * list authorized UDF by resource id

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/ResourceProcessDefinitionUtils.java

+/*
+ * 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.dolphinscheduler.dao.utils;
+
+import org.apache.dolphinscheduler.common.utils.CollectionUtils;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * resource process definition utils
+ */
+public class ResourceProcessDefinitionUtils {
+    /**
+     * get resource process map key is resource id,value is the set of process definition
+     * @param list the map key is process definition id and value is resource_ids
+     * @return resource process definition map
+     */
+    public static Map<Integer, Set<Integer>> getResourceProcessDefinitionMap(List<Map<String, Object>> list) {
+        Map<Integer, String> map = new HashMap<>();
+        Map<Integer, Set<Integer>> result = new HashMap<>();
+        if (CollectionUtils.isNotEmpty(list)) {
+            for (Map<String, Object> tempMap : list) {
+
+                map.put((Integer) tempMap.get("id"), (String)tempMap.get("resource_ids"));
+            }
+        }
+
+        for (Map.Entry<Integer, String> entry : map.entrySet()) {
+            Integer mapKey = entry.getKey();
+            String[] arr = entry.getValue().split(",");
+            Set<Integer> mapValues = Arrays.stream(arr).map(Integer::parseInt).collect(Collectors.toSet());
+            for (Integer value : mapValues) {
+                if (result.containsKey(value)) {
+                    Set<Integer> set = result.get(value);
+                    set.add(mapKey);
+                    result.put(value, set);
+                } else {
+                    Set<Integer> set = new HashSet<>();
+                    set.add(mapKey);
+                    result.put(value, set);
+                }
+            }
+        }
+        return result;
+    }
+}

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ResourceMapper.xml

@@ -39,7 +39,11 @@
         <if test="type != -1">
             and type=#{type}
         </if>
-        <if test="userId != 0">
+        <if test="userId != 0 and perm != 0">
+            and id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId} and perm=#{perm}
+            union select id as resources_id  from t_ds_resources where user_id=#{userId})
+        </if>
+        <if test="userId != 0 and perm == 0">
             and id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId}
             union select id as resources_id  from t_ds_resources where user_id=#{userId})
         </if>
@@ -48,7 +52,7 @@
         select *
         from t_ds_resources
         where type=#{type} and pid=#{id}
-        <if test="userId != 0 and id == -1">
+        <if test="userId != 0">
             and id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId}
             union select id as resources_id  from t_ds_resources where user_id=#{userId})
         </if>
@@ -60,7 +64,7 @@
     <select id="queryAuthorizedResourceList" resultType="org.apache.dolphinscheduler.dao.entity.Resource">
         select r.*
         from  t_ds_resources r,t_ds_relation_resources_user rel
-        where r.id = rel.resources_id AND rel.user_id = #{userId}
+        where r.id = rel.resources_id AND rel.user_id = #{userId} and perm=7
     </select>
     <select id="queryResourceExceptUserId" resultType="org.apache.dolphinscheduler.dao.entity.Resource">
         select *
@@ -77,7 +81,7 @@
         select *
         from t_ds_resources
         where type=0
-        and id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId}
+        and id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId} and perm=7
         union select id as resources_id  from t_ds_resources where user_id=#{userId})
         <if test="resNames != null and resNames != ''">
             and full_name in
@@ -89,7 +93,7 @@
     <select id="listAuthorizedResourceById" resultType="org.apache.dolphinscheduler.dao.entity.Resource">
         select *
         from t_ds_resources
-        where id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId}
+        where id in (select resources_id from  t_ds_relation_resources_user where user_id=#{userId} and perm=7
         union select id as resources_id  from t_ds_resources where user_id=#{userId})
         <if test="resIds != null and resIds != ''">
             and id in

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ResourceUserMapper.xml

@@ -29,4 +29,17 @@
             and resources_id = #{resourceId}
         </if>
     </delete>
+
+    <delete id="deleteResourceUserArray" parameterType="java.lang.Integer">
+        delete
+        from t_ds_relation_resources_user
+        where 1 = 1
+        <if test="userId != 0">
+            and user_id = #{userId}
+        </if>
+        and resources_id in
+        <foreach collection="resIds" item="i" open="(" close=")" separator=",">
+            #{i}
+        </foreach>
+    </delete>
 </mapper>
\ No newline at end of file

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ProcessDefinitionMapperTest.java

@@ -17,6 +17,7 @@
 package org.apache.dolphinscheduler.dao.mapper;
 
 
+import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.common.enums.UserType;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -32,6 +33,7 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 @RunWith(SpringRunner.class)
 @SpringBootTest
@@ -215,4 +217,13 @@ public class ProcessDefinitionMapperTest {
         );
         Assert.assertNotEquals(processDefinitions.size(), 0);
     }
+
+    @Test
+    public void listResourcesTest(){
+        ProcessDefinition processDefinition = insertOne();
+        processDefinition.setResourceIds("3,5");
+        processDefinition.setReleaseState(ReleaseState.ONLINE);
+        List<Map<String, Object>> maps = processDefinitionMapper.listResources();
+        Assert.assertNotNull(maps);
+    }
 }
\ No newline at end of file

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ResourceMapperTest.java

@@ -19,6 +19,7 @@ package org.apache.dolphinscheduler.dao.mapper;
 
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.ResourceType;
 import org.apache.dolphinscheduler.common.enums.UserType;
 import org.apache.dolphinscheduler.dao.entity.Resource;
@@ -138,6 +139,7 @@ public class ResourceMapperTest {
         resourcesUser.setUpdateTime(new Date());
         resourcesUser.setUserId(user.getId());
         resourcesUser.setResourcesId(resource.getId());
+        resourcesUser.setPerm(7);
         resourceUserMapper.insert(resourcesUser);
         return resourcesUser;
     }
@@ -247,6 +249,7 @@ public class ResourceMapperTest {
 
         resourcesUser.setResourcesId(resource.getId());
         resourcesUser.setUserId(1110);
+        resourcesUser.setPerm(Constants.AUTHORIZE_WRITABLE_PERM);
         resourceUserMapper.insert(resourcesUser);
 
         List<Resource> resources1 = resourceMapper.queryAuthorizedResourceList(1110);
@@ -351,4 +354,34 @@ public class ResourceMapperTest {
         int result = resourceMapper.deleteIds(resourceList.toArray(new Integer[resourceList.size()]));
         Assert.assertEquals(result,2);
     }
+
+    @Test
+    public void queryResourceListAuthoredTest(){
+        // create a general user
+        User generalUser1 = createGeneralUser("user1");
+        User generalUser2 = createGeneralUser("user2");
+        // create resource
+        Resource resource = createResource(generalUser1);
+        createResourcesUser(resource, generalUser2);
+
+        List<Resource> resourceList = resourceMapper.queryResourceListAuthored(generalUser2.getId(), ResourceType.FILE.ordinal(), 0);
+        Assert.assertNotNull(resourceList);
+
+        resourceList = resourceMapper.queryResourceListAuthored(generalUser2.getId(), ResourceType.FILE.ordinal(), 4);
+        Assert.assertFalse(resourceList.contains(resource));
+    }
+
+    @Test
+    public void batchUpdateResourceTest(){
+        // create a general user
+        User generalUser1 = createGeneralUser("user1");
+        // create resource
+        Resource resource = createResource(generalUser1);
+        resource.setFullName(String.format("%s-update",resource.getFullName()));
+        resource.setUpdateTime(new Date());
+        List<Resource> resourceList = new ArrayList<>();
+        resourceList.add(resource);
+        int result = resourceMapper.batchUpdateResource(resourceList);
+        Assert.assertTrue(result>0);
+    }
 }
\ No newline at end of file

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ResourceUserMapperTest.java

@@ -17,6 +17,7 @@
 package org.apache.dolphinscheduler.dao.mapper;
 
 
+import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.dao.entity.ResourcesUser;
 import org.junit.Assert;
 import org.junit.Test;
@@ -47,13 +48,14 @@ public class ResourceUserMapperTest {
      */
     private ResourcesUser insertOne(){
         //insertOne
-        ResourcesUser queue = new ResourcesUser();
-        queue.setCreateTime(new Date());
-        queue.setUpdateTime(new Date());
-        queue.setUserId(11111);
-        queue.setResourcesId(1110);
-        resourceUserMapper.insert(queue);
-        return queue;
+        ResourcesUser resourcesUser = new ResourcesUser();
+        resourcesUser.setCreateTime(new Date());
+        resourcesUser.setUpdateTime(new Date());
+        resourcesUser.setUserId(11111);
+        resourcesUser.setResourcesId(1110);
+        resourcesUser.setPerm(Constants.AUTHORIZE_WRITABLE_PERM);
+        resourceUserMapper.insert(resourcesUser);
+        return resourcesUser;
     }
 
     /**
@@ -102,4 +104,18 @@ public class ResourceUserMapperTest {
                 queue.getResourcesId());
         Assert.assertNotEquals(delete, 0);
     }
+
+    /**
+     * test delete
+     */
+    @Test
+    public void testDeleteResourceUserArray() {
+
+        ResourcesUser resourcesUser = insertOne();
+        Integer[] resourceIdArray = new Integer[]{resourcesUser.getResourcesId()};
+        int delete = resourceUserMapper.deleteResourceUserArray(
+                resourcesUser.getUserId(),
+                resourceIdArray);
+        Assert.assertNotEquals(delete, 0);
+    }
 }
\ No newline at end of file

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/ResourceProcessDefinitionUtilsTest.java

+/*
+ * 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.dolphinscheduler.dao.utils;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * resource process definition utils test
+ */
+public class ResourceProcessDefinitionUtilsTest {
+    @Test
+    public void getResourceProcessDefinitionMapTest(){
+        List<Map<String,Object>> mapList = new ArrayList<>();
+        Map<String,Object> map = new HashMap();
+        map.put("id",1);
+        map.put("resource_ids","1,2,3");
+        mapList.add(map);
+        Assert.assertNotNull(ResourceProcessDefinitionUtils.getResourceProcessDefinitionMap(mapList));
+    }
+
+}
\ No newline at end of file