From cf4edcea8de5d38b559a3cd392c1c649677a34a0 Mon Sep 17 00:00:00 2001
From: Dennis Keck <26092524+fellhorn@users.noreply.github.com>
Date: Mon, 10 Jun 2024 11:30:25 +0200
Subject: [PATCH] Handle inputs in subworkflows

Signed-off-by: Dennis Keck <26092524+fellhorn@users.noreply.github.com>
---
 flytectl/cmd/compile/compile.go               |  73 +++++++++++++++---
 .../cmd/compile/testdata/launchplan-in-wf.tgz | Bin 777 -> 875 bytes
 2 files changed, 61 insertions(+), 12 deletions(-)

diff --git a/flytectl/cmd/compile/compile.go b/flytectl/cmd/compile/compile.go
index f64991f941b..16d1fcf9a72 100644
--- a/flytectl/cmd/compile/compile.go
+++ b/flytectl/cmd/compile/compile.go
@@ -81,24 +81,15 @@ func compileFromPackage(packagePath string) error {
 	}
 
 	var providers []common.InterfaceProvider
-	for _, plan := range plans {
-		providers = append(providers, compiler.NewLaunchPlanInterfaceProvider(*plan))
-	}
+	var compiledWorkflows = map[string]*core.CompiledWorkflowClosure{}
 
 	// compile workflows
