[AIRFLOW-6946] Switch to MySQL 5.7 in 2.0 as base

Dockerfile

@@ -99,7 +99,7 @@ RUN KEY="A4A9406876FCBD3C456770C88C718D3B5072E1F5" \
     && gpgconf --kill all \
     rm -rf "${GNUPGHOME}"; \
     apt-key list > /dev/null \
-    && echo "deb http://repo.mysql.com/apt/debian/ stretch mysql-5.6" | tee -a /etc/apt/sources.list.d/mysql.list \
+    && echo "deb http://repo.mysql.com/apt/debian/ stretch mysql-5.7" | tee -a /etc/apt/sources.list.d/mysql.list \
     && apt-get update \
     && apt-get install --no-install-recommends -y \
         libmysqlclient-dev \

airflow/config_templates/config.yml

@@ -74,6 +74,16 @@
       type: string
       example: ~
       default: "utf-8"
+    - name: sql_engine_collation_for_ids
+      description: |
+        Collation for `dag_id`, `task_id`, `key` columns in case they have different encoding.
+        This is particularly useful in case of mysql with utf8mb4 encoding because
+        primary keys for XCom table has too big size and `sql_engine_collation_for_ids` should
+        be set to `utf8mb3_general_ci`
+      version_added: 2.0.0
+      type: string
+      example: ~
+      default: ~
     - name: sql_alchemy_pool_enabled
       description: |
         If SqlAlchemy should pool database connections.

airflow/config_templates/default_airflow.cfg

@@ -59,6 +59,12 @@ sql_alchemy_conn = sqlite:///{AIRFLOW_HOME}/airflow.db
 # The encoding for the databases
 sql_engine_encoding = utf-8
 
+# Collation for `dag_id`, `task_id`, `key` columns in case they have different encoding.
+# This is particularly useful in case of mysql with utf8mb4 encoding because
+# primary keys for XCom table has too big size and `sql_engine_collation_for_ids` should
+# be set to `utf8mb3_general_ci`
+# sql_engine_collation_for_ids =
+
 # If SqlAlchemy should pool database connections.
 sql_alchemy_pool_enabled = True
 

airflow/migrations/versions/0a2a5b66e19d_add_task_reschedule_table.py

@@ -27,6 +27,8 @@
 from sqlalchemy.dialects import mysql
 
 # revision identifiers, used by Alembic.
+from airflow.models.base import COLLATION_ARGS
+
 revision = '0a2a5b66e19d'
 down_revision = '9635ae0956e7'
 branch_labels = None
@@ -64,8 +66,8 @@ def upgrade():
     op.create_table(
         TABLE_NAME,
         sa.Column('id', sa.Integer(), nullable=False),
-        sa.Column('task_id', sa.String(length=250), nullable=False),
-        sa.Column('dag_id', sa.String(length=250), nullable=False),
+        sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
+        sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
         # use explicit server_default=None otherwise mysql implies defaults for first timestamp column
         sa.Column('execution_date', timestamp(), nullable=False, server_default=None),
         sa.Column('try_number', sa.Integer(), nullable=False),

airflow/migrations/versions/64de9cddf6c9_add_task_fails_journal_table.py

@@ -27,6 +27,8 @@
 from alembic import op
 
 # revision identifiers, used by Alembic.
+from airflow.models.base import COLLATION_ARGS
+
 revision = '64de9cddf6c9'
 down_revision = '211e584da130'
 branch_labels = None
@@ -37,8 +39,8 @@ def upgrade():
     op.create_table(
         'task_fail',
         sa.Column('id', sa.Integer(), nullable=False),
-        sa.Column('task_id', sa.String(length=250), nullable=False),
-        sa.Column('dag_id', sa.String(length=250), nullable=False),
+        sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
+        sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
         sa.Column('execution_date', sa.DateTime(), nullable=False),
         sa.Column('start_date', sa.DateTime(), nullable=True),
         sa.Column('end_date', sa.DateTime(), nullable=True),

airflow/migrations/versions/6e96a59344a4_make_taskinstance_pool_not_nullable.py

@@ -30,6 +30,7 @@
 from sqlalchemy import Column, Float, Integer, PickleType, String
 from sqlalchemy.ext.declarative import declarative_base
 
+from airflow.models.base import COLLATION_ARGS
 from airflow.utils.session import create_session
 from airflow.utils.sqlalchemy import UtcDateTime
 
@@ -59,8 +60,8 @@ class TaskInstance(Base):
 
     __tablename__ = "task_instance"
 
-    task_id = Column(String(ID_LEN), primary_key=True)
-    dag_id = Column(String(ID_LEN), primary_key=True)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
     execution_date = Column(UtcDateTime, primary_key=True)
     start_date = Column(UtcDateTime)
     end_date = Column(UtcDateTime)

airflow/migrations/versions/cc1e65623dc7_add_max_tries_column_to_task_instance.py

@@ -31,8 +31,9 @@
 
 from airflow import settings
 from airflow.models import DagBag
-
 # revision identifiers, used by Alembic.
+from airflow.models.base import COLLATION_ARGS
+
 revision = 'cc1e65623dc7'
 down_revision = '127d2bf2dfa7'
 branch_labels = None
@@ -46,8 +47,8 @@
 class TaskInstance(Base):
     __tablename__ = "task_instance"
 
-    task_id = Column(String(ID_LEN), primary_key=True)
-    dag_id = Column(String(ID_LEN), primary_key=True)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
     execution_date = Column(sa.DateTime, primary_key=True)
     max_tries = Column(Integer)
     try_number = Column(Integer, default=0)

airflow/migrations/versions/e3a246e0dc1_current_schema.py

@@ -30,6 +30,8 @@
 from sqlalchemy.engine.reflection import Inspector
 
 # revision identifiers, used by Alembic.
+from airflow.models.base import COLLATION_ARGS
+
 revision = 'e3a246e0dc1'
 down_revision = None
 branch_labels = None
@@ -115,8 +117,8 @@ def upgrade():
             'log',
             sa.Column('id', sa.Integer(), nullable=False),
             sa.Column('dttm', sa.DateTime(), nullable=True),
-            sa.Column('dag_id', sa.String(length=250), nullable=True),
-            sa.Column('task_id', sa.String(length=250), nullable=True),
+            sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=True),
+            sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=True),
             sa.Column('event', sa.String(length=30), nullable=True),
             sa.Column('execution_date', sa.DateTime(), nullable=True),
             sa.Column('owner', sa.String(length=500), nullable=True),
