code: def _move_duplicate_data_to_new_table(
      session,
      source_table: Table,
      subquery: Query,
      uniqueness: list[str],
      target_table_name: str,
  ):
      """
      When adding a uniqueness constraint we first should ensure that there are no duplicate rows.
  
      This function accepts a subquery that should return one record for each row with duplicates (e.g.
      a group by with having count(*) > 1).  We select from ``source_table`` getting all rows matching the
      subquery result and store in ``target_table_name``.  Then to purge the duplicates from the source table,
      we do a DELETE FROM with a join to the target table (which now contains the dupes).
  
      :param session: sqlalchemy session for metadata db
      :param source_table: table to purge dupes from
      :param subquery: the subquery that returns the duplicate rows
      :param uniqueness: the string list of columns used to define the uniqueness for the table. used in
          building the DELETE FROM join condition.
      :param target_table_name: name of the table in which to park the duplicate rows
      """
      bind = session.get_bind()
      dialect_name = bind.dialect.name
  
      query = (
          select(
              *(
                  source_table.c[x.name].label(str(x.name))
                  for x in source_table.columns
              )
          )
          .select_from(source_table)
          .join(
              subquery,
              and_(
                  *(
                      source_table.c[x] == subquery.c[x]
                      for x in uniqueness
                  )
              ),
          )
      )
  
      _create_table_as(
          session=session,
          dialect_name=dialect_name,
          source_query=query,
          target_table_name=target_table_name,
          source_table_name=source_table.name,
      )
  
      # we must ensure that the CTAS table is created prior to the DELETE step since we have to join to it
      session.commit()
  
      metadata = reflect_tables([target_table_name], session)
      target_table = metadata.tables[target_table_name]
      where_clause = and_(
          *(source_table.c[x] == target_table.c[x] for x in uniqueness)
      )
  
      if dialect_name == "sqlite":
          subq = query.selectable.with_only_columns(
              [text(f"{source_table}.ROWID")]
          )
          delete = source_table.delete().where(
              column("ROWID").in_(subq)
          )
      else:
          delete = source_table.delete(where_clause)
  
      session.execute(delete)
  
imports: from __future__ import annotations
  import contextlib
  import enum
  import json
  import logging
  import os
  import sys
  import time
  import warnings
  from dataclasses import dataclass
  from tempfile import gettempdir
  from typing import TYPE_CHECKING, Callable, Generator, Iterable
  from sqlalchemy import Table, and_, column, delete, exc, func, inspect, literal, or_, select, table, text, tuple_
  import airflow
  from airflow import settings
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.models import import_all_models
  from airflow.utils import helpers
  from airflow.utils.session import NEW_SESSION, create_session, provide_session
label_desc: Use image node_id 27 for calling a global function with example usage: _move_duplicate_data_to_new_table(session, source_table, subquery, uniqueness, target_table_name) without return types