def METHOD_NAME(
    total_row_count: int,
    batch_size: int,
    dry_run: bool,
    table_name: str,
    query_id: str,
    update_query: str,
    update_timeout: int,
    batch_start_var: str,
    postgres_conn_id: str = POSTGRES_CONN_ID,
    task: AbstractOperator = None,
    **kwargs,
):
    if total_row_count == 0:
        return 0
    # Progress is tracked in an Airflow variable. When the task run starts, we resume
    # from the start point set by this variable (defaulted to 0). This prevents the
    # task from starting over at the beginning on retries.
    initial_batch_start = Variable.get(batch_start_var, 0, deserialize_json=True)
    logger.info(f"Starting at {initial_batch_start:,}")
    updated_count = 0
    batch_start = initial_batch_start
    while batch_start <= total_row_count:
        batch_end = batch_start + batch_size
        logger.info(f"Updating rows with id {batch_start:,} through {batch_end:,}.")
        count = run_sql.function(
            dry_run=dry_run,
            sql_template=constants.UPDATE_BATCH_QUERY,
            query_id=query_id,
            # Only log the query the first time, so as not to spam the logs
            log_sql=batch_start == initial_batch_start,
            postgres_conn_id=postgres_conn_id,
            task=task,
            timeout=update_timeout,
            table_name=table_name,
            update_query=update_query,
            batch_start=batch_start,
            batch_end=batch_end,
        )
        updated_count += count
        batch_start = batch_end
        # Update the Airflow variable to the next value of batch_start.
        Variable.set(batch_start_var, batch_end)
        percent_complete = (min(batch_end, total_row_count) / total_row_count) * 100
        logger.info(
            f"Updated {updated_count:,} rows. {percent_complete:.2f}% complete."
        )
    return updated_count