@@ -125,8 +127,8 @@ def upgrade():
     if 'sla_miss' not in tables:
         op.create_table(
             'sla_miss',
-            sa.Column('task_id', sa.String(length=250), nullable=False),
-            sa.Column('dag_id', sa.String(length=250), nullable=False),
+            sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
+            sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
             sa.Column('execution_date', sa.DateTime(), nullable=False),
             sa.Column('email_sent', sa.Boolean(), nullable=True),
             sa.Column('timestamp', sa.DateTime(), nullable=True),
@@ -146,8 +148,8 @@ def upgrade():
     if 'task_instance' not in tables:
         op.create_table(
             'task_instance',
-            sa.Column('task_id', sa.String(length=250), nullable=False),
-            sa.Column('dag_id', sa.String(length=250), nullable=False),
+            sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
+            sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
             sa.Column('execution_date', sa.DateTime(), nullable=False),
             sa.Column('start_date', sa.DateTime(), nullable=True),
             sa.Column('end_date', sa.DateTime(), nullable=True),
@@ -224,16 +226,16 @@ def upgrade():
         op.create_table(
             'xcom',
             sa.Column('id', sa.Integer(), nullable=False),
-            sa.Column('key', sa.String(length=512), nullable=True),
+            sa.Column('key', sa.String(length=512, **COLLATION_ARGS), nullable=True),
             sa.Column('value', sa.PickleType(), nullable=True),
             sa.Column(
                 'timestamp',
                 sa.DateTime(),
                 default=func.now(),
                 nullable=False),
             sa.Column('execution_date', sa.DateTime(), nullable=False),
-            sa.Column('task_id', sa.String(length=250), nullable=False),
-            sa.Column('dag_id', sa.String(length=250), nullable=False),
+            sa.Column('task_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
+            sa.Column('dag_id', sa.String(length=250, **COLLATION_ARGS), nullable=False),
             sa.PrimaryKeyConstraint('id')
         )
 

airflow/models/base.py

@@ -38,3 +38,14 @@
 # used for typing
 class Operator:
     pass
+
+
+def get_id_collation_args():
+    collation = conf.get('core', 'sql_engine_collation_for_ids', fallback=None)
+    if collation:
+        return {'collation': collation}
+    else:
+        return {}
+
+
+COLLATION_ARGS = get_id_collation_args()

airflow/models/log.py

@@ -18,7 +18,7 @@
 
 from sqlalchemy import Column, Index, Integer, String, Text
 
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils import timezone
 from airflow.utils.sqlalchemy import UtcDateTime
 
@@ -32,8 +32,8 @@ class Log(Base):
 
     id = Column(Integer, primary_key=True)
     dttm = Column(UtcDateTime)
-    dag_id = Column(String(ID_LEN))
-    task_id = Column(String(ID_LEN))
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS))
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS))
     event = Column(String(30))
     execution_date = Column(UtcDateTime)
     owner = Column(String(500))

airflow/models/slamiss.py

