Reversing resolved inputs and outputs key and velue on NuItaipuFacet. dataset path is now the key and dataset name is now the value

Author: jrosend

integration/spark/shared/src/main/java/io/openlineage/spark/agent/facets/NuItaipuFacet.java

@@ -11,6 +11,7 @@
 import java.util.Map;
 
 import static io.openlineage.spark.agent.util.NuFacetsUtils.getConfigValues;
+import static io.openlineage.spark.agent.util.NuFacetsUtils.keyToValue;
 
 @Getter
 @Slf4j
@@ -27,7 +28,7 @@ public class NuItaipuFacet extends NuRunFacet{
 
     public NuItaipuFacet(@NonNull SparkSession sparkSession) {
         super(Versions.OPEN_LINEAGE_PRODUCER_URI);
-        this.resolvedInputs = getConfigValues(RESOLVED_INPUTS_PREFIX, sparkSession.conf());
-        this.resolvedOutputs = getConfigValues(RESOLVED_OUTPUTS_PREFIX, sparkSession.conf());
+        this.resolvedInputs = keyToValue(getConfigValues(RESOLVED_INPUTS_PREFIX, sparkSession.conf()));
+        this.resolvedOutputs = keyToValue(getConfigValues(RESOLVED_OUTPUTS_PREFIX, sparkSession.conf()));
     }
 }

integration/spark/shared/src/main/java/io/openlineage/spark/agent/util/NuFacetsUtils.java

@@ -40,4 +40,9 @@ public static Map<String, String> getConfigValues(String prefix, RuntimeConfig c
                     return new AbstractMap.SimpleEntry<>(datasetName, datasetPath);
                 }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
+
+    public static Map<String, String> keyToValue(Map<String, String> map) {
+        return map.entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
+    }
 }

integration/spark/shared/src/test/java/io/openlineage/spark/agent/facets/builder/NuFacetBuilderTest.java

@@ -92,16 +92,14 @@ void testBuildItaipuFacet() {
         });
 
         Map<String, String> expectedResolvedInputs = new HashMap<>();
-        expectedResolvedInputs.put("inputName_1", "s3a://path/to/inputName_1");
-        expectedResolvedInputs.put("inputName_2", "s3a://path/to/inputName_2");
-        expectedResolvedInputs.put("inputName_3", "s3a://path/to/inputName_3");
-
+        expectedResolvedInputs.put("s3a://path/to/inputName_1", "inputName_1");
+        expectedResolvedInputs.put("s3a://path/to/inputName_2", "inputName_2");
+        expectedResolvedInputs.put("s3a://path/to/inputName_3", "inputName_3");
 
         Map<String, String> expectedResolvedOutputs = new HashMap<>();
-        expectedResolvedOutputs.put("outputName_1", "s3a://path/to/outputName_1");
-        expectedResolvedOutputs.put("outputName_2", "s3a://path/to/outputName_2");
-        expectedResolvedOutputs.put("outputName_3", "s3a://path/to/outputName_3");
-
+        expectedResolvedOutputs.put("s3a://path/to/outputName_1", "outputName_1");
+        expectedResolvedOutputs.put("s3a://path/to/outputName_2", "outputName_2");
+        expectedResolvedOutputs.put("s3a://path/to/outputName_3", "outputName_3");
 
         NuFacetBuilder builder = new NuFacetBuilder(openLineageContext);
         Map<String, RunFacet> runFacetMap = new HashMap<>();