Skip to content

Commit b757bd8

Browse files
potiukpotiuk
authored
Temporarily revert dag versioning changes (#43730)
* Revert "Delete the Serialized Dag and DagCode before DagVersion migration (#43700)" This reverts commit 438f71d. * Revert "AIP-65: Add DAG versioning support (#42913)" This reverts commit 1116f28.
1 parent 26eaeed commit b757bd8

46 files changed

Lines changed: 2366 additions & 3056 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎airflow/api/common/trigger_dag.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
from airflow.api_internal.internal_api_call import internal_api_call
2626
from airflow.exceptions import DagNotFound, DagRunAlreadyExists
2727
from airflow.models import DagBag, DagModel, DagRun
28-
from airflow.models.dag_version import DagVersion
2928
from airflow.utils import timezone
3029
from airflow.utils.session import NEW_SESSION, provide_session
3130
from airflow.utils.state import DagRunState
@@ -93,14 +92,14 @@ def _trigger_dag(
9392
run_conf = None
9493
if conf:
9594
run_conf = conf if isinstance(conf, dict) else json.loads(conf)
96-
dag_version = DagVersion.get_latest_version(dag.dag_id)
95+
9796
dag_run = dag.create_dagrun(
9897
run_id=run_id,
9998
execution_date=execution_date,
10099
state=DagRunState.QUEUED,
101100
conf=run_conf,
102101
external_trigger=True,
103-
dag_version=dag_version,
102+
dag_hash=dag_bag.dags_hash.get(dag_id),
104103
data_interval=data_interval,
105104
triggered_by=triggered_by,
106105
)

‎airflow/api_connexion/endpoints/dag_run_endpoint.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,6 @@
6161
from airflow.auth.managers.models.resource_details import DagAccessEntity
6262
from airflow.exceptions import ParamValidationError
6363
from airflow.models import DagModel, DagRun
64-
from airflow.models.dag_version import DagVersion
6564
from airflow.timetables.base import DataInterval
6665
from airflow.utils.airflow_flask_app import get_airflow_app
6766
from airflow.utils.api_migration import mark_fastapi_migration_done
@@ -342,7 +341,7 @@ def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
342341
)
343342
else:
344343
data_interval = dag.timetable.infer_manual_data_interval(run_after=logical_date)
345-
dag_version = DagVersion.get_latest_version(dag.dag_id)
344+
346345
dag_run = dag.create_dagrun(
347346
run_type=DagRunType.MANUAL,
348347
run_id=run_id,
@@ -351,7 +350,7 @@ def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
351350
state=DagRunState.QUEUED,
352351
conf=post_body.get("conf"),
353352
external_trigger=True,
354-
dag_version=dag_version,
353+
dag_hash=get_airflow_app().dag_bag.dags_hash.get(dag_id),
355354
session=session,
356355
triggered_by=DagRunTriggeredByType.REST_API,
357356
)

‎airflow/dag_processing/manager.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
from airflow.models.dagwarning import DagWarning
5151
from airflow.models.db_callback_request import DbCallbackRequest
5252
from airflow.models.errors import ParseImportError
53+
from airflow.models.serialized_dag import SerializedDagModel
5354
from airflow.secrets.cache import SecretCache
5455
from airflow.stats import Stats
5556
from airflow.traces.tracer import Trace, add_span
@@ -538,6 +539,10 @@ def deactivate_stale_dags(
538539
if deactivated:
539540
cls.logger().info("Deactivated %i DAGs which are no longer present in file.", deactivated)
540541

542+
for dag_id in to_deactivate:
543+
SerializedDagModel.remove_dag(dag_id)
544+
cls.logger().info("Deleted DAG %s in serialized_dag table", dag_id)
545+
541546
def _run_parsing_loop(self):
542547
# In sync mode we want timeout=None -- wait forever until a message is received
543548
if self._async_mode:
@@ -814,10 +819,20 @@ def _iter_dag_filelocs(fileloc: str) -> Iterator[str]:
814819

815820
dag_filelocs = {full_loc for path in self._file_paths for full_loc in _iter_dag_filelocs(path)}
816821

822+
from airflow.models.dagcode import DagCode
823+
824+
SerializedDagModel.remove_deleted_dags(
825+
alive_dag_filelocs=dag_filelocs,
826+
processor_subdir=self.get_dag_directory(),
827+
)
817828
DagModel.deactivate_deleted_dags(
818829
dag_filelocs,
819830
processor_subdir=self.get_dag_directory(),
820831
)
832+
DagCode.remove_deleted_code(
833+
dag_filelocs,
834+
processor_subdir=self.get_dag_directory(),
835+
)
821836

822837
return True
823838
return False

‎airflow/example_dags/plugins/event_listener.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -164,10 +164,9 @@ def on_dag_run_running(dag_run: DagRun, msg: str):
164164
"""
165165
print("Dag run in running state")
166166
queued_at = dag_run.queued_at
167+
dag_hash_info = dag_run.dag_hash
167168

168-
version = dag_run.dag_version.version
169-
170-
print(f"Dag information Queued at: {queued_at} version: {version}")
169+
print(f"Dag information Queued at: {queued_at} hash info: {dag_hash_info}")
171170

172171

173172
# [END howto_listen_dagrun_running_task]

‎airflow/jobs/scheduler_job_runner.py‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -54,10 +54,10 @@
5454
)
5555
from airflow.models.backfill import Backfill
5656
from airflow.models.dag import DAG, DagModel
57-
from airflow.models.dag_version import DagVersion
5857
from airflow.models.dagbag import DagBag
5958
from airflow.models.dagrun import DagRun
6059
from airflow.models.dagwarning import DagWarning, DagWarningType
60+
from airflow.models.serialized_dag import SerializedDagModel
6161
from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance
6262
from airflow.stats import Stats
6363
from airflow.ti_deps.dependencies_states import EXECUTION_STATES
@@ -1338,7 +1338,7 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -
13381338
self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
13391339
continue
13401340

1341-
latest_dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
1341+
dag_hash = self.dagbag.dags_hash.get(dag.dag_id)
13421342

13431343
data_interval = dag.get_next_data_interval(dag_model)
13441344
# Explicitly check if the DagRun already exists. This is an edge case
@@ -1358,7 +1358,7 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -
13581358
data_interval=data_interval,
13591359
external_trigger=False,
13601360
session=session,
1361-
dag_version=latest_dag_version,
1361+
dag_hash=dag_hash,
13621362
creating_job_id=self.job.id,
13631363
triggered_by=DagRunTriggeredByType.TIMETABLE,
13641364
)
@@ -1417,7 +1417,7 @@ def _create_dag_runs_asset_triggered(
14171417
)
14181418
continue
14191419

1420-
latest_dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
1420+
dag_hash = self.dagbag.dags_hash.get(dag.dag_id)
14211421

14221422
# Explicitly check if the DagRun already exists. This is an edge case
14231423
# where a Dag Run is created but `DagModel.next_dagrun` and `DagModel.next_dagrun_create_after`
@@ -1472,7 +1472,7 @@ def _create_dag_runs_asset_triggered(
14721472
state=DagRunState.QUEUED,
14731473
external_trigger=False,
14741474
session=session,
1475-
dag_version=latest_dag_version,
1475+
dag_hash=dag_hash,
14761476
creating_job_id=self.job.id,
14771477
triggered_by=DagRunTriggeredByType.ASSET,
14781478
)
@@ -1750,20 +1750,18 @@ def _verify_integrity_if_dag_changed(self, dag_run: DagRun, session: Session) ->
17501750
17511751
Return True if we determine that DAG still exists.
17521752
"""
1753-
latest_dag_version = DagVersion.get_latest_version(dag_run.dag_id, session=session)
1754-
if TYPE_CHECKING:
1755-
assert latest_dag_version
1756-
if dag_run.dag_version_id == latest_dag_version.id:
1753+
latest_version = SerializedDagModel.get_latest_version_hash(dag_run.dag_id, session=session)
1754+
if dag_run.dag_hash == latest_version:
17571755
self.log.debug("DAG %s not changed structure, skipping dagrun.verify_integrity", dag_run.dag_id)
17581756
return True
17591757

1758+
dag_run.dag_hash = latest_version
1759+
17601760
# Refresh the DAG
17611761
dag_run.dag = self.dagbag.get_dag(dag_id=dag_run.dag_id, session=session)
17621762
if not dag_run.dag:
17631763
return False
17641764

1765-
dag_run.dag_version = latest_dag_version
1766-
17671765
# Verify integrity also takes care of session.flush
17681766
dag_run.verify_integrity(session=session)
17691767
return True
@@ -2043,6 +2041,7 @@ def _cleanup_stale_dags(self, session: Session = NEW_SESSION) -> None:
20432041
20442042
In case one of DagProcessors is stopped (in case there are multiple of them
20452043
for different dag folders), its dags are never marked as inactive.
2044+
Also remove dags from SerializedDag table.
20462045
Executed on schedule only if [scheduler]standalone_dag_processor is True.
20472046
"""
20482047
self.log.debug("Checking dags not parsed within last %s seconds.", self._dag_stale_not_seen_duration)
@@ -2057,6 +2056,7 @@ def _cleanup_stale_dags(self, session: Session = NEW_SESSION) -> None:
20572056
self.log.info("Found (%d) stales dags not parsed after %s.", len(stale_dags), limit_lpt)
20582057
for dag in stale_dags:
20592058
dag.is_active = False
2059+
SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
20602060
session.flush()
20612061

20622062
@provide_session

‎airflow/migrations/versions/0047_3_0_0_add_dag_versioning.py‎

Lines changed: 0 additions & 158 deletions
This file was deleted.

‎airflow/models/__init__.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,6 @@ def import_all_models():
5757

5858
import airflow.models.asset
5959
import airflow.models.backfill
60-
import airflow.models.dag_version
6160
import airflow.models.dagwarning
6261
import airflow.models.errors
6362
import airflow.models.serialized_dag

‎airflow/models/backfill.py‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,6 @@
4343
from airflow.api_connexion.exceptions import NotFound
4444
from airflow.exceptions import AirflowException
4545
from airflow.models.base import Base, StringID
46-
from airflow.models.dag_version import DagVersion
4746
from airflow.settings import json
4847
from airflow.utils import timezone
4948
from airflow.utils.session import create_session
@@ -201,7 +200,7 @@ def _create_backfill_dag_run(
201200
)
202201
)
203202
return
204-
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
203+
205204
dr = dag.create_dagrun(
206205
triggered_by=DagRunTriggeredByType.BACKFILL,
207206
execution_date=info.logical_date,
@@ -214,7 +213,6 @@ def _create_backfill_dag_run(
214213
creating_job_id=None,
215214
session=session,
216215
backfill_id=backfill_id,
217-
dag_version=dag_version,
218216
)
219217
session.add(
220218
BackfillDagRun(
@@ -255,7 +253,7 @@ def _create_backfill(
255253
from airflow.models.serialized_dag import SerializedDagModel
256254

257255
with create_session() as session:
258-
serdag = session.scalar(SerializedDagModel.latest_item_select_object(dag_id))
256+
serdag = session.get(SerializedDagModel, dag_id)
259257
if not serdag:
260258
raise NotFound(f"Could not find dag {dag_id}")
261259
# todo: if dag has no schedule, raise

0 commit comments

Comments
 (0)