return Variable.get("arg1", deserialize_json=True)
    with dag_maker(session=session) as dag:
        @task
        def task_2(arg2):
            ...
        task_2.expand(arg2=task_1())
    dr = dag_maker.create_dagrun()
    ti = dr.get_task_instance(task_id="task_1")
    ti.run()
    dr.task_instance_scheduling_decisions()
    tis = dr.get_task_instances()
    indices = [(ti.map_index, ti.state) for ti in tis if ti.map_index >= 0]
    assert sorted(indices) == [
        (0, State.NONE),
        (1, State.NONE),
        (2, State.NONE),
    ]
    
    dag.clear()
    Variable.set(key="arg1", value=[1, 2, 3])
    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")