Remove DAG parsing from StandardTaskRunner

[ephraimbuddy]

This makes the starting of StandardTaskRunner faster as the parsing of DAG will now be done at task_run.
Also removed parsing of example dags when running a task

[ashb]

Fixes the slowdown part of #26573

[tirkarthi]

include_examples is retrieved by default from configuration in get_dag. Can this change to have it always False be removed so that users who don't want examples to load can do so by using configuration? I was having a test case in my development PR that worked fine running an example dag and now it fails after rebase. This will be an issue to users too who might want to explore tasks command in cli with examples that worked before 2.4.0.

[ephraimbuddy]

Sorry for this late reply. Can you share the test case?

[tirkarthi]

I was trying out a patch for this issue : #26555 . I just added an example of dynamic task mapping dag in this PR as a fixture : #26678 . I used to use this dag in my test case calling airflow tasks render which used to pass earlier but failed after rebase. I don't have the PR created yet.

[ephraimbuddy]

diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index 982aa31fd5..ab05bb56a9 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -578,13 +578,13 @@ def task_render(args):
         task, args.map_index, exec_date_or_run_id=args.execution_date_or_run_id, create_if_necessary="memory"
     )
     ti.render_templates()
-    for attr in task.__class__.template_fields:
+    for attr in task.template_fields:
         print(
             textwrap.dedent(
                 f"""        # ----------------------------------------------------------
         # property: {attr}
         # ----------------------------------------------------------
-        {getattr(task, attr)}
+        {getattr(ti.task, attr)}
         """
             )
         )
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index c565f601d7..f3673fb2dd 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -388,6 +388,54 @@ class TestCliTasks:
         assert 'echo "2016-01-01"' in output
         assert 'echo "2016-01-08"' in output
 
+    def test_mapped_task_render(self):
+        """
+        tasks render should render and displays templated fields for a given mapped task
+        """
+        with redirect_stdout(io.StringIO()) as stdout:
+            task_command.task_render(
+                self.parser.parse_args(
+                    [
+                        "tasks",
+                        "render",
+                        "test_mapped_classic",
+                        "consumer_literal",
+                        "2022-01-01",
+                        "--map-index",
+                        "0",
+                    ]
+                )
+            )
+        # the dag test_mapped_classic has op_args=[[1], [2], [3]], so the first mapped task should have op_args=[1]
+        output = stdout.getvalue()
+        assert "[1]" in output
+        assert "[2]" not in output
+        assert "[3]" not in output
+        assert "property: op_args" in output
+
+    def test_mapped_task_render_with_template(self):
+        """
+        tasks render should render and displays templated fields for a given mapped task
+        """
+        with redirect_stdout(io.StringIO()) as stdout:
+            task_command.task_render(
+                self.parser.parse_args(
+                    [
+                        "tasks",
+                        "render",
+                        "test_mapped_task_with_template",
+                        "some_command",
+                        "2022-01-01",
+                        "--map-index",
+                        "0",
+                    ]
+                )
+            )
+        # the dag test_mapped_classic has op_args=[[1], [2], [3]], so the first mapped task should have op_args=[1]
+        output = stdout.getvalue()
+        assert 'echo "2022-01-01"' in output
+        assert 'echo "2022-01-08"' in output
+
     def test_cli_run_when_pickle_and_dag_cli_method_selected(self):
         """
         tasks run should return an AirflowException when invalid pickle_id is passed
diff --git a/tests/dags/test_mapped_task_with_templates.py b/tests/dags/test_mapped_task_with_templates.py
new file mode 100644
index 0000000000..fe0b8d0c49
--- /dev/null
+++ b/tests/dags/test_mapped_task_with_templates.py
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import datetime
+from textwrap import dedent
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+
+with DAG(dag_id="test_mapped_task_with_template", start_date=datetime.datetime(2022, 1, 1)) as dag:
+    templated_command = dedent(
+        """
+    {% for i in range(5) %}
+        echo "{{ ds }}"
+        echo "{{ macros.ds_add(ds, 7)}}"
+    {% endfor %}
+    """
+    )
+    commands = [templated_command, "echo 1"]
+
+    BashOperator.partial(task_id="some_command").expand(bash_command=commands)

[tirkarthi]

Thanks @ephraimbuddy for the patch. I will use it and also add you as co-author.

[pingzh]

hi @ephraimbuddy could you please give more context about why this should be removed?

it looks like your PR only tried to optimize when executing a task via fork. there is also an important part https://github.com/apache/airflow/pull/26750/files#diff-8b336b0706abbada03e8dce519e122384da2c913cc79ee4d638d0427a1342d41L47-L48

i think we should need dag = get_dag_by_deserialization(args.dag_id) please let me know your thoughts

[ephraimbuddy]

hi @ephraimbuddy could you please give more context about why this should be removed?

it looks like your PR only tried to optimize when executing a task via fork. there is also an important part https://github.com/apache/airflow/pull/26750/files#diff-8b336b0706abbada03e8dce519e122384da2c913cc79ee4d638d0427a1342d41L47-L48

i think we should need dag = get_dag_by_deserialization(args.dag_id) please let me know your thoughts

We noticed a significant increase in task run duration with the previous change. Using get_dag_by_deserialization only delays the dag parsing(for _start_by_fork). Looking at exec I think we should devise another way to get the serialized dag for use because using the previous change, we had to parse the dag in TaskRunner process for fork which causes delays.