<?xml version="1.0" encoding="UTF-8" ?>
<!--
  ~ 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.
  -->

<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
<mapper namespace="org.apache.dolphinscheduler.dao.mapper.WorkFlowLineageMapper">

    <select id="queryWorkFlowLineageByName" resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
        select tepd.code as work_flow_code,tepd.name as work_flow_name
        from t_ds_process_definition tepd
        left join t_ds_schedules tes on tepd.code = tes.process_definition_code
        where tepd.project_code = #{projectCode}
        <if test="workFlowName != null and workFlowName != ''">
            and tepd.name like concat('%', #{workFlowName}, '%')
        </if>
    </select>

    <select id="queryWorkFlowLineageByCode" resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
        select tepd.code as work_flow_code,tepd.name as work_flow_name,
               '' as source_work_flow_code,
                tepd.release_state as work_flow_publish_status,
                tes.start_time as schedule_start_time,
                tes.end_time as schedule_end_time,
                tes.crontab as crontab,
                tes.release_state as schedule_publish_status
        from t_ds_process_definition tepd
        left join t_ds_schedules tes on tepd.code = tes.process_definition_code
        where tepd.project_code = #{projectCode} and tepd.code = #{workFlowCode}
    </select>

    <select id="queryWorkFlowLineageByProcessDefinitionCodes" resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
        select tepd.code as work_flow_code,
               tepd.name as work_flow_name,
               tepd.release_state as work_flow_publish_status,
               tes.start_time as schedule_start_time,
               tes.end_time as schedule_end_time,
               tes.crontab as crontab,
               tes.release_state as schedule_publish_status
        from t_ds_process_definition tepd
        left join t_ds_schedules tes on tepd.code = tes.process_definition_code
        where 1=1
        <if test="workFlowCodes != null and workFlowCodes.size() != 0 ">
            and tepd.code in
            <foreach collection="workFlowCodes" item="code" index="index" open="(" close=")" separator=",">
                #{code}
            </foreach>
        </if>
    </select>

    <select id="queryWorkFlowLineageByLineage" resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
        select tepd.code as work_flow_code,tepd.name as work_flow_name,
               '' as source_work_flow_code,
                tepd.release_state as work_flow_publish_status,
                tes.start_time as schedule_start_time,
                tes.end_time as schedule_end_time,
                tes.crontab as crontab,
                tes.release_state as schedule_publish_status
        from t_ds_process_definition tepd
        left join t_ds_schedules tes on tepd.code = tes.process_definition_code
        where 1=1
        <if test="processLineages != null and processLineages.size != 0">
            and
            <foreach collection="processLineages" index="index" item="item" open="(" separator=" or " close=")">
                (tepd.project_code = #{item.projectCode}
                and tepd.code = #{item.processDefinitionCode})
            </foreach>
        </if>
    </select>

    <select id="queryProcessLineage" resultType="org.apache.dolphinscheduler.dao.entity.ProcessLineage">
    select ptr.project_code,
        ptr.post_task_code,
        ptr.post_task_version,
        ptr.pre_task_code,
        ptr.pre_task_version,
        ptr.process_definition_code,
        ptr.process_definition_version
        from t_ds_process_definition pd
        join t_ds_process_task_relation ptr on pd.code = ptr.process_definition_code
        and pd.version = ptr.process_definition_version
        where pd.project_code = #{projectCode}
    </select>

    <select id="queryProcessLineageByCode" resultType="org.apache.dolphinscheduler.dao.entity.ProcessLineage">
        select project_code,
               post_task_code,
               post_task_version,
               pre_task_code,
               pre_task_version,
               process_definition_code,
               process_definition_version
         from t_ds_process_task_relation
         where project_code = #{projectCode}
                and process_definition_code = #{processDefinitionCode}
    </select>

    <select id="queryUpstreamDependentParamsByProcessDefinitionCode"
            resultType="org.apache.dolphinscheduler.dao.entity.DependentProcessDefinition">
        SELECT
            DISTINCT c.task_params
        FROM
        t_ds_process_definition a
        JOIN t_ds_process_task_relation b ON a.code = b.process_definition_code AND a.version = b.process_definition_version AND a.project_code = b.project_code
        JOIN t_ds_task_definition c ON c.code = b.pre_task_code and c.version = b.pre_task_version
        WHERE 1=1
        AND a.code = #{code}
        AND c.task_type = #{taskType}
        ;
    </select>

    <select id="queryDownstreamLineageByProcessDefinitionCode"
            resultType="org.apache.dolphinscheduler.dao.entity.WorkFlowLineage">
        SELECT
            c.code AS work_flow_code
            ,c.name AS work_flow_name
            ,c.release_state AS work_flow_publish_status
            ,d.start_time AS schedule_start_time
            ,d.end_time AS schedule_end_time
            ,d.crontab AS crontab
            ,d.release_state AS schedule_publish_status
            ,'' AS source_work_flow_code
        FROM
        t_ds_task_definition a
        JOIN t_ds_process_task_relation b ON a.code	= b.pre_task_code and a.version = b.pre_task_version
        JOIN t_ds_process_definition c ON c.code = b.process_definition_code AND c.version = b.process_definition_version AND c.project_code = b.project_code
        LEFT JOIN t_ds_schedules d ON d.process_definition_code = c.code
        WHERE 1=1
        <if test="code != null and code != ''">
            AND a.task_params LIKE concat('%', #{code}, '%')
        </if>
        AND a.task_type = #{taskType}
    </select>

    <select id="queryDependentProcessDefinitionByProcessDefinitionCode" resultType="DependentProcessDefinition">
        SELECT
        c.code AS process_definition_code
        ,c.name AS process_definition_name
        ,c.version as process_definition_version
        ,a.code AS task_definition_code
        ,a.task_params
        FROM
        t_ds_task_definition a
        JOIN t_ds_process_task_relation b ON a.code	= b.pre_task_code and a.version = b.pre_task_version
        JOIN t_ds_process_definition c ON c.code = b.process_definition_code AND c.version = b.process_definition_version AND c.project_code = b.project_code
        WHERE 1=1
        <if test="code != null and code != ''">
            AND a.task_params LIKE concat('%', #{code}, '%')
        </if>
        AND a.task_type = 'DEPENDENT'
        AND c.release_state = 1
    </select>

    <select id="queryTaskSubProcessDepOnProcess" resultType="org.apache.dolphinscheduler.dao.entity.TaskMainInfo">
        select td.id
            , td.name as taskName
            , td.code as taskCode
            , td.version as taskVersion
            , td.task_type as taskType
            , ptr.process_definition_code as processDefinitionCode
            , pd.name as processDefinitionName
            , pd.version as processDefinitionVersion
            , pd.release_state as processReleaseState
        from t_ds_task_definition td
        join t_ds_process_task_relation ptr on ptr.post_task_code = td.code and td.version = ptr.post_task_version
        join t_ds_process_definition pd on pd.code = ptr.process_definition_code and pd.version = ptr.process_definition_version
        <where>
            <if test="projectCode != 0">
                and ptr.project_code = #{projectCode}
            </if>
            <!-- ptr.process_definition_code != #{processDefinitionCode} query task not in current workflow -->
            <!-- For subprocess task type, using `concat('%"processDefinitionCode":', #{processDefinitionCode}, '%')` -->
            <if test="processDefinitionCode != 0">
                and td.task_type = 'SUB_PROCESS'
                and ptr.process_definition_code != #{processDefinitionCode}
                and td.task_params like concat('%"processDefinitionCode":', #{processDefinitionCode}, '%')
            </if>
        </where>
    </select>

    <select id="queryTaskDependentDepOnProcess" resultType="org.apache.dolphinscheduler.dao.entity.TaskMainInfo">
        select td.id
        , td.name as taskName
        , td.code as taskCode
        , td.version as taskVersion
        , td.task_type as taskType
        , ptr.process_definition_code as processDefinitionCode
        , pd.name as processDefinitionName
        , pd.version as processDefinitionVersion
        , pd.release_state as processReleaseState
        from t_ds_task_definition td
        join t_ds_process_task_relation ptr on ptr.post_task_code = td.code and td.version = ptr.post_task_version
        join t_ds_process_definition pd on pd.code = ptr.process_definition_code and pd.version = ptr.process_definition_version
        <where>
            <if test="projectCode != 0">
                and ptr.project_code = #{projectCode}
            </if>
            <!-- ptr.process_definition_code != #{processDefinitionCode} query task not in current workflow -->
            <!-- For dependnet task type, using `like concat('%"definitionCode":', #{processDefinitionCode}, '%')` -->
            <if test="processDefinitionCode != 0">
                and td.task_type = 'DEPENDENT'
                and ptr.process_definition_code != #{processDefinitionCode}
                and td.task_params like concat('%"definitionCode":', #{processDefinitionCode}, '%')
            </if>
        </where>
    </select>

    <select id="queryTaskDepOnTask" resultType="org.apache.dolphinscheduler.dao.entity.TaskMainInfo">
        select td.id
             , td.name as taskName
             , td.code as taskCode
             , td.version as taskVersion
             , td.task_type as taskType
             , ptr.process_definition_code as processDefinitionCode
             , pd.name as processDefinitionName
             , pd.version as processDefinitionVersion
             , pd.release_state as processReleaseState
        from t_ds_task_definition td
        join t_ds_process_task_relation ptr on ptr.post_task_code = td.code and td.version = ptr.post_task_version
        join t_ds_process_definition pd on pd.code = ptr.process_definition_code and pd.version = ptr.process_definition_version
        <where>
            <if test="projectCode != 0">
                and ptr.project_code = #{projectCode}
            </if>
            <!-- ptr.process_definition_code != #{processDefinitionCode} query task not in current workflow -->
            <if test="processDefinitionCode != 0">
                and ptr.process_definition_code != #{processDefinitionCode}
                and td.task_params like concat('%"definitionCode":', #{processDefinitionCode}, '%')
            </if>
            <if test="taskCode != 0">
                and td.task_params like concat('%"depTaskCode":', #{taskCode}, '%')
            </if>
        </where>
    </select>
</mapper>
