]
    
    dag.clear()
    Variable.set(key="arg1", value=[1, 2, 3, 4, 5])
    with dag:
        task_2.expand(arg2=task_1()).operator
    
    
    serialized_dag = SerializedDAG.from_dict(SerializedDAG.to_dict(dag))
    dr.dag = serialized_dag
    
    ti = dr.get_task_instance(task_id="task_1")
    task1 = dag.get_task("task_1")
    ti.refresh_from_task(task1)
    ti.run()
    task2 = dag.get_task("task_2")
    for ti in dr.get_task_instances():
        if ti.map_index < 0:
            ti.task = task1
        else:
            ti.task = task2
        session.merge(ti)
    session.flush()
    
    dr.task_instance_scheduling_decisions()
    
    dr.verify_integrity()
    tis = dr.get_task_instances()
    indices = [(ti.map_index, ti.state) for ti in tis if ti.map_index >= 0]
    assert sorted(indices) == [