<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">When using a <code class="notranslate">@task.group</code> for the depth first expansion functionality, a <code class="notranslate">@task.short_circuit</code> in that group throws a ForeignKeyViolation</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 1905, in _execute_context
    self.dialect.do_execute(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py&quot;, line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.ForeignKeyViolation: insert or update on table &quot;xcom&quot; violates foreign key constraint &quot;xcom_task_instance_fkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table &quot;task_instance&quot;.
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/decorators/base.py&quot;, line 220, in execute
    return_value = super().execute(context)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/operators/python.py&quot;, line 268, in execute
    self.skip(dag_run, execution_date, downstream_tasks)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py&quot;, line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/models/skipmixin.py&quot;, line 139, in skip
    XCom.set(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py&quot;, line 73, in wrapper
    return func(*args, **kwargs)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py&quot;, line 264, in set
    session.flush()
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py&quot;, line 3449, in flush
    self._flush(objects)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py&quot;, line 3588, in _flush
    with util.safe_reraise():
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 70, in __exit__
    compat.raise_(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py&quot;, line 211, in raise_
    raise exception
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py&quot;, line 3549, in _flush
    flush_context.execute()
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 456, in execute
    rec.execute(self)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py&quot;, line 630, in execute
    util.preloaded.orm_persistence.save_obj(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py&quot;, line 245, in save_obj
    _emit_insert_statements(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 1710, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/sql/elements.py&quot;, line 334, in _execute_on_connection
    return connection._execute_clauseelement(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 1577, in _execute_clauseelement
    ret = self._execute_context(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 1948, in _execute_context
    self._handle_dbapi_exception(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 2129, in _handle_dbapi_exception
    util.raise_(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py&quot;, line 211, in raise_
    raise exception
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py&quot;, line 1905, in _execute_context
    self.dialect.do_execute(
  File &quot;/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py&quot;, line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table &quot;xcom&quot; violates foreign key constraint &quot;xcom_task_instance_fkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table &quot;task_instance&quot;.
[SQL: INSERT INTO xcom (dag_run_id, task_id, map_index, key, dag_id, run_id, value, timestamp) VALUES (%(dag_run_id)s, %(task_id)s, %(map_index)s, %(key)s, %(dag_id)s, %(run_id)s, %(value)s, %(timestamp)s)]
[parameters: {'dag_run_id': 93, 'task_id': 'expansion_group.short_circuit', 'map_index': -1, 'key': 'skipmixin_key', 'dag_id': 'test_short_circuit', 'run_id': 'manual__2023-05-27T22:49:24.543812+00:00', 'value': &lt;psycopg2.extensions.Binary object at 0xffff7ae48b10&gt;, 'timestamp': datetime.datetime(2023, 5, 27, 22, 49, 57, 760293, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
[2023-05-27, 22:49:57 UTC] {taskinstance.py:1345} INFO - Marking task as FAILED. dag_id=test_short_circuit, task_id=expansion_group.short_circuit, map_index=0, execution_date=20230527T224924, start_date=20230527T224955, end_date=20230527T224957
[2023-05-27, 22:49:57 UTC] {standard_task_runner.py:104} ERROR - Failed to execute job 1765 for task expansion_group.short_circuit ((psycopg2.errors.ForeignKeyViolation) insert or update on table &quot;xcom&quot; violates foreign key constraint &quot;xcom_task_instance_fkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table &quot;task_instance&quot;.
[SQL: INSERT INTO xcom (dag_run_id, task_id, map_index, key, dag_id, run_id, value, timestamp) VALUES (%(dag_run_id)s, %(task_id)s, %(map_index)s, %(key)s, %(dag_id)s, %(run_id)s, %(value)s, %(timestamp)s)]
[parameters: {'dag_run_id': 93, 'task_id': 'expansion_group.short_circuit', 'map_index': -1, 'key': 'skipmixin_key', 'dag_id': 'test_short_circuit', 'run_id': 'manual__2023-05-27T22:49:24.543812+00:00', 'value': &lt;psycopg2.extensions.Binary object at 0xffff7ae48b10&gt;, 'timestamp': datetime.datetime(2023, 5, 27, 22, 49, 57, 760293, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj); 104)"><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context
    self.dialect.do_execute(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.ForeignKeyViolation: insert or update on table "xcom" violates foreign key constraint "xcom_task_instance_fkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table "task_instance".
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/decorators/base.py", line 220, in execute
    return_value = super().execute(context)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/operators/python.py", line 268, in execute
    self.skip(dag_run, execution_date, downstream_tasks)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/skipmixin.py", line 139, in skip
    XCom.set(
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 73, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/xcom.py", line 264, in set
    session.flush()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3449, in flush
    self._flush(objects)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3588, in _flush
    with util.safe_reraise():
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3549, in _flush
    flush_context.execute()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute
    rec.execute(self)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 630, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 245, in save_obj
    _emit_insert_statements(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1710, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement
    ret = self._execute_context(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1948, in _execute_context
    self._handle_dbapi_exception(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2129, in _handle_dbapi_exception
    util.raise_(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1905, in _execute_context
    self.dialect.do_execute(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table "xcom" violates foreign key constraint "xcom_task_instance_fkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table "task_instance".
[SQL: INSERT INTO xcom (dag_run_id, task_id, map_index, key, dag_id, run_id, value, timestamp) VALUES (%(dag_run_id)s, %(task_id)s, %(map_index)s, %(key)s, %(dag_id)s, %(run_id)s, %(value)s, %(timestamp)s)]
[parameters: {'dag_run_id': 93, 'task_id': 'expansion_group.short_circuit', 'map_index': -1, 'key': 'skipmixin_key', 'dag_id': 'test_short_circuit', 'run_id': 'manual__2023-05-27T22:49:24.543812+00:00', 'value': &lt;psycopg2.extensions.Binary object at 0xffff7ae48b10&gt;, 'timestamp': datetime.datetime(2023, 5, 27, 22, 49, 57, 760293, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
[2023-05-27, 22:49:57 UTC] {taskinstance.py:1345} INFO - Marking task as FAILED. dag_id=test_short_circuit, task_id=expansion_group.short_circuit, map_index=0, execution_date=20230527T224924, start_date=20230527T224955, end_date=20230527T224957
[2023-05-27, 22:49:57 UTC] {standard_task_runner.py:104} ERROR - Failed to execute job 1765 for task expansion_group.short_circuit ((psycopg2.errors.ForeignKeyViolation) insert or update on table "xcom" violates foreign key constraint "xcom_task_instance_fkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_short_circuit, expansion_group.short_circuit, manual__2023-05-27T22:49:24.543812+00:00, -1) is not present in table "task_instance".
[SQL: INSERT INTO xcom (dag_run_id, task_id, map_index, key, dag_id, run_id, value, timestamp) VALUES (%(dag_run_id)s, %(task_id)s, %(map_index)s, %(key)s, %(dag_id)s, %(run_id)s, %(value)s, %(timestamp)s)]
[parameters: {'dag_run_id': 93, 'task_id': 'expansion_group.short_circuit', 'map_index': -1, 'key': 'skipmixin_key', 'dag_id': 'test_short_circuit', 'run_id': 'manual__2023-05-27T22:49:24.543812+00:00', 'value': &lt;psycopg2.extensions.Binary object at 0xffff7ae48b10&gt;, 'timestamp': datetime.datetime(2023, 5, 27, 22, 49, 57, 760293, tzinfo=Timezone('UTC'))}]
(Background on this error at: https://sqlalche.me/e/14/gkpj); 104)
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The short circuit to have short circuited (or not) as appropriate</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">This DAG causes the behavior:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="@dag(
    dag_id=&quot;test_short_circuit&quot;,
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz=&quot;UTC&quot;),
    catchup=False,
)
def test_short_circuit() -&gt; None:
    @task
    def add_one(datum: int) -&gt; int:
        return datum + 1

    @task.short_circuit
    def short_circuit(datum: int) -&gt; bool:
        return datum &gt; 3

    @task
    def subtract_one(datum: int) -&gt; int:
        return datum - 1

    @task_group
    def expansion_group(datum: int) -&gt; None:
        plus_one = add_one(datum)
        foo = short_circuit(plus_one)
        bar = subtract_one(plus_one)
        foo &gt;&gt; bar

    stuff = range(1, 10)
    expansion_group.expand(datum=stuff)

test_short_circuit()"><pre class="notranslate"><code class="notranslate">@dag(
    dag_id="test_short_circuit",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
)
def test_short_circuit() -&gt; None:
    @task
    def add_one(datum: int) -&gt; int:
        return datum + 1

    @task.short_circuit
    def short_circuit(datum: int) -&gt; bool:
        return datum &gt; 3

    @task
    def subtract_one(datum: int) -&gt; int:
        return datum - 1

    @task_group
    def expansion_group(datum: int) -&gt; None:
        plus_one = add_one(datum)
        foo = short_circuit(plus_one)
        bar = subtract_one(plus_one)
        foo &gt;&gt; bar

    stuff = range(1, 10)
    expansion_group.expand(datum=stuff)

test_short_circuit()
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">MacOS Ventura</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-common-sql==1.2.0<br>
apache-airflow-providers-ftp==3.1.0<br>
apache-airflow-providers-google==8.4.0<br>
apache-airflow-providers-http==4.0.0<br>
apache-airflow-providers-imap==3.0.0<br>
apache-airflow-providers-postgres==5.2.2<br>
apache-airflow-providers-sendgrid==3.0.0<br>
apache-airflow-providers-slack==6.0.0<br>
apache-airflow-providers-sqlite==3.2.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other 3rd-party Helm chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">It's every time. If I convert to an old fashioned <code class="notranslate">.expand()</code> (i.e. serial <code class="notranslate">.expand()</code> calls), it does not happen</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.6.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I have a short circuit task which is in a task group that is expanded. The task work correctly when it returns true but the task fails when it returns false with the following error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table &quot;xcom&quot; violates foreign key constraint &quot;xcom_task_instance_fkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(pipeline_output_to_s3, transfer_output_file.already_in_manifest_short_circuit, manual__2023-05-24T20:21:35.420606+00:00, -1) is not present in table &quot;task_instance&quot;."><pre class="notranslate"><code class="notranslate">sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table "xcom" violates foreign key constraint "xcom_task_instance_fkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(pipeline_output_to_s3, transfer_output_file.already_in_manifest_short_circuit, manual__2023-05-24T20:21:35.420606+00:00, -1) is not present in table "task_instance".
</code></pre></div>
<p dir="auto">It looks like it sets the map-index to -1 when false is returned which is causing the issue.<br>
If one task fails this way in the the task group, all other mapped tasks fail as well, even if the short circuit returns true.<br>
When this error occurs, all subsequent DAG runs will be stuck indefinitely in the running state unless the DAG is deleted.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Returning false from the short circuit operator should skip downstream tasks without affecting other task groups / map indexes or subsequent DAG runs.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">include a short circuit operator in a mapped task group and have it return false.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Red Hat Enterprise Linux 7</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>