Fix: Support older versions of Airflow (#1587)

Author: izeigerman

sqlmesh/schedulers/airflow/dag_generator.py

@@ -7,7 +7,6 @@
 import pendulum
 from airflow import DAG
 from airflow.models import BaseOperator, baseoperator
-from airflow.operators.empty import EmptyOperator
 from airflow.operators.python import PythonOperator
 from airflow.sensors.base import BaseSensorOperator
 
@@ -33,6 +32,11 @@
 from sqlmesh.utils.date import TimeLike, to_datetime, yesterday_timestamp
 from sqlmesh.utils.errors import SQLMeshError
 
+try:
+    from airflow.operators.empty import EmptyOperator
+except ImportError:
+    from airflow.operators.dummy import DummyOperator as EmptyOperator  # type: ignore
+
 logger = logging.getLogger(__name__)
 
 

sqlmesh/schedulers/airflow/operators/hwm_sensor.py

@@ -1,15 +1,20 @@
+from __future__ import annotations
+
 import logging
 import typing as t
 from datetime import datetime
 
 from airflow.models import DagRun
-from airflow.sensors.base import BaseSensorOperator, PokeReturnValue
+from airflow.sensors.base import BaseSensorOperator
 from airflow.utils.context import Context
 
 from sqlmesh.core.snapshot import Snapshot, SnapshotTableInfo
 from sqlmesh.schedulers.airflow import util
 from sqlmesh.utils.date import now, to_datetime
 
+if t.TYPE_CHECKING:
+    from airflow.sensors.base import PokeReturnValue
+
 logger = logging.getLogger(__name__)
 
 

sqlmesh/utils/migration.py

@@ -2,7 +2,7 @@
 
 # Sizes based on a composite key/index of two text fields with 4 bytes per characters.
 MAX_TEXT_INDEX_LENGTH = {
-    "mysql": "380",  # 380 characters per column, <= 3070 byte index size limit
+    "mysql": "250",  # 250 characters per column, <= 767 byte index size limit
     "tsql": "450",  # 450 bytes per column, <= 900 byte index size limit
 }