repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasRelationshipDefStoreV1.java - incubator-atlas - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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 org.apache.atlas.repository.store.graph.v1;

 import org.apache.atlas.ApplicationProperties;
 import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.AtlasException;
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.typedef.AtlasRelationshipDef;
 import org.apache.atlas.model.typedef.AtlasRelationshipDef.RelationshipCategory;
 import org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags;
 import org.apache.atlas.model.typedef.AtlasRelationshipEndDef;
 import org.apache.atlas.query.QueryParser;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.graphdb.AtlasEdge;
 import org.apache.atlas.repository.graphdb.AtlasVertex;
 import org.apache.atlas.repository.store.graph.AtlasRelationshipDefStore;
 import org.apache.atlas.type.AtlasRelationshipType;
 import org.apache.atlas.type.AtlasType;
 import org.apache.atlas.type.AtlasTypeRegistry;
 import org.apache.atlas.typesystem.types.DataTypes.TypeCategory;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import javax.inject.Inject;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;

 /**
  * RelationshipDef store in v1 format.
  */
 public class AtlasRelationshipDefStoreV1 extends AtlasAbstractDefStoreV1 implements AtlasRelationshipDefStore {
     private static final Logger LOG = LoggerFactory.getLogger(AtlasRelationshipDefStoreV1.class);

     @Inject
     public AtlasRelationshipDefStoreV1(AtlasTypeDefGraphStoreV1 typeDefStore, AtlasTypeRegistry typeRegistry) {
         super(typeDefStore, typeRegistry);
     }

     @Override
     public AtlasVertex preCreate(AtlasRelationshipDef relationshipDef) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.preCreate({})", relationshipDef);
         }

         validateType(relationshipDef);

         AtlasType type = typeRegistry.getType(relationshipDef.getName());

         if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
         }

         AtlasVertex relationshipDefVertex = typeDefStore.findTypeVertexByName(relationshipDef.getName());

         if (relationshipDefVertex != null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_ALREADY_EXISTS, relationshipDef.getName());
         }

         relationshipDefVertex = typeDefStore.createTypeVertex(relationshipDef);

         updateVertexPreCreate(relationshipDef, (AtlasRelationshipType) type, relationshipDefVertex);

         final AtlasRelationshipEndDef endDef1        = relationshipDef.getEndDef1();
         final AtlasRelationshipEndDef endDef2        = relationshipDef.getEndDef2();
         final String                  type1          = endDef1.getType();
         final String                  type2          = endDef2.getType();
         final String                  name1          = endDef1.getName();
         final String                  name2          = endDef2.getName();
         final AtlasVertex             end1TypeVertex = typeDefStore.findTypeVertexByName(type1);
         final AtlasVertex             end2TypeVertex = typeDefStore.findTypeVertexByName(type2);

         if (end1TypeVertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND, relationshipDef.getName(), type1);
         }

         if (end2TypeVertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND, relationshipDef.getName(), type2);
         }

         // create an edge between the relationshipDef and each of the entityDef vertices.
         AtlasEdge edge1 = typeDefStore.getOrCreateEdge(relationshipDefVertex, end1TypeVertex, AtlasGraphUtilsV1.RELATIONSHIPTYPE_EDGE_LABEL);

         /*
         Where edge1 and edge2 have the same names and types we do not need a second edge.
         We are not invoking the equals method on the AtlasRelationshipedDef, as we only want 1 edge even if propagateTags or other properties are different.
         */

         if (type1.equals(type2) && name1.equals(name2)) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("AtlasRelationshipDefStoreV1.preCreate({}): created relationshipDef vertex {}," +
                         " and one edge as {}, because end1 and end2 have the same type and name", relationshipDef, relationshipDefVertex, edge1);
             }

         } else {
             AtlasEdge edge2 = typeDefStore.getOrCreateEdge(relationshipDefVertex, end2TypeVertex, AtlasGraphUtilsV1.RELATIONSHIPTYPE_EDGE_LABEL);
             if (LOG.isDebugEnabled()) {
                 LOG.debug("AtlasRelationshipDefStoreV1.preCreate({}): created relationshipDef vertex {}," +
                         " edge1 as {}, edge2 as {} ", relationshipDef, relationshipDefVertex, edge1, edge2);
             }

         }
         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.preCreate({}): {}", relationshipDef, relationshipDefVertex);
         }
         return relationshipDefVertex;
     }

     @Override
     public AtlasRelationshipDef create(AtlasRelationshipDef relationshipDef, Object preCreateResult)
             throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.create({}, {})", relationshipDef, preCreateResult);
         }

         AtlasVertex vertex;

         if (preCreateResult == null || !(preCreateResult instanceof AtlasVertex)) {
             vertex = preCreate(relationshipDef);
         } else {
             vertex = (AtlasVertex) preCreateResult;
         }

         AtlasRelationshipDef ret = toRelationshipDef(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.create({}, {}): {}", relationshipDef, preCreateResult, ret);
         }

         return ret;
     }

     @Override
     public List<AtlasRelationshipDef> getAll() throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.getAll()");
         }

         List<AtlasRelationshipDef> ret = new ArrayList<>();
         Iterator<AtlasVertex> vertices = typeDefStore.findTypeVerticesByCategory(TypeCategory.RELATIONSHIP);

         while (vertices.hasNext()) {
             ret.add(toRelationshipDef(vertices.next()));
         }

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.getAll(): count={}", ret.size());
         }

         return ret;
     }

     @Override
     public AtlasRelationshipDef getByName(String name) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.getByName({})", name);
         }

         AtlasVertex vertex = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

         if (vertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
         }

         vertex.getProperty(Constants.TYPE_CATEGORY_PROPERTY_KEY, TypeCategory.class);

         AtlasRelationshipDef ret = toRelationshipDef(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.getByName({}): {}", name, ret);
         }

         return ret;
     }

     @Override
     public AtlasRelationshipDef getByGuid(String guid) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.getByGuid({})", guid);
         }

         AtlasVertex vertex = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

         if (vertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
         }

         AtlasRelationshipDef ret = toRelationshipDef(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.getByGuid({}): {}", guid, ret);
         }

         return ret;
     }

     @Override
     public AtlasRelationshipDef update(AtlasRelationshipDef relationshipDef) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.update({})", relationshipDef);
         }

         validateType(relationshipDef);

         AtlasRelationshipDef ret = StringUtils.isNotBlank(relationshipDef.getGuid())
                 ? updateByGuid(relationshipDef.getGuid(), relationshipDef)
                 : updateByName(relationshipDef.getName(), relationshipDef);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.update({}): {}", relationshipDef, ret);
         }

         return ret;
     }

     @Override
     public AtlasRelationshipDef updateByName(String name, AtlasRelationshipDef relationshipDef)
             throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.updateByName({}, {})", name, relationshipDef);
         }

         validateType(relationshipDef);

         AtlasType type = typeRegistry.getType(relationshipDef.getName());

         if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
         }

         AtlasVertex vertex = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

         if (vertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
         }

         preUpdateCheck(relationshipDef, (AtlasRelationshipType) type, vertex);

         AtlasRelationshipDef ret = toRelationshipDef(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.updateByName({}, {}): {}", name, relationshipDef, ret);
         }

         return ret;
     }

     @Override
     public AtlasRelationshipDef updateByGuid(String guid, AtlasRelationshipDef relationshipDef)
             throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.updateByGuid({})", guid);
         }

         validateType(relationshipDef);

         AtlasType type = typeRegistry.getTypeByGuid(guid);

         if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
         }

         AtlasVertex vertex = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

         if (vertex == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
         }

         preUpdateCheck(relationshipDef, (AtlasRelationshipType) type, vertex);
         // updates should not effect the edges between the types as we do not allow updates that change the endpoints.

         AtlasRelationshipDef ret = toRelationshipDef(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.updateByGuid({}): {}", guid, ret);
         }

         return ret;
     }

     @Override
     public AtlasVertex preDeleteByName(String name) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.preDeleteByName({})", name);
         }

         AtlasVertex ret = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

         if (ret == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
         }

         if (AtlasGraphUtilsV1.relationshipTypeHasInstanceEdges(name)) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_HAS_REFERENCES, name);
         }

         typeDefStore.deleteTypeVertexOutEdges(ret);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.preDeleteByName({}): {}", name, ret);
         }

         return ret;
     }

     @Override
     public void deleteByName(String name, Object preDeleteResult) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.deleteByName({}, {})", name, preDeleteResult);
         }

         AtlasVertex vertex;

         if (preDeleteResult == null || !(preDeleteResult instanceof AtlasVertex)) {
             vertex = preDeleteByName(name);
         } else {
             vertex = (AtlasVertex) preDeleteResult;
         }

         typeDefStore.deleteTypeVertex(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.deleteByName({}, {})", name, preDeleteResult);
         }
     }

     @Override
     public AtlasVertex preDeleteByGuid(String guid) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.preDeleteByGuid({})", guid);
         }

         AtlasVertex ret = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

         if (ret == null) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
         }

         String typeName = AtlasGraphUtilsV1.getProperty(ret, Constants.TYPENAME_PROPERTY_KEY, String.class);

         if (AtlasGraphUtilsV1.relationshipTypeHasInstanceEdges(typeName)) {
             throw new AtlasBaseException(AtlasErrorCode.TYPE_HAS_REFERENCES, typeName);
         }

         typeDefStore.deleteTypeVertexOutEdges(ret);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.preDeleteByGuid({}): {}", guid, ret);
         }

         return ret;
     }

     @Override
     public void deleteByGuid(String guid, Object preDeleteResult) throws AtlasBaseException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> AtlasRelationshipDefStoreV1.deleteByGuid({}, {})", guid, preDeleteResult);
         }

         AtlasVertex vertex;

         if (preDeleteResult == null || !(preDeleteResult instanceof AtlasVertex)) {
             vertex = preDeleteByGuid(guid);
         } else {
             vertex = (AtlasVertex) preDeleteResult;
         }

         typeDefStore.deleteTypeVertex(vertex);

         if (LOG.isDebugEnabled()) {
             LOG.debug("<== AtlasRelationshipDefStoreV1.deleteByGuid({}, {})", guid, preDeleteResult);
         }
     }

     private void updateVertexPreCreate(AtlasRelationshipDef relationshipDef, AtlasRelationshipType relationshipType,
                                        AtlasVertex vertex) throws AtlasBaseException {
         AtlasRelationshipEndDef end1 = relationshipDef.getEndDef1();
         AtlasRelationshipEndDef end2 = relationshipDef.getEndDef2();

         // check whether the names added on the relationship Ends are reserved if required.
         final boolean allowReservedKeywords;
         try {
             allowReservedKeywords = ApplicationProperties.get().getBoolean(ALLOW_RESERVED_KEYWORDS, true);
         } catch (AtlasException e) {
             throw new AtlasBaseException(e);
         }

         if (!allowReservedKeywords) {
             if (QueryParser.isKeyword(end1.getName())) {
                 throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END1_NAME_INVALID, end1.getName());
             }

             if (QueryParser.isKeyword(end2.getName())) {
                 throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END2_NAME_INVALID, end2.getName());
             }
         }

         AtlasStructDefStoreV1.updateVertexPreCreate(relationshipDef, relationshipType, vertex, typeDefStore);
         // Update ends
         setVertexPropertiesFromRelationshipDef(relationshipDef, vertex);
     }

     private void preUpdateCheck(AtlasRelationshipDef newRelationshipDef, AtlasRelationshipType relationshipType,
                                 AtlasVertex vertex) throws AtlasBaseException {
         // We will not support an update to endpoints or category key
         AtlasRelationshipDef existingRelationshipDef = toRelationshipDef(vertex);

         preUpdateCheck(newRelationshipDef, existingRelationshipDef);
         // we do allow change to tag propagation and the addition of new attributes.

         AtlasStructDefStoreV1.updateVertexPreUpdate(newRelationshipDef, relationshipType, vertex, typeDefStore);

         setVertexPropertiesFromRelationshipDef(newRelationshipDef, vertex);
     }

     /**
      * Check ends are the same and relationshipCategory is the same.
      *
      * We do this by comparing 2 relationshipDefs to avoid exposing the AtlasVertex to unit testing.
      *
      * @param newRelationshipDef
      * @param existingRelationshipDef
      * @throws AtlasBaseException
      */
     public static void preUpdateCheck(AtlasRelationshipDef newRelationshipDef, AtlasRelationshipDef existingRelationshipDef) throws AtlasBaseException {
         // do not allow renames of the Def.
         String existingName = existingRelationshipDef.getName();
         String newName      = newRelationshipDef.getName();

         if (!existingName.equals(newName)) {
             throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_NAME_UPDATE,
                     newRelationshipDef.getGuid(),existingName, newName);
         }

         RelationshipCategory existingRelationshipCategory = existingRelationshipDef.getRelationshipCategory();
         RelationshipCategory newRelationshipCategory      = newRelationshipDef.getRelationshipCategory();

         if ( !existingRelationshipCategory.equals(newRelationshipCategory)){
             throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_CATEGORY_UPDATE,
                     newRelationshipDef.getName(),newRelationshipCategory.name(),
                     existingRelationshipCategory.name() );
         }

         AtlasRelationshipEndDef existingEnd1 = existingRelationshipDef.getEndDef1();
         AtlasRelationshipEndDef newEnd1      = newRelationshipDef.getEndDef1();

         if ( !newEnd1.equals(existingEnd1) ) {
             throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_END1_UPDATE,
                                          newRelationshipDef.getName(), newEnd1.toString(), existingEnd1.toString());
         }

         AtlasRelationshipEndDef existingEnd2 = existingRelationshipDef.getEndDef2();
         AtlasRelationshipEndDef newEnd2      = newRelationshipDef.getEndDef2();

         if ( !newEnd2.equals(existingEnd2) ) {
                 throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_END2_UPDATE,
                                          newRelationshipDef.getName(), newEnd2.toString(), existingEnd2.toString());
         }
     }

     public static void setVertexPropertiesFromRelationshipDef(AtlasRelationshipDef relationshipDef, AtlasVertex vertex) {
         vertex.setProperty(Constants.RELATIONSHIPTYPE_END1_KEY, AtlasType.toJson(relationshipDef.getEndDef1()));
         vertex.setProperty(Constants.RELATIONSHIPTYPE_END2_KEY, AtlasType.toJson(relationshipDef.getEndDef2()));
         // Update RelationshipCategory
         vertex.setProperty(Constants.RELATIONSHIPTYPE_CATEGORY_KEY, relationshipDef.getRelationshipCategory().name());

         if (relationshipDef.getPropagateTags() == null) {
             vertex.setProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, AtlasRelationshipDef.PropagateTags.NONE.name());
         } else {
             vertex.setProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, relationshipDef.getPropagateTags().name());
         }
     }

     private AtlasRelationshipDef toRelationshipDef(AtlasVertex vertex) throws AtlasBaseException {
         AtlasRelationshipDef ret = null;

         if (vertex != null && typeDefStore.isTypeVertex(vertex, TypeCategory.RELATIONSHIP)) {
             String name         = vertex.getProperty(Constants.TYPENAME_PROPERTY_KEY, String.class);
             String description  = vertex.getProperty(Constants.TYPEDESCRIPTION_PROPERTY_KEY, String.class);
             String version      = vertex.getProperty(Constants.TYPEVERSION_PROPERTY_KEY, String.class);
             String end1Str = vertex.getProperty(Constants.RELATIONSHIPTYPE_END1_KEY, String.class);
             String end2Str = vertex.getProperty(Constants.RELATIONSHIPTYPE_END2_KEY, String.class);
             String relationStr  = vertex.getProperty(Constants.RELATIONSHIPTYPE_CATEGORY_KEY, String.class);
             String propagateStr = vertex.getProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, String.class);

             // set the ends
             AtlasRelationshipEndDef endDef1 = AtlasType.fromJson(end1Str, AtlasRelationshipEndDef.class);
             AtlasRelationshipEndDef endDef2 = AtlasType.fromJson(end2Str, AtlasRelationshipEndDef.class);

             // set the relationship Category
             RelationshipCategory relationshipCategory = null;
             for (RelationshipCategory value : RelationshipCategory.values()) {
                 if (value.name().equals(relationStr)) {
                     relationshipCategory = value;
                 }
             }

             // set the propagateTags
             PropagateTags propagateTags = null;
             for (PropagateTags value : PropagateTags.values()) {
                 if (value.name().equals(propagateStr)) {
                     propagateTags = value;
                 }
             }

             ret = new AtlasRelationshipDef(name, description, version, relationshipCategory,  propagateTags, endDef1, endDef2);

             // add in the attributes
             AtlasStructDefStoreV1.toStructDef(vertex, ret, typeDefStore);
         }

         return ret;
     }

 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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 org.apache.atlas.repository.store.graph.v1;

	import org.apache.atlas.ApplicationProperties;
	import org.apache.atlas.AtlasErrorCode;
	import org.apache.atlas.AtlasException;
	import org.apache.atlas.exception.AtlasBaseException;
	import org.apache.atlas.model.typedef.AtlasRelationshipDef;
	import org.apache.atlas.model.typedef.AtlasRelationshipDef.RelationshipCategory;
	import org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags;
	import org.apache.atlas.model.typedef.AtlasRelationshipEndDef;
	import org.apache.atlas.query.QueryParser;
	import org.apache.atlas.repository.Constants;
	import org.apache.atlas.repository.graphdb.AtlasEdge;
	import org.apache.atlas.repository.graphdb.AtlasVertex;
	import org.apache.atlas.repository.store.graph.AtlasRelationshipDefStore;
	import org.apache.atlas.type.AtlasRelationshipType;
	import org.apache.atlas.type.AtlasType;
	import org.apache.atlas.type.AtlasTypeRegistry;
	import org.apache.atlas.typesystem.types.DataTypes.TypeCategory;
	import org.apache.commons.lang.StringUtils;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import javax.inject.Inject;
	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;

	/**
	* RelationshipDef store in v1 format.
	*/
	public class AtlasRelationshipDefStoreV1 extends AtlasAbstractDefStoreV1 implements AtlasRelationshipDefStore {
	private static final Logger LOG = LoggerFactory.getLogger(AtlasRelationshipDefStoreV1.class);

	@Inject
	public AtlasRelationshipDefStoreV1(AtlasTypeDefGraphStoreV1 typeDefStore, AtlasTypeRegistry typeRegistry) {
	super(typeDefStore, typeRegistry);
	}

	@Override
	public AtlasVertex preCreate(AtlasRelationshipDef relationshipDef) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.preCreate({})", relationshipDef);
	}

	validateType(relationshipDef);

	AtlasType type = typeRegistry.getType(relationshipDef.getName());

	if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
	}

	AtlasVertex relationshipDefVertex = typeDefStore.findTypeVertexByName(relationshipDef.getName());

	if (relationshipDefVertex != null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_ALREADY_EXISTS, relationshipDef.getName());
	}

	relationshipDefVertex = typeDefStore.createTypeVertex(relationshipDef);

	updateVertexPreCreate(relationshipDef, (AtlasRelationshipType) type, relationshipDefVertex);

	final AtlasRelationshipEndDef endDef1 = relationshipDef.getEndDef1();
	final AtlasRelationshipEndDef endDef2 = relationshipDef.getEndDef2();
	final String type1 = endDef1.getType();
	final String type2 = endDef2.getType();
	final String name1 = endDef1.getName();
	final String name2 = endDef2.getName();
	final AtlasVertex end1TypeVertex = typeDefStore.findTypeVertexByName(type1);
	final AtlasVertex end2TypeVertex = typeDefStore.findTypeVertexByName(type2);

	if (end1TypeVertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND, relationshipDef.getName(), type1);
	}

	if (end2TypeVertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND, relationshipDef.getName(), type2);
	}

	// create an edge between the relationshipDef and each of the entityDef vertices.
	AtlasEdge edge1 = typeDefStore.getOrCreateEdge(relationshipDefVertex, end1TypeVertex, AtlasGraphUtilsV1.RELATIONSHIPTYPE_EDGE_LABEL);

	/*
	Where edge1 and edge2 have the same names and types we do not need a second edge.
	We are not invoking the equals method on the AtlasRelationshipedDef, as we only want 1 edge even if propagateTags or other properties are different.
	*/

	if (type1.equals(type2) && name1.equals(name2)) {
	if (LOG.isDebugEnabled()) {
	LOG.debug("AtlasRelationshipDefStoreV1.preCreate({}): created relationshipDef vertex {}," +
	" and one edge as {}, because end1 and end2 have the same type and name", relationshipDef, relationshipDefVertex, edge1);
	}

	} else {
	AtlasEdge edge2 = typeDefStore.getOrCreateEdge(relationshipDefVertex, end2TypeVertex, AtlasGraphUtilsV1.RELATIONSHIPTYPE_EDGE_LABEL);
	if (LOG.isDebugEnabled()) {
	LOG.debug("AtlasRelationshipDefStoreV1.preCreate({}): created relationshipDef vertex {}," +
	" edge1 as {}, edge2 as {} ", relationshipDef, relationshipDefVertex, edge1, edge2);
	}

	}
	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.preCreate({}): {}", relationshipDef, relationshipDefVertex);
	}
	return relationshipDefVertex;
	}

	@Override
	public AtlasRelationshipDef create(AtlasRelationshipDef relationshipDef, Object preCreateResult)
	throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.create({}, {})", relationshipDef, preCreateResult);
	}

	AtlasVertex vertex;

	if (preCreateResult == null \|\| !(preCreateResult instanceof AtlasVertex)) {
	vertex = preCreate(relationshipDef);
	} else {
	vertex = (AtlasVertex) preCreateResult;
	}

	AtlasRelationshipDef ret = toRelationshipDef(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.create({}, {}): {}", relationshipDef, preCreateResult, ret);
	}

	return ret;
	}

	@Override
	public List<AtlasRelationshipDef> getAll() throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.getAll()");
	}

	List<AtlasRelationshipDef> ret = new ArrayList<>();
	Iterator<AtlasVertex> vertices = typeDefStore.findTypeVerticesByCategory(TypeCategory.RELATIONSHIP);

	while (vertices.hasNext()) {
	ret.add(toRelationshipDef(vertices.next()));
	}

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.getAll(): count={}", ret.size());
	}

	return ret;
	}

	@Override
	public AtlasRelationshipDef getByName(String name) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.getByName({})", name);
	}

	AtlasVertex vertex = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

	if (vertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
	}

	vertex.getProperty(Constants.TYPE_CATEGORY_PROPERTY_KEY, TypeCategory.class);

	AtlasRelationshipDef ret = toRelationshipDef(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.getByName({}): {}", name, ret);
	}

	return ret;
	}

	@Override
	public AtlasRelationshipDef getByGuid(String guid) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.getByGuid({})", guid);
	}

	AtlasVertex vertex = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

	if (vertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
	}

	AtlasRelationshipDef ret = toRelationshipDef(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.getByGuid({}): {}", guid, ret);
	}

	return ret;
	}

	@Override
	public AtlasRelationshipDef update(AtlasRelationshipDef relationshipDef) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.update({})", relationshipDef);
	}

	validateType(relationshipDef);

	AtlasRelationshipDef ret = StringUtils.isNotBlank(relationshipDef.getGuid())
	? updateByGuid(relationshipDef.getGuid(), relationshipDef)
	: updateByName(relationshipDef.getName(), relationshipDef);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.update({}): {}", relationshipDef, ret);
	}

	return ret;
	}

	@Override
	public AtlasRelationshipDef updateByName(String name, AtlasRelationshipDef relationshipDef)
	throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.updateByName({}, {})", name, relationshipDef);
	}

	validateType(relationshipDef);

	AtlasType type = typeRegistry.getType(relationshipDef.getName());

	if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
	}

	AtlasVertex vertex = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

	if (vertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
	}

	preUpdateCheck(relationshipDef, (AtlasRelationshipType) type, vertex);

	AtlasRelationshipDef ret = toRelationshipDef(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.updateByName({}, {}): {}", name, relationshipDef, ret);
	}

	return ret;
	}

	@Override
	public AtlasRelationshipDef updateByGuid(String guid, AtlasRelationshipDef relationshipDef)
	throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.updateByGuid({})", guid);
	}

	validateType(relationshipDef);

	AtlasType type = typeRegistry.getTypeByGuid(guid);

	if (type.getTypeCategory() != org.apache.atlas.model.TypeCategory.RELATIONSHIP) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_MATCH_FAILED, relationshipDef.getName(), TypeCategory.RELATIONSHIP.name());
	}

	AtlasVertex vertex = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

	if (vertex == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
	}

	preUpdateCheck(relationshipDef, (AtlasRelationshipType) type, vertex);
	// updates should not effect the edges between the types as we do not allow updates that change the endpoints.

	AtlasRelationshipDef ret = toRelationshipDef(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.updateByGuid({}): {}", guid, ret);
	}

	return ret;
	}

	@Override
	public AtlasVertex preDeleteByName(String name) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.preDeleteByName({})", name);
	}

	AtlasVertex ret = typeDefStore.findTypeVertexByNameAndCategory(name, TypeCategory.RELATIONSHIP);

	if (ret == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_NAME_NOT_FOUND, name);
	}

	if (AtlasGraphUtilsV1.relationshipTypeHasInstanceEdges(name)) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_HAS_REFERENCES, name);
	}

	typeDefStore.deleteTypeVertexOutEdges(ret);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.preDeleteByName({}): {}", name, ret);
	}

	return ret;
	}

	@Override
	public void deleteByName(String name, Object preDeleteResult) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.deleteByName({}, {})", name, preDeleteResult);
	}

	AtlasVertex vertex;

	if (preDeleteResult == null \|\| !(preDeleteResult instanceof AtlasVertex)) {
	vertex = preDeleteByName(name);
	} else {
	vertex = (AtlasVertex) preDeleteResult;
	}

	typeDefStore.deleteTypeVertex(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.deleteByName({}, {})", name, preDeleteResult);
	}
	}

	@Override
	public AtlasVertex preDeleteByGuid(String guid) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.preDeleteByGuid({})", guid);
	}

	AtlasVertex ret = typeDefStore.findTypeVertexByGuidAndCategory(guid, TypeCategory.RELATIONSHIP);

	if (ret == null) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_GUID_NOT_FOUND, guid);
	}

	String typeName = AtlasGraphUtilsV1.getProperty(ret, Constants.TYPENAME_PROPERTY_KEY, String.class);

	if (AtlasGraphUtilsV1.relationshipTypeHasInstanceEdges(typeName)) {
	throw new AtlasBaseException(AtlasErrorCode.TYPE_HAS_REFERENCES, typeName);
	}

	typeDefStore.deleteTypeVertexOutEdges(ret);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.preDeleteByGuid({}): {}", guid, ret);
	}

	return ret;
	}

	@Override
	public void deleteByGuid(String guid, Object preDeleteResult) throws AtlasBaseException {
	if (LOG.isDebugEnabled()) {
	LOG.debug("==> AtlasRelationshipDefStoreV1.deleteByGuid({}, {})", guid, preDeleteResult);
	}

	AtlasVertex vertex;

	if (preDeleteResult == null \|\| !(preDeleteResult instanceof AtlasVertex)) {
	vertex = preDeleteByGuid(guid);
	} else {
	vertex = (AtlasVertex) preDeleteResult;
	}

	typeDefStore.deleteTypeVertex(vertex);

	if (LOG.isDebugEnabled()) {
	LOG.debug("<== AtlasRelationshipDefStoreV1.deleteByGuid({}, {})", guid, preDeleteResult);
	}
	}

	private void updateVertexPreCreate(AtlasRelationshipDef relationshipDef, AtlasRelationshipType relationshipType,
	AtlasVertex vertex) throws AtlasBaseException {
	AtlasRelationshipEndDef end1 = relationshipDef.getEndDef1();
	AtlasRelationshipEndDef end2 = relationshipDef.getEndDef2();

	// check whether the names added on the relationship Ends are reserved if required.
	final boolean allowReservedKeywords;
	try {
	allowReservedKeywords = ApplicationProperties.get().getBoolean(ALLOW_RESERVED_KEYWORDS, true);
	} catch (AtlasException e) {
	throw new AtlasBaseException(e);
	}

	if (!allowReservedKeywords) {
	if (QueryParser.isKeyword(end1.getName())) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END1_NAME_INVALID, end1.getName());
	}

	if (QueryParser.isKeyword(end2.getName())) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_END2_NAME_INVALID, end2.getName());
	}
	}

	AtlasStructDefStoreV1.updateVertexPreCreate(relationshipDef, relationshipType, vertex, typeDefStore);
	// Update ends
	setVertexPropertiesFromRelationshipDef(relationshipDef, vertex);
	}

	private void preUpdateCheck(AtlasRelationshipDef newRelationshipDef, AtlasRelationshipType relationshipType,
	AtlasVertex vertex) throws AtlasBaseException {
	// We will not support an update to endpoints or category key
	AtlasRelationshipDef existingRelationshipDef = toRelationshipDef(vertex);

	preUpdateCheck(newRelationshipDef, existingRelationshipDef);
	// we do allow change to tag propagation and the addition of new attributes.

	AtlasStructDefStoreV1.updateVertexPreUpdate(newRelationshipDef, relationshipType, vertex, typeDefStore);

	setVertexPropertiesFromRelationshipDef(newRelationshipDef, vertex);
	}

	/**
	* Check ends are the same and relationshipCategory is the same.
	*
	* We do this by comparing 2 relationshipDefs to avoid exposing the AtlasVertex to unit testing.
	*
	* @param newRelationshipDef
	* @param existingRelationshipDef
	* @throws AtlasBaseException
	*/
	public static void preUpdateCheck(AtlasRelationshipDef newRelationshipDef, AtlasRelationshipDef existingRelationshipDef) throws AtlasBaseException {
	// do not allow renames of the Def.
	String existingName = existingRelationshipDef.getName();
	String newName = newRelationshipDef.getName();

	if (!existingName.equals(newName)) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_NAME_UPDATE,
	newRelationshipDef.getGuid(),existingName, newName);
	}

	RelationshipCategory existingRelationshipCategory = existingRelationshipDef.getRelationshipCategory();
	RelationshipCategory newRelationshipCategory = newRelationshipDef.getRelationshipCategory();

	if ( !existingRelationshipCategory.equals(newRelationshipCategory)){
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_CATEGORY_UPDATE,
	newRelationshipDef.getName(),newRelationshipCategory.name(),
	existingRelationshipCategory.name() );
	}

	AtlasRelationshipEndDef existingEnd1 = existingRelationshipDef.getEndDef1();
	AtlasRelationshipEndDef newEnd1 = newRelationshipDef.getEndDef1();

	if ( !newEnd1.equals(existingEnd1) ) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_END1_UPDATE,
	newRelationshipDef.getName(), newEnd1.toString(), existingEnd1.toString());
	}

	AtlasRelationshipEndDef existingEnd2 = existingRelationshipDef.getEndDef2();
	AtlasRelationshipEndDef newEnd2 = newRelationshipDef.getEndDef2();

	if ( !newEnd2.equals(existingEnd2) ) {
	throw new AtlasBaseException(AtlasErrorCode.RELATIONSHIPDEF_INVALID_END2_UPDATE,
	newRelationshipDef.getName(), newEnd2.toString(), existingEnd2.toString());
	}
	}

	public static void setVertexPropertiesFromRelationshipDef(AtlasRelationshipDef relationshipDef, AtlasVertex vertex) {
	vertex.setProperty(Constants.RELATIONSHIPTYPE_END1_KEY, AtlasType.toJson(relationshipDef.getEndDef1()));
	vertex.setProperty(Constants.RELATIONSHIPTYPE_END2_KEY, AtlasType.toJson(relationshipDef.getEndDef2()));
	// Update RelationshipCategory
	vertex.setProperty(Constants.RELATIONSHIPTYPE_CATEGORY_KEY, relationshipDef.getRelationshipCategory().name());

	if (relationshipDef.getPropagateTags() == null) {
	vertex.setProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, AtlasRelationshipDef.PropagateTags.NONE.name());
	} else {
	vertex.setProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, relationshipDef.getPropagateTags().name());
	}
	}

	private AtlasRelationshipDef toRelationshipDef(AtlasVertex vertex) throws AtlasBaseException {
	AtlasRelationshipDef ret = null;

	if (vertex != null && typeDefStore.isTypeVertex(vertex, TypeCategory.RELATIONSHIP)) {
	String name = vertex.getProperty(Constants.TYPENAME_PROPERTY_KEY, String.class);
	String description = vertex.getProperty(Constants.TYPEDESCRIPTION_PROPERTY_KEY, String.class);
	String version = vertex.getProperty(Constants.TYPEVERSION_PROPERTY_KEY, String.class);
	String end1Str = vertex.getProperty(Constants.RELATIONSHIPTYPE_END1_KEY, String.class);
	String end2Str = vertex.getProperty(Constants.RELATIONSHIPTYPE_END2_KEY, String.class);
	String relationStr = vertex.getProperty(Constants.RELATIONSHIPTYPE_CATEGORY_KEY, String.class);
	String propagateStr = vertex.getProperty(Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, String.class);

	// set the ends
	AtlasRelationshipEndDef endDef1 = AtlasType.fromJson(end1Str, AtlasRelationshipEndDef.class);
	AtlasRelationshipEndDef endDef2 = AtlasType.fromJson(end2Str, AtlasRelationshipEndDef.class);

	// set the relationship Category
	RelationshipCategory relationshipCategory = null;
	for (RelationshipCategory value : RelationshipCategory.values()) {
	if (value.name().equals(relationStr)) {
	relationshipCategory = value;
	}
	}

	// set the propagateTags
	PropagateTags propagateTags = null;
	for (PropagateTags value : PropagateTags.values()) {
	if (value.name().equals(propagateStr)) {
	propagateTags = value;
	}
	}

	ret = new AtlasRelationshipDef(name, description, version, relationshipCategory, propagateTags, endDef1, endDef2);

	// add in the attributes
	AtlasStructDefStoreV1.toStructDef(vertex, ret, typeDefStore);
	}

	return ret;
	}

	}