ATLAS-621 Introduce entity state in Id object (shwethags)

notification/src/test/java/org/apache/atlas/notification/hook/HookNotificationTest.java

@@ -51,15 +51,13 @@ public class HookNotificationTest {
 
     @Test
     public void testBackwardCompatibility() throws Exception {
-        /**
+        //Code to generate the json, use it for hard-coded json used later in this test
         Referenceable entity = new Referenceable("sometype");
         entity.set("attr", "value");
-        String user = "user";
         HookNotification.EntityCreateRequest request = new HookNotification.EntityCreateRequest(null, entity);
 
-        String notificationJson = AbstractNotificationConsumer.GSON.toJson(request);
-        System.out.println(notificationJson);
-         **/
+        String notificationJsonFromCode = AbstractNotificationConsumer.GSON.toJson(request);
+        System.out.println(notificationJsonFromCode);
 
         //Json without user and assert that the string can be deserialised
         String notificationJson = "{\n"
@@ -68,9 +66,10 @@ public class HookNotificationTest {
                 + "      \"jsonClass\": \"org.apache.atlas.typesystem.json.InstanceSerialization$_Reference\",\n"
                 + "      \"id\": {\n"
                 + "        \"jsonClass\": \"org.apache.atlas.typesystem.json.InstanceSerialization$_Id\",\n"
-                + "        \"id\": \"-1457685864305243000\",\n"
+                + "        \"id\": \"-1459493350903186000\",\n"
                 + "        \"version\": 0,\n"
-                + "        \"typeName\": \"sometype\"\n"
+                + "        \"typeName\": \"sometype\",\n"
+                + "        \"state\": \"ACTIVE\"\n"
                 + "      },\n"
                 + "      \"typeName\": \"sometype\",\n"
                 + "      \"values\": {\n"

release-log.txt

@@ -3,6 +3,7 @@ Apache Atlas Release Notes
 
 --trunk - unreleased
 INCOMPATIBLE CHANGES:
+ATLAS-621 Introduce entity state in Id object (shwethags)
 ATLAS-474 Server does not start if the type is updated with same super type class information (dkantor via shwethags)
 ATLAS-479 Add description for different types during create time (guptaneeru via shwethags)
 ATLAS-521 Support Alter Table column commands (suma.shivaprasad via shwethags)

repository/src/main/java/org/apache/atlas/repository/Constants.java

@@ -57,6 +57,8 @@ public final class Constants {
     public static final String TRAIT_NAMES_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "traitNames";
 
     public static final String VERSION_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "version";
+    public static final String STATE_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "state";
+
     public static final String TIMESTAMP_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "timestamp";
     public static final String MODIFICATION_TIMESTAMP_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "modificationTimestamp";
 

repository/src/main/java/org/apache/atlas/repository/DiscoverInstances.java

@@ -60,7 +60,7 @@ public class DiscoverInstances implements ObjectGraphWalker.NodeProcessor {
         if (id != null) {
             if (id.isUnassigned()) {
                 if (!idToNewIdMap.containsKey(id)) {
-                    idToNewIdMap.put(id, repository.newId(id.className));
+                    idToNewIdMap.put(id, repository.newId(id.typeName));
                 }
                 if (ref != null && idToInstanceMap.containsKey(ref)) {
                     // Oops

repository/src/main/java/org/apache/atlas/repository/graph/GraphHelper.java

@@ -85,6 +85,9 @@ public final class GraphHelper {
         // add version information
         setProperty(vertexWithIdentity, Constants.VERSION_PROPERTY_KEY, typedInstance.getId().version);
 
+        // add state information
+        setProperty(vertexWithIdentity, Constants.STATE_PROPERTY_KEY, Id.EntityState.ACTIVE.name());
+
         return vertexWithIdentity;
     }
 

repository/src/main/java/org/apache/atlas/repository/graph/GraphToTypedInstanceMapper.java

@@ -64,7 +64,8 @@ public final class GraphToTypedInstanceMapper {
         String typeName = instanceVertex.getProperty(Constants.ENTITY_TYPE_PROPERTY_KEY);
         List<String> traits = GraphHelper.getTraitNames(instanceVertex);
 
-        Id id = new Id(guid, instanceVertex.<Integer>getProperty(Constants.VERSION_PROPERTY_KEY), typeName);
+        Id id = new Id(guid, instanceVertex.<Integer>getProperty(Constants.VERSION_PROPERTY_KEY), typeName,
+                instanceVertex.<String>getProperty(Constants.STATE_PROPERTY_KEY));
         LOG.debug("Created id {} for instance type {}", id, typeName);
 
         ClassType classType = typeSystem.getDataType(ClassType.class, typeName);

repository/src/test/java/org/apache/atlas/discovery/GraphBackedDiscoveryServiceTest.java

@@ -87,7 +87,7 @@ public class GraphBackedDiscoveryServiceTest extends BaseHiveRepositoryTest {
     }
 
     @Test
-    public void testSearchByDSL() throws Exception {
+    public void testSearchByDSLReturnsEntity() throws Exception {
         String dslQuery = "from Department";
 
         String jsonResults = discoveryService.searchByDSL(dslQuery);
@@ -109,6 +109,10 @@ public class GraphBackedDiscoveryServiceTest extends BaseHiveRepositoryTest {
         JSONArray rows = results.getJSONArray("rows");
         Assert.assertNotNull(rows);
         Assert.assertEquals(rows.length(), 1);
+
+        //Assert that entity state is set in the result entities
+        String entityState = rows.getJSONObject(0).getJSONObject("$id$").getString("state");
+        Assert.assertEquals(entityState, Id.EntityState.ACTIVE.name());
     }
 
     @Test(expectedExceptions = Throwable.class)

repository/src/test/java/org/apache/atlas/repository/graph/GraphBackedMetadataRepositoryTest.java

@@ -134,6 +134,9 @@ public class GraphBackedMetadataRepositoryTest {
     public void testGetEntityDefinitionForDepartment() throws Exception {
         ITypedReferenceableInstance entity = repositoryService.getEntityDefinition(guid);
         Assert.assertNotNull(entity);
+
+        //entity state should be active by default
+        Assert.assertEquals(entity.getId().getState(), Id.EntityState.ACTIVE);
     }
 
     @Test(expectedExceptions = EntityNotFoundException.class)

typesystem/src/main/java/org/apache/atlas/typesystem/Referenceable.java

@@ -78,6 +78,21 @@ public class Referenceable extends Struct implements IReferenceableInstance {
     }
 
     /**
+     * Not public - only use during deserialization
+     * @param id      entity id
+     * @param typeName  the type name
+     * @param values    the entity attribute values
+     */
+    @InterfaceAudience.Private
+    public Referenceable(Id id, String typeName, Map<String, Object> values, List<String> _traitNames,
+                         Map<String, IStruct> _traits) {
+        super(typeName, values);
+        this.id = id;
+        traitNames = ImmutableList.copyOf(_traitNames);
+        traits = ImmutableMap.copyOf(_traits);
+    }
+
+    /**
      * Construct a Referenceable from the given IReferenceableInstance.
      *
      * @param instance  the referenceable instance to copy

typesystem/src/main/java/org/apache/atlas/typesystem/persistence/Id.java

@@ -35,23 +35,41 @@ import java.util.Map;
 import java.util.UUID;
 
 public class Id implements ITypedReferenceableInstance {
+    public enum EntityState {
+        ACTIVE, DELETED
+    }
 
     public final String id;
-    public final String className;
+    public final String typeName;
     public final int version;
+    public EntityState state;
 
-    public Id(String id, int version, String className) {
-        ParamChecker.notEmpty(className, "id");
-        ParamChecker.notEmpty(className, "className");
+    public Id(String id, int version, String typeName, String state) {
+        ParamChecker.notEmpty(id, "id");
+        ParamChecker.notEmpty(typeName, "typeName");
+        ParamChecker.notEmptyIfNotNull(state, "state");
         this.id = id;
-        this.className = className;
+        this.typeName = typeName;
         this.version = version;
+        if (state == null) {
+            this.state = EntityState.ACTIVE;
+        } else {
+            this.state = EntityState.valueOf(state.toUpperCase());
+        }
+    }
+
+    public Id(String id, int version, String className) {
+        this(id, version, className, null);
     }
 
     public Id(long id, int version, String className) {
         this("" + id, version, className);
     }
 
+    public Id(long id, int version, String className, String state) {
+        this("" + id, version, className, state);
+    }
+
     public Id(String className) {
         this("" + (-System.nanoTime()), 0, className);
     }
@@ -76,11 +94,11 @@ public class Id implements ITypedReferenceableInstance {
     }
 
     public String toString() {
-        return String.format("(type: %s, id: %s)", className, isUnassigned() ? "<unassigned>" : "" + id);
+        return String.format("(type: %s, id: %s)", typeName, isUnassigned() ? "<unassigned>" : "" + id);
     }
 
     public String getClassName() {
-        return className;
+        return typeName;
     }
 
     public int getVersion() {
@@ -91,6 +109,14 @@ public class Id implements ITypedReferenceableInstance {
         return id;
     }
 
+    public EntityState getState() {
+        return state;
+    }
+
+    public String getStateAsString() {
+        return state == null ? null : state.name();
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -105,7 +131,7 @@ public class Id implements ITypedReferenceableInstance {
         if (version != id1.version) {
             return false;
         }
-        if (!className.equals(id1.className)) {
+        if (!typeName.equals(id1.typeName)) {
             return false;
         }
         if (!id.equals(id1.id)) {
@@ -118,7 +144,7 @@ public class Id implements ITypedReferenceableInstance {
     @Override
     public int hashCode() {
         int result = id.hashCode();
-        result = 31 * result + className.hashCode();
+        result = 31 * result + typeName.hashCode();
         result = 31 * result + version;
         return result;
     }
@@ -140,7 +166,7 @@ public class Id implements ITypedReferenceableInstance {
 
     @Override
     public String getTypeName() {
-        return className;
+        return typeName;
     }
 
     @Override
@@ -258,7 +284,7 @@ public class Id implements ITypedReferenceableInstance {
     @Override
     public String getSignatureHash(MessageDigest digester) throws AtlasException {
         digester.update(id.getBytes(Charset.forName("UTF-8")));
-        digester.update(className.getBytes(Charset.forName("UTF-8")));
+        digester.update(typeName.getBytes(Charset.forName("UTF-8")));
         byte[] digest = digester.digest();
         return MD5Utils.toString(digest);
     }

typesystem/src/main/java/org/apache/atlas/typesystem/types/ClassType.java

@@ -67,7 +67,7 @@ public class ClassType extends HierarchicalType<ClassType, IReferenceableInstanc
 
     public void validateId(Id id) throws AtlasException {
         if (id != null) {
-            ClassType cType = typeSystem.getDataType(ClassType.class, id.className);
+            ClassType cType = typeSystem.getDataType(ClassType.class, id.typeName);
             if (isSubType(cType.getName())) {
                 return;
             }
@@ -150,7 +150,7 @@ public class ClassType extends HierarchicalType<ClassType, IReferenceableInstanc
                 validateId(((ReferenceableInstance) val).getId());
                 return (ReferenceableInstance) val;
             } else {
-                throw new ValueConversionException(this, val);
+                throw new ValueConversionException(this, val, "value's class is " + val.getClass().getName());
             }
         }
         if (!m.nullAllowed()) {

typesystem/src/main/scala/org/apache/atlas/typesystem/json/InstanceSerialization.scala

@@ -31,7 +31,7 @@ import scala.collection.JavaConverters._
 
 object InstanceSerialization {
 
-  case class _Id(id : String, version : Int, typeName : String)
+  case class _Id(id : String, version : Int, typeName : String, state : String)
   case class _Struct(typeName : String, values : Map[String, AnyRef])
   case class _Reference(id : _Id,
                         typeName : String,
@@ -72,6 +72,14 @@ object InstanceSerialization {
     }
 
     /**
+     * validate and extract 'state' attribute from Map
+     * @return
+     */
+    def state: Option[String] = {
+      jsonMap.get("state").filter(_.isInstanceOf[String]).flatMap(v => Some(v.asInstanceOf[String]))
+    }
+
+    /**
      * validate and extract 'version' attribute from Map
      * @return
      */
@@ -93,11 +101,12 @@ object InstanceSerialization {
      */
     def convertId : Option[_Id] = {
       for {
-        refClass <- idClass;
-        typNm <- typeName;
-        i <- id;
+        refClass <- idClass
+        typNm <- typeName
+        i <- id
+        s <- state
         v <- version
-      } yield _Id(i, v, typNm)
+      } yield _Id(i, v, typNm, s)
     }
 
     /**
@@ -151,8 +160,8 @@ object InstanceSerialization {
      */
     def struct: Option[_Struct] = {
       for {
-        refClass <- structureClass;
-        typNm <- typeName;
+        refClass <- structureClass
+        typNm <- typeName
         values <- valuesMap
       } yield _Struct(typNm, values)
     }
@@ -218,11 +227,11 @@ object InstanceSerialization {
      */
     def reference : Option[_Reference] = {
       for {
-        refClass <- referenceClass;
-        typNm <- typeName;
-        i <- idObject;
-        values <- valuesMap;
-        traitNms <- traitNames;
+        refClass <- referenceClass
+        typNm <- typeName
+        i <- idObject
+        values <- valuesMap
+        traitNms <- traitNames
         ts <- traits
       } yield _Reference(i, typNm, values, traitNms.toList, ts)
     }
@@ -249,10 +258,10 @@ object InstanceSerialization {
   }
 
   def asJava(v : Any)(implicit format: Formats) : Any = v match {
-    case i : _Id => new Id(i.id, i.version, i.typeName)
+    case i : _Id => new Id(i.id, i.version, i.typeName, i.state)
     case s : _Struct => new Struct(s.typeName, asJava(s.values).asInstanceOf[java.util.Map[String, Object]])
     case r : _Reference => {
-      new Referenceable(r.id.asInstanceOf[_Id].id,
+      new Referenceable(new Id(r.id.id, r.id.version, r.id.typeName, r.id.state),
         r.typeName,
         asJava(r.values).asInstanceOf[java.util.Map[String, Object]],
         asJava(r.traitNames).asInstanceOf[java.util.List[String]],
@@ -271,7 +280,7 @@ object InstanceSerialization {
   }
 
   def asScala(v : Any) : Any = v match {
-    case i : Id => _Id(i._getId(), i.getVersion, i.getClassName)
+    case i : Id => _Id(i._getId(), i.getVersion, i.getClassName, i.getStateAsString)
     case r : IReferenceableInstance => {
       val traits = r.getTraits.map { tName =>
         val t = r.getTrait(tName).asInstanceOf[IStruct]

typesystem/src/main/scala/org/apache/atlas/typesystem/json/Serialization.scala

@@ -49,20 +49,20 @@ class BigIntegerSerializer extends CustomSerializer[java.math.BigInteger](format
 
 class IdSerializer extends CustomSerializer[Id](format => ( {
     case JObject(JField("id", JInt(id)) ::
-        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-        JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
-    case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
+        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+        JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, typeName)
+    case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
         JField("id", JInt(id)) ::
-        JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
+        JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, typeName)
     case JObject(JField("id", JString(id)) ::
-        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-        JField("version", JString(version)) :: Nil) => new Id(id, version.toInt, className)
-    case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
+        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+        JField("version", JString(version)) :: Nil) => new Id(id, version.toInt, typeName)
+    case JObject(JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
         JField("id", JString(id)) ::
-        JField("version", JString(version)) :: Nil) => new Id(id, version.toInt, className)
+        JField("version", JString(version)) :: Nil) => new Id(id, version.toInt, typeName)
 }, {
     case id: Id => JObject(JField("id", JString(id.id)),
-        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(id.className)),
+        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(id.typeName)),
         JField("version", JInt(id.version)))
 }
     ))
@@ -106,11 +106,13 @@ class TypedReferenceableInstanceSerializer()
     def deserialize(implicit format: Formats) = {
         case (TypeInfo(clazz, ptype), json) if classOf[ITypedReferenceableInstance].isAssignableFrom(clazz) => json match {
             case JObject(JField("id", JInt(id)) ::
-                JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-                JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
+                JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+                JField("version", JInt(version)) ::
+                JField("state", JString(state)) :: Nil) => new Id(id.toLong, version.toInt, typeName, state)
             case JObject(JField("id", JString(id)) ::
-                JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-                JField("version", JInt(version)) :: Nil) => new Id(id, version.toInt, className)
+                JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+                JField("version", JInt(version)) ::
+                JField("state", JString(state)) :: Nil) => new Id(id, version.toInt, typeName, state)
             case JObject(fs) =>
                 var typField: Option[JField] = None
                 var idField: Option[JField] = None
@@ -213,8 +215,8 @@ object Serialization {
     }
 
     def serializeId(id: Id) = JObject(JField("id", JString(id.id)),
-        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(id.className)),
-        JField("version", JInt(id.version)))
+        JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(id.typeName)),
+        JField("version", JInt(id.version)), JField("state", JString(id.state.name())))
 
     def serializeFields(e: ITypedInstance)(implicit format: Formats) = e.fieldMapping.fields.map {
         case (fName, info) => {
@@ -261,11 +263,13 @@ object Serialization {
 
     def deserializeId(value: JValue)(implicit format: Formats) = value match {
         case JObject(JField("id", JInt(id)) ::
-            JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-            JField("version", JInt(version)) :: Nil) => new Id(id.toLong, version.toInt, className)
+            JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+            JField("version", JInt(version)) ::
+            JField("state", JString(state)) :: Nil) => new Id(id.toLong, version.toInt, typeName, state)
         case JObject(JField("id", JString(id)) ::
-            JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(className)) ::
-            JField("version", JInt(version)) :: Nil) => new Id(id, version.toInt, className)
+            JField(Serialization.STRUCT_TYPE_FIELD_NAME, JString(typeName)) ::
+            JField("version", JInt(version)) ::
+            JField("state", JString(state)) :: Nil) => new Id(id, version.toInt, typeName, state)
     }
 
     def toJson(value: ITypedReferenceableInstance): String = {

typesystem/src/test/java/org/apache/atlas/typesystem/types/BaseTest.java

@@ -40,7 +40,6 @@ public abstract class BaseTest {
     public static final String STRUCT_TYPE_1 = "t1";
     public static final String STRUCT_TYPE_2 = "t2";
     public static final String TEST_DATE = "2014-12-11T02:35:58.440Z";
-    public static final long TEST_DATE_IN_LONG = 1418265358440L;
 
     public static Struct createStruct() throws AtlasException {
         StructType structType = TypeSystem.getInstance().getDataType(StructType.class, STRUCT_TYPE_1);

typesystem/src/test/scala/org/apache/atlas/typesystem/json/SerializationTest.scala

@@ -19,6 +19,7 @@
 package org.apache.atlas.typesystem.json
 
 import com.google.common.collect.ImmutableList
+import org.apache.atlas.typesystem.persistence.Id.EntityState
 import org.apache.atlas.typesystem.persistence.{Id, ReferenceableInstance, StructInstance}
 import org.apache.atlas.typesystem.types._
 import org.apache.atlas.typesystem.types.utils.TypesUtil
@@ -29,6 +30,7 @@ import org.json4s.{NoTypeHints, _}
 import org.testng.Assert
 import org.testng.annotations.{BeforeMethod,Test}
 import com.google.common.collect.ImmutableSet
+import org.testng.Assert.assertEquals
 
 class SerializationTest extends BaseTest {
 
@@ -241,4 +243,21 @@ class SerializationTest extends BaseTest {
 
   }
 
+  @Test def testIdSerde: Unit = {
+
+    val ts: TypeSystem = getTypeSystem
+    defineHRTypes(ts)
+    val hrDept: Referenceable = defineHRDept()
+    //default state is actiev by default
+    assertEquals(hrDept.getId.getState, EntityState.ACTIVE)
+
+    val deptType: ClassType = ts.getDataType(classOf[ClassType], "Department")
+    val hrDept2: ITypedReferenceableInstance = deptType.convert(hrDept, Multiplicity.REQUIRED)
+    hrDept2.getId.state = EntityState.DELETED
+
+    //updated state should be maintained correctly after serialisation-deserialisation
+    val deptJson: String = InstanceSerialization.toJson(hrDept2, true)
+    val deserDept: Referenceable = InstanceSerialization.fromJsonReferenceable(deptJson, true)
+    assertEquals(deserDept.getId.getState, EntityState.DELETED)
+  }
 }
\ No newline at end of file