You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ds...@apache.org on 2022/08/28 06:02:23 UTC
[airflow] branch main updated: Rename DatasetDagRef to DagScheduleDatasetReference (#25920)
This is an automated email from the ASF dual-hosted git repository.
dstandish pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 5f333b4004 Rename DatasetDagRef to DagScheduleDatasetReference (#25920)
5f333b4004 is described below
commit 5f333b4004660eaac9b68f72f8dd611f8482b05e
Author: Daniel Standish <15...@users.noreply.github.com>
AuthorDate: Sat Aug 27 23:02:11 2022 -0700
Rename DatasetDagRef to DagScheduleDatasetReference (#25920)
---
airflow/api_connexion/openapi/v1.yaml | 4 ++--
airflow/api_connexion/schemas/dataset_schema.py | 10 ++++-----
airflow/jobs/scheduler_job.py | 9 +++++---
.../versions/0114_2_4_0_add_dataset_model.py | 10 ++++-----
airflow/models/dag.py | 24 +++++++++++-----------
airflow/models/dataset.py | 16 +++++++--------
airflow/www/static/js/types/api-generated.ts | 6 +++---
airflow/www/views.py | 10 ++++-----
tests/test_utils/db.py | 4 ++--
tests/utils/test_db_cleanup.py | 2 +-
10 files changed, 49 insertions(+), 46 deletions(-)
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index c040efbb32..1ba052b7e0 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -3522,7 +3522,7 @@ components:
consuming_dags:
type: array
items:
- $ref: '#/components/schemas/DatasetDagRef'
+ $ref: '#/components/schemas/DagScheduleDatasetReference'
producing_tasks:
type: array
items:
@@ -3553,7 +3553,7 @@ components:
description: The dataset update time
nullable: false
- DatasetDagRef:
+ DagScheduleDatasetReference:
description: |
A datasets reference to a downstream DAG.
diff --git a/airflow/api_connexion/schemas/dataset_schema.py b/airflow/api_connexion/schemas/dataset_schema.py
index cc6feef4a3..65d14f97ab 100644
--- a/airflow/api_connexion/schemas/dataset_schema.py
+++ b/airflow/api_connexion/schemas/dataset_schema.py
@@ -21,7 +21,7 @@ from marshmallow import Schema, fields
from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
from airflow.api_connexion.schemas.common_schema import JsonObjectField
-from airflow.models.dataset import DatasetDagRef, DatasetEvent, DatasetModel, DatasetTaskRef
+from airflow.models.dataset import DagScheduleDatasetReference, DatasetEvent, DatasetModel, DatasetTaskRef
class DatasetTaskRefSchema(SQLAlchemySchema):
@@ -38,13 +38,13 @@ class DatasetTaskRefSchema(SQLAlchemySchema):
updated_at = auto_field()
-class DatasetDagRefSchema(SQLAlchemySchema):
- """DatasetDagRef DB schema"""
+class DagScheduleDatasetReferenceSchema(SQLAlchemySchema):
+ """DagScheduleDatasetReference DB schema"""
class Meta:
"""Meta"""
- model = DatasetDagRef
+ model = DagScheduleDatasetReference
dag_id = auto_field()
created_at = auto_field()
@@ -65,7 +65,7 @@ class DatasetSchema(SQLAlchemySchema):
created_at = auto_field()
updated_at = auto_field()
producing_tasks = fields.List(fields.Nested(DatasetTaskRefSchema))
- consuming_dags = fields.List(fields.Nested(DatasetDagRefSchema))
+ consuming_dags = fields.List(fields.Nested(DagScheduleDatasetReferenceSchema))
class DatasetCollection(NamedTuple):
diff --git a/airflow/jobs/scheduler_job.py b/airflow/jobs/scheduler_job.py
index 6a63aaa474..62907e7d7d 100644
--- a/airflow/jobs/scheduler_job.py
+++ b/airflow/jobs/scheduler_job.py
@@ -47,7 +47,7 @@ from airflow.models import DAG
from airflow.models.dag import DagModel
from airflow.models.dagbag import DagBag
from airflow.models.dagrun import DagRun
-from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue, DatasetEvent
+from airflow.models.dataset import DagScheduleDatasetReference, DatasetDagRunQueue, DatasetEvent
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import SimpleTaskInstance, TaskInstance, TaskInstanceKey
from airflow.stats import Stats
@@ -1118,14 +1118,17 @@ class SchedulerJob(BaseJob):
.first()
)
dataset_event_filters = [
- DatasetDagRef.dag_id == dag.dag_id,
+ DagScheduleDatasetReference.dag_id == dag.dag_id,
DatasetEvent.timestamp <= exec_date,
]
if previous_dag_run:
dataset_event_filters.append(DatasetEvent.timestamp > previous_dag_run.execution_date)
dataset_events = (
session.query(DatasetEvent)
- .join(DatasetDagRef, DatasetEvent.dataset_id == DatasetDagRef.dataset_id)
+ .join(
+ DagScheduleDatasetReference,
+ DatasetEvent.dataset_id == DagScheduleDatasetReference.dataset_id,
+ )
.join(DatasetEvent.source_dag_run)
.filter(*dataset_event_filters)
.all()
diff --git a/airflow/migrations/versions/0114_2_4_0_add_dataset_model.py b/airflow/migrations/versions/0114_2_4_0_add_dataset_model.py
index 8ab6f4f962..3c3e8b0a08 100644
--- a/airflow/migrations/versions/0114_2_4_0_add_dataset_model.py
+++ b/airflow/migrations/versions/0114_2_4_0_add_dataset_model.py
@@ -63,9 +63,9 @@ def _create_dataset_table():
op.create_index('idx_uri_unique', 'dataset', ['uri'], unique=True)
-def _create_dataset_dag_ref_table():
+def _create_dag_schedule_dataset_reference_table():
op.create_table(
- 'dataset_dag_ref',
+ 'dag_schedule_dataset_reference',
sa.Column('dataset_id', Integer, primary_key=True, nullable=False),
sa.Column('dag_id', String(250), primary_key=True, nullable=False),
sa.Column('created_at', TIMESTAMP, default=func.now, nullable=False),
@@ -73,7 +73,7 @@ def _create_dataset_dag_ref_table():
sa.ForeignKeyConstraint(
('dataset_id',),
['dataset.id'],
- name="datasetdagref_dataset_fkey",
+ name="dsdr_dataset_fkey",
ondelete="CASCADE",
),
sqlite_autoincrement=True, # ensures PK values not reused
@@ -161,7 +161,7 @@ def _create_dataset_event_dag_run_table():
def upgrade():
"""Apply Add Dataset model"""
_create_dataset_table()
- _create_dataset_dag_ref_table()
+ _create_dag_schedule_dataset_reference_table()
_create_dataset_task_ref_table()
_create_dataset_dag_run_queue_table()
_create_dataset_event_table()
@@ -170,7 +170,7 @@ def upgrade():
def downgrade():
"""Unapply Add Dataset model"""
- op.drop_table('dataset_dag_ref')
+ op.drop_table('dag_schedule_dataset_reference')
op.drop_table('dataset_task_ref')
op.drop_table('dataset_dag_run_queue')
op.drop_table('dagrun_dataset_event')
diff --git a/airflow/models/dag.py b/airflow/models/dag.py
index c86d954f39..933816954b 100644
--- a/airflow/models/dag.py
+++ b/airflow/models/dag.py
@@ -79,7 +79,7 @@ from airflow.models.dagbag import DagBag
from airflow.models.dagcode import DagCode
from airflow.models.dagpickle import DagPickle
from airflow.models.dagrun import DagRun
-from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+from airflow.models.dataset import DagScheduleDatasetReference, DatasetDagRunQueue as DDRQ
from airflow.models.operator import Operator
from airflow.models.param import DagParam, ParamsDict
from airflow.models.taskinstance import Context, TaskInstance, TaskInstanceKey, clear_task_instances
@@ -206,20 +206,20 @@ def get_dataset_triggered_next_run_info(dag_ids: List[str], *, session: Session)
return {
x.dag_id: f"{x.ready} of {x.total} datasets updated"
for x in session.query(
- DatasetDagRef.dag_id,
+ DagScheduleDatasetReference.dag_id,
func.count().label("total"),
func.sum(case((DDRQ.target_dag_id.is_not(None), 1), else_=0)).label("ready"),
)
.join(
DDRQ,
and_(
- DDRQ.dataset_id == DatasetDagRef.dataset_id,
- DDRQ.target_dag_id == DatasetDagRef.dag_id,
+ DDRQ.dataset_id == DagScheduleDatasetReference.dataset_id,
+ DDRQ.target_dag_id == DagScheduleDatasetReference.dag_id,
),
isouter=True,
)
- .group_by(DatasetDagRef.dag_id)
- .filter(DatasetDagRef.dag_id.in_(dag_ids))
+ .group_by(DagScheduleDatasetReference.dag_id)
+ .filter(DagScheduleDatasetReference.dag_id.in_(dag_ids))
.all()
}
@@ -2659,7 +2659,7 @@ class DAG(LoggingMixin):
DagCode.bulk_sync_to_db(filelocs, session=session)
from airflow.datasets import Dataset
- from airflow.models.dataset import DatasetDagRef, DatasetModel, DatasetTaskRef
+ from airflow.models.dataset import DagScheduleDatasetReference, DatasetModel, DatasetTaskRef
class OutletRef(NamedTuple):
dag_id: str
@@ -2704,7 +2704,7 @@ class DAG(LoggingMixin):
# store dag-schedule-on-dataset references
for dag_ref in dag_references:
session.merge(
- DatasetDagRef(
+ DagScheduleDatasetReference(
dataset_id=stored_datasets[dag_ref.uri].id,
dag_id=dag_ref.dag_id,
)
@@ -3179,18 +3179,18 @@ class DagModel(Base):
you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
transaction is committed it will be unlocked.
"""
- from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+ from airflow.models.dataset import DagScheduleDatasetReference, DatasetDagRunQueue as DDRQ
# these dag ids are triggered by datasets, and they are ready to go.
dataset_triggered_dag_info_list = {
x.dag_id: (x.first_event_time, x.last_event_time)
for x in session.query(
- DatasetDagRef.dag_id,
+ DagScheduleDatasetReference.dag_id,
func.max(DDRQ.created_at).label('last_event_time'),
func.max(DDRQ.created_at).label('first_event_time'),
)
- .join(DatasetDagRef.queue_records, isouter=True)
- .group_by(DatasetDagRef.dag_id)
+ .join(DagScheduleDatasetReference.queue_records, isouter=True)
+ .group_by(DagScheduleDatasetReference.dag_id)
.having(func.count() == func.sum(case((DDRQ.target_dag_id.is_not(None), 1), else_=0)))
.all()
}
diff --git a/airflow/models/dataset.py b/airflow/models/dataset.py
index 1ab48ec9c9..eee1b567b8 100644
--- a/airflow/models/dataset.py
+++ b/airflow/models/dataset.py
@@ -65,7 +65,7 @@ class DatasetModel(Base):
created_at = Column(UtcDateTime, default=timezone.utcnow, nullable=False)
updated_at = Column(UtcDateTime, default=timezone.utcnow, onupdate=timezone.utcnow, nullable=False)
- consuming_dags = relationship("DatasetDagRef", back_populates="dataset")
+ consuming_dags = relationship("DagScheduleDatasetReference", back_populates="dataset")
producing_tasks = relationship("DatasetTaskRef", back_populates="dataset")
__tablename__ = "dataset"
@@ -101,8 +101,8 @@ class DatasetModel(Base):
return f"{self.__class__.__name__}(uri={self.uri!r}, extra={self.extra!r})"
-class DatasetDagRef(Base):
- """References from a DAG to an upstream dataset."""
+class DagScheduleDatasetReference(Base):
+ """References from a DAG to a dataset of which it is a consumer."""
dataset_id = Column(Integer, primary_key=True, nullable=False)
dag_id = Column(String(ID_LEN), primary_key=True, nullable=False)
@@ -113,18 +113,18 @@ class DatasetDagRef(Base):
queue_records = relationship(
"DatasetDagRunQueue",
primaryjoin="""and_(
- DatasetDagRef.dataset_id == foreign(DatasetDagRunQueue.dataset_id),
- DatasetDagRef.dag_id == foreign(DatasetDagRunQueue.target_dag_id),
+ DagScheduleDatasetReference.dataset_id == foreign(DatasetDagRunQueue.dataset_id),
+ DagScheduleDatasetReference.dag_id == foreign(DatasetDagRunQueue.target_dag_id),
)""",
)
- __tablename__ = "dataset_dag_ref"
+ __tablename__ = "dag_schedule_dataset_reference"
__table_args__ = (
- PrimaryKeyConstraint(dataset_id, dag_id, name="datasetdagref_pkey", mssql_clustered=True),
+ PrimaryKeyConstraint(dataset_id, dag_id, name="dsdr_pkey", mssql_clustered=True),
ForeignKeyConstraint(
(dataset_id,),
["dataset.id"],
- name='datasetdagref_dataset_fkey',
+ name='dsdr_dataset_fkey',
ondelete="CASCADE",
),
)
diff --git a/airflow/www/static/js/types/api-generated.ts b/airflow/www/static/js/types/api-generated.ts
index 5296e0e8d6..a1b95bdd73 100644
--- a/airflow/www/static/js/types/api-generated.ts
+++ b/airflow/www/static/js/types/api-generated.ts
@@ -1477,7 +1477,7 @@ export interface components {
created_at?: string;
/** @description The dataset update time */
updated_at?: string;
- consuming_dags?: components["schemas"]["DatasetDagRef"][];
+ consuming_dags?: components["schemas"]["DagScheduleDatasetReference"][];
producing_tasks?: components["schemas"]["DatasetTaskRef"][];
};
/**
@@ -1500,7 +1500,7 @@ export interface components {
*
* *New in version 2.4.0*
*/
- DatasetDagRef: {
+ DagScheduleDatasetReference: {
/** @description The DAG ID that depends on the dataset. */
dag_id?: string | null;
/** @description The dataset reference creation time */
@@ -4113,7 +4113,7 @@ export type Resource = CamelCasedPropertiesDeep<components['schemas']['Resource'
export type ActionResource = CamelCasedPropertiesDeep<components['schemas']['ActionResource']>;
export type Dataset = CamelCasedPropertiesDeep<components['schemas']['Dataset']>;
export type DatasetTaskRef = CamelCasedPropertiesDeep<components['schemas']['DatasetTaskRef']>;
-export type DatasetDagRef = CamelCasedPropertiesDeep<components['schemas']['DatasetDagRef']>;
+export type DagScheduleDatasetReference = CamelCasedPropertiesDeep<components['schemas']['DagScheduleDatasetReference']>;
export type DatasetCollection = CamelCasedPropertiesDeep<components['schemas']['DatasetCollection']>;
export type DatasetEvent = CamelCasedPropertiesDeep<components['schemas']['DatasetEvent']>;
export type DatasetEventCollection = CamelCasedPropertiesDeep<components['schemas']['DatasetEventCollection']>;
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 6f547fb1db..e7cdc3aeb8 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -121,7 +121,7 @@ from airflow.models.abstractoperator import AbstractOperator
from airflow.models.dag import DAG, get_dataset_triggered_next_run_info
from airflow.models.dagcode import DagCode
from airflow.models.dagrun import DagRun, DagRunType
-from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue, DatasetModel
+from airflow.models.dataset import DagScheduleDatasetReference, DatasetDagRunQueue, DatasetModel
from airflow.models.operator import Operator
from airflow.models.serialized_dag import SerializedDagModel
from airflow.models.taskinstance import TaskInstance
@@ -3659,16 +3659,16 @@ class Airflow(AirflowBaseView):
DatasetModel.uri,
DatasetDagRunQueue.created_at,
)
- .join(DatasetDagRef, DatasetModel.id == DatasetDagRef.dataset_id)
+ .join(DagScheduleDatasetReference, DatasetModel.id == DagScheduleDatasetReference.dataset_id)
.join(
DatasetDagRunQueue,
and_(
- DatasetDagRunQueue.dataset_id == DatasetDagRef.dataset_id,
- DatasetDagRunQueue.target_dag_id == DatasetDagRef.dag_id,
+ DatasetDagRunQueue.dataset_id == DagScheduleDatasetReference.dataset_id,
+ DatasetDagRunQueue.target_dag_id == DagScheduleDatasetReference.dag_id,
),
isouter=True,
)
- .filter(DatasetDagRef.dag_id == dag_id)
+ .filter(DagScheduleDatasetReference.dag_id == dag_id)
.order_by(DatasetModel.id)
.all()
]
diff --git a/tests/test_utils/db.py b/tests/test_utils/db.py
index e175e4ece1..f55f783e29 100644
--- a/tests/test_utils/db.py
+++ b/tests/test_utils/db.py
@@ -39,7 +39,7 @@ from airflow.models import (
)
from airflow.models.dagcode import DagCode
from airflow.models.dataset import (
- DatasetDagRef,
+ DagScheduleDatasetReference,
DatasetDagRunQueue,
DatasetEvent,
DatasetModel,
@@ -65,7 +65,7 @@ def clear_db_datasets():
session.query(DatasetEvent).delete()
session.query(DatasetModel).delete()
session.query(DatasetDagRunQueue).delete()
- session.query(DatasetDagRef).delete()
+ session.query(DagScheduleDatasetReference).delete()
session.query(DatasetTaskRef).delete()
diff --git a/tests/utils/test_db_cleanup.py b/tests/utils/test_db_cleanup.py
index 64a4461624..cce3c9de1d 100644
--- a/tests/utils/test_db_cleanup.py
+++ b/tests/utils/test_db_cleanup.py
@@ -264,7 +264,7 @@ class TestDBCleanup:
'dag_warning', # self-maintaining
'connection', # leave alone
'slot_pool', # leave alone
- 'dataset_dag_ref', # leave alone for now
+ 'dag_schedule_dataset_reference', # leave alone for now
'dataset_task_ref', # leave alone for now
'dataset_dag_run_queue', # self-managed
'dataset_event_dag_run', # foreign keys