Repository: finos/tracdap (language: python)

Issue:
out_attrs property in flow definition
## Feature Request

### Description of Problem:
It is not possible to add key attributes or titles (or any attribute) to the outputs of a job. This makes understanding and presenting the outputs of a calculation difficult.

### Potential Solutions:

- A out_attrs property added to the flow definition that allows attributes to be added to the data outputs from a job is needed.
- This is in addition to the job attributes already in the API that add these attributes to all outputs from a job.


Discussion / hints:
Agree this is needed. I think the attrs applied to output nodes shoudl be called "nodeAttrs", so flow nodes can either have "nodeSearch" (for input and model nodes) or "nodeAttrs" (for output nodes). This might also be clearer when it's used with a job, because the job already has "outputAttrs" which are applied to all output nodes.
See also #221 

Fix patch:
diff --git a/tracdap-api/tracdap-metadata/src/main/proto/tracdap/metadata/flow.proto b/tracdap-api/tracdap-metadata/src/main/proto/tracdap/metadata/flow.proto
index 48f63bbb3..f16de066c 100644
--- a/tracdap-api/tracdap-metadata/src/main/proto/tracdap/metadata/flow.proto
+++ b/tracdap-api/tracdap-metadata/src/main/proto/tracdap/metadata/flow.proto
@@ -23,6 +23,7 @@ option java_multiple_files = true;
 import "tracdap/metadata/type.proto";
 import "tracdap/metadata/model.proto";
 import "tracdap/metadata/search.proto";
+import "tracdap/metadata/tag_update.proto";
 
 
 enum FlowNodeType {
@@ -40,6 +41,7 @@ message FlowNode {
   repeated string outputs = 3;
 
   SearchExpression nodeSearch = 4;
+  repeated TagUpdate nodeAttrs = 5;
 
   reserved "parameters;";
 }
diff --git a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunFlowJob.java b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunFlowJob.java
index 3f05fb13a..5151ea533 100644
--- a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunFlowJob.java
+++ b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunFlowJob.java
@@ -24,10 +24,7 @@
 import org.finos.tracdap.config.PlatformConfig;
 import org.finos.tracdap.metadata.*;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 
 
@@ -64,7 +61,8 @@ public Map<String, MetadataWriteRequest> newResultIds(
 
         var runFlow = job.getRunFlow();
 
-        var outputs = flowOutputs(runFlow.getFlow(), resources, resourceMapping);
+        var outputFlowNodes = getFlowOutputNodes(runFlow.getFlow(), resources, resourceMapping);
+        var outputs = getFlowOutputNames(outputFlowNodes);
 
         return newResultIds(tenant, outputs, runFlow.getPriorOutputsMap());
     }
@@ -77,24 +75,27 @@ public Map<String, TagHeader> priorResultIds(
 
         var runFlow = job.getRunFlow();
 
-        var outputs = flowOutputs(runFlow.getFlow(), resources, resourceMapping);
+        var outputFlowNodes = getFlowOutputNodes(runFlow.getFlow(), resources, resourceMapping);
+        var outputs = getFlowOutputNames(outputFlowNodes);
 
         return priorResultIds(outputs, runFlow.getPriorOutputsMap(), resources, resourceMapping);
     }
 
-    private Set<String> flowOutputs(
+    private static Set<String> getFlowOutputNames(Map<String, FlowNode> outputFlowNodes) {
+        return new HashSet<>(outputFlowNodes.keySet());
+    }
+
+    private static Map<String, FlowNode> getFlowOutputNodes(
             TagSelector flowSelector,
             Map<String, ObjectDefinition> resources,
             Map<String, TagHeader> resourceMapping) {
-
         var flowKey = MetadataUtil.objectKey(flowSelector);
         var flowId = resourceMapping.get(flowKey);
         var flowDef = resources.get(MetadataUtil.objectKey(flowId)).getFlow();
 
         return flowDef.getNodesMap().entrySet().stream()
                 .filter(nodeEntry -> nodeEntry.getValue().getNodeType() == FlowNodeType.OUTPUT_NODE)
-                .map(Map.Entry::getKey)
-                .collect(Collectors.toSet());
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
     @Override
@@ -103,7 +104,8 @@ public JobDefinition setResultIds(
             Map<String, ObjectDefinition> resources,
             Map<String, TagHeader> resourceMapping) {
 
-        var flowOutputNames = flowOutputs(job.getRunFlow().getFlow(), resources, resourceMapping);
+        var outputFlowNodes = getFlowOutputNodes(job.getRunFlow().getFlow(), resources, resourceMapping);
+        var flowOutputNames = getFlowOutputNames(outputFlowNodes);
 
         var flowOutputSelectors = setResultIds(flowOutputNames, resultMapping);
 
@@ -120,9 +122,18 @@ public JobDefinition setResultIds(
     public List<MetadataWriteRequest> buildResultMetadata(String tenant, JobConfig jobConfig, JobResult jobResult) {
 
         var runFlow = jobConfig.getJob().getRunFlow();
+        var outputFlowNodes = getFlowOutputNodes(
+                jobConfig.getJob().getRunFlow().getFlow(),
+                jobConfig.getResourcesMap(),
+                jobConfig.getResourceMappingMap()
+        );
+
+        var perNodeOutputAttrs = outputFlowNodes.entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getNodeAttrsList()));
 
         return buildResultMetadata(
                 tenant, runFlow.getOutputsMap(), runFlow.getPriorOutputsMap(),
-                runFlow.getOutputAttrsList(), jobConfig, jobResult);
+                runFlow.getOutputAttrsList(), perNodeOutputAttrs,
+                jobConfig, jobResult);
     }
 }
diff --git a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelJob.java b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelJob.java
index 0388bdca4..292740466 100644
--- a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelJob.java
+++ b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelJob.java
@@ -115,6 +115,7 @@ public List<MetadataWriteRequest> buildResultMetadata(String tenant, JobConfig j
 
         return buildResultMetadata(
                 tenant, runModel.getOutputsMap(), runModel.getPriorOutputsMap(),
-                runModel.getOutputAttrsList(), jobConfig, jobResult);
+                runModel.getOutputAttrsList(), Map.of(),
+                jobConfig, jobResult);
     }
 }
diff --git a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelOrFlow.java b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelOrFlow.java
index e3c369bf8..11f2da0a4 100644
--- a/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelOrFlow.java
+++ b/tracdap-services/tracdap-svc-orch/src/main/java/org/finos/tracdap/svc/orch/jobs/RunModelOrFlow.java
@@ -129,7 +129,8 @@ public Map<String, TagSelector> setResultIds(
 
     public List<MetadataWriteRequest> buildResultMetadata(
             String tenant, Map<String, TagSelector> outputs, Map<String, TagSelector> priorOutputs,
-            List<TagUpdate> outputAttrs, JobConfig jobConfig, JobResult jobResult) {
+            List<TagUpdate> outputAttrs, Map<String, List<TagUpdate>> perNodeOutputAttrs,
+            JobConfig jobConfig, JobResult jobResult) {
 
         var updates = new ArrayList<MetadataWriteRequest>();
 
@@ -163,6 +164,11 @@ public List<MetadataWriteRequest> buildResultMetadata(
                     .setValue(MetadataCodec.encodeValue(outputName))
                     .build());
 
+            var nodeOutputAttrs = perNodeOutputAttrs.get(outputName);
+            if (nodeOutputAttrs == null) {
+                nodeOutputAttrs = List.of();
+            }
+
             var dataUpdate = MetadataWriteRequest.newBuilder()
                     .setTenant(tenant)
                     .setObjectType(ObjectType.DATA)
@@ -170,6 +176,7 @@ public List<MetadataWriteRequest> buildResultMetadata(
                     .setDefinition(dataObj)
                     .addAllTagUpdates(controlledAttrs)
                     .addAllTagUpdates(outputAttrs)
+                    .addAllTagUpdates(nodeOutputAttrs)
                     .build();
 
             updates.add(dataUpdate);


Test patch:
diff --git a/tracdap-services/tracdap-svc-orch/src/test/java/org/finos/tracdap/svc/orch/jobs/RunFlowTest.java b/tracdap-services/tracdap-svc-orch/src/test/java/org/finos/tracdap/svc/orch/jobs/RunFlowTest.java
index a76e4dcb8..f86c6d1d2 100644
--- a/tracdap-services/tracdap-svc-orch/src/test/java/org/finos/tracdap/svc/orch/jobs/RunFlowTest.java
+++ b/tracdap-services/tracdap-svc-orch/src/test/java/org/finos/tracdap/svc/orch/jobs/RunFlowTest.java
@@ -36,6 +36,7 @@
 import java.time.LocalDate;
 import java.util.Arrays;
 import java.util.List;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import static org.finos.tracdap.svc.orch.jobs.Helpers.runJob;
@@ -51,6 +52,7 @@ public class RunFlowTest {
 
     private static final String LOANS_INPUT_PATH = "examples/models/python/data/inputs/loan_final313_100_shortform.csv";
     private static final String CURRENCY_INPUT_PATH = "examples/models/python/data/inputs/currency_data_sample.csv";
+    public static final String ANALYSIS_TYPE = "analysis_type";
 
     // Only test E2E run model using the local repo
     // E2E model loading with different repo types is tested in ImportModelTest
@@ -79,6 +81,15 @@ void createFlow() {
 
         var metaClient = platform.metaClientBlocking();
 
+        var pi_value = Value.newBuilder()
+                .setType(
+                        TypeDescriptor.newBuilder()
+                                .setBasicType(BasicType.FLOAT)
+                                .build()
+                )
+                .setFloatValue(3.14)
+                .build();
+
         var flowDef = FlowDefinition.newBuilder()
                 .putNodes("customer_loans", FlowNode.newBuilder().setNodeType(FlowNodeType.INPUT_NODE).build())
                 .putNodes("currency_data", FlowNode.newBuilder().setNodeType(FlowNodeType.INPUT_NODE).build())
@@ -93,7 +104,18 @@ void createFlow() {
                         .addInputs("preprocessed_data")
                         .addOutputs("profit_by_region")
                         .build())
-                .putNodes("profit_by_region", FlowNode.newBuilder().setNodeType(FlowNodeType.OUTPUT_NODE).build())
+                .putNodes("profit_by_region", FlowNode.newBuilder()
+                        .setNodeType(FlowNodeType.OUTPUT_NODE)
+                        .addAllNodeAttrs(List.of(
+                                TagUpdate.newBuilder()
+                                        .setAttrName(ANALYSIS_TYPE)
+                                        .setValue(pi_value)
+                                        .build()
+                        ))
+                        .build())
+                .putNodes("preprocessed_data", FlowNode.newBuilder()
+                        .setNodeType(FlowNodeType.OUTPUT_NODE)
+                        .build())
                 .addEdges(FlowEdge.newBuilder()
                         .setSource(FlowSocket.newBuilder().setNode("customer_loans"))
                         .setTarget(FlowSocket.newBuilder().setNode("model_1").setSocket("customer_loans")))
@@ -106,6 +128,9 @@ void createFlow() {
                 .addEdges(FlowEdge.newBuilder()
                         .setSource(FlowSocket.newBuilder().setNode("model_2").setSocket("profit_by_region"))
                         .setTarget(FlowSocket.newBuilder().setNode("profit_by_region")))
+                .addEdges(FlowEdge.newBuilder()
+                        .setSource(FlowSocket.newBuilder().setNode("model_1").setSocket("preprocessed_data"))
+                        .setTarget(FlowSocket.newBuilder().setNode("preprocessed_data")))
                 .build();
 
         var createReq = MetadataWriteRequest.newBuilder()
@@ -325,9 +350,22 @@ void runFlow() {
 
         var dataSearchResult = metaClient.search(dataSearch);
 
-        Assertions.assertEquals(1, dataSearchResult.getSearchResultCount());
+        Assertions.assertEquals(2, dataSearchResult.getSearchResultCount());
 
-        var searchResult = dataSearchResult.getSearchResult(0);
+        var indexedResult = dataSearchResult.getSearchResultList().stream()
+                .collect(Collectors.toMap(RunFlowTest::getJobOutput, Function.identity()));
+
+        var profitByRegionAnalysisType = indexedResult.get("profit_by_region")
+                .getAttrsOrThrow(ANALYSIS_TYPE)
+                .getFloatValue();
+        Assertions.assertEquals(3.14, profitByRegionAnalysisType);
+
+        Assertions.assertFalse(
+            indexedResult.get("preprocessed_data")
+                    .getAttrsMap().containsKey(ANALYSIS_TYPE)
+        );
+
+        var searchResult = indexedResult.get("profit_by_region");
         var dataReq = MetadataReadRequest.newBuilder()
                 .setTenant(TEST_TENANT)
                 .setSelector(MetadataUtil.selectorFor(searchResult.getHeader()))
@@ -343,6 +381,11 @@ void runFlow() {
         outputDataId = dataTag.getHeader();
     }
 
+    private static String getJobOutput(org.finos.tracdap.metadata.Tag t) {
+        var attr = t.getAttrsOrThrow("trac_job_output");
+        return attr.getStringValue();
+    }
+
     @Test @Order(5)
     void checkOutputData() {