Add variable_map to GetDataResponse

Signed-off-by: mao3267 <chenvincent610@gmail.com>

flyteadmin/pkg/manager/impl/node_execution_manager.go

@@ -538,24 +538,39 @@ func (m *NodeExecutionManager) GetNodeExecutionData(
 
 	var outputs *core.LiteralMap
 	var outputURLBlob *admin.UrlBlob
+	var outputVariableMap *core.VariableMap
 	group.Go(func() error {
 		var err error
 		outputs, outputURLBlob, err = util.GetOutputs(groupCtx, m.urlData, m.config.ApplicationConfiguration().GetRemoteDataConfig(),
 			m.storageClient, nodeExecution.GetClosure())
 		return err
 	})
 
+	// TODO: Get the output variable map from the node execution model
+	// group.Go(func() error {
+	// 	var err error
+
+	// 	modelNode, err := m.db.NodeExecutionRepo().Get(groupCtx, repoInterfaces.NodeExecutionResource{
+	// 		NodeExecutionIdentifier: request.GetId(),
+	// 	})
+
+	// 	node, err := transformers.FromNodeExecutionModel(modelNode, transformers.DefaultExecutionTransformerOptions)
+
+	// 	return err
+	// })
+
 	err = group.Wait()
 	if err != nil {
 		return nil, err
 	}
 
 	response := &admin.NodeExecutionGetDataResponse{
-		Inputs:      inputURLBlob,
-		Outputs:     outputURLBlob,
-		FullInputs:  inputs,
-		FullOutputs: outputs,
-		FlyteUrls:   common.FlyteURLsFromNodeExecutionID(request.GetId(), nodeExecution.GetClosure() != nil && nodeExecution.GetClosure().GetDeckUri() != ""),
+		Inputs:            inputURLBlob,
+		Outputs:           outputURLBlob,
+		FullInputs:        inputs,
+		FullOutputs:       outputs,
+		FlyteUrls:         common.FlyteURLsFromNodeExecutionID(request.GetId(), nodeExecution.GetClosure() != nil && nodeExecution.GetClosure().GetDeckUri() != ""),
+		OutputVariableMap: outputVariableMap,
 	}
 
 	if len(nodeExecutionModel.DynamicWorkflowRemoteClosureReference) > 0 {

flyteadmin/pkg/manager/impl/task_execution_manager.go

@@ -328,24 +328,41 @@ func (m *TaskExecutionManager) GetTaskExecutionData(
 
 	var outputs *core.LiteralMap
 	var outputURLBlob *admin.UrlBlob
+	var outputVariableMap *core.VariableMap
 	group.Go(func() error {
 		var err error
 		outputs, outputURLBlob, err = util.GetOutputs(groupCtx, m.urlData, m.config.ApplicationConfiguration().GetRemoteDataConfig(),
 			m.storageClient, taskExecution.GetClosure())
 		return err
 	})
 
+	group.Go(func() error {
+		var err error
+		taskModel, err := m.db.TaskRepo().Get(groupCtx, repoInterfaces.Identifier{
+			Project: taskExecution.GetId().GetTaskId().GetProject(),
+			Domain:  taskExecution.GetId().GetTaskId().GetDomain(),
+			Name:    taskExecution.GetId().GetTaskId().GetName(),
+			Version: taskExecution.GetId().GetTaskId().GetVersion(),
+		})
+
+		task, err := transformers.FromTaskModel(taskModel)
+		outputVariableMap = task.GetClosure().GetCompiledTask().GetTemplate().GetInterface().Outputs
+
+		return err
+	})
+
 	err = group.Wait()
 	if err != nil {
 		return nil, err
 	}
 
 	response := &admin.TaskExecutionGetDataResponse{
-		Inputs:      inputURLBlob,
-		Outputs:     outputURLBlob,
-		FullInputs:  inputs,
-		FullOutputs: outputs,
-		FlyteUrls:   common.FlyteURLsFromTaskExecutionID(request.GetId(), false),
+		Inputs:            inputURLBlob,
+		Outputs:           outputURLBlob,
+		FullInputs:        inputs,
+		FullOutputs:       outputs,
+		FlyteUrls:         common.FlyteURLsFromTaskExecutionID(request.GetId(), false),
+		OutputVariableMap: outputVariableMap,
 	}
 
 	m.metrics.TaskExecutionInputBytes.Observe(float64(response.GetInputs().GetBytes()))