### Apache Airflow version

main (development)

### What happened

Task failing with error "missing upstream values" when running a dag with dynamic task group mapping
error -
```
[2022-11-23, 15:50:07 UTC] {abstractoperator.py:456} ERROR - Cannot expand <Task(_PythonDecoratedOperator): increment_and_verify.hello_there> for run manual__2022-11-23T15:49:59.771915+00:00; missing upstream values: ['x']
```

### What you think should happen instead

It should succeed 

### How to reproduce

Run the below dag code 
```
from airflow import DAG
from airflow.models.taskinstance import TaskInstance
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta
from airflow.decorators import task, task_group
with DAG(
    dag_id="taskmap_taskgroup",
    tags=["AIP_42"],
    start_date=datetime(1970, 1, 1),
    schedule_interval=None,
) as dag:

    @task
    def onetwothree():
        return [1,2,3]

    @task
    def hello_there(arg):
        print(arg)

    @task_group
    def increment_and_verify(x):
        hello_there(x)

    increment_and_verify.expand(x=onetwothree())>>DummyOperator(task_id="done")
```

### Operating System

mac os

### Versions of Apache Airflow Providers

_No response_

### Deployment

Virtualenv installation

### Deployment details

_No response_

### Anything else

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

"""
        This function checks if there are any tasks in the dagrun (or all) that
        have a schedule or queued states but are not known by the executor. If
        it finds those it will reset the state to None so they will get picked
        up again.  The batch option is for performance reasons as the queries
        are made in sequence.
        :param filter_by_dag_run: the dag_run we want to process, None if all
        :return: the number of TIs reset
        """
        queued_tis = self.executor.queued_tasks
        
        running_tis = self.executor.running
        
        resettable_states = [TaskInstanceState.SCHEDULED, TaskInstanceState.QUEUED]
        if filter_by_dag_run is None:
            resettable_tis = (
                session.query(TaskInstance)
                .join(TaskInstance.dag_run)
                .filter(
                    DagRun.state == DagRunState.RUNNING,
                    DagRun.run_type != DagRunType.BACKFILL_JOB,
                    TaskInstance.state.in_(resettable_states),
                )
            ).all()
        else:
            resettable_tis = filter_by_dag_run.get_task_instances(state=resettable_states, session=session)
        tis_to_reset = [ti for ti in resettable_tis if ti.key not in queued_tis and ti.key not in running_tis]
        if not tis_to_reset:
            return 0
        def query(result, items):