-	for wfName, workflow := range workflows {
-
-		fmt.Println("\nCompiling workflow:", wfName)
+	for _, workflow := range workflows {
+		providers, err = handleWorkflow(workflow, compiledTasks, compiledWorkflows, providers, plans, workflows)
 
-		_, err := compiler.CompileWorkflow(workflow.Template,
-			workflow.SubWorkflows,
-			compiledTasks,
-			providers)
 		if err != nil {
-			fmt.Println(":( Error Compiling workflow:", wfName)
 			return err
 		}
-
 	}
 
 	fmt.Println("All Workflows compiled successfully!")
@@ -109,6 +100,64 @@ func compileFromPackage(packagePath string) error {
 	return nil
 }
 
+func handleWorkflow(
+	workflow *admin.WorkflowSpec,
+	compiledTasks []*core.CompiledTask,
+	compiledWorkflows map[string]*core.CompiledWorkflowClosure,
+	compiledLaunchPlanProviders []common.InterfaceProvider,
+	plans map[string]*admin.LaunchPlan,
+	workflows map[string]*admin.WorkflowSpec) ([]common.InterfaceProvider, error) {
+	reqs, _ := compiler.GetRequirements(workflow.Template, workflow.SubWorkflows)
+	wfName := workflow.Template.Id.Name
+
+	// Check if all the subworkflows referenced by launchplan are compiled
+	for _, lpId := range reqs.GetRequiredLaunchPlanIds() {
+		lpWfName := plans[lpId.Name].Spec.WorkflowId.Name
+		missingWorkflow := workflows[lpWfName]
+		if compiledWorkflows[lpWfName] == nil {
+			// Recursively compile the missing workflow first
+			err := error(nil)
+			compiledLaunchPlanProviders, err = handleWorkflow(missingWorkflow, compiledTasks, compiledWorkflows, compiledLaunchPlanProviders, plans, workflows)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	fmt.Println("\nCompiling workflow:", wfName)
+
+	wf, err := compiler.CompileWorkflow(workflow.Template,
+		workflow.SubWorkflows,
+		compiledTasks,
+		compiledLaunchPlanProviders)
+
+	if err != nil {
+		fmt.Println(":( Error Compiling workflow:", wfName)
+		return nil, err
+	}
+	compiledWorkflows[wfName] = wf
+
+	// Update the expected inputs and outputs for the launchplans which reference this workflow
+	for _, plan := range plans {
+		if plan.Spec.WorkflowId.Name == wfName {
+			plan.Closure.ExpectedOutputs = wf.Primary.Template.Interface.Outputs
+			newMap := make(map[string]*core.Parameter)
+
+			for key, value := range wf.Primary.Template.Interface.Inputs.Variables {
+				newMap[key] = &core.Parameter{
+					Var: value,
+				}
+			}
+			plan.Closure.ExpectedInputs = &core.ParameterMap{
+				Parameters: newMap,
+			}
+			compiledLaunchPlanProviders = append(compiledLaunchPlanProviders, compiler.NewLaunchPlanInterfaceProvider(*plan))
+		}
+	}
+
+	return compiledLaunchPlanProviders, nil
+}
+
 const (
 	compileShort = `Validate flyte packages without registration needed.`
 	compileLong  = `
diff --git a/flytectl/cmd/compile/testdata/launchplan-in-wf.tgz b/flytectl/cmd/compile/testdata/launchplan-in-wf.tgz
index 1a78ceaf22acedd4f12d0dccec9e08b50cbb3aa4..b297277ce0b807eb705c203a0fcd6039e6072e4d 100644
GIT binary patch
literal 875
zcmV-x1C;z9iwFq$#%5*$|7L7?bY(4YVPk7yXJsyQXL<nbn#*q5Komd+NE0$?UJi=r
zBvK+4T~K-A@v}kdLuwXPH_0DZE)0zY*vOawin8sZQvac!*iYzS2Vx*0R7zZ>xo6{H
z=8ny4ICHO|NA&39nbZHwof;3&qoBmxbyO5hyJ-*mnW8d=4*Mttx}DGo#O2ZRM70iE
zq1)o`n8n^IjF~#4n#xSm!U*{P(e0ieH0n*i*P*>S^}8Vt#$C2b+g~4M{hOvfJTl8r
zlE(;DH4)QIjTyS8Sxl^di&+*rRMK4kHST$C=kB=s#-AwoX9geSv!%T`Ujqq|0!Gpn
z-oRO3k@7g3&q$(GdXKZ7A{Fq4>v@UzGExv($2qUtl9ut32rpN5@OC~|TKAk5xBUwq
z1a6I&x1#f5l8??K+7(2WNVb@Rb(t0v{15&I|36zad*UbA)*i(8f9HRu8&m$*RRjE=
zj{j%1EPc6UM=)u4u_M^WBsTmGUX3;cN!^!_iYn+wM(A9G!YTv*ga4n9|Ct5;e<m<{
z&d+>W`M<$t@_!xtpN{{3WfrmhYb@cd*!KzB&sGFJM49{=UTF>N(CIYFh-}2uR>+HJ
zlDdzJ`An%WalPwk2|-B!{15(z{2xdOV{ZLlOU3_8wWj^ADh&LeivJHp{y(?)PZzmA
zfFR=s|AYS_{|BBHihWxmX0wO94RD|TEo(aeHw?oB|EJ>r-?=&Y_*#1qp8K9Wd_Z<3
zNo4uUj50<QvK$YUQE9>2fcia$-&6xyZw^9E{ovbC)EYM$Zdhp#G<qywUtzO7hz4Zd
z=T))cAbDAKy>>V35PW$_qvq8W*_36m9T9E09q_u_C);8$?m3G(L`+eZ1E)8Nt2v=F
z3NDIZF?1S@oQQ$(oNZaIp7Uy>?Yo{hw-cU|JrR7<cRG0u5iibpk(fkYiTGjPr*YRf
z8jEMdJAU(m2jqD4`f1e<IE~*}+v$eBUG=@taXlW;8m~LuW*9He@zks=leJ-&-12MP
zCMPeGpCeg~hK6@{$fzPyLQ=Ax0{#d8ga4l|>fHK&O8LKG%+&u49sK`~^8cyg|H&fv
z2X=8G-T|y9{tZyfKve($0000000000000000000000000007{z`2$XxUKs#T006ft
Bt-$~Q

literal 777
zcmV+k1NQtMiwFqb!dYel|7L7?bY(4YVPk7yXJsyQXL<nbnn6$7Fc^kSi&z&l5ZQ*n
zyP%#nCcAOcrhqhQ(stMlPP;}~>yD^x5~U7pg^+gM&)P4Wo6^!!x}k|>1Kx*2-0%C`
z0*&)Hc6fzX-k<n`PttG7g!~n(G0TpY<v3UEQCwKI5M*yaav<#~KhYmg9w&Bvud9Nt
zd?)Jqr*`1lHg|2|+6M*!51*RXZ}oV;#iLZo#Op^%yVZ&Mp2OoWcRBuDw>mnCdd-?W
zW(eUBQFU$CsS3xb>+xT!S1qz<<vISFG7N*>^19#U?<vM^saSq_<2hiy*2%)MB%!2C
zY;r`tuMoOAdI|0!9R7#@;r~aAnD{>k!&(2^dHUb67W{A3;s0Fxe=;Ze`LjS<ebpc)
zDZeoap=GWcPr5@-`Mvf&v9}Ggv{K%<>~|SA^?ZP?(!x#=BMbhA|KWe|e`9&`S8z|<
zq4;m<|N6rHf7KB#{GW^ee-v*N`FcE5r%}j8FX*<ZOUyOFh2R!l8>jY3Ih!2V--^cD
zM#5X2p^`jGzE-l<w9yXKQ9N|`A-n9q84t4rlLOgEm82!cf-p{1j~eIaJZoND&`rjo
zROvR06WIy|bW114k&`T-Yjzk*{Qjh_=_`LyyrYYC>Nv}M)QM@IEyfzBveAyCAk^>e
zt5f<)7hlDW=RZT&>-RjH=|p#R{V4Wf8k$xc`ZtojsB<P0dN}#}c_T_BAOG7tKUI;}
zh(hHDdalKrvgM~8H6DD^t_8;E`Y6P@Q8VpGx;FlP=+i71`dvvUld&=}|9r{?|HJ?A
z|Kr8+M%NNA(ifd1Xv#ZX_rLZ1zqR0hyXL_Ex%hvpR4hN6I|r_<`!6gBVBH`0|L{Nj
z4<5qzzhwRYU+@158~)G5|Fir5*Hn*r0Q?XC!~fvnQeE==U%vbQ?&9-*xc|?`|DS*T
zkl%R!zmo0$zc4oDCjtA!B1g|}SVF*400000000000000000000000000D18ff@*-0
H08jt`ngp7{