@@ -18,7 +18,7 @@
 
 from sqlalchemy import Boolean, Column, Index, String, Text
 
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils.sqlalchemy import UtcDateTime
 
 
@@ -30,8 +30,8 @@ class SlaMiss(Base):
     """
     __tablename__ = "sla_miss"
 
-    task_id = Column(String(ID_LEN), primary_key=True)
-    dag_id = Column(String(ID_LEN), primary_key=True)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
     execution_date = Column(UtcDateTime, primary_key=True)
     email_sent = Column(Boolean, default=False)
     timestamp = Column(UtcDateTime)

airflow/models/taskfail.py

@@ -18,7 +18,7 @@
 """Taskfail tracks the failed run durations of each task instance"""
 from sqlalchemy import Column, Index, Integer, String
 
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils.sqlalchemy import UtcDateTime
 
 
@@ -30,8 +30,8 @@ class TaskFail(Base):
     __tablename__ = "task_fail"
 
     id = Column(Integer, primary_key=True)
-    task_id = Column(String(ID_LEN), nullable=False)
-    dag_id = Column(String(ID_LEN), nullable=False)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
     execution_date = Column(UtcDateTime, nullable=False)
     start_date = Column(UtcDateTime)
     end_date = Column(UtcDateTime)

airflow/models/taskinstance.py

@@ -42,7 +42,7 @@
 from airflow.exceptions import (
     AirflowException, AirflowRescheduleException, AirflowSkipException, AirflowTaskTimeout,
 )
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.models.log import Log
 from airflow.models.taskfail import TaskFail
 from airflow.models.taskreschedule import TaskReschedule
@@ -146,8 +146,8 @@ class TaskInstance(Base, LoggingMixin):
 
     __tablename__ = "task_instance"
 
-    task_id = Column(String(ID_LEN), primary_key=True)
-    dag_id = Column(String(ID_LEN), primary_key=True)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
     execution_date = Column(UtcDateTime, primary_key=True)
     start_date = Column(UtcDateTime)
     end_date = Column(UtcDateTime)

airflow/models/taskreschedule.py

@@ -18,7 +18,7 @@
 """TaskReschedule tracks rescheduled task instances."""
 from sqlalchemy import Column, ForeignKeyConstraint, Index, Integer, String, asc
 
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils.session import provide_session
 from airflow.utils.sqlalchemy import UtcDateTime
 
@@ -31,8 +31,8 @@ class TaskReschedule(Base):
     __tablename__ = "task_reschedule"
 
     id = Column(Integer, primary_key=True)
-    task_id = Column(String(ID_LEN), nullable=False)
-    dag_id = Column(String(ID_LEN), nullable=False)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
     execution_date = Column(UtcDateTime, nullable=False)
     try_number = Column(Integer, nullable=False)
     start_date = Column(UtcDateTime, nullable=False)

airflow/models/xcom.py

@@ -26,7 +26,7 @@
 from sqlalchemy.orm import Query, Session, reconstructor
 
 from airflow.configuration import conf
-from airflow.models.base import ID_LEN, Base
+from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
 from airflow.utils import timezone
 from airflow.utils.helpers import is_container
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -47,15 +47,15 @@ class XCom(Base, LoggingMixin):
     """
     __tablename__ = "xcom"
 
-    key = Column(String(512), primary_key=True)
+    key = Column(String(512, **COLLATION_ARGS), primary_key=True)
     value = Column(LargeBinary)
     timestamp = Column(
         UtcDateTime, default=timezone.utcnow, nullable=False)
     execution_date = Column(UtcDateTime, primary_key=True)
 
     # source information
-    task_id = Column(String(ID_LEN), primary_key=True)
-    dag_id = Column(String(ID_LEN), primary_key=True)
+    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
+    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
 
     """
     TODO: "pickling" has been deprecated and JSON is preferred.

scripts/ci/docker-compose/backend-mysql.yml

@@ -20,12 +20,14 @@ services:
   airflow-testing:
     environment:
       - BACKEND=mysql
-      - AIRFLOW__CORE__SQL_ALCHEMY_CONN=mysql://root@mysql/airflow
-      - AIRFLOW__CELERY__RESULT_BACKEND=db+mysql://root@mysql/airflow
+      - AIRFLOW__CORE__SQL_ALCHEMY_CONN=mysql://root@mysql/airflow?charset=utf8mb4
+      - AIRFLOW__CORE__SQL_ENGINE_COLLATION_FOR_IDS=utf8mb3_general_ci
+      - AIRFLOW__CELERY__RESULT_BACKEND=db+mysql://root@mysql/airflow?charset=utf8mb4
     depends_on:
       - mysql
   mysql:
-    image: mysql:5.6
+    image: mysql:5.7
+    command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
     environment:
       - MYSQL_ALLOW_EMPTY_PASSWORD=true
       - MYSQL_ROOT_HOST=%

scripts/ci/in_container/airflow_ci.cfg

@@ -20,6 +20,7 @@
 dags_folder = ~/airflow/dags
 executor = LocalExecutor
 sql_alchemy_conn = # overridden by the startup scripts
+#sql_engine_collation_for_ids = overridden by the startup scripts
 unit_test_mode = True
 load_examples = True
 donot_pickle = False

scripts/ci/in_container/entrypoint_ci.sh

@@ -46,6 +46,10 @@ echo
 echo "Airflow home: ${AIRFLOW_HOME}"
 echo "Airflow sources: ${AIRFLOW_SOURCES}"
 echo "Airflow core SQL connection: ${AIRFLOW__CORE__SQL_ALCHEMY_CONN:=}"
+if [[ -n "${AIRFLOW__CORE__SQL_ENGINE_COLLATION_FOR_IDS:=}" ]]; then
+    echo "Airflow collation for IDs: ${AIRFLOW__CORE__SQL_ENGINE_COLLATION_FOR_IDS}"
+fi
+
 echo
 
 ARGS=( "$@" )