include_upstream=False,
                include_downstream=True,
            )
            result.update(
                p_dag._get_task_instances(
                    task_ids=task_ids,
                    start_date=start_date,
                    end_date=end_date,
                    run_id=None,
                    state=state,
                    include_subdags=include_subdags,
                    include_parentdag=False,
                    include_dependent_dags=include_dependent_dags,
                    as_pk_tuple=True,
                    exclude_task_ids=exclude_task_ids,
                    session=session,
                    dag_bag=dag_bag,
                    recursion_depth=recursion_depth,
                    max_recursion_depth=max_recursion_depth,
                    visited_external_tis=visited_external_tis,
                )
            )
        if include_dependent_dags:
            
            from airflow.sensors.external_task import ExternalTaskMarker
            query = tis
            if as_pk_tuple:
                condition = TI.filter_for_tis(TaskInstanceKey(*cols) for cols in tis.all())
                if condition is not None:
                    query = session.query(TI).filter(condition)