Fix ORM vs migration files inconsistencies

[ephraimbuddy]

There have been some inconsistences between ORM and migration files but it doesn't fail in tests. This is an attempt to fix the inconsistency and also have it fail in tests

[kaxil]

Old sqlalchemy versions don't support drop table if exists. Should we have the _xcom_archive table in the ORM? cc @kaxil see: https://github.com/apache/airflow/actions/runs/12034647961/job/33552179419?pr=44221#step:7:57121

oh, huh, we can remove the if_exists then. We can use almebic get tables to do it

def get_current_table_names():
    connection = op.get_bind()

    metadata = MetaData()
    metadata.reflect(bind=connection)

    return metadata.tables.keys()

or

airflow/airflow/utils/db.py

Lines 1008 to 1029 in 761cedd

	def reflect_tables(tables: list[MappedClassProtocol | str] | None, session):
	"""
	When running checks prior to upgrades, we use reflection to determine current state of the database.
	This function gets the current state of each table in the set of models
	provided and returns a SqlAlchemy metadata object containing them.
	"""
	import sqlalchemy.schema
	bind = session.bind
	metadata = sqlalchemy.schema.MetaData()
	if tables is None:
	metadata.reflect(bind=bind, resolve_fks=False)
	else:
	for tbl in tables:
	try:
	table_name = tbl if isinstance(tbl, str) else tbl.__tablename__
	metadata.reflect(bind=bind, only=[table_name], extend_existing=True, resolve_fks=False)
	except exc.InvalidRequestError:
	continue
	return metadata

[ephraimbuddy]

Old sqlalchemy versions don't support drop table if exists. Should we have the _xcom_archive table in the ORM? cc @kaxil see: https://github.com/apache/airflow/actions/runs/12034647961/job/33552179419?pr=44221#step:7:57121

oh, huh, we can remove the if_exists then. We can use almebic get tables to do it

def get_current_table_names():
    connection = op.get_bind()

    metadata = MetaData()
    metadata.reflect(bind=connection)

    return metadata.tables.keys()

or

airflow/airflow/utils/db.py

Lines 1008 to 1029 in 761cedd

	def reflect_tables(tables: list[MappedClassProtocol | str] | None, session):
	"""
	When running checks prior to upgrades, we use reflection to determine current state of the database.
	This function gets the current state of each table in the set of models
	provided and returns a SqlAlchemy metadata object containing them.
	"""
	import sqlalchemy.schema
	bind = session.bind
	metadata = sqlalchemy.schema.MetaData()
	if tables is None:
	metadata.reflect(bind=bind, resolve_fks=False)
	else:
	for tbl in tables:
	try:
	table_name = tbl if isinstance(tbl, str) else tbl.__tablename__
	metadata.reflect(bind=bind, only=[table_name], extend_existing=True, resolve_fks=False)
	except exc.InvalidRequestError:
	continue
	return metadata

I think that would not work well with offline sql scripts. I’ll verify

[ephraimbuddy]

Didn't think of using raw SQL. I just used it now. Let's see how it goes

[kaxil]

Just one comment https://github.com/apache/airflow/pull/44221/files#r1854515334