[AIRFLOW-4867] check if table exists before creating

[dszakallas]

Make sure you have checked all steps below.

Jira

• My PR addresses the following AIRFLOW-4867 issues and references them in the PR title.

Description

• Here are some details about my PR, including screenshots of any UI changes:

needed to downgrade to Airflow 1.10.2 (db head rev 41f5f12752f8) from Airflow 1.10.4b2 (db head rev 004c1210f153).

When running

alembic downgrade 41f5f12752f8

I faced the following error

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
[2019-06-28 20:28:02,375] {migration.py:357} INFO - Running downgrade 004c1210f153 -> 939bb1e647c8, increase queue name size limit
[2019-06-28 20:28:03,719] {migration.py:357} INFO - Running downgrade 939bb1e647c8 -> 4ebbffe0a39a, task reschedule fk on cascade delete
[2019-06-28 20:28:04,241] {migration.py:357} INFO - Running downgrade 4ebbffe0a39a -> dd4ecb8fbee3, cf5dc11e79ad, a56c9515abdc, Merge heads
[2019-06-28 20:29:22,124] {migration.py:357} INFO - Running downgrade dd4ecb8fbee3 -> c8ffec048a3b, Add schedule interval to dag
[2019-06-28 20:29:22,468] {migration.py:357} INFO - Running downgrade cf5dc11e79ad -> 41f5f12752f8, drop_user_and_chart
Traceback (most recent call last):
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
    cursor, statement, parameters, context
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 550, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.DuplicateTable: relation "users" already exists


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/bin/alembic", line 11, in <module>
    load_entry_point('alembic==1.0.11', 'console_scripts', 'alembic')()
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/config.py", line 540, in main
    CommandLine(prog=prog).main(argv=argv)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/config.py", line 534, in main
    self.run_cmd(cfg, options)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/config.py", line 514, in run_cmd
    **dict((k, getattr(options, k, None)) for k in kwarg)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/command.py", line 317, in downgrade
    script.run_env()
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/script/base.py", line 475, in run_env
    util.load_python_file(self.dir, "env.py")
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/util/pyfiles.py", line 90, in load_python_file
    module = load_module_py(module_id, path)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/util/compat.py", line 177, in load_module_py
    spec.loader.exec_module(module)
  File "<frozen importlib._bootstrap_external>", line 678, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "migrations/env.py", line 75, in <module>
    run_migrations_online()
  File "migrations/env.py", line 69, in run_migrations_online
    context.run_migrations()
  File "<string>", line 8, in run_migrations
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/runtime/environment.py", line 839, in run_migrations
    self.get_context().run_migrations(**kw)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/runtime/migration.py", line 362, in run_migrations
    step.migration_fn(**kw)
  File "/Users/dszakallas/Projects/airflow/airflow/migrations/versions/cf5dc11e79ad_drop_user_and_chart.py", line 66, in downgrade
    sa.UniqueConstraint('username')
  File "<string>", line 8, in create_table
  File "<string>", line 3, in create_table
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/operations/ops.py", line 1248, in create_table
    return operations.invoke(op)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/operations/base.py", line 345, in invoke
    return fn(self, operation)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/operations/toimpl.py", line 101, in create_table
    operations.impl.create_table(table)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/ddl/impl.py", line 252, in create_table
    self._exec(schema.CreateTable(table))
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/alembic/ddl/impl.py", line 134, in _exec
    return conn.execute(construct, *multiparams, **params)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 988, in execute
    return meth(self, multiparams, params)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/sql/ddl.py", line 72, in _execute_on_connection
    return connection._execute_ddl(self, multiparams, params)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1050, in _execute_ddl
    compiled,
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1248, in _execute_context
    e, statement, parameters, cursor, context
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1466, in _handle_dbapi_exception
    util.raise_from_cause(sqlalchemy_exception, exc_info)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 399, in raise_from_cause
    reraise(type(exception), exception, tb=exc_tb, cause=cause)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 153, in reraise
    raise value.with_traceback(tb)
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1244, in _execute_context
    cursor, statement, parameters, context
  File "/Users/dszakallas/.pyenv/versions/3.6.6/envs/pds-airflow/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 550, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.ProgrammingError: (psycopg2.errors.DuplicateTable) relation "users" already exists

[SQL:
CREATE TABLE users (
    id SERIAL NOT NULL,
 69             sa.UniqueConstraint('username')
    username VARCHAR(250),
    email VARCHAR(500),
    password VARCHAR(255),
    superuser BOOLEAN,
    PRIMARY KEY (id),
    UNIQUE (username)
)

]
(Background on this error at: http://sqlalche.me/e/f405)

I 'fixed' the migration script with only creating the tables if they exist:
After this fix the migrations successfully ran for me.

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

I don't think this can be unit tested.

Commits

• My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain docstrings that explain what it does
  • If you implement backwards incompatible changes, please leave a note in the Updating.md so we can assign it to a appropriate release

Code Quality

• Passes flake8

[ashb]

The "up" migration for this table drops those two tables. Do you know how you ended up with this migration being recorded as applied when it hadn't dropped the tables?

[dszakallas]

I updated the airflow version from 1.10.2 to 1.10.4b2. I didn't do any manual migrations. initdb should have taken care of it right?

[codecov-io]

Codecov Report

Merging #5592 into master will increase coverage by 0.1%.
The diff coverage is n/a.

@@            Coverage Diff            @@
##           master    #5592     +/-   ##
=========================================
+ Coverage   78.95%   79.05%   +0.1%     
=========================================
  Files         489      489             
  Lines       30684    30684             
=========================================
+ Hits        24226    24257     +31     
+ Misses       6458     6427     -31

Impacted Files	Coverage Δ
airflow/utils/dag_processing.py	58.25% <0%> (+1.06%)	⬆️
airflow/jobs/scheduler_job.py	70.38% <0%> (+1.24%)	⬆️
airflow/executors/__init__.py	66.66% <0%> (+4.16%)	⬆️
airflow/utils/sqlalchemy.py	79.06% <0%> (+4.65%)	⬆️
airflow/executors/sequential_executor.py	100% <0%> (+52.38%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b33b989...0d4e883. Read the comment docs.

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.