1fanwang commented on code in PR #70235:
URL: https://github.com/apache/airflow/pull/70235#discussion_r3731600219
##########
airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py:
##########
@@ -39,40 +39,57 @@
airflow_version = "2.9.2"
-def upgrade():
- """Apply Update missing constraints."""
- conn = op.get_bind()
- if conn.dialect.name == "mysql":
- # TODO: Rewrite these queries to use alembic when lowest MYSQL version
supports IF EXISTS
- conn.execute(
- sa.text("""
- set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS
WHERE
- CONSTRAINT_SCHEMA = DATABASE() AND
- TABLE_NAME = 'connection' AND
- CONSTRAINT_NAME = 'unique_conn_id' AND
- CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE connection
- DROP INDEX unique_conn_id','select 1');
+def _mysql_drop_unique_constraint_if_exists(conn, table: str, index_name: str)
-> None:
+ """
+ Drop a MySQL unique constraint only if it is actually present.
- prepare stmt from @var;
- execute stmt;
- deallocate prepare stmt;
- """)
- )
- # Dropping the below and recreating cause there's no IF NOT EXISTS in
mysql
+ MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not support the
+ ``prepare``/``execute``/``deallocate prepare`` sequence in a single
+ ``cursor.execute()`` call, so the existence check and the drop are issued
as two
Review Comment:
```suggestion
MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not enable
``CLIENT.MULTI_STATEMENTS``, so it cannot run the
``prepare``/``execute``/``deallocate prepare`` script in a single
``cursor.execute()`` call, so the existence check and the drop are
issued as two
```
Small correction so this doesn't mislead later: PyMySQL runs a bare
`PREPARE` fine. What it can't do is send more than one statement per
`execute()`, because it never sets `CLIENT.MULTI_STATEMENTS`
(`Connection.__init__` starts at `client_flag=0` and only ORs in `LOCAL_FILES`,
`SSL`, `CAPABILITIES`, `CONNECT_WITH_DB`). Checked on 8.4:
```
FAIL plain multi-statement (no PREPARE): SELECT 1; SELECT 2; -> 1064
OK single PREPARE only: PREPARE s FROM 'SELECT 1'
FAIL SET then SELECT (2 statements): set @x=1; SELECT @x; -> 1064
```
##########
airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py:
##########
@@ -39,40 +39,57 @@
airflow_version = "2.9.2"
-def upgrade():
- """Apply Update missing constraints."""
- conn = op.get_bind()
- if conn.dialect.name == "mysql":
- # TODO: Rewrite these queries to use alembic when lowest MYSQL version
supports IF EXISTS
- conn.execute(
- sa.text("""
- set @var=if((SELECT true FROM information_schema.TABLE_CONSTRAINTS
WHERE
- CONSTRAINT_SCHEMA = DATABASE() AND
- TABLE_NAME = 'connection' AND
- CONSTRAINT_NAME = 'unique_conn_id' AND
- CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE connection
- DROP INDEX unique_conn_id','select 1');
+def _mysql_drop_unique_constraint_if_exists(conn, table: str, index_name: str)
-> None:
+ """
+ Drop a MySQL unique constraint only if it is actually present.
- prepare stmt from @var;
- execute stmt;
- deallocate prepare stmt;
- """)
- )
- # Dropping the below and recreating cause there's no IF NOT EXISTS in
mysql
+ MySQL has no ``DROP INDEX IF EXISTS``, and PyMySQL does not support the
+ ``prepare``/``execute``/``deallocate prepare`` sequence in a single
+ ``cursor.execute()`` call, so the existence check and the drop are issued
as two
+ separate single statements. In offline (``--sql``) mode there is no live
connection
+ to query information_schema against, so the guarded dynamic SQL is emitted
as literal
+ script text instead, to be run later through a real SQL client that
supports
+ multi-statement scripts.
+ """
+ if context.is_offline_mode():
conn.execute(
- sa.text("""
+ sa.text(f"""
set @var=if((SELECT true FROM
information_schema.TABLE_CONSTRAINTS WHERE
CONSTRAINT_SCHEMA = DATABASE() AND
- TABLE_NAME = 'connection' AND
- CONSTRAINT_NAME = 'connection_conn_id_uq' AND
- CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE
connection
- DROP INDEX connection_conn_id_uq','select 1');
+ TABLE_NAME = '{table}' AND
+ CONSTRAINT_NAME = '{index_name}' AND
+ CONSTRAINT_TYPE = 'UNIQUE') = true,'ALTER TABLE {table}
+ DROP INDEX {index_name}','select 1');
prepare stmt from @var;
execute stmt;
deallocate prepare stmt;
""")
)
+ return
+ existing_indexes = {
+ row[0]
+ for row in conn.execute(
+ sa.text(f"""
+ SELECT CONSTRAINT_NAME FROM
information_schema.TABLE_CONSTRAINTS
+ WHERE CONSTRAINT_SCHEMA = DATABASE()
+ AND TABLE_NAME = '{table}'
+ AND CONSTRAINT_TYPE = 'UNIQUE'
+ """)
Review Comment:
```suggestion
sa.text("""
SELECT CONSTRAINT_NAME FROM
information_schema.TABLE_CONSTRAINTS
WHERE CONSTRAINT_SCHEMA = DATABASE()
AND TABLE_NAME = :table
AND CONSTRAINT_TYPE = 'UNIQUE'
"""),
{"table": table},
```
`TABLE_NAME` is a value here, not an identifier, so it can be bound.
Only the `ALTER TABLE` below needs interpolation. Every call site passes a
literal today so there's no live injection,
https://github.com/apache/airflow/issues/34252 is open about exactly this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
