You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by jo...@apache.org on 2018/06/21 00:42:49 UTC
[1/2] incubator-airflow git commit: [AIRFLOW-2606] Fix DB schema and
SQLAlchemy model
Repository: incubator-airflow
Updated Branches:
refs/heads/master c9a6fec09 -> 3162e1c9a
[AIRFLOW-2606] Fix DB schema and SQLAlchemy model
* Add test that verifies that database schema and SQLAlchemy model are in sync
* Add exception for users.password that doesn't exist in model and tables created by other tests
* Add migration script to merge the two heads
* Add migration script to fix not-null constrains for MySQL that were lost by 0e2a74e0fc9f_add_time_zone_awareness
* Add migration script to fix FK constraint for existing SQLite DBs
* Enable ForeignKey support for SQLite, otherwise 2e82aab8ef20_rename_user_table won't change FK in chart and known_event tables
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/680651f0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/680651f0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/680651f0
Branch: refs/heads/master
Commit: 680651f0ae2a314f8e9882a6bc38f4fa3795cdbe
Parents: 3dade54
Author: Stefan Seelmann <ma...@stefan-seelmann.de>
Authored: Sun Jun 17 09:52:45 2018 +0200
Committer: Stefan Seelmann <ma...@stefan-seelmann.de>
Committed: Mon Jun 18 23:29:29 2018 +0200
----------------------------------------------------------------------
airflow/jobs.py | 1 +
.../versions/05f30312d566_merge_heads.py | 43 ++++++++++
.../856955da8476_fix_sqlite_foreign_key.py | 90 ++++++++++++++++++++
...23433877c24_fix_mysql_not_null_constraint.py | 54 ++++++++++++
airflow/models.py | 22 +++--
airflow/utils/sqlalchemy.py | 6 ++
tests/utils/test_db.py | 71 +++++++++++++++
7 files changed, 278 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/jobs.py
----------------------------------------------------------------------
diff --git a/airflow/jobs.py b/airflow/jobs.py
index ad114ab..2045ae5 100644
--- a/airflow/jobs.py
+++ b/airflow/jobs.py
@@ -96,6 +96,7 @@ class BaseJob(Base, LoggingMixin):
__table_args__ = (
Index('job_type_heart', job_type, latest_heartbeat),
+ Index('idx_job_state_heartbeat', state, latest_heartbeat),
)
def __init__(
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/migrations/versions/05f30312d566_merge_heads.py
----------------------------------------------------------------------
diff --git a/airflow/migrations/versions/05f30312d566_merge_heads.py b/airflow/migrations/versions/05f30312d566_merge_heads.py
new file mode 100644
index 0000000..78d5652
--- /dev/null
+++ b/airflow/migrations/versions/05f30312d566_merge_heads.py
@@ -0,0 +1,43 @@
+# flake8: noqa
+#
+# 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.
+
+"""merge heads
+
+Revision ID: 05f30312d566
+Revises: 86770d1215c0, 0e2a74e0fc9f
+Create Date: 2018-06-17 10:47:23.339972
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '05f30312d566'
+down_revision = ('86770d1215c0', '0e2a74e0fc9f')
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+ pass
+
+
+def downgrade():
+ pass
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/migrations/versions/856955da8476_fix_sqlite_foreign_key.py
----------------------------------------------------------------------
diff --git a/airflow/migrations/versions/856955da8476_fix_sqlite_foreign_key.py b/airflow/migrations/versions/856955da8476_fix_sqlite_foreign_key.py
new file mode 100644
index 0000000..5b11dc7
--- /dev/null
+++ b/airflow/migrations/versions/856955da8476_fix_sqlite_foreign_key.py
@@ -0,0 +1,90 @@
+# flake8: noqa
+#
+# 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.
+
+"""fix sqlite foreign key
+
+Revision ID: 856955da8476
+Revises: f23433877c24
+Create Date: 2018-06-17 15:54:53.844230
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '856955da8476'
+down_revision = 'f23433877c24'
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+ conn = op.get_bind()
+ if conn.dialect.name == 'sqlite':
+ # Fix broken foreign-key constraint for existing SQLite DBs.
+ #
+ # Re-define tables and use copy_from to avoid reflection
+ # which would fail because referenced user table doesn't exist.
+ #
+ # Use batch_alter_table to support SQLite workaround.
+ chart_table = sa.Table('chart',
+ sa.MetaData(),
+ sa.Column('id', sa.Integer(), nullable=False),
+ sa.Column('label', sa.String(length=200), nullable=True),
+ sa.Column('conn_id', sa.String(length=250), nullable=False),
+ sa.Column('user_id', sa.Integer(), nullable=True),
+ sa.Column('chart_type', sa.String(length=100), nullable=True),
+ sa.Column('sql_layout', sa.String(length=50), nullable=True),
+ sa.Column('sql', sa.Text(), nullable=True),
+ sa.Column('y_log_scale', sa.Boolean(), nullable=True),
+ sa.Column('show_datatable', sa.Boolean(), nullable=True),
+ sa.Column('show_sql', sa.Boolean(), nullable=True),
+ sa.Column('height', sa.Integer(), nullable=True),
+ sa.Column('default_params', sa.String(length=5000), nullable=True),
+ sa.Column('x_is_date', sa.Boolean(), nullable=True),
+ sa.Column('iteration_no', sa.Integer(), nullable=True),
+ sa.Column('last_modified', sa.DateTime(), nullable=True),
+ sa.PrimaryKeyConstraint('id')
+ )
+ with op.batch_alter_table('chart', copy_from=chart_table) as batch_op:
+ batch_op.create_foreign_key('chart_user_id_fkey', 'users',
+ ['user_id'], ['id'])
+
+ known_event_table = sa.Table('known_event',
+ sa.MetaData(),
+ sa.Column('id', sa.Integer(), nullable=False),
+ sa.Column('label', sa.String(length=200), nullable=True),
+ sa.Column('start_date', sa.DateTime(), nullable=True),
+ sa.Column('end_date', sa.DateTime(), nullable=True),
+ sa.Column('user_id', sa.Integer(), nullable=True),
+ sa.Column('known_event_type_id', sa.Integer(), nullable=True),
+ sa.Column('description', sa.Text(), nullable=True),
+ sa.ForeignKeyConstraint(['known_event_type_id'],
+ ['known_event_type.id'], ),
+ sa.PrimaryKeyConstraint('id')
+ )
+ with op.batch_alter_table('chart', copy_from=known_event_table) as batch_op:
+ batch_op.create_foreign_key('known_event_user_id_fkey', 'users',
+ ['user_id'], ['id'])
+
+
+def downgrade():
+ # Downgrade would fail because the broken FK constraint can't be re-created.
+ pass
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/migrations/versions/f23433877c24_fix_mysql_not_null_constraint.py
----------------------------------------------------------------------
diff --git a/airflow/migrations/versions/f23433877c24_fix_mysql_not_null_constraint.py b/airflow/migrations/versions/f23433877c24_fix_mysql_not_null_constraint.py
new file mode 100644
index 0000000..44edeef
--- /dev/null
+++ b/airflow/migrations/versions/f23433877c24_fix_mysql_not_null_constraint.py
@@ -0,0 +1,54 @@
+# flake8: noqa
+#
+# 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.
+
+"""fix mysql not null constraint
+
+Revision ID: f23433877c24
+Revises: 05f30312d566
+Create Date: 2018-06-17 10:16:31.412131
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'f23433877c24'
+down_revision = '05f30312d566'
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+def upgrade():
+ conn = op.get_bind()
+ if conn.dialect.name == 'mysql':
+ conn.execute("SET time_zone = '+00:00'")
+ op.alter_column('task_fail', 'execution_date', existing_type=mysql.TIMESTAMP(fsp=6), nullable=False)
+ op.alter_column('xcom', 'execution_date', existing_type=mysql.TIMESTAMP(fsp=6), nullable=False)
+ op.alter_column('xcom', 'timestamp', existing_type=mysql.TIMESTAMP(fsp=6), nullable=False)
+
+
+def downgrade():
+ conn = op.get_bind()
+ if conn.dialect.name == 'mysql':
+ conn.execute("SET time_zone = '+00:00'")
+ op.alter_column('xcom', 'timestamp', existing_type=mysql.TIMESTAMP(fsp=6), nullable=True)
+ op.alter_column('xcom', 'execution_date', existing_type=mysql.TIMESTAMP(fsp=6), nullable=True)
+ op.alter_column('task_fail', 'execution_date', existing_type=mysql.TIMESTAMP(fsp=6), nullable=True)
+
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/models.py
----------------------------------------------------------------------
diff --git a/airflow/models.py b/airflow/models.py
index 4706c2d..8f83586 100755
--- a/airflow/models.py
+++ b/airflow/models.py
@@ -58,7 +58,7 @@ from urllib.parse import urlparse, quote, parse_qsl
from sqlalchemy import (
Column, Integer, String, DateTime, Text, Boolean, ForeignKey, PickleType,
- Index, Float, LargeBinary)
+ Index, Float, LargeBinary, UniqueConstraint)
from sqlalchemy import func, or_, and_, true as sqltrue
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import reconstructor, relationship, synonym
@@ -861,7 +861,7 @@ class TaskInstance(Base, LoggingMixin):
max_tries = Column(Integer)
hostname = Column(String(1000))
unixname = Column(String(1000))
- job_id = Column(Integer, index=True)
+ job_id = Column(Integer)
pool = Column(String(50))
queue = Column(String(50))
priority_weight = Column(Integer)
@@ -875,6 +875,7 @@ class TaskInstance(Base, LoggingMixin):
Index('ti_state', state),
Index('ti_state_lkp', dag_id, task_id, execution_date, state),
Index('ti_pool', pool, state, priority_weight),
+ Index('ti_job_id', job_id),
)
def __init__(self, task, execution_date, state=None):
@@ -2048,12 +2049,13 @@ class TaskFail(Base):
__tablename__ = "task_fail"
- task_id = Column(String(ID_LEN), primary_key=True)
- dag_id = Column(String(ID_LEN), primary_key=True)
- execution_date = Column(UtcDateTime, primary_key=True)
+ id = Column(Integer, primary_key=True)
+ task_id = Column(String(ID_LEN), nullable=False)
+ dag_id = Column(String(ID_LEN), nullable=False)
+ execution_date = Column(UtcDateTime, nullable=False)
start_date = Column(UtcDateTime)
end_date = Column(UtcDateTime)
- duration = Column(Float)
+ duration = Column(Integer)
def __init__(self, task, execution_date, start_date, end_date):
self.dag_id = task.dag_id
@@ -4650,8 +4652,8 @@ class DagStat(Base):
dag_id = Column(String(ID_LEN), primary_key=True)
state = Column(String(50), primary_key=True)
- count = Column(Integer, default=0)
- dirty = Column(Boolean, default=False)
+ count = Column(Integer, default=0, nullable=False)
+ dirty = Column(Boolean, default=False, nullable=False)
def __init__(self, dag_id, state, count=0, dirty=False):
self.dag_id = dag_id
@@ -4784,7 +4786,9 @@ class DagRun(Base, LoggingMixin):
dag = None
__table_args__ = (
- Index('dr_run_id', dag_id, run_id, unique=True),
+ Index('dag_id_state', dag_id, _state),
+ UniqueConstraint('dag_id', 'execution_date'),
+ UniqueConstraint('dag_id', 'run_id'),
)
def __repr__(self):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/airflow/utils/sqlalchemy.py
----------------------------------------------------------------------
diff --git a/airflow/utils/sqlalchemy.py b/airflow/utils/sqlalchemy.py
index a00fe17..4dab322 100644
--- a/airflow/utils/sqlalchemy.py
+++ b/airflow/utils/sqlalchemy.py
@@ -101,6 +101,12 @@ def setup_event_handlers(
def connect(dbapi_connection, connection_record):
connection_record.info['pid'] = os.getpid()
+ @event.listens_for(engine, "connect")
+ def set_sqlite_pragma(dbapi_connection, connection_record):
+ if 'sqlite3.Connection' in str(type(dbapi_connection)):
+ cursor = dbapi_connection.cursor()
+ cursor.execute("PRAGMA foreign_keys=ON")
+ cursor.close()
@event.listens_for(engine, "checkout")
def checkout(dbapi_connection, connection_record, connection_proxy):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/680651f0/tests/utils/test_db.py
----------------------------------------------------------------------
diff --git a/tests/utils/test_db.py b/tests/utils/test_db.py
new file mode 100644
index 0000000..cb7b737
--- /dev/null
+++ b/tests/utils/test_db.py
@@ -0,0 +1,71 @@
+# -*- coding: utf-8 -*-
+#
+# 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.
+
+import unittest
+
+from airflow.models import Base as airflow_base
+from flask_appbuilder.models.sqla import Base as fab_base
+
+from airflow.settings import engine, RBAC
+from alembic.autogenerate import compare_metadata
+from alembic.migration import MigrationContext
+from sqlalchemy import MetaData
+
+
+class DbTest(unittest.TestCase):
+
+ def test_database_schema_and_sqlalchemy_model_are_in_sync(self):
+ # combine Airflow and Flask-AppBuilder (if rbac enabled) models
+ all_meta_data = MetaData()
+ for (table_name, table) in airflow_base.metadata.tables.items():
+ all_meta_data._add_table(table_name, table.schema, table)
+ if RBAC:
+ for (table_name, table) in fab_base.metadata.tables.items():
+ all_meta_data._add_table(table_name, table.schema, table)
+
+ # create diff between database schema and SQLAlchemy model
+ mc = MigrationContext.configure(engine.connect())
+ diff = compare_metadata(mc, all_meta_data)
+
+ # known diffs to ignore
+ ignores = [
+ # users.password is not part of User model,
+ # otherwise it would show up in (old) UI
+ lambda t: (t[0] == 'remove_column' and
+ t[2] == 'users' and
+ t[3].name == 'password'),
+ # ignore tables created by other tests
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 't'),
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 'test_airflow'),
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 'test_postgres_to_postgres'),
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 'test_mysql_to_mysql'),
+ # ignore tables created by celery
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 'celery_taskmeta'),
+ lambda t: (t[0] == 'remove_table' and
+ t[1].name == 'celery_tasksetmeta'),
+ ]
+ for ignore in ignores:
+ diff = [d for d in diff if not ignore(d)]
+
+ self.assertFalse(diff, 'Database schema and SQLAlchemy model are not in sync')
[2/2] incubator-airflow git commit: Merge pull request #3516 from
seelmann/AIRFLOW-2606-db-schema-model
Posted by jo...@apache.org.
Merge pull request #3516 from seelmann/AIRFLOW-2606-db-schema-model
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/3162e1c9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/3162e1c9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/3162e1c9
Branch: refs/heads/master
Commit: 3162e1c9ac37c0b0998f4b5499b5beef4d26511d
Parents: c9a6fec 680651f
Author: Joy Gao <Jo...@apache.org>
Authored: Wed Jun 20 17:42:44 2018 -0700
Committer: Joy Gao <Jo...@apache.org>
Committed: Wed Jun 20 17:42:44 2018 -0700
----------------------------------------------------------------------
airflow/jobs.py | 1 +
.../versions/05f30312d566_merge_heads.py | 43 ++++++++++
.../856955da8476_fix_sqlite_foreign_key.py | 90 ++++++++++++++++++++
...23433877c24_fix_mysql_not_null_constraint.py | 54 ++++++++++++
airflow/models.py | 22 +++--
airflow/utils/sqlalchemy.py | 6 ++
tests/utils/test_db.py | 71 +++++++++++++++
7 files changed, 278 insertions(+), 9 deletions(-)
----------------------------------------------------------------------