You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by pi...@apache.org on 2023/06/10 10:05:00 UTC
[airflow] branch v2-6-test updated (d7ba536645 -> 91cbb84bc9)
This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a change to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
from d7ba536645 Update release notes
new e19ff6f620 Cascade update of taskinstance to TaskMap table (#31445)
new 91cbb84bc9 Update release note and revision head
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
RELEASE_NOTES.rst | 1 +
... 0125_2_6_2_add_onupdate_cascade_to_taskmap.py} | 40 +++++++++++++---------
airflow/models/taskmap.py | 1 +
airflow/utils/db.py | 2 +-
docs/apache-airflow/img/airflow_erd.sha256 | 2 +-
docs/apache-airflow/img/airflow_erd.svg | 8 ++---
docs/apache-airflow/migrations-ref.rst | 4 ++-
tests/models/test_taskinstance.py | 35 ++++++++++++++++++-
8 files changed, 68 insertions(+), 25 deletions(-)
copy airflow/migrations/versions/{0042_1_10_3_task_reschedule_fk_on_cascade_delete.py => 0125_2_6_2_add_onupdate_cascade_to_taskmap.py} (53%)
[airflow] 01/02: Cascade update of taskinstance to TaskMap table (#31445)
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit e19ff6f620a2d0d682a2b243593119ca5eb501fb
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed May 24 11:54:45 2023 +0100
Cascade update of taskinstance to TaskMap table (#31445)
(cherry picked from commit f6bb4746efbc6a94fa17b6c77b31d9fb17305ffc)
---
.../0125_2_6_2_add_onupdate_cascade_to_taskmap.py | 62 ++++++++++++++++++++++
airflow/models/taskmap.py | 1 +
docs/apache-airflow/img/airflow_erd.sha256 | 2 +-
docs/apache-airflow/img/airflow_erd.svg | 8 +--
docs/apache-airflow/migrations-ref.rst | 4 +-
tests/models/test_taskinstance.py | 35 +++++++++++-
6 files changed, 105 insertions(+), 7 deletions(-)
diff --git a/airflow/migrations/versions/0125_2_6_2_add_onupdate_cascade_to_taskmap.py b/airflow/migrations/versions/0125_2_6_2_add_onupdate_cascade_to_taskmap.py
new file mode 100644
index 0000000000..cdc21ddaa1
--- /dev/null
+++ b/airflow/migrations/versions/0125_2_6_2_add_onupdate_cascade_to_taskmap.py
@@ -0,0 +1,62 @@
+#
+# 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.
+
+"""Add ``onupdate`` cascade to ``task_map`` table
+
+Revision ID: c804e5c76e3e
+Revises: 98ae134e6fff
+Create Date: 2023-05-19 23:30:57.368617
+
+"""
+from __future__ import annotations
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "c804e5c76e3e"
+down_revision = "98ae134e6fff"
+branch_labels = None
+depends_on = None
+airflow_version = "2.6.2"
+
+
+def upgrade():
+ """Apply Add onupdate cascade to taskmap"""
+ with op.batch_alter_table("task_map") as batch_op:
+ batch_op.drop_constraint("task_map_task_instance_fkey", type_="foreignkey")
+ batch_op.create_foreign_key(
+ "task_map_task_instance_fkey",
+ "task_instance",
+ ["dag_id", "task_id", "run_id", "map_index"],
+ ["dag_id", "task_id", "run_id", "map_index"],
+ ondelete="CASCADE",
+ onupdate="CASCADE",
+ )
+
+
+def downgrade():
+ """Unapply Add onupdate cascade to taskmap"""
+ with op.batch_alter_table("task_map") as batch_op:
+ batch_op.drop_constraint("task_map_task_instance_fkey", type_="foreignkey")
+ batch_op.create_foreign_key(
+ "task_map_task_instance_fkey",
+ "task_instance",
+ ["dag_id", "task_id", "run_id", "map_index"],
+ ["dag_id", "task_id", "run_id", "map_index"],
+ ondelete="CASCADE",
+ )
diff --git a/airflow/models/taskmap.py b/airflow/models/taskmap.py
index e7abcc1b6e..9704cfb5cc 100644
--- a/airflow/models/taskmap.py
+++ b/airflow/models/taskmap.py
@@ -72,6 +72,7 @@ class TaskMap(Base):
],
name="task_map_task_instance_fkey",
ondelete="CASCADE",
+ onupdate="CASCADE",
),
)
diff --git a/docs/apache-airflow/img/airflow_erd.sha256 b/docs/apache-airflow/img/airflow_erd.sha256
index 8a74bd97ab..f6a4f6967c 100644
--- a/docs/apache-airflow/img/airflow_erd.sha256
+++ b/docs/apache-airflow/img/airflow_erd.sha256
@@ -1 +1 @@
-c9c60f41925503ca504eca8b6688ca7ec2d8d7cbbd711285ae5c711cc2599f0e
\ No newline at end of file
+b8e715d65aba5965f3ae0134ee7c38dcff539ebf27442db9771145ed7fb18186
\ No newline at end of file
diff --git a/docs/apache-airflow/img/airflow_erd.svg b/docs/apache-airflow/img/airflow_erd.svg
index 017bff5c8f..b531759ac4 100644
--- a/docs/apache-airflow/img/airflow_erd.svg
+++ b/docs/apache-airflow/img/airflow_erd.svg
@@ -1225,28 +1225,28 @@
<g id="edge48" class="edge">
<title>task_instance--xcom</title>
<path fill="none" stroke="#7f7f7f" stroke-dasharray="5,2" d="M1137.01,-488.65C1161.12,-465.83 1186.07,-443.06 1210,-422 1216.27,-416.48 1222.72,-410.89 1229.27,-405.29"/>
-<text text-anchor="start" x="1219.27" y="-394.09" font-family="Times,serif" font-size="14.00">1</text>
+<text text-anchor="start" x="1198.27" y="-394.09" font-family="Times,serif" font-size="14.00">0..N</text>
<text text-anchor="start" x="1137.01" y="-477.45" font-family="Times,serif" font-size="14.00">1</text>
</g>
<!-- task_instance--xcom -->
<g id="edge49" class="edge">
<title>task_instance--xcom</title>
<path fill="none" stroke="#7f7f7f" stroke-dasharray="5,2" d="M1137.01,-506.4C1161.12,-483.83 1186.07,-461.06 1210,-440 1216.27,-434.48 1222.72,-428.89 1229.27,-423.28"/>
-<text text-anchor="start" x="1219.27" y="-412.08" font-family="Times,serif" font-size="14.00">1</text>
+<text text-anchor="start" x="1198.27" y="-412.08" font-family="Times,serif" font-size="14.00">0..N</text>
<text text-anchor="start" x="1137.01" y="-495.2" font-family="Times,serif" font-size="14.00">1</text>
</g>
<!-- task_instance--xcom -->
<g id="edge50" class="edge">
<title>task_instance--xcom</title>
<path fill="none" stroke="#7f7f7f" stroke-dasharray="5,2" d="M1137.01,-524.15C1161.12,-501.83 1186.07,-479.06 1210,-458 1216.66,-452.14 1223.53,-446.19 1230.5,-440.21"/>
-<text text-anchor="start" x="1199.5" y="-444.01" font-family="Times,serif" font-size="14.00">0..N</text>
+<text text-anchor="start" x="1220.5" y="-444.01" font-family="Times,serif" font-size="14.00">1</text>
<text text-anchor="start" x="1137.01" y="-512.95" font-family="Times,serif" font-size="14.00">1</text>
</g>
<!-- task_instance--xcom -->
<g id="edge51" class="edge">
<title>task_instance--xcom</title>
<path fill="none" stroke="#7f7f7f" stroke-dasharray="5,2" d="M1137.01,-541.9C1161.12,-519.83 1186.07,-497.06 1210,-476 1223.32,-464.28 1237.48,-452.19 1251.63,-440.12"/>
-<text text-anchor="start" x="1251.63" y="-443.92" font-family="Times,serif" font-size="14.00">0..N</text>
+<text text-anchor="start" x="1241.63" y="-443.92" font-family="Times,serif" font-size="14.00">1</text>
<text text-anchor="start" x="1137.01" y="-530.7" font-family="Times,serif" font-size="14.00">1</text>
</g>
<!-- log_template -->
diff --git a/docs/apache-airflow/migrations-ref.rst b/docs/apache-airflow/migrations-ref.rst
index 08365cd250..f33243391b 100644
--- a/docs/apache-airflow/migrations-ref.rst
+++ b/docs/apache-airflow/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
+---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description |
+=================================+===================+===================+==============================================================+
-| ``98ae134e6fff`` (head) | ``6abdffdd4815`` | ``2.6.0`` | Increase length of user identifier columns in ``ab_user`` |
+| ``c804e5c76e3e`` (head) | ``98ae134e6fff`` | ``2.6.2`` | Add ``onupdate`` cascade to ``task_map`` table |
++---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
+| ``98ae134e6fff`` | ``6abdffdd4815`` | ``2.6.0`` | Increase length of user identifier columns in ``ab_user`` |
| | | | and ``ab_register_user`` tables |
+---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
| ``6abdffdd4815`` | ``290244fb8b83`` | ``2.6.0`` | add dttm index on log table |
diff --git a/tests/models/test_taskinstance.py b/tests/models/test_taskinstance.py
index e0ed6de42b..cc8346999b 100644
--- a/tests/models/test_taskinstance.py
+++ b/tests/models/test_taskinstance.py
@@ -69,7 +69,7 @@ from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
from airflow.sensors.base import BaseSensorOperator
from airflow.sensors.python import PythonSensor
-from airflow.serialization.serialized_objects import SerializedBaseOperator
+from airflow.serialization.serialized_objects import SerializedBaseOperator, SerializedDAG
from airflow.settings import TIMEZONE
from airflow.stats import Stats
from airflow.ti_deps.dep_context import DepContext
@@ -3484,6 +3484,39 @@ class TestTaskInstanceRecordTaskMapXComPush:
assert task_map.length == expected_length
assert task_map.keys == expected_keys
+ def test_no_error_on_changing_from_non_mapped_to_mapped(self, dag_maker, session):
+ """If a task changes from non-mapped to mapped, don't fail on integrity error."""
+ with dag_maker(dag_id="test_no_error_on_changing_from_non_mapped_to_mapped") as dag:
+
+ @dag.task()
+ def add_one(x):
+ return [x + 1]
+
+ @dag.task()
+ def add_two(x):
+ return x + 2
+
+ task1 = add_one(2)
+ add_two.expand(x=task1)
+
+ dr = dag_maker.create_dagrun()
+ ti = dr.get_task_instance(task_id="add_one")
+ ti.run()
+ assert ti.state == TaskInstanceState.SUCCESS
+ dag._remove_task("add_one")
+ with dag:
+ task1 = add_one.expand(x=[1, 2, 3]).operator
+ serialized_dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))
+
+ dr.dag = serialized_dag
+ dr.verify_integrity(session=session)
+ ti = dr.get_task_instance(task_id="add_one")
+ assert ti.state == TaskInstanceState.REMOVED
+ dag.clear()
+ ti.refresh_from_task(task1)
+ # This should not raise an integrity error
+ dr.task_instance_scheduling_decisions()
+
class TestMappedTaskInstanceReceiveValue:
@pytest.mark.parametrize(
[airflow] 02/02: Update release note and revision head
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pierrejeambrun pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 91cbb84bc924543b6b3754556410509ff902508b
Author: pierrejeambrun <pi...@gmail.com>
AuthorDate: Sat Jun 10 12:04:19 2023 +0200
Update release note and revision head
---
RELEASE_NOTES.rst | 1 +
airflow/utils/db.py | 2 +-
2 files changed, 2 insertions(+), 1 deletion(-)
diff --git a/RELEASE_NOTES.rst b/RELEASE_NOTES.rst
index 49d20ade53..003e404228 100644
--- a/RELEASE_NOTES.rst
+++ b/RELEASE_NOTES.rst
@@ -31,6 +31,7 @@ No significant changes.
Bug Fixes
^^^^^^^^^
+- Cascade update of TaskInstance to TaskMap table (#31445)
- Fix Kubernetes executors detection of deleted pods (#31274)
- Use keyword parameters for migration methods for mssql (#31309)
- Control permissibility of driver config in extra from airflow.cfg (#31754)
diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index aec0a6fe6d..ad505f33ad 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -81,7 +81,7 @@ REVISION_HEADS_MAP = {
"2.5.3": "290244fb8b83",
"2.6.0": "98ae134e6fff",
"2.6.1": "98ae134e6fff",
- "2.6.2": "98ae134e6fff",
+ "2.6.2": "c804e5c76e3e",
}