<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version (please specify below)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Maybe similar to <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="849655118" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/15174" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/15174/hovercard" href="https://github.com/apache/airflow/pull/15174">#15174</a></p>
<p dir="auto">Trying to trigger the same dag_id  within the same second.  I get:</p>
<h2 dir="auto">Traceback (most recent call last):</h2>
<p dir="auto">File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1802, in _execute_context<br>
self.dialect.do_execute(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 719, in do_execute<br>
cursor.execute(statement, parameters)<br>
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"<br>
DETAIL:  Key (dag_id, execution_date)=(shawn-shotgun3-shotgun, 2023-02-16 22:47:21+00) already exists.<br>
The above exception was the direct cause of the following exception:<br>
Traceback (most recent call last):<br>
File "/usr/local/airflow/.local/bin/airflow", line 8, in <br>
sys.exit(main())<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/<strong>main</strong>.py", line 39, in main<br>
args.func(args)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/cli/cli_parser.py", line 52, in command<br>
return func(*args, **kwargs)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/utils/cli.py", line 103, in wrapper<br>
return f(*args, **kwargs)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/cli/commands/dag_command.py", line 145, in dag_trigger<br>
message = api_client.trigger_dag(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/api/client/local_client.py", line 32, in trigger_dag<br>
dag_run = trigger_dag.trigger_dag(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/api/common/trigger_dag.py", line 124, in trigger_dag<br>
triggers = _trigger_dag(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/api/common/trigger_dag.py", line 89, in _trigger_dag<br>
dag_run = _dag.create_dagrun(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 75, in wrapper<br>
return func(*args, session=session, **kwargs)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/airflow/models/dag.py", line 2540, in create_dagrun<br>
session.flush()<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3345, in flush<br>
self.<em>flush(objects)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3484, in <em>flush<br>
with util.safe_reraise():<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py", line 70, in <strong>exit</strong><br>
compat.raise</em>(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 207, in raise</em><br>
raise exception<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3445, in _flush<br>
flush_context.execute()<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute<br>
rec.execute(self)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 630, in execute<br>
util.preloaded.orm_persistence.save_obj(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 244, in save_obj<br>
_emit_insert_statements(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 1221, in _emit_insert_statements<br>
result = connection._execute_20(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1614, in _execute_20<br>
return meth(self, args_10style, kwargs_10style, execution_options)<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 325, in _execute_on_connection<br>
return connection._execute_clauseelement(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1481, in _execute_clauseelement<br>
ret = self._execute_context(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1845, in _execute_context<br>
self.<em>handle_dbapi_exception(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2026, in <em>handle_dbapi_exception<br>
util.raise</em>(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 207, in raise</em><br>
raise exception<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1802, in _execute_context<br>
self.dialect.do_execute(<br>
File "/usr/local/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 719, in do_execute<br>
cursor.execute(statement, parameters)<br>
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"<br>
DETAIL:  Key (dag_id, execution_date)=(shawn-shotgun3-shotgun, 2023-02-16 22:47:21+00) already exists.<br>
[SQL: INSERT INTO dag_run (dag_id, queued_at, execution_date, start_date, end_date, state, run_id, creating_job_id, external_trigger, run_type, conf, data_interval_start, data_interval_end, last_scheduling_decision, dag_hash, log_template_id) VALUES (%(dag_id)s, %(queued_at)s, %(execution_date)s, %(start_date)s, %(end_date)s, %(state)s, %(run_id)s, %(creating_job_id)s, %(external_trigger)s, %(run_type)s, %(conf)s, %(data_interval_start)s, %(data_interval_end)s, %(last_scheduling_decision)s, %(dag_hash)s, (SELECT max(log_template.id) AS max_1<br>
FROM log_template)) RETURNING dag_run.id]<br>
[parameters: {'dag_id': 'shawn-shotgun3-shotgun', 'queued_at': datetime.datetime(2023, 2, 16, 22, 47, 22, 43099, tzinfo=Timezone('UTC')), 'execution_date': DateTime(2023, 2, 16, 22, 47, 21, tzinfo=Timezone('UTC')), 'start_date': None, 'end_date': None, 'state': &lt;DagRunState.QUEUED: 'queued'&gt;, 'run_id': 'shotgun3_20230216224703986721', 'creating_job_id': None, 'external_trigger': True, 'run_type': &lt;DagRunType.MANUAL: 'manual'&gt;, 'conf': &lt;psycopg2.extensions.Binary object at 0x7fab34ddb840&gt;, 'data_interval_start': DateTime(2023, 2, 16, 22, 47, 21, tzinfo=Timezone('UTC')), 'data_interval_end': DateTime(2023, 2, 16, 22, 47, 21, tzinfo=Timezone('UTC')), 'last_scheduling_decision': None, 'dag_hash': '7ab0e818c048cc52d84ba94f0df44258'}]<br>
(Background on this error at: <a href="https://sqlalche.me/e/14/gkpj" rel="nofollow">https://sqlalche.me/e/14/gkpj</a>)</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">It should be allowed to run the same dag many times without any issue of how fast it is being run.  The error message above seems to suggest that there is a unique key constraint with columns:</p>
<p dir="auto">dag_id, execution_date</p>
<p dir="auto">which prevent this since the execution_date only has resolution to the second.  I would have thought dag_id and run_id should be the unique key constraint and leave it up to the caller to guarantee unique run_ids.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">AWS MWAA 2.4.3</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">MWAA</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>
<h3 dir="auto">Description</h3>
<p dir="auto">I would like to build a graph like this:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="              ┌───►  task_2_1  ─────►  task_2_1
              │
              │
              │
task 1   ───────►  task_2_2  ─────►  task_3_2
              │
              │
              │
              │
              └──►   task_2_N  ─────►  task_3_N"><pre class="notranslate"><code class="notranslate">              ┌───►  task_2_1  ─────►  task_2_1
              │
              │
              │
task 1   ───────►  task_2_2  ─────►  task_3_2
              │
              │
              │
              │
              └──►   task_2_N  ─────►  task_3_N
</code></pre></div>
<p dir="auto">Where the number of vertical tasks (N) is controlled dynamically by the first step computation. A possible solution would be:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from airflow import DAG
from airflow.decorators import task, task_group
from pendulum import datetime, now


@task
def task_1():
    return list(range(5))


@task
def task_2(task_num):
    return task_num


@task
def task_3(task_num):
    return task_num


with DAG(dag_id=&quot;my_dag&quot;, start_date=now(), schedule_interval=None) as dag:
    task_3.expand(task_num=task_2.expand(task_num=task_1()))"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">task</span>, <span class="pl-s1">task_group</span>
<span class="pl-k">from</span> <span class="pl-s1">pendulum</span> <span class="pl-k">import</span> <span class="pl-s1">datetime</span>, <span class="pl-s1">now</span>


<span class="pl-en">@<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">task_1</span>():
    <span class="pl-k">return</span> <span class="pl-en">list</span>(<span class="pl-en">range</span>(<span class="pl-c1">5</span>))


<span class="pl-en">@<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">task_2</span>(<span class="pl-s1">task_num</span>):
    <span class="pl-k">return</span> <span class="pl-s1">task_num</span>


<span class="pl-en">@<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">task_3</span>(<span class="pl-s1">task_num</span>):
    <span class="pl-k">return</span> <span class="pl-s1">task_num</span>


<span class="pl-k">with</span> <span class="pl-v">DAG</span>(<span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">"my_dag"</span>, <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">now</span>(), <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:
    <span class="pl-s1">task_3</span>.<span class="pl-en">expand</span>(<span class="pl-s1">task_num</span><span class="pl-c1">=</span><span class="pl-s1">task_2</span>.<span class="pl-en">expand</span>(<span class="pl-s1">task_num</span><span class="pl-c1">=</span><span class="pl-en">task_1</span>()))</pre></div>
<p dir="auto">But the task_3 instances start to run only when all the task_2 instance have finished. I would rather group tasks by instances of task_2 and task_3 sharing the same task_num.<br>
An intuitive solution would have been to use a taskgroup, but these do not have the expand method.</p>
<h3 dir="auto">Use case/motivation</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Related issues</h3>
<p dir="auto"><a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1303831971" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/25032" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/25032/hovercard" href="https://github.com/apache/airflow/issues/25032">#25032</a></p>
<h3 dir="auto">Are you willing to submit a 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>