Dev-iL opened a new issue, #51109:
URL: https://github.com/apache/airflow/issues/51109

   ### Apache Airflow version
   
   2.11.0
   
   ### If "Other Airflow 2 version" selected, which one?
   
   _No response_
   
   ### What happened?
   
   The mapping length in `expandinput` is determined incorrectly to be the 
**number of returns from a task**, instead of the number of elements in the 
sequence we map on (which is one of the values returned by the aforementioned 
task).
   
   Note how `all_lengths` is always `2` regardless of the contents of `value`:
   
   
![Image](https://github.com/user-attachments/assets/8a15409d-7024-4446-acd3-e4215664980f)
 
   
   
![Image](https://github.com/user-attachments/assets/9b8d2494-ea88-4e10-9021-89dc3a9752be)
   
   Depending on the relative size of the determined and actual map size, the 
dag either reports the error below, works correctly, or ends up not processing 
the later elements.
   
   ```none
   .../.venv/lib64/python3.11/site-packages/airflow/models/expandinput.py", 
line 197, in _expand_mapped_field
       return value[found_index]
              ~~~~~^^^^^^^^^^^^^
   IndexError: list index out of range
   ```
   
   ### What you think should happen instead?
   
   `all_length` should correctly reflect the number of elements we're mapping 
on.
   
   ### How to reproduce
   
   ```python
   from __future__ import annotations
   
   import pendulum
   from airflow.decorators import task, task_group
   from airflow.models.dag import DAG
   
   TYPE_CHECKING: bool = False
   if TYPE_CHECKING:
       from datetime import datetime
       from typing import Any
   
   
   @task.python(multiple_outputs=True)
   def determine_run_params() -> dict[str, Any]:
       return {
           "targets": [f"target{n}" for n in range(1)],  # <<<<<<<<<< Modify 
for different results
           "date": pendulum.now(),
           # "str1": "1",  # Uncomment these for different results
           # "str2": "2",
           # "str3": "3",
       }
   
   
   @task.python
   def do_something(
       target: str,
       date: datetime,
   ) -> str:
       return f"All good for {target} on {date}!"
   
   
   @task_group
   def process(
       date: datetime,
       target: str,
   ) -> None:
       do_something(target, date)
   
   
   with DAG(
       "mcve",
       start_date=pendulum.datetime(2020, 1, 1),
   ) as dag:
       run_params = determine_run_params()
       
process.partial(date=run_params["date"]).expand(target=run_params["targets"])
   
   
   if __name__ == "__main__":
       dag.test()
   ```
   
   ### Operating System
   
   Linux
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Other
   
   ### Deployment details
   
   _No response_
   
   ### Anything else?
   
   When the number of map elements (length of `targets`) is equal to the number 
of outputs from `determine_run_params`, the code works correctly. Perhaps 
Airflow's tests related to this functionality are hitting this exact sweet spot.
   
   ### 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 is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to