task group mapped_ti_count is incorrected calculated

[Lee-W]

Apache Airflow version

3.0.1

If "Other Airflow 2 version" selected, which one?

No response

What happened?

The following dag fails.

from __future__ import annotations

from airflow.sdk import dag, task, task_group


@dag
def dag1():
    @task
    def task1():
        return [1, 2, 3, 4, 5]

    @task_group
    def task_group_1(arg1):
        @task
        def group1_task_1(arg1):
            return {"a": arg1}

        @task
        def group1_task_2(arg2):
            return arg2

        group1_task_2(group1_task_1(arg1))

    @task
    def task2():
        return None

    task_group_1.expand(arg1=task1()) >> task2()


dag1()

What you think should happen instead?

It should not fail.

How to reproduce

Run the above dag.

Operating System

macOS

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

breeze

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Lee-W]

It's most likely wrong here

airflow/airflow-core/src/airflow/models/expandinput.py

Lines 69 to 76 in 08cc57d

	def get_parse_time_mapped_ti_count(self) -> int:
	if not self.value:
	return 0
	literal_values = [len(v) for _, v in self._iter_parse_time_resolved_kwargs()]
	if len(literal_values) != len(self.value):
	literal_keys = (k for k, _ in self._iter_parse_time_resolved_kwargs())
	raise NotFullyPopulated(set(self.value).difference(literal_keys))
	return functools.reduce(operator.mul, literal_values, 1)

[vatsrahul1001]

@Lee-W how are we progressing on this?

[Lee-W]

will wrap up the PR today. test will be there but refactor will be the next