diff --git a/airflow-core/newsfragments/71249.improvement.rst b/airflow-core/newsfragments/71249.improvement.rst new file mode 100644 index 0000000000000..43085779f4f35 --- /dev/null +++ b/airflow-core/newsfragments/71249.improvement.rst @@ -0,0 +1 @@ +The collation used for the asset ``name``, ``uri`` and ``group`` columns on MySQL is now configurable through ``[database] sql_engine_collation_for_asset_names``. It still defaults to ``latin1_general_cs``, so existing databases are unaffected. Set it if your MySQL-compatible engine does not provide that collation. diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index c065b277716c4..10022889dd1fc 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/airflow-core/src/airflow/config_templates/config.yml @@ -638,6 +638,17 @@ database: type: string example: ~ default: ~ + sql_engine_collation_for_asset_names: + description: | + Collation for the ``name``, ``uri`` and ``group`` columns of the asset tables on + ``mysql`` and ``mariadb``. These columns hold ASCII values and are indexed at 1500 + characters, so a single-byte charset is used to stay within the maximum index size. + Override this if your database engine does not provide ``latin1_general_cs`` -- + for example TiDB, which supports ``latin1_bin`` instead. + version_added: 3.4.0 + type: string + example: "latin1_bin" + default: "latin1_general_cs" sql_alchemy_pool_enabled: description: | If SQLAlchemy should pool database connections. diff --git a/airflow-core/src/airflow/models/asset.py b/airflow-core/src/airflow/models/asset.py index 750aac8d2b73b..ccfbc72c268c4 100644 --- a/airflow-core/src/airflow/models/asset.py +++ b/airflow-core/src/airflow/models/asset.py @@ -30,7 +30,6 @@ Index, Integer, PrimaryKeyConstraint, - String, Table, delete, select, @@ -39,7 +38,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship from airflow._shared.timezones import timezone -from airflow.models.base import Base, StringID +from airflow.models.base import ASSET_STR_FIELD, Base, StringID from airflow.utils.sqlalchemy import UtcDateTime if TYPE_CHECKING: @@ -147,15 +146,7 @@ class AssetWatcherModel(Base): """A table to store asset watchers.""" name: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) asset_id: Mapped[int] = mapped_column(Integer, primary_key=True, nullable=False) @@ -198,27 +189,11 @@ class AssetAliasModel(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) group: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, default="", nullable=False, ) @@ -279,39 +254,15 @@ class AssetModel(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) uri: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) group: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, default=str, nullable=False, ) @@ -408,27 +359,11 @@ class AssetActive(Base): """ name: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) uri: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, nullable=False, ) @@ -456,15 +391,7 @@ class DagScheduleAssetNameReference(Base): """Reference from a DAG to an asset name reference of which it is a consumer.""" name: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, primary_key=True, nullable=False, ) @@ -502,15 +429,7 @@ class DagScheduleAssetUriReference(Base): """Reference from a DAG to an asset URI reference of which it is a consumer.""" uri: Mapped[str] = mapped_column( - String(length=1500).with_variant( - String( - length=1500, - # latin1 allows for more indexed length in mysql - # and this field should only be ascii chars - collation="latin1_general_cs", - ), - "mysql", - ), + ASSET_STR_FIELD, primary_key=True, nullable=False, ) diff --git a/airflow-core/src/airflow/models/base.py b/airflow-core/src/airflow/models/base.py index 1c7af7b275ab3..2840bf7e8f8a8 100644 --- a/airflow-core/src/airflow/models/base.py +++ b/airflow-core/src/airflow/models/base.py @@ -83,6 +83,23 @@ def get_id_collation_args(): COLLATION_ARGS: dict[str, Any] = get_id_collation_args() +def get_asset_str_field(length: int = 1500) -> String: + """ + Build the string type used for asset name/uri/group columns. + + On MySQL these carry an explicit latin1 collation: the values are ASCII, and + a 1-byte-per-character charset keeps the 1500-char unique indexes inside the + 3072-byte index limit that utf8mb4 would blow past. The collation is + overridable because MySQL-compatible engines do not all ship + ``latin1_general_cs`` (TiDB, for one, accepts only ``latin1_bin``). + """ + collation = conf.get("database", "sql_engine_collation_for_asset_names", fallback="latin1_general_cs") + return String(length=length).with_variant(String(length=length, collation=collation), "mysql") + + +ASSET_STR_FIELD: String = get_asset_str_field() + + def StringID(*, length=ID_LEN, **kwargs) -> String: return String(length=length, **kwargs, **COLLATION_ARGS)