Merge pull request #24 in BACIBBD/spline from feature/SL-59-updating-Atlas-persistence-layer to feature/SL-43-new-data-model

* commit '947e742dabbeedc49df5b118beb1d71029e2f590':
  SL-59 Adding comment parameters
  SL-59 Refactoring Atlas persistence layer according to the latest Splline data model.
  SL-59 Renaming AtlasDataLineagePersistor to AtlasDataLineageWriter

Author: wajda

persistence/atlas/src/main/atlas/spline-meta-model.json

@@ -104,6 +104,14 @@
         "Process", "spark_entity"
       ],
       "typeVersion": "1.0",
+      "attributeDefs": []
+    },
+    {
+      "name": "spark_generic_operation",
+      "superTypes": [
+        "Process", "spark_operation"
+      ],
+      "typeVersion": "1.0",
       "attributeDefs": [
         {
           "name": "rawString",
@@ -200,7 +208,7 @@
       "attributeDefs": [
         {
           "name": "attributes",
-          "typeName": "array<spark_dataset_attribute>",
+          "typeName": "array<spark_attribute>",
           "cardinality": "LIST",
           "constraints": [
             {
@@ -260,34 +268,6 @@
         "schemaElementsAttribute": "attributes"
       }
     },
-    {
-      "name": "spark_dataset_attribute",
-      "superTypes": [
-        "spark_attribute"
-      ],
-      "typeVersion": "1.0",
-      "attributeDefs": [
-        {
-          "name": "dataset",
-          "typeName": "spark_dataset",
-          "cardinality": "SINGLE",
-          "constraints": [
-            {
-              "type": "inverseRef",
-              "params": {
-                "attribute": "attributes"
-              }
-            }
-          ],
-          "isIndexable": false,
-          "isOptional": true,
-          "isUnique": false
-        }
-      ],
-      "options": {
-        "schemaAttributes": "[\"name\", \"type\"]"
-      }
-    },
     {
       "name": "spark_data_type",
       "superTypes": [
@@ -467,7 +447,7 @@
       "attributeDefs": [
         {
           "name": "attributeId",
-          "typeName": "long",
+          "typeName": "string",
           "cardinality": "SINGLE",
           "isIndexable": false,
           "isOptional": true,
@@ -480,6 +460,14 @@
           "isIndexable": false,
           "isOptional": false,
           "isUnique": false
+        },
+        {
+          "name": "attribute",
+          "typeName": "spark_attribute",
+          "cardinality": "SINGLE",
+          "isIndexable": false,
+          "isOptional": false,
+          "isUnique": false
         }
       ]
     },

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/conversion/AttributeConverter.scala

@@ -16,46 +16,22 @@
 
 package za.co.absa.spline.persistence.atlas.conversion
 
-import za.co.absa.spline.model.Schema
 import za.co.absa.spline.persistence.atlas.model._
-import za.co.absa.spline.persistence.atlas.model.Dataset
 
 /**
   * The object is responsible for conversion of [[za.co.absa.spline.model.Attribute Spline attributes]] to [[za.co.absa.spline.persistence.atlas.model.Attribute Atlas attributes]].
   */
 object AttributeConverter {
 
-  /**
-    * The method converts an options of [[za.co.absa.spline.model.Schema Spline attributes]] to a sequence [[za.co.absa.spline.persistence.atlas.model.Attribute Atlas attributes]].
- *
-    * @param splineAttributes An Input option of Spline attributes
-    * @param dataset A dataset that the attributes are part of
-    * @return Sequence of Atlas attributes
-    */
-  def convert(splineAttributes : Option[Schema], dataset : Dataset) : Seq[Attribute] = splineAttributes match {
-    case None => Seq.empty
-    case Some(x) => convert(x, dataset)
-  }
-
-  /**
-    * The method converts [[za.co.absa.spline.model.Schema  Spline attributes]] to a sequence [[za.co.absa.spline.persistence.atlas.model.Attribute Atlas attributes]].
- *
-    * @param splineAttributes Input Spline attributes
-    * @param dataset A dataset that the attributes are part of
-    * @return Sequence of Atlas attributes
-    */
-  def convert(splineAttributes: Schema, dataset : Dataset) : Seq[Attribute] = splineAttributes.attrs.map(i => ??? /*convert(i, dataset)*/)
-
   /**
     * The method converts an [[za.co.absa.spline.model.Attribute Spline attribute]] to an [[za.co.absa.spline.persistence.atlas.model.Attribute Atlas attribute]].
     * @param splineAttribute An input Spline attribute
-    * @param dataset A dataset that the attribute is part of
     * @return An Atlas attributes
     */
-  def convert(splineAttribute : za.co.absa.spline.model.Attribute, dataset : Dataset) : Attribute = {
-    val attributeQualifiedName = dataset.qualifiedName + "@" + splineAttribute.name
+  def convert(splineAttribute : za.co.absa.spline.model.Attribute) : Attribute = {
+    val attributeQualifiedName = splineAttribute.id.toString
     val dataType = DataTypeConverter.convert(splineAttribute.dataType, attributeQualifiedName)
-    new Attribute(splineAttribute.name, attributeQualifiedName, dataType, dataset.getId)
+    new Attribute(splineAttribute.name, splineAttribute.id, dataType)
   }
 
 }

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/conversion/DataLineageToTypeSystemConverter.scala

@@ -17,7 +17,7 @@
 package za.co.absa.spline.persistence.atlas.conversion
 
 import org.apache.atlas.typesystem.Referenceable
-import za.co.absa.spline.model._
+import za.co.absa.spline.model.DataLineage
 import za.co.absa.spline.persistence.atlas.model._
 
 /**
@@ -31,15 +31,16 @@ object DataLineageToTypeSystemConverter {
     * @return Atlas entities
     */
   def convert(lineage: DataLineage): Seq[Referenceable] = {
-    val hashSuffix = "_" + ???
-    val nodesWithIndexes = lineage.operations.zipWithIndex
-    val datasets = createDatasets(nodesWithIndexes, hashSuffix)
-    val operations = createOperations(nodesWithIndexes, hashSuffix, datasets)
-    val process = createProcess(lineage, hashSuffix, operations, datasets)
-    datasets ++ operations ++ Seq(process)
+    val attributes = lineage.attributes.map(i => AttributeConverter.convert(i))
+    val attributeIdMap = attributes.map(i => i.qualifiedName -> i.getId).toMap
+    val datasets = DatasetConverter.convert(lineage.operations, lineage.datasets, attributeIdMap)
+    val datasetIdMap = datasets.map(i => i.qualifiedName -> i.getId).toMap
+    val operations = OperationConverter.convert(lineage.operations, datasetIdMap, attributeIdMap)
+    val process = createProcess(lineage, operations, datasets, attributes)
+    attributes ++ datasets ++ operations :+ process
   }
 
-  private def createProcess(lineage: DataLineage, hashSuffix: String, operations : Seq[Operation] , datasets : Seq[Dataset]) : Referenceable = {
+  private def createProcess(lineage: DataLineage, operations : Seq[Operation] , datasets : Seq[Dataset], attributes: Seq[Attribute]) : Referenceable = {
     val (inputDatasets, outputDatasets) = datasets
       .filter(_.isInstanceOf[EndpointDataset])
       .map(_.asInstanceOf[EndpointDataset])
@@ -48,7 +49,7 @@ object DataLineageToTypeSystemConverter {
     new Job(
       lineage.id.toString,
       lineage.appName,
-      lineage.appName + hashSuffix,
+      lineage.id.toString,
       operations.map(_.getId),
       datasets.map(_.getId),
       inputDatasets.map(_.getId),
@@ -57,52 +58,4 @@ object DataLineageToTypeSystemConverter {
       outputDatasets.map(_.endpoint.getId)
     )
   }
-  private def createDatasets(nodesWithIndexes: Seq[Tuple2[op.Operation,Int]], hashSuffix: String) : Seq[Dataset] =
-    nodesWithIndexes.map(i =>
-    {
-      val datasetSuffix = "_Dataset"
-      val name = i._1.mainProps.name + datasetSuffix
-      val operationIdSuffix = "_op" + i._2.toString
-      val qualifiedName = name + operationIdSuffix + hashSuffix
-      val dataset = i._1 match {
-        case op.Source(m, st, paths) =>
-          val path = paths.mkString(", ")
-          new EndpointDataset(name, qualifiedName, new FileEndpoint(path, path), EndpointType.file, EndpointDirection.input, st)
-        case op.Destination(m, dt, path) => new EndpointDataset(name, qualifiedName, new FileEndpoint(path, path), EndpointType.file, EndpointDirection.output, dt)
-        case _ => new Dataset(name, qualifiedName)
-      }
-      ??? //dataset.addAttributes(AttributeConverter.convert(i._1.mainProps.output, dataset))
-      dataset
-    })
-
-  private def createOperations(nodesWithIndexes: Seq[Tuple2[op.Operation,Int]], hashSuffix: String, datasets : Seq[Dataset]) : Seq[Operation] = {
-    val operations = nodesWithIndexes.map(i =>
-        {
-          val operationIdSuffix = "_op" + i._2
-          val commonProperties = OperationCommonProperties(
-            i._2,
-            i._1.mainProps.name,
-            i._1.mainProps.name + operationIdSuffix + hashSuffix,
-            ???,
-            ??? //i._1.mainProps.childRefs
-          )
-          i._1 match {
-            case op.Join(_, c, t) =>  new JoinOperation(commonProperties, t, c.map(j => ExpressionConverter.convert(commonProperties.qualifiedName, j)).get)
-            case op.Filter(_, c) => new FilterOperation(commonProperties, ExpressionConverter.convert(commonProperties.qualifiedName, c))
-            case op.Projection(_, t) => new ProjectOperation(commonProperties, t.zipWithIndex.map(j => ExpressionConverter.convert(commonProperties.qualifiedName + "@" + j._2, j._1)))
-            case op.Alias(_, a) => new AliasOperation(commonProperties, a)
-            case _ => new Operation(commonProperties)
-          }
-        }
-      )
-
-    val datasetIds = datasets.map(_.getId)
-
-    for(o <- operations){
-      o.resolveInputDatasets(datasetIds)
-      o.resolveOutputDatasets(datasetIds)
-    }
-
-    operations
   }
-}

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/conversion/DatasetConverter.scala

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2017 Barclays Africa Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spline.persistence.atlas.conversion
+
+import java.util.UUID
+
+import org.apache.atlas.typesystem.persistence.Id
+import za.co.absa.spline.model.{MetaDataset, op}
+import za.co.absa.spline.persistence.atlas.model._
+
+/**
+  * The object is responsible for conversion of [[za.co.absa.spline.model.MetaDataset Spline meta data sets]] to [[za.co.absa.spline.persistence.atlas.model.Dataset Atlas data sets]].
+  */
+object DatasetConverter {
+  val datasetSuffix = "_Dataset"
+
+  /**
+    * The method converts [[za.co.absa.spline.model.MetaDataset Spline meta data sets]] to [[za.co.absa.spline.persistence.atlas.model.Dataset Atlas data sets]].
+    * @param operations A sequence of [[za.co.absa.spline.model.op.Operation Spline operations]]
+    * @param datasets A sequence of [[za.co.absa.spline.model.MetaDataset Spline meta data sets]]
+    * @param attributeIdMap A map of Spline attribute ids to Atlas ids
+    * @return A sequence of [[za.co.absa.spline.persistence.atlas.model.Dataset Atlas data sets]]
+    */
+  def convert(operations: Seq[op.Operation], datasets : Seq[MetaDataset], attributeIdMap: Map[UUID,Id]) : Seq[Dataset] =
+    for(
+      operation <- operations;
+      dataset <- datasets.withFilter(d => d.id == operation.mainProps.output);
+      name = operation.mainProps.name + datasetSuffix;
+      qualifiedName = dataset.id;
+      attributes = dataset.schema.attrs.map(i => attributeIdMap(i))
+    )
+    yield operation match {
+      case op.Source(m, st, paths) =>
+        val path = paths.mkString(", ")
+        new EndpointDataset(name, qualifiedName, attributes, new FileEndpoint(path, path), EndpointType.file, EndpointDirection.input, st)
+      case op.Destination(m, dt, path) => new EndpointDataset(name, qualifiedName, attributes, new FileEndpoint(path, path), EndpointType.file, EndpointDirection.output, dt)
+      case _ => new Dataset(name, qualifiedName, attributes)
+    }
+}

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/conversion/ExpressionConverter.scala

@@ -16,6 +16,9 @@
 
 package za.co.absa.spline.persistence.atlas.conversion
 
+import java.util.UUID
+
+import org.apache.atlas.typesystem.persistence.Id
 import za.co.absa.spline.model._
 import za.co.absa.spline.persistence.atlas.model._
 
@@ -29,11 +32,12 @@ object ExpressionConverter {
     *
     * @param qualifiedNamePrefix A prefix helping to ensure uniqueness of qualified names of created expressions
     * @param expression          An input Spline expression
+    * @param attributeIdMap A map of Spline attribute ids to Atlas ids
     * @return An Atlas expression
     */
-  def convert(qualifiedNamePrefix: String, expression: expr.Expression): Expression = {
+  def convert(qualifiedNamePrefix: String, expression: expr.Expression, attributeIdMap: Map[UUID, Id]): Expression = {
     val qualifiedName = qualifiedNamePrefix + "@" + expression.text
-    val children = expression.children.zipWithIndex.map(i => convert(qualifiedName + "@" + i._2, i._1))
+    val children = expression.children.zipWithIndex.map(i => convert(qualifiedName + "@" + i._2, i._1, attributeIdMap))
     val mainProperties = ExpressionCommonProperties(
       qualifiedName,
       expression.text,
@@ -44,7 +48,7 @@ object ExpressionConverter {
 
     expression match {
       case expr.Binary(_, symbol, _, _, _) => new BinaryExpression(mainProperties, symbol)
-      case expr.AttributeReference(attributeId, attributeName, _, _) => new AttributeReferenceExpression(mainProperties, attributeId, attributeName)
+      case expr.AttributeReference(attributeId, attributeName, _, _) => new AttributeReferenceExpression(mainProperties, attributeId, attributeName, attributeIdMap(attributeId))
       case expr.UserDefinedFunction(name, _, _, _) => new UserDefinedFunctionExpression(mainProperties, name)
       case _ => new Expression(mainProperties)
     }

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/conversion/OperationConverter.scala

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2017 Barclays Africa Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spline.persistence.atlas.conversion
+
+import java.util.UUID
+
+import org.apache.atlas.typesystem.persistence.Id
+import za.co.absa.spline.model.op
+import za.co.absa.spline.persistence.atlas.model._
+
+/**
+  * The object is responsible for conversion of [[za.co.absa.spline.model.op.Operation Spline operations]] to [[za.co.absa.spline.persistence.atlas.model.Operation Atlas operations]].
+  */
+object OperationConverter {
+
+  /**
+    * The method converts [[za.co.absa.spline.model.op.Operation Spline operations]] to [[za.co.absa.spline.persistence.atlas.model.Operation Atlas operations]].
+    * @param operations A sequence of [[za.co.absa.spline.model.op.Operation Spline operations]]
+    * @param datasetIdMap A map of Spline data set ids to Atlas ids
+    * @param attributeIdMap A map of Spline attribute ids to Atlas ids
+    * @return A sequence of [[za.co.absa.spline.persistence.atlas.model.Operation Atlas operations]]
+    */
+  def convert(operations: Seq[op.Operation], datasetIdMap : Map[UUID, Id], attributeIdMap: Map[UUID, Id]) : Seq[Operation] =
+    operations.map{o =>
+      val commonProperties = OperationCommonProperties(
+        o.mainProps.name,
+        o.mainProps.id.toString,
+        o.mainProps.inputs.map(i => datasetIdMap(i)),
+        Seq(datasetIdMap(o.mainProps.output))
+      )
+      o match {
+        case op.Join(_, c, t) => new JoinOperation(commonProperties, t, c.map(j => ExpressionConverter.convert(commonProperties.qualifiedName, j, attributeIdMap)).get)
+        case op.Filter(_, c) => new FilterOperation(commonProperties, ExpressionConverter.convert(commonProperties.qualifiedName, c, attributeIdMap))
+        case op.Projection(_, t) => new ProjectOperation(commonProperties, t.zipWithIndex.map(j => ExpressionConverter.convert(commonProperties.qualifiedName + "@" + j._2, j._1, attributeIdMap)))
+        case op.Alias(_, a) => new AliasOperation(commonProperties, a)
+        case op.Generic(_, r) => new GenericOperation(commonProperties, r)
+        case _ => new Operation(commonProperties)
+      }
+    }
+}

persistence/atlas/src/main/scala/za/co/absa/spline/persistence/atlas/model/Attribute.scala

@@ -16,6 +16,8 @@
 
 package za.co.absa.spline.persistence.atlas.model
 
+import java.util.UUID
+
 import org.apache.atlas.AtlasClient
 import org.apache.atlas.typesystem.persistence.Id
 import org.apache.atlas.typesystem.{Referenceable, Struct}
@@ -25,15 +27,13 @@ import org.apache.atlas.typesystem.{Referenceable, Struct}
   * @param name A name
   * @param qualifiedName An unique identifier
   * @param dataType A data type
-  * @param dataset A data set that the attribute is part of
   */
-class Attribute(name : String, qualifiedName: String, dataType : DataType, dataset: Id) extends Referenceable(
+class Attribute(val name : String, val qualifiedName: UUID, dataType : DataType) extends Referenceable(
   SparkDataTypes.Attribute,
   new java.util.HashMap[String, Object]{
     put(AtlasClient.NAME, name)
-    put(AtlasClient.REFERENCEABLE_ATTRIBUTE_NAME, qualifiedName)
+    put(AtlasClient.REFERENCEABLE_ATTRIBUTE_NAME, qualifiedName.toString)
     put("type", dataType.name)
     put("typeRef", dataType)
-    put("dataset", dataset)
   }
 )