Fix implicit cartessian product in AirflowSecurityManagerV2 (#38913)

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>
(cherry picked from commit e16dc0f)

airflow/www/security_manager.py

@@ -199,8 +199,8 @@ def get_dag_id_from_task_instance(resource_pk):
  if not resource_pk:
  return None
  composite_pk = json.loads(resource_pk)
- ti = session.scalar(
- select(DagRun)
+ dag_id = session.scalar(
+ select(TaskInstance.dag_id)
  .where(
  TaskInstance.dag_id == composite_pk[0],
  TaskInstance.task_id == composite_pk[1],
@@ -209,9 +209,9 @@ def get_dag_id_from_task_instance(resource_pk):
  )
  .limit(1)
  )
- if not ti:
+ if not dag_id:
  raise AirflowException("Task instance not found")
- return ti.dag_id
+ return dag_id
 
  def get_pool_name(resource_pk):
  if not resource_pk:

pyproject.toml

@@ -478,6 +478,8 @@ norecursedirs = [
 log_level = "INFO"
 filterwarnings = [
  "error::pytest.PytestCollectionWarning",
+ # Avoid building cartesian product which might impact performance
+ "error:SELECT statement has a cartesian product between FROM:sqlalchemy.exc.SAWarning:airflow",
  "ignore::DeprecationWarning:flask_appbuilder.filemanager",
  "ignore::DeprecationWarning:flask_appbuilder.widgets",
  # https://github.com/dpgaspar/Flask-AppBuilder/pull/1940