Compile the workflow only at compile time

flytekit/core/workflow.py

@@ -731,7 +731,10 @@ def wrapper(fn):
             default_metadata=workflow_metadata_defaults,
             docstring=Docstring(callable_=fn),
         )
-        workflow_instance.compile()
+
+        ctx = FlyteContextManager.current_context()
+        if ctx.execution_state.mode != ctx.execution_state.Mode.TASK_EXECUTION:
+            workflow_instance.compile()
         update_wrapper(workflow_instance, fn)
         return workflow_instance
 

tests/flytekit/unit/core/test_workflows.py

@@ -7,8 +7,9 @@
 from typing_extensions import Annotated  # type: ignore
 
 import flytekit.configuration
-from flytekit import StructuredDataset, kwtypes
+from flytekit import FlyteContextManager, StructuredDataset, kwtypes
 from flytekit.configuration import Image, ImageConfig
+from flytekit.core import context_manager
 from flytekit.core.condition import conditional
 from flytekit.core.task import task
 from flytekit.core.workflow import WorkflowFailurePolicy, WorkflowMetadata, WorkflowMetadataDefaults, workflow
@@ -320,3 +321,18 @@ def test_structured_dataset_wf():
     assert_frame_equal(sd_to_schema_wf(), superset_df)
     assert_frame_equal(schema_to_sd_wf()[0], subset_df)
     assert_frame_equal(schema_to_sd_wf()[1], subset_df)
+
+
+def test_compile_wf_at_compile_time():
+    ctx = FlyteContextManager.current_context()
+    with FlyteContextManager.with_context(
+        ctx.with_execution_state(
+            ctx.new_execution_state().with_params(mode=context_manager.ExecutionState.Mode.TASK_EXECUTION)
+        )
+    ):
+
+        @workflow
+        def wf():
+            t4()
+
+        assert len(wf.nodes) == 0