code: def _update_value_from_dag_run(
      dialect_name: str,
      target_table: sa.Table,
      target_column: ColumnElement,
      join_columns: list[str],
  ) -> Update:
      """
      Grabs a value from the source table ``dag_run`` and updates target with this value.
      :param dialect_name: dialect in use
      :param target_table: the table to update
      :param target_column: the column to update
      """
      # for run_id:  dag_id, execution_date
      # otherwise:  dag_id, run_id
      condition_list = [
          getattr(dag_run.c, x) == getattr(target_table.c, x)
          for x in join_columns
      ]
      condition = and_(*condition_list)
  
      if dialect_name == "sqlite":
          # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
          # need to do a Correlated subquery update
          sub_q = select(dag_run.c[target_column.name]).where(condition)
  
          return target_table.update().values({target_column: sub_q})
      else:
          return (
              target_table.update()
              .where(condition)
              .values({target_column: dag_run.c[target_column.name]})
          )
  
file_path: airflow/airflow/migrations/versions/0105_2_3_0_add_map_index_to_taskfail.py
imports: from __future__ import annotations
  import sqlalchemy
  from alembic import op
  from sqlalchemy.sql import ColumnElement, Update, and_, select
  from airflow.migrations.db_types import TIMESTAMP, StringID
label_desc: Use image node_id 2 for calling a global function with example usage: _update_value_from_dag_run(dialect_name, target_table, target_column, join_columns) and returns: target_table, target_table