Repository: odpi/egeria-connector-ibm-information-server
Issue: Lineage mapping output not properly limited for stores that have multiple readers or writers

When a stage interacts with a data store which itself may have multiple stages writing into it or reading from it, the lineage mapping processing is not currently limiting its output to only the relevant portion of those mappings for the particular stage being processed.

The result is that _all_ inputs or outputs for that data store are included in the lineage mappings for a stage-level Process, which is incorrect.

Changed files:
datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/model/DataStageCache.java
datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/DataStageConnector.java
datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/LineageMappingMapping.java
datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/ProcessMapping.java

diff --git a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/DataStageConnector.java b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/DataStageConnector.java
index 1d8b20c3..f125ae96 100644
--- a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/DataStageConnector.java
+++ b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/DataStageConnector.java
@@ -382,8 +382,8 @@ public class DataStageConnector extends DataEngineConnectorBase {
         List<Process> processes = new ArrayList<>();
         log.debug("Translating processes for each stage...");
         for (Stage stage : job.getAllStages()) {
-            ProcessMapping processMapping = new ProcessMapping(dataStageCache, job, stage);
-            Process process = processMapping.getProcess();
+            ProcessMapping processMapping = new ProcessMapping(dataStageCache);
+            Process process = processMapping.getForStage(stage, job);
             if (process != null) {
                 try {
                     log.debug(" ... process: {}", objectMapper.writeValueAsString(process));
diff --git a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/LineageMappingMapping.java b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/LineageMappingMapping.java
index db9c4059..11b47293 100644
--- a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/LineageMappingMapping.java
+++ b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/LineageMappingMapping.java
@@ -189,11 +189,18 @@ class LineageMappingMapping extends BaseMapping {
      *
      * @param fields list of IGC field objects (data_file_field or database_column)
      * @param job the job for which to create the LineageMappings
+     * @param stageRid the RID of the stage for which we are building mappings
+     * @param knownLinks set of known link RIDs
      * @param bSource true if processing a source link, false if a target link
      * @param fullyQualifiedStageName the fully qualifiedName of the stage itself
      * @return {@code Set<LineageMapping>}
      */
-    Set<LineageMapping> getForDataStoreFields(List<Classificationenabledgroup> fields, DataStageJob job, boolean bSource, String fullyQualifiedStageName) {
+    Set<LineageMapping> getForDataStoreFieldsInStage(List<Classificationenabledgroup> fields,
+                                                     DataStageJob job,
+                                                     String stageRid,
+                                                     Set<String> knownLinks,
+                                                     boolean bSource,
+                                                     String fullyQualifiedStageName) {
         final String methodName = "getForDataStoreFields";
         Set<LineageMapping> lineageMappings = new HashSet<>();
         // For each field in the data store...
@@ -202,6 +209,8 @@ class LineageMappingMapping extends BaseMapping {
                 try {
                     String field1QN = getFullyQualifiedName(fieldObj);
                     if (field1QN != null) {
+                        // relatedStageCols could have more than just this stage's inputs / outputs (a given data store
+                        // may be written or read by many stages)
                         ItemList<InformationAsset> relatedStageCols;
                         String propertyName;
                         if (bSource) {
@@ -217,7 +226,8 @@ class LineageMappingMapping extends BaseMapping {
                             // For each object that reads / writes to that field...
                             for (InformationAsset stageColRef : allRelatedStageCols) {
                                 ColumnLevelLineage stageColFull = job.getColumnLevelLineageByRid(stageColRef.getId());
-                                if (stageColFull != null) {
+                                // Limit the details we capture to only the stage we're processing
+                                if (stageColumnForKnownLink(stageColFull, stageRid, knownLinks)) {
                                     String field1EmbeddedQN = getFullyQualifiedName(fieldObj, fullyQualifiedStageName);
                                     String field2EmbeddedQN = getFullyQualifiedName(stageColFull, fullyQualifiedStageName);
                                     if (bSource) {
@@ -236,7 +246,7 @@ class LineageMappingMapping extends BaseMapping {
                                         lineageMappings.add(oneToOne);
                                     }
                                 } else {
-                                    log.error("Unable to find referenced stage column: {}", stageColRef);
+                                    log.debug("Found a stage column for a link not listed as a known link for this stage -- ignoring: {}", stageColFull);
                                 }
                             }
                         } else {
diff --git a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/ProcessMapping.java b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/ProcessMapping.java
index 0cd51ac5..50c1a1f2 100644
--- a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/ProcessMapping.java
+++ b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/mapping/ProcessMapping.java
@@ -24,18 +24,25 @@ import java.util.stream.Stream;
 public class ProcessMapping extends BaseMapping {
 
     private static final Logger log = LoggerFactory.getLogger(ProcessMapping.class);
-    private Process process;
 
     /**
-     * Create a new process from a DataStageJob that represents either a job or a sequence.
+     * Default constructor to pass in the cache for re-use.
      *
      * @param cache used by this mapping
-     * @param job the job or sequence from which to create a process
      */
-    public ProcessMapping(DataStageCache cache, DataStageJob job) {
+    public ProcessMapping(DataStageCache cache) {
         super(cache);
-        final String methodName = "ProcessMapping";
-        process = null;
+    }
+
+    /**
+     * Create a new process from a DataStageJob that represents either a job or a sequence.
+     *
+     * @param job the job or sequence from which to create a process
+     * @return Process
+     */
+    public Process getForJob(DataStageJob job) {
+        final String methodName = "getForJob";
+        Process process;
         if (job.getType().equals(DataStageJob.JobType.SEQUENCE)) {
             Dsjob jobObj = job.getJobObject();
             process = getSkeletonProcess(jobObj);
@@ -94,19 +101,19 @@ public class ProcessMapping extends BaseMapping {
                 }
             }
         }
+        return process;
     }
 
     /**
      * Create a new process from the provided DataStage stage.
      *
-     * @param cache used by this mapping
-     * @param job the job within which the stage exists
      * @param stage the stage from which to create a process
+     * @param job the job within which the stage exists
+     * @return Process
      */
-    public ProcessMapping(DataStageCache cache, DataStageJob job, Stage stage) {
-        super(cache);
-        final String methodName = "ProcessMapping";
-        process = getSkeletonProcess(stage);
+    public Process getForStage(Stage stage, DataStageJob job) {
+        final String methodName = "getForStage";
+        Process process = getSkeletonProcess(stage);
         if (process != null) {
             Set<PortImplementation> portImplementations = new HashSet<>();
             Set<LineageMapping> lineageMappings = new HashSet<>();
@@ -117,11 +124,11 @@ public class ProcessMapping extends BaseMapping {
                 log.debug("Adding input links: {}", allInputLinks);
                 addImplementationDetails(job, stage, allInputLinks, allLinkRids, PortType.INPUT_PORT, portImplementations, lineageMappings);
                 log.debug("Adding input stores: {}", stage.getReadsFromDesign());
-                addDataStoreDetails(job, stage, "reads_from_(design)", stage.getReadsFromDesign(), PortType.INPUT_PORT, portImplementations, lineageMappings);
+                addDataStoreDetails(job, stage, "reads_from_(design)", stage.getReadsFromDesign(), allLinkRids, PortType.INPUT_PORT, portImplementations, lineageMappings);
                 log.debug("Adding output links: {}", allOutputLinks);
                 addImplementationDetails(job, stage, allOutputLinks, allLinkRids, PortType.OUTPUT_PORT, portImplementations, lineageMappings);
                 log.debug("Adding output stores: {}", stage.getWritesToDesign());
-                addDataStoreDetails(job, stage, "writes_to_(design)", stage.getWritesToDesign(), PortType.OUTPUT_PORT, portImplementations, lineageMappings);
+                addDataStoreDetails(job, stage, "writes_to_(design)", stage.getWritesToDesign(), allLinkRids, PortType.OUTPUT_PORT, portImplementations, lineageMappings);
                 log.debug("Adding stage variables");
                 addStageVariableDetails(job, stage, portImplementations, lineageMappings);
                 process.setPortImplementations(new ArrayList<>(portImplementations));
@@ -146,15 +153,9 @@ public class ProcessMapping extends BaseMapping {
                         e);
             }
         }
+        return process;
     }
 
-    /**
-     * Retrieve the mapped process.
-     *
-     * @return Process
-     */
-    public Process getProcess() { return this.process; }
-
     /**
      * Construct a minimal Process object from the provided IGC object (stage, job, sequence, etc).
      *
@@ -271,6 +272,7 @@ public class ProcessMapping extends BaseMapping {
      * @param stage the stage for which to add implementation details
      * @param propertyName the name of the property from which stores were retrieved
      * @param stores the stores for which to create the implementation details
+     * @param linkRids the RIDs of all links known by this stage (both input and output)
      * @param portType the type of port
      * @param portImplementations the list of PortImplementations to append to with implementation details
      * @param lineageMappings the list of LineageMappings to append to with implementation details
@@ -279,6 +281,7 @@ public class ProcessMapping extends BaseMapping {
                                      Stage stage,
                                      String propertyName,
                                      ItemList<InformationAsset> stores,
+                                     Set<String> linkRids,
                                      PortType portType,
                                      Set<PortImplementation> portImplementations,
                                      Set<LineageMapping> lineageMappings) {
@@ -295,7 +298,7 @@ public class ProcessMapping extends BaseMapping {
                     portImplementations.add(portImplementationMapping.getForDataStoreFields(fieldsForStore, stage, portType, fullyQualifiedStageName));
                     log.debug("Adding lineage mappings for fields as {}: {}", portType.getName(), fieldsForStore);
                     LineageMappingMapping lineageMappingMapping = new LineageMappingMapping(cache);
-                    lineageMappings.addAll(lineageMappingMapping.getForDataStoreFields(fieldsForStore, job, portType.equals(PortType.INPUT_PORT), fullyQualifiedStageName));
+                    lineageMappings.addAll(lineageMappingMapping.getForDataStoreFieldsInStage(fieldsForStore, job, stage.getId(), linkRids, portType.equals(PortType.INPUT_PORT), fullyQualifiedStageName));
                 }
             } else {
                 log.error("Unable to determine identity for stage -- not including: {}", stage);
diff --git a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/model/DataStageCache.java b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/model/DataStageCache.java
index 50e2ee9c..d5cb7236 100644
--- a/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/model/DataStageCache.java
+++ b/datastage-adapter/src/main/java/org/odpi/egeria/connectors/ibm/datastage/dataengineconnector/model/DataStageCache.java
@@ -141,8 +141,8 @@ public class DataStageCache {
             log.debug("(cache miss) -- building Process for job: {}", rid);
             DataStageJob job = getJobByRid(rid);
             if (job != null) {
-                ProcessMapping processMapping = new ProcessMapping(this, job);
-                process = processMapping.getProcess();
+                ProcessMapping processMapping = new ProcessMapping(this);
+                process = processMapping.getForJob(job);
                 if (process != null) {
                     ridToProcess.put(rid, process);
